pthread lock: Invalid argument

[silence-coding]

use https://github.com/rust-rocksdb/rust-rocksdb

Expected behavior

no err

Actual behavior

pthread lock: Invalid argument

Steps to reproduce the behavior

just cargo run

main.rs

use std::sync::mpsc::{channel, Sender};
use rocksdb::DB;

fn main() {
    let tx = open_client();
    tx.send("sss".to_string()).unwrap();
}

fn open_client() -> Sender<String> {
    let dir = std::env::temp_dir().join("db");
    let db = DB::open_default(&dir).unwrap();
    let (tx, rx) = channel();
    std::thread::spawn(move || {
        while let Ok(k) = rx.recv() {
            if let Some(data) = db.get(k).unwrap() {
                println!("xxxx: {}", String::from_utf8_lossy(&data));
            }
        }
    });
    tx
}

Cargo.toml

[dependencies]
rocksdb = "0.20.1"

(gdb) b rocksdb/port/port_posix.cc:54
(gdb) run
Thread 20 "db" hit Breakpoint 6, rocksdb::port::PthreadCall (label=0x555555d58a0e "lock", result=22) at rocksdb/port/port_posix.cc:54
54	    abort();
(gdb) bt
#0  rocksdb::port::PthreadCall (label=0x555555d58a0e "lock", result=22) at rocksdb/port/port_posix.cc:54
#1  0x0000555555924124 in rocksdb::port::Mutex::Lock (this=0x5555560177a8 <rocksdb::PeriodicTaskScheduler::Default()::timer+8>) at rocksdb/port/port_posix.cc:80
#2  0x00005555558b49c2 in rocksdb::InstrumentedMutex::LockInternal (this=0x5555560177a8 <rocksdb::PeriodicTaskScheduler::Default()::timer+8>)
    at rocksdb/monitoring/instrumented_mutex.cc:55
#3  0x00005555558b4978 in rocksdb::InstrumentedMutex::Lock (this=0x5555560177a8 <rocksdb::PeriodicTaskScheduler::Default()::timer+8>) at rocksdb/monitoring/instrumented_mutex.cc:31
#4  0x00005555556098ec in rocksdb::InstrumentedMutexLock::InstrumentedMutexLock (this=0x7fffe3ff58c8, mutex=0x5555560177a8 <rocksdb::PeriodicTaskScheduler::Default()::timer+8>)
    at rocksdb/monitoring/instrumented_mutex.h:74
#5  0x000055555578958e in rocksdb::Timer::Cancel (this=0x5555560177a0 <rocksdb::PeriodicTaskScheduler::Default()::timer>, fn_name="dump_st0") at rocksdb/util/timer.h:86
#6  0x000055555578a644 in rocksdb::PeriodicTaskScheduler::Unregister (this=0x555556067368, task_type=rocksdb::PeriodicTaskType::kDumpStats) at rocksdb/db/periodic_task_scheduler.cc:88
#7  0x000055555561595a in rocksdb::DBImpl::CancelAllBackgroundWork (this=0x555556065c40, wait=false) at rocksdb/db/db_impl/db_impl.cc:503
#8  0x0000555555615f7d in rocksdb::DBImpl::CloseHelper (this=0x555556065c40) at rocksdb/db/db_impl/db_impl.cc:575
#9  0x0000555555616d81 in rocksdb::DBImpl::CloseImpl (this=0x555556065c40) at rocksdb/db/db_impl/db_impl.cc:743
#10 0x0000555555616e33 in rocksdb::DBImpl::~DBImpl (this=0x555556065c40, __in_chrg=<optimized out>) at rocksdb/db/db_impl/db_impl.cc:761
#11 0x00005555556176c4 in rocksdb::DBImpl::~DBImpl (this=0x555556065c40, __in_chrg=<optimized out>) at rocksdb/db/db_impl/db_impl.cc:763
#12 0x00005555555f27b6 in rocksdb_close (db=0x55555608bc90) at rocksdb/db/c.cc:852
#13 0x00005555555e477e in <rocksdb::db::DBWithThreadModeInner as core::ops::drop::Drop>::drop (self=0x7fffe3ff61c8) at src/db.rs:290
#14 0x00005555555bcf6b in core::ptr::drop_in_place<rocksdb::db::DBWithThreadModeInner> () at /rustc/2c8cc343237b8f7d5a3c3703e3a87f2eb2c54a74/library/core/src/ptr/mod.rs:490
#15 0x00005555555bc49b in core::ptr::drop_in_place<rocksdb::db::DBCommon<rocksdb::db::SingleThreaded,rocksdb::db::DBWithThreadModeInner>> ()
    at /rustc/2c8cc343237b8f7d5a3c3703e3a87f2eb2c54a74/library/core/src/ptr/mod.rs:490
#16 0x00005555555bd49f in core::ptr::drop_in_place<db::open_client::{{closure}}> () at /rustc/2c8cc343237b8f7d5a3c3703e3a87f2eb2c54a74/library/core/src/ptr/mod.rs:490
#17 0x00005555555d16e5 in db::open_client::{{closure}} () at src/main.rs:21
#18 0x00005555555a7143 in std::sys_common::backtrace::__rust_begin_short_backtrace (f=...) at /rustc/2c8cc343237b8f7d5a3c3703e3a87f2eb2c54a74/library/std/src/sys_common/backtrace.rs:121
#19 0x00005555555acbe0 in std::thread::Builder::spawn_unchecked_::{{closure}}::{{closure}} () at /rustc/2c8cc343237b8f7d5a3c3703e3a87f2eb2c54a74/library/std/src/thread/mod.rs:558
#20 0x00005555555d5e30 in <core::panic::unwind_safe::AssertUnwindSafe<F> as core::ops::function::FnOnce<()>>::call_once (self=..., _args=())
    at /rustc/2c8cc343237b8f7d5a3c3703e3a87f2eb2c54a74/library/core/src/panic/unwind_safe.rs:271
#21 0x00005555555ab6ad in std::panicking::try::do_call (data=0x7fffe3ff6590 "\220\274\bVUU") at /rustc/2c8cc343237b8f7d5a3c3703e3a87f2eb2c54a74/library/std/src/panicking.rs:483
#22 0x00005555555ab8eb in __rust_try () at library/core/src/panicking.rs:212
#23 0x00005555555ab4f6 in std::panicking::try (f=...) at /rustc/2c8cc343237b8f7d5a3c3703e3a87f2eb2c54a74/library/std/src/panicking.rs:447
#24 0x00005555555cde70 in std::panic::catch_unwind (f=...) at /rustc/2c8cc343237b8f7d5a3c3703e3a87f2eb2c54a74/library/std/src/panic.rs:140
#25 0x00005555555ac9b6 in std::thread::Builder::spawn_unchecked_::{{closure}} () at /rustc/2c8cc343237b8f7d5a3c3703e3a87f2eb2c54a74/library/std/src/thread/mod.rs:557
#26 0x00005555555bc07e in core::ops::function::FnOnce::call_once{{vtable-shim}} () at /rustc/2c8cc343237b8f7d5a3c3703e3a87f2eb2c54a74/library/core/src/ops/function.rs:250
#27 0x0000555555d18cf3 in <alloc::boxed::Box<F,A> as core::ops::function::FnOnce<Args>>::call_once () at library/alloc/src/boxed.rs:1988
#28 <alloc::boxed::Box<F,A> as core::ops::function::FnOnce<Args>>::call_once () at library/alloc/src/boxed.rs:1988
#29 std::sys::unix::thread::Thread::new::thread_start () at library/std/src/sys/unix/thread.rs:108
#30 0x00007ffff7df5f3b in ?? () from /usr/lib64/libpthread.so.0
#31 0x00007ffff7ba3570 in clone () from /usr/lib64/libc.so.6

[silence-coding]

The other old version 6.22.1：

(gdb) bt
#0  0x00007ffff655c81b in raise () from /usr/lib64/libc.so.6
#1  0x00007ffff655db41 in abort () from /usr/lib64/libc.so.6
#2  0x00007ffff7af5536 in rocksdb::port::PthreadCall (label=0x7ffff7c3976b "lock", result=22) at rocksdb/port/port_posix.cc:52
#3  0x00007ffff7af5668 in rocksdb::port::Mutex::Lock (this=0x7ffff7f8c2a0 <rocksdb::(anonymous namespace)::mutex_locked_files>)
    at rocksdb/port/port_posix.cc:80
#4  0x00007ffff7afe559 in rocksdb::(anonymous namespace)::PosixFileSystem::UnlockFile (this=0x7ffff7f8c3a0 <rocksdb::FileSystem::Default()::default_fs>, lock=0x7fffec01b470)
    at rocksdb/env/fs_posix.cc:849
#5  0x00007ffff779ccd1 in rocksdb::CompositeEnv::UnlockFile (this=0x7ffff7f8bd60 <rocksdb::Env::Default()::default_env>, l=0x7fffec01b470)
    at rocksdb/env/composite_env_wrapper.h:160
#6  0x00007ffff759afb0 in rocksdb::DBImpl::CloseHelper (this=0x7fffec00b950) at rocksdb/db/db_impl/db_impl.cc:644
#7  0x00007ffff759b438 in rocksdb::DBImpl::~DBImpl (this=0x7fffec00b950, __in_chrg=<optimized out>) at rocksdb/db/db_impl/db_impl.cc:686
#8  0x00007ffff759b804 in rocksdb::DBImpl::~DBImpl (this=0x7fffec00b950, __in_chrg=<optimized out>) at rocksdb/db/db_impl/db_impl.cc:688
#9  0x00007ffff74e241f in rocksdb_close (db=0x7fffec031a10) at rocksdb/db/c.cc:816

[aleksuss]

The problem has no relation to this project at all. There is an error in your code.

1. The main problem is that the thread you create immediately dies after creation. There are no blocking operations in your main function. After sending a value sss into the channel, the database no longer exists.
2. If you want to use rocksdb in multithreaded program you should add features multi-threaded-cf.
   rocksdb = { version = "0.20", features = ["multi-threaded-cf"] }

I guess you would like to write something like this:

use std::sync::mpsc::{channel, Sender};
use rocksdb::DB;

fn main() {
    let (handle, tx) = open_client();
    
    tx.send("sss".to_string()).unwrap();
    handle.join();
}

fn open_client() -> (std::thread::JoinHandle<()>, Sender<String>) {
    let dir = std::env::temp_dir().join("db");
    let db = DB::open_default(&dir).unwrap();
    let (tx, rx) = channel();

    let handle = std::thread::spawn(move || {
        while let Ok(k) = rx.recv() {
            if let Some(data) = db.get(&k).unwrap() {
                println!("xxxx: {}", String::from_utf8_lossy(&data));
            } else {
                println!("there is no value for key: {k}");
            }
        }
    });

    (handle, tx)
}

???

[silence-coding]

@aleksuss This is only a demo. The actual application scenario is that I want to encapsulate the database as an asynchronous operation through the channel of tokio, so I do not want to block the main thread through handle.join(). I'll try enabling the multi-threaded-cf feature.

[silence-coding]

@aleksuss Enabling the multi-threaded-cf feature does not solve this problem.

[dependencies]
rocksdb = { version = "0.20", features = ["multi-threaded-cf"] }

use std::sync::mpsc::{channel, Sender};
use rocksdb::DB;

fn main() {
    let tx = open_client();
    let (data_tx, rx) = channel();
    tx.send(("key".to_string(), data_tx)).unwrap();
    let data = rx.recv().unwrap();
    println!("{:?}", data);
}


fn open_client() -> Sender<(String, Sender<Option<Vec<u8>>>)> {
    let dir = std::env::temp_dir().join("db");
    let db = DB::open_default(&dir).unwrap();
    let (tx, rx) = channel::<(String, Sender<Option<Vec<u8>>>)>();
    std::thread::spawn(move || {
        while let Ok((k, data_tx)) = rx.recv() {
            let data = db.get(k).unwrap();
            if let Some(data) = &data {
                println!("xxxx: {}", String::from_utf8_lossy(data));
            }
            data_tx.send(data).unwrap();
        }
    });
    tx
}

[aleksuss]

You don't listen to me. Learn how threads work.

[silence-coding]

You don't listen to me. Learn how threads work.

I don't understand you very well. I'm particularly unfamiliar with this. I've turned on the multi-threaded-cf feature, but the code still has this problem. Does that mean I shouldn't let the spawn thread die?

[aleksuss]

Does that mean I shouldn't let the spawn thread die?

Yes. It's your main error in the example.

[silence-coding]

Does that mean I shouldn't let the spawn thread die?

Yes. It's your main error in the example.
Is there any other way to avoid this problem than handle.join?

[pum-purum-pum-pum]

Just wanted to share some info for my case where I've encountered sporadic pthread lock: Invalid argument and terminate called without an active exception.

The issue is one can't just drop when closing the database. Rocksdb spawns workers under the hood so one has to wait until they're done.
After I added cancel_all_background_work
before dropping the database the sporadic failures with pthread lock: Invalid argument and terminate called without an active exception were resolved.

I'm not sure if this API is intended (I guess because having a blocking drop might be problematic for various reasons), but it's not something I would expect from a Rust crate to do.

Perhaps this could be a separate issue in Rust bindings repo.