tokio-postgres occasionally hangs when pipe lining in a transaction

[erikjohnston]

I have a little app that occasionally stalls due to waiting on a futex. This happens when doing multiple txn.execute_raw at once and doing try_join_all (I'm not sure whether that is something that I'm meant to be doing?). I'm using v0.7.0 of tokio postgres, but I didn't see any relevant changes since then.

Looking at the stacktrace from GDB it appears that the InnerClient::state mutex is trying to be locked while its already held (I've included the full stack trace and app code at the bottom).

If I'm reading the stack trace correctly what's happening is that we first take out the lock in InnerClient::set_typeinfo_composite:

rust-postgres/tokio-postgres/src/client.rs

Lines 96 to 98 in 4fb6fd9

	pub fn set_typeinfo_composite(&self, statement: &Statement) {
	self.state.lock().typeinfo_composite = Some(statement.clone());
	}

which then drops a Statement (presumably whatever was in State::typeinfo_composite before) while holding the lock (as the drop happens during assignment), calling:

rust-postgres/tokio-postgres/src/statement.rs

Lines 19 to 28 in 4fb6fd9

	fn drop(&mut self) {
	if let Some(client) = self.client.upgrade() {
	let buf = client.with_buf(|buf| {
	frontend::close(b'S', &self.name, buf).unwrap();
	frontend::sync(buf);
	buf.split().freeze()
	});
	let _ = client.send(RequestMessages::Single(FrontendMessage::Raw(buf)));
	}
	}

And then with_buf tries taking out the InnerClient::state mutex again:

rust-postgres/tokio-postgres/src/client.rs

Lines 116 to 124 in 4fb6fd9

	pub fn with_buf<F, R>(&self, f: F) -> R
	where
	F: FnOnce(&mut BytesMut) -> R,
	{
	let mut state = self.state.lock();
	let r = f(&mut state.buf);
	state.buf.clear();
	r
	}

Causing a deadlock. (Reading between the lines of Amanieu/parking_lot#212 it sounds like this won't necessarily always cause a deadlock 🤷)

One fix is to use Option::replace in InnerClient::set_typeinfo_composite to force that we get the statement and can ensure we only drop it after dropping the lock? It feels quite fragile though.

Condensed stack trace

#9  tokio_postgres::client::InnerClient::with_buf(...)
    at tokio-postgres-0.7.0/src/client.rs:120
#10 tokio_postgres::statement::{{impl}}::drop (...)
    at tokio-postgres-0.7.0/src/statement.rs:21
...
#16 core::ptr::drop_in_place<core::option::Option<tokio_postgres::statement::Statement>> ()
    at rustlib/src/rust/library/core/src/ptr/mod.rs:187
#17 tokio_postgres::client::InnerClient::set_typeinfo_composite (...)
    at tokio-postgres-0.7.0/src/client.rs:97
...
#31 tokio_postgres::client::{{impl}}::prepare::{{closure}} () 
    at tokio-postgres-0.7.0/src/client.rs:189
...
#35 tokio_postgres::client::{{impl}}::execute_raw()
    at tokio-postgres-0.7.0/src/client.rs:400

Full Stack Trace

#0  syscall () at ../sysdeps/unix/sysv/linux/x86_64/syscall.S:38
#1  0x000055ad1b616871 in parking_lot_core::thread_parker::imp::ThreadParker::futex_wait (self=0x7fdda472f778, ts=...)
    at /home/erikj/.cargo/registry/src/github.com-1ecc6299db9ec823/parking_lot_core-0.8.3/src/thread_parker/linux.rs:112
#2  0x000055ad1b61667c in parking_lot_core::thread_parker::imp::{{impl}}::park (self=0x7fdda472f778)
    at /home/erikj/.cargo/registry/src/github.com-1ecc6299db9ec823/parking_lot_core-0.8.3/src/thread_parker/linux.rs:66
#3  0x000055ad1b61d10c in parking_lot_core::parking_lot::park::{{closure}}<closure-0,closure-1,closure-2> (thread_data=0x7fdda472f758)
    at /home/erikj/.cargo/registry/src/github.com-1ecc6299db9ec823/parking_lot_core-0.8.3/src/parking_lot.rs:611
#4  0x000055ad1b61b42a in parking_lot_core::parking_lot::with_thread_data<parking_lot_core::parking_lot::ParkResult,closure-0> (f=...)
    at /home/erikj/.cargo/registry/src/github.com-1ecc6299db9ec823/parking_lot_core-0.8.3/src/parking_lot.rs:183
#5  parking_lot_core::parking_lot::park<closure-0,closure-1,closure-2> (key=94202014844568, validate=..., before_sleep=..., timed_out=..., park_token=..., timeout=...)
    at /home/erikj/.cargo/registry/src/github.com-1ecc6299db9ec823/parking_lot_core-0.8.3/src/parking_lot.rs:576
#6  0x000055ad1b617c56 in parking_lot::raw_mutex::RawMutex::lock_slow (self=0x55ad1da1c698, timeout=...) at /home/erikj/.cargo/registry/src/github.com-1ecc6299db9ec823/parking_lot-0.11.1/src/raw_mutex.rs:262
#7  0x000055ad1b3f8a59 in parking_lot::raw_mutex::{{impl}}::lock (self=0x55ad1da1c698) at /home/erikj/.cargo/registry/src/github.com-1ecc6299db9ec823/parking_lot-0.11.1/src/raw_mutex.rs:72
#8  0x000055ad1b4051a3 in lock_api::mutex::Mutex<parking_lot::raw_mutex::RawMutex, tokio_postgres::client::State>::lock<parking_lot::raw_mutex::RawMutex,tokio_postgres::client::State> (self=0x55ad1da1c698)
    at /home/erikj/.cargo/registry/src/github.com-1ecc6299db9ec823/lock_api-0.4.2/src/mutex.rs:207
#9  0x000055ad1b427b79 in tokio_postgres::client::InnerClient::with_buf<closure-0,bytes::bytes::Bytes> (self=0x55ad1da1c690, f=...)
    at /home/erikj/.cargo/registry/src/github.com-1ecc6299db9ec823/tokio-postgres-0.7.0/src/client.rs:120
#10 0x000055ad1b3e431c in tokio_postgres::statement::{{impl}}::drop (self=0x55ad1db089d0) at /home/erikj/.cargo/registry/src/github.com-1ecc6299db9ec823/tokio-postgres-0.7.0/src/statement.rs:21
#11 0x000055ad1b3cf097 in core::ptr::drop_in_place<tokio_postgres::statement::StatementInner> ()
    at /home/erikj/.rustup/toolchains/nightly-x86_64-unknown-linux-gnu/lib/rustlib/src/rust/library/core/src/ptr/mod.rs:187
#12 0x000055ad1b3dfdc4 in alloc::sync::Arc<tokio_postgres::statement::StatementInner>::drop_slow<tokio_postgres::statement::StatementInner> (self=0x55ad1da1c6a8)
    at /home/erikj/.rustup/toolchains/nightly-x86_64-unknown-linux-gnu/lib/rustlib/src/rust/library/alloc/src/sync.rs:1039
#13 0x000055ad1b3e09fc in alloc::sync::{{impl}}::drop<tokio_postgres::statement::StatementInner> (self=0x55ad1da1c6a8)
    at /home/erikj/.rustup/toolchains/nightly-x86_64-unknown-linux-gnu/lib/rustlib/src/rust/library/alloc/src/sync.rs:1571
#14 0x000055ad1b3d034e in core::ptr::drop_in_place<alloc::sync::Arc<tokio_postgres::statement::StatementInner>> ()
    at /home/erikj/.rustup/toolchains/nightly-x86_64-unknown-linux-gnu/lib/rustlib/src/rust/library/core/src/ptr/mod.rs:187
#15 0x000055ad1b3cedae in core::ptr::drop_in_place<tokio_postgres::statement::Statement> ()
    at /home/erikj/.rustup/toolchains/nightly-x86_64-unknown-linux-gnu/lib/rustlib/src/rust/library/core/src/ptr/mod.rs:187
#16 0x000055ad1b3d02c3 in core::ptr::drop_in_place<core::option::Option<tokio_postgres::statement::Statement>> ()
    at /home/erikj/.rustup/toolchains/nightly-x86_64-unknown-linux-gnu/lib/rustlib/src/rust/library/core/src/ptr/mod.rs:187
#17 0x000055ad1b427593 in tokio_postgres::client::InnerClient::set_typeinfo_composite (self=0x55ad1da1c690, statement=0x7ffc63238980)
    at /home/erikj/.cargo/registry/src/github.com-1ecc6299db9ec823/tokio-postgres-0.7.0/src/client.rs:97
#18 0x000055ad1b40e1d9 in tokio_postgres::prepare::typeinfo_composite_statement::{{closure}} () at /home/erikj/.cargo/registry/src/github.com-1ecc6299db9ec823/tokio-postgres-0.7.0/src/prepare.rs:260
#19 0x000055ad1b403500 in core::future::from_generator::{{impl}}::poll<generator-0> (self=..., cx=0x7ffc63252488)
    at /home/erikj/.rustup/toolchains/nightly-x86_64-unknown-linux-gnu/lib/rustlib/src/rust/library/core/src/future/mod.rs:80
#20 0x000055ad1b40cb74 in tokio_postgres::prepare::get_composite_fields::{{closure}} () at /home/erikj/.cargo/registry/src/github.com-1ecc6299db9ec823/tokio-postgres-0.7.0/src/prepare.rs:235
#21 0x000055ad1b4029f3 in core::future::from_generator::{{impl}}::poll<generator-0> (self=..., cx=0x7ffc63252488)
    at /home/erikj/.rustup/toolchains/nightly-x86_64-unknown-linux-gnu/lib/rustlib/src/rust/library/core/src/future/mod.rs:80
#22 0x000055ad1b409fac in tokio_postgres::prepare::get_type::{{closure}} () at /home/erikj/.cargo/registry/src/github.com-1ecc6299db9ec823/tokio-postgres-0.7.0/src/prepare.rs:168
#23 0x000055ad1b402eb5 in core::future::from_generator::{{impl}}::poll<generator-0> (self=..., cx=0x7ffc63252488)
    at /home/erikj/.rustup/toolchains/nightly-x86_64-unknown-linux-gnu/lib/rustlib/src/rust/library/core/src/future/mod.rs:80
#24 0x000055ad1b3e3ec9 in core::future::future::{{impl}}::poll<alloc::boxed::Box<Future, alloc::alloc::Global>> (self=..., cx=0x7ffc63252488)
    at /home/erikj/.rustup/toolchains/nightly-x86_64-unknown-linux-gnu/lib/rustlib/src/rust/library/core/src/future/future.rs:119
#25 0x000055ad1b409d91 in tokio_postgres::prepare::get_type::{{closure}} () at /home/erikj/.cargo/registry/src/github.com-1ecc6299db9ec823/tokio-postgres-0.7.0/src/prepare.rs:165
#26 0x000055ad1b402eb5 in core::future::from_generator::{{impl}}::poll<generator-0> (self=..., cx=0x7ffc63252488)
    at /home/erikj/.rustup/toolchains/nightly-x86_64-unknown-linux-gnu/lib/rustlib/src/rust/library/core/src/future/mod.rs:80
#27 0x000055ad1b406a13 in tokio_postgres::prepare::prepare::{{closure}} () at /home/erikj/.cargo/registry/src/github.com-1ecc6299db9ec823/tokio-postgres-0.7.0/src/prepare.rs:89
#28 0x000055ad1b402b20 in core::future::from_generator::{{impl}}::poll<generator-0> (self=..., cx=0x7ffc63252488)
    at /home/erikj/.rustup/toolchains/nightly-x86_64-unknown-linux-gnu/lib/rustlib/src/rust/library/core/src/future/mod.rs:80
#29 0x000055ad1b428a88 in tokio_postgres::client::{{impl}}::prepare_typed::{{closure}} () at /home/erikj/.cargo/registry/src/github.com-1ecc6299db9ec823/tokio-postgres-0.7.0/src/client.rs:201
#30 0x000055ad1b4036c0 in core::future::from_generator::{{impl}}::poll<generator-0> (self=..., cx=0x7ffc63252488)
    at /home/erikj/.rustup/toolchains/nightly-x86_64-unknown-linux-gnu/lib/rustlib/src/rust/library/core/src/future/mod.rs:80
#31 0x000055ad1b42876c in tokio_postgres::client::{{impl}}::prepare::{{closure}} () at /home/erikj/.cargo/registry/src/github.com-1ecc6299db9ec823/tokio-postgres-0.7.0/src/client.rs:189
#32 0x000055ad1b403330 in core::future::from_generator::{{impl}}::poll<generator-0> (self=..., cx=0x7ffc63252488)
--Type <RET> for more, q to quit, c to continue without paging--
    at /home/erikj/.rustup/toolchains/nightly-x86_64-unknown-linux-gnu/lib/rustlib/src/rust/library/core/src/future/mod.rs:80
#33 0x000055ad1a230d66 in tokio_postgres::to_statement::private::{{impl}}::into_statement::{{closure}} ()
    at /home/erikj/.cargo/registry/src/github.com-1ecc6299db9ec823/tokio-postgres-0.7.0/src/to_statement.rs:18
#34 0x000055ad1a31784b in core::future::from_generator::{{impl}}::poll<generator-0> (self=..., cx=0x7ffc63252488)
    at /home/erikj/.rustup/toolchains/nightly-x86_64-unknown-linux-gnu/lib/rustlib/src/rust/library/core/src/future/mod.rs:80
#35 0x000055ad1a28a94f in tokio_postgres::client::{{impl}}::execute_raw::{{closure}}<str,&ToSql,alloc::vec::Vec<&ToSql, alloc::alloc::Global>> ()
    at /home/erikj/.cargo/registry/src/github.com-1ecc6299db9ec823/tokio-postgres-0.7.0/src/client.rs:400
#36 0x000055ad1a31b9d1 in core::future::from_generator::{{impl}}::poll<generator-0> (self=..., cx=0x7ffc63252488)
    at /home/erikj/.rustup/toolchains/nightly-x86_64-unknown-linux-gnu/lib/rustlib/src/rust/library/core/src/future/mod.rs:80
#37 0x000055ad1a3215de in tokio_postgres::transaction::{{impl}}::execute_raw::{{closure}}<&ToSql,alloc::vec::Vec<&ToSql, alloc::alloc::Global>,str> ()
    at /home/erikj/.cargo/registry/src/github.com-1ecc6299db9ec823/tokio-postgres-0.7.0/src/transaction.rs:172
#38 0x000055ad1a31c5a1 in core::future::from_generator::{{impl}}::poll<generator-0> (self=..., cx=0x7ffc63252488)
    at /home/erikj/.rustup/toolchains/nightly-x86_64-unknown-linux-gnu/lib/rustlib/src/rust/library/core/src/future/mod.rs:80
#39 0x000055ad1a316313 in futures_core::future::{{impl}}::try_poll<core::future::from_generator::GenFuture<generator-0>,u64,tokio_postgres::error::Error> (self=..., cx=0x7ffc63252488)
    at /home/erikj/.cargo/registry/src/github.com-1ecc6299db9ec823/futures-core-0.3.13/src/future.rs:84
#40 0x000055ad1a23bbd0 in futures_util::future::try_maybe_done::{{impl}}::poll<core::future::from_generator::GenFuture<generator-0>> (self=..., cx=0x7ffc63252488)
    at /home/erikj/.cargo/registry/src/github.com-1ecc6299db9ec823/futures-util-0.3.13/src/future/try_maybe_done.rs:80
#41 0x000055ad1a23c253 in futures_core::future::{{impl}}::try_poll<futures_util::future::try_maybe_done::TryMaybeDone<core::future::from_generator::GenFuture<generator-0>>,(),tokio_postgres::error::Error> (
    self=..., cx=0x7ffc63252488) at /home/erikj/.cargo/registry/src/github.com-1ecc6299db9ec823/futures-core-0.3.13/src/future.rs:84
#42 0x000055ad1a2ed24c in futures_util::future::try_join_all::{{impl}}::poll<core::future::from_generator::GenFuture<generator-0>> (self=..., cx=0x7ffc63252488)
    at /home/erikj/.cargo/registry/src/github.com-1ecc6299db9ec823/futures-util-0.3.13/src/future/try_join_all.rs:111
#43 0x000055ad1a3e2d2e in calendar_bot::database::{{impl}}::insert_events::{{closure}} () at src/database.rs:271

App code

        futures::future::try_join_all(events.iter().map(|event| {
            txn.execute_raw(
                r#"
                    INSERT INTO events (calendar_id, event_id, summary, description, location, attendees)
                    VALUES ($1, $2, $3, $4, $5, $6)
                    ON CONFLICT (calendar_id, event_id)
                    DO UPDATE SET
                        summary = EXCLUDED.summary,
                        description = EXCLUDED.description,
                        location = EXCLUDED.location,
                        attendees = EXCLUDED.attendees
                "#,
                vec![
                    &calendar_id as &dyn ToSql,
                    &event.event_id,
                    &event.summary,
                    &event.description,
                    &event.location,
                    &event.attendees,
                ],
            )
        }))
        .await?;

[sfackler]

Oh, that is an interesting failure mode. Rather than using replace to defer the destruction of the old statement, we could alternatively just not overwrite the cached statement if it already exists.

[erikjohnston]

Rather than using replace to defer the destruction of the old statement, we could alternatively just not overwrite the cached statement if it already exists.

Am I understanding the code correctly that the typeinfo statements in the client are always going to be for the same query, and it's just a race that we've prepared it multiple times? So ideally ideally we'd try and rewrite things so that we only prepared them once, but for now the easiest fix is to just not overwrite the cache?

I'll throw up a quick PR for that.