Fix flake in buffer_unordered

[cramertj]

seen on this change: #1789
in this job: https://travis-ci.com/rust-lang-nursery/futures-rs/jobs/224053387

[taiki-e]

I tested it locally but could not reproduce it (Windows and Linux(WSL-Ubuntu)).
As far as I look at history (pr and master), this problem seems not to occur in Linux, so I think this is an issue specific to macOS or a Travis issue.

Recent Travis's macOS build is very unstable regardless of this test (#1787 (comment)), so for now, I'm guessing this is a Travis issue.

[taiki-e]

(The OS looks irrelevant, but I have never been able to reproduce it locally.)

[kentfredric]

I've been reproducing this on linux in futures 0.1.29, the problem is its very ephemeral and prone to randomly working 99% of the time.

I can't make it do this on nightly rust, but I can do it on current stable rust.

Attaching gdb once it deadlocks gives me:

gdb thread apply all bt full

Thread 4 (Thread 0x7f9d6d721700 (LWP 9458)):
#0  futex_wait_cancelable (private=0, expected=0, futex_word=0x7f9d64000cac) at ../sysdeps/unix/sysv/linux/futex-internal.h:88
        __ret = -512
        oldtype = 0
        err = <optimized out>
        oldtype = <optimized out>
        err = <optimized out>
        __ret = <optimized out>
        resultvar = <optimized out>
        __arg4 = <optimized out>
        __arg3 = <optimized out>
        __arg2 = <optimized out>
        __arg1 = <optimized out>
        _a4 = <optimized out>
        _a3 = <optimized out>
        _a2 = <optimized out>
        _a1 = <optimized out>
#1  __pthread_cond_wait_common (abstime=0x0, mutex=0x7f9d64000b20, cond=0x7f9d64000c80) at pthread_cond_wait.c:502
        spin = 0
        buffer = {__routine = 0x7f9d6de63b80 <__condvar_cleanup_waiting>, __arg = 0x7f9d6d720a10, __canceltype = 1836190128, __prev = 0x0}
        cbuffer = {wseq = 11, cond = 0x7f9d64000c80, mutex = 0x7f9d64000b20, private = 0}
        rt = <optimized out>
        err = <optimized out>
        g = 1
        flags = <optimized out>
        g1_start = <optimized out>
        signals = <optimized out>
        result = 0
        wseq = 11
        seq = 5
        private = 0
        maxspin = <optimized out>
        err = <optimized out>
        result = <optimized out>
        wseq = <optimized out>
        g = <optimized out>
        seq = <optimized out>
        flags = <optimized out>
        private = <optimized out>
        signals = <optimized out>
        g1_start = <optimized out>
        spin = <optimized out>
        buffer = <optimized out>
        cbuffer = <optimized out>
        rt = <optimized out>
        s = <optimized out>
#2  __pthread_cond_wait (cond=0x7f9d64000c80, mutex=0x7f9d64000b20) at pthread_cond_wait.c:655
No locals.
#3  0x0000561614377c8d in futures::task_impl::std::ThreadNotify::park ()
No symbol table info available.
#4  0x000056161432c36a in std::sys_common::backtrace::__rust_begin_short_backtrace ()
No symbol table info available.
#5  0x000056161432e1bc in std::panicking::try::do_call ()
No symbol table info available.
#6  0x0000561614388a4a in __rust_maybe_catch_panic () at src/libpanic_unwind/lib.rs:80
No locals.
#7  0x000056161433a55f in core::ops::function::FnOnce::call_once{{vtable-shim}} ()
No symbol table info available.
#8  0x000056161437bcdf in <alloc::boxed::Box<F> as core::ops::function::FnOnce<A>>::call_once () at /rustc/4560ea788cb760f0a34127156c78e2552949f734/src/liballoc/boxed.rs:922
No locals.
#9  0x0000561614388110 in <alloc::boxed::Box<F> as core::ops::function::FnOnce<A>>::call_once () at /rustc/4560ea788cb760f0a34127156c78e2552949f734/src/liballoc/boxed.rs:922
No locals.
#10 std::sys_common::thread::start_thread () at src/libstd/sys_common/thread.rs:13
No locals.
#11 std::sys::unix::thread::Thread::new::thread_start () at src/libstd/sys/unix/thread.rs:79
No locals.
#12 0x00007f9d6de5d3a7 in start_thread (arg=<optimized out>) at pthread_create.c:486
        ret = <optimized out>
        pd = <optimized out>
        now = <optimized out>
        unwind_buf = {cancel_jmp_buf = {{jmp_buf = {140314122786560, -2814752794425973897, 140314126984062, 140314126984063, 140314122784448, 140314126984320, 2870244160037954423, 2870243169132979063}, mask_was_saved = 0}}, priv = {pad = {0x0, 0x0, 0x0, 0x0}, data = {prev = 0x0, cleanup = 0x0, canceltype = 0}}}
        not_first_call = <optimized out>
#13 0x00007f9d6dd720ff in clone () at ../sysdeps/unix/sysv/linux/x86_64/clone.S:95
No locals.

Thread 3 (Thread 0x7f9d6d922700 (LWP 9457)):
#0  futex_wait_cancelable (private=0, expected=0, futex_word=0x7f9d60000cac) at ../sysdeps/unix/sysv/linux/futex-internal.h:88
        __ret = -512
        oldtype = 0
        err = <optimized out>
        oldtype = <optimized out>
        err = <optimized out>
        __ret = <optimized out>
        resultvar = <optimized out>
        __arg4 = <optimized out>
        __arg3 = <optimized out>
        __arg2 = <optimized out>
        __arg1 = <optimized out>
        _a4 = <optimized out>
        _a3 = <optimized out>
        _a2 = <optimized out>
        _a1 = <optimized out>
#1  __pthread_cond_wait_common (abstime=0x0, mutex=0x7f9d60000b20, cond=0x7f9d60000c80) at pthread_cond_wait.c:502
        spin = 0
        buffer = {__routine = 0x7f9d6de63b80 <__condvar_cleanup_waiting>, __arg = 0x7f9d6d921890, __canceltype = 339515576, __prev = 0x0}
        cbuffer = {wseq = 7, cond = 0x7f9d60000c80, mutex = 0x7f9d60000b20, private = 0}
        rt = <optimized out>
        err = <optimized out>
        g = 1
        flags = <optimized out>
        g1_start = <optimized out>
        signals = <optimized out>
        result = 0
        wseq = 7
        seq = 3
        private = 0
        maxspin = <optimized out>
        err = <optimized out>
        result = <optimized out>
        wseq = <optimized out>
        g = <optimized out>
        seq = <optimized out>
        flags = <optimized out>
        private = <optimized out>
        signals = <optimized out>
        g1_start = <optimized out>
        spin = <optimized out>
        buffer = <optimized out>
        cbuffer = <optimized out>
        rt = <optimized out>
        s = <optimized out>
#2  __pthread_cond_wait (cond=0x7f9d60000c80, mutex=0x7f9d60000b20) at pthread_cond_wait.c:655
No locals.
#3  0x0000561614377c8d in futures::task_impl::std::ThreadNotify::park ()
No symbol table info available.
#4  0x000056161432d263 in std::thread::local::LocalKey<T>::with ()
No symbol table info available.
#5  0x000056161432ea05 in futures::future::Future::wait ()
No symbol table info available.
#6  0x000056161432cc95 in std::sys_common::backtrace::__rust_begin_short_backtrace ()
No symbol table info available.
#7  0x000056161432e1fe in std::panicking::try::do_call ()
No symbol table info available.
#8  0x0000561614388a4a in __rust_maybe_catch_panic () at src/libpanic_unwind/lib.rs:80
No locals.
#9  0x000056161433a733 in core::ops::function::FnOnce::call_once{{vtable-shim}} ()
No symbol table info available.
#10 0x000056161437bcdf in <alloc::boxed::Box<F> as core::ops::function::FnOnce<A>>::call_once () at /rustc/4560ea788cb760f0a34127156c78e2552949f734/src/liballoc/boxed.rs:922
No locals.
#11 0x0000561614388110 in <alloc::boxed::Box<F> as core::ops::function::FnOnce<A>>::call_once () at /rustc/4560ea788cb760f0a34127156c78e2552949f734/src/liballoc/boxed.rs:922
No locals.
#12 std::sys_common::thread::start_thread () at src/libstd/sys_common/thread.rs:13
No locals.
#13 std::sys::unix::thread::Thread::new::thread_start () at src/libstd/sys/unix/thread.rs:79
No locals.
#14 0x00007f9d6de5d3a7 in start_thread (arg=<optimized out>) at pthread_create.c:486
        ret = <optimized out>
        pd = <optimized out>
        now = <optimized out>
        unwind_buf = {cancel_jmp_buf = {{jmp_buf = {140314124887808, -2814752794425973897, 140314126984014, 140314126984015, 140314124885696, 140314126984272, 2870242235355734903, 2870243169132979063}, mask_was_saved = 0}}, priv = {pad = {0x0, 0x0, 0x0, 0x0}, data = {prev = 0x0, cleanup = 0x0, canceltype = 0}}}
        not_first_call = <optimized out>
#15 0x00007f9d6dd720ff in clone () at ../sysdeps/unix/sysv/linux/x86_64/clone.S:95
No locals.

Thread 2 (Thread 0x7f9d6db23700 (LWP 9456)):
#0  futex_wait_cancelable (private=0, expected=0, futex_word=0x561615133cd8) at ../sysdeps/unix/sysv/linux/futex-internal.h:88
        __ret = -512
        oldtype = 0
        err = <optimized out>
        oldtype = <optimized out>
        err = <optimized out>
        __ret = <optimized out>
        resultvar = <optimized out>
        __arg4 = <optimized out>
        __arg3 = <optimized out>
        __arg2 = <optimized out>
        __arg1 = <optimized out>
        _a4 = <optimized out>
        _a3 = <optimized out>
        _a2 = <optimized out>
        _a1 = <optimized out>
#1  __pthread_cond_wait_common (abstime=0x0, mutex=0x561615132c30, cond=0x561615133cb0) at pthread_cond_wait.c:502
        spin = 0
        buffer = {__routine = 0x7f9d6de63b80 <__condvar_cleanup_waiting>, __arg = 0x7f9d6db22340, __canceltype = 0, __prev = 0x0}
        cbuffer = {wseq = 16, cond = 0x561615133cb0, mutex = 0x561615132c30, private = 0}
        rt = <optimized out>
        err = <optimized out>
        g = 0
        flags = <optimized out>
        g1_start = <optimized out>
        signals = <optimized out>
        result = 0
        wseq = 16
        seq = 8
        private = 0
        maxspin = <optimized out>
        err = <optimized out>
        result = <optimized out>
        wseq = <optimized out>
        g = <optimized out>
        seq = <optimized out>
        flags = <optimized out>
        private = <optimized out>
        signals = <optimized out>
        g1_start = <optimized out>
        spin = <optimized out>
        buffer = <optimized out>
        cbuffer = <optimized out>
        rt = <optimized out>
        s = <optimized out>
#2  __pthread_cond_wait (cond=0x561615133cb0, mutex=0x561615132c30) at pthread_cond_wait.c:655
No locals.
#3  0x000056161437c793 in std::sys::unix::condvar::Condvar::wait () at src/libstd/sys/unix/condvar.rs:71
No locals.
#4  std::sys_common::condvar::Condvar::wait () at src/libstd/sys_common/condvar.rs:41
No locals.
#5  std::sync::condvar::Condvar::wait () at src/libstd/sync/condvar.rs:204
No locals.
#6  std::thread::park () at src/libstd/thread/mod.rs:911
No locals.
#7  0x00005616143828d2 in std::sync::mpsc::blocking::WaitToken::wait () at src/libstd/sync/mpsc/blocking.rs:71
No locals.
#8  0x00005616143249b1 in std::sync::mpsc::stream::Packet<T>::recv ()
No symbol table info available.
#9  0x0000561614333028 in std::sync::mpsc::Receiver<T>::recv ()
No symbol table info available.
#10 0x0000561614328607 in core::ops::function::FnOnce::call_once ()
No symbol table info available.
#11 0x000056161434637f in <alloc::boxed::Box<F> as core::ops::function::FnOnce<A>>::call_once () at /rustc/4560ea788cb760f0a34127156c78e2552949f734/src/liballoc/boxed.rs:922
No locals.
#12 0x0000561614388a4a in __rust_maybe_catch_panic () at src/libpanic_unwind/lib.rs:80
No locals.
#13 0x0000561614360b8e in std::panicking::try () at /rustc/4560ea788cb760f0a34127156c78e2552949f734/src/libstd/panicking.rs:271
No locals.
#14 std::panic::catch_unwind () at /rustc/4560ea788cb760f0a34127156c78e2552949f734/src/libstd/panic.rs:394
No locals.
#15 test::run_test::run_test_inner::{{closure}} () at src/libtest/lib.rs:1413
No locals.
#16 0x000056161433bbf5 in std::sys_common::backtrace::__rust_begin_short_backtrace () at /rustc/4560ea788cb760f0a34127156c78e2552949f734/src/libstd/sys_common/backtrace.rs:126
No locals.
#17 0x000056161433fdf5 in std::thread::Builder::spawn_unchecked::{{closure}}::{{closure}} () at /rustc/4560ea788cb760f0a34127156c78e2552949f734/src/libstd/thread/mod.rs:470
No locals.
#18 <std::panic::AssertUnwindSafe<F> as core::ops::function::FnOnce<()>>::call_once () at /rustc/4560ea788cb760f0a34127156c78e2552949f734/src/libstd/panic.rs:315
No locals.
#19 std::panicking::try::do_call () at /rustc/4560ea788cb760f0a34127156c78e2552949f734/src/libstd/panicking.rs:292
No locals.
#20 0x0000561614388a4a in __rust_maybe_catch_panic () at src/libpanic_unwind/lib.rs:80
No locals.
#21 0x0000561614340362 in std::panicking::try () at /rustc/4560ea788cb760f0a34127156c78e2552949f734/src/libstd/panicking.rs:271
No locals.
#22 std::panic::catch_unwind () at /rustc/4560ea788cb760f0a34127156c78e2552949f734/src/libstd/panic.rs:394
No locals.
#23 std::thread::Builder::spawn_unchecked::{{closure}} () at /rustc/4560ea788cb760f0a34127156c78e2552949f734/src/libstd/thread/mod.rs:469
No locals.
#24 core::ops::function::FnOnce::call_once{{vtable-shim}} () at /rustc/4560ea788cb760f0a34127156c78e2552949f734/src/libcore/ops/function.rs:227
No locals.
#25 0x000056161437bcdf in <alloc::boxed::Box<F> as core::ops::function::FnOnce<A>>::call_once () at /rustc/4560ea788cb760f0a34127156c78e2552949f734/src/liballoc/boxed.rs:922
No locals.
#26 0x0000561614388110 in <alloc::boxed::Box<F> as core::ops::function::FnOnce<A>>::call_once () at /rustc/4560ea788cb760f0a34127156c78e2552949f734/src/liballoc/boxed.rs:922
No locals.
#27 std::sys_common::thread::start_thread () at src/libstd/sys_common/thread.rs:13
No locals.
#28 std::sys::unix::thread::Thread::new::thread_start () at src/libstd/sys/unix/thread.rs:79
No locals.
#29 0x00007f9d6de5d3a7 in start_thread (arg=<optimized out>) at pthread_create.c:486
        ret = <optimized out>
        pd = <optimized out>
        now = <optimized out>
        unwind_buf = {cancel_jmp_buf = {{jmp_buf = {140314126989056, -2814752794425973897, 140734155202382, 140734155202383, 140314126986944, 140734155202640, 2870242511844254583, 2870243169132979063}, mask_was_saved = 0}}, priv = {pad = {0x0, 0x0, 0x0, 0x0}, data = {prev = 0x0, cleanup = 0x0, canceltype = 0}}}
        not_first_call = <optimized out>
#30 0x00007f9d6dd720ff in clone () at ../sysdeps/unix/sysv/linux/x86_64/clone.S:95
No locals.

Thread 1 (Thread 0x7f9d6db24840 (LWP 9455)):
#0  futex_reltimed_wait_cancelable (private=0, reltime=0x7fff39542410, expected=0, futex_word=0x561615132a98) at ../sysdeps/unix/sysv/linux/futex-internal.h:142
        __ret = -516
        oldtype = 0
        err = <optimized out>
        oldtype = <optimized out>
        err = <optimized out>
        __ret = <optimized out>
        resultvar = <optimized out>
        __arg4 = <optimized out>
        __arg3 = <optimized out>
        __arg2 = <optimized out>
        __arg1 = <optimized out>
        _a4 = <optimized out>
        _a3 = <optimized out>
        _a2 = <optimized out>
        _a1 = <optimized out>
#1  __pthread_cond_wait_common (abstime=0x7fff395424e8, mutex=0x561615132a20, cond=0x561615132a70) at pthread_cond_wait.c:533
        rt = {tv_sec = 59, tv_nsec = 999910039}
        spin = 0
        buffer = {__routine = 0x7f9d6de63b80 <__condvar_cleanup_waiting>, __arg = 0x7fff39542440, __canceltype = 1844400740, __prev = 0x0}
        cbuffer = {wseq = 0, cond = 0x561615132a70, mutex = 0x561615132a20, private = 0}
        err = <optimized out>
        g = 0
        flags = 2
        g1_start = <optimized out>
        maxspin = 0
        signals = <optimized out>
        result = 0
        wseq = 0
        seq = 0
        private = 0
        maxspin = <optimized out>
        err = <optimized out>
        result = <optimized out>
        wseq = <optimized out>
        g = <optimized out>
        seq = <optimized out>
        flags = <optimized out>
        private = <optimized out>
        signals = <optimized out>
        g1_start = <optimized out>
        spin = <optimized out>
        buffer = <optimized out>
        cbuffer = <optimized out>
        rt = <optimized out>
        s = <optimized out>
#2  __pthread_cond_timedwait (cond=0x561615132a70, mutex=0x561615132a20, abstime=0x7fff395424e8) at pthread_cond_wait.c:667
No locals.
#3  0x0000561614387061 in std::sys::unix::condvar::Condvar::wait_timeout () at src/libstd/sys/unix/condvar.rs:102
No locals.
#4  0x000056161437cb52 in std::sys_common::condvar::Condvar::wait_timeout () at src/libstd/sys_common/condvar.rs:51
No locals.
#5  std::sync::condvar::Condvar::wait_timeout () at src/libstd/sync/condvar.rs:405
No locals.
#6  std::thread::park_timeout () at src/libstd/thread/mod.rs:1006
No locals.
#7  0x00005616143829fd in std::sync::mpsc::blocking::WaitToken::wait_max_until () at src/libstd/sync/mpsc/blocking.rs:82
No locals.
#8  0x000056161433e56e in std::sync::mpsc::shared::Packet<T>::recv () at /rustc/4560ea788cb760f0a34127156c78e2552949f734/src/libstd/sync/mpsc/shared.rs:229
No locals.
#9  0x0000561614358d78 in std::sync::mpsc::Receiver<T>::recv_deadline () at /rustc/4560ea788cb760f0a34127156c78e2552949f734/src/libstd/sync/mpsc/mod.rs:1403
No locals.
#10 std::sync::mpsc::Receiver<T>::recv_timeout () at /rustc/4560ea788cb760f0a34127156c78e2552949f734/src/libstd/sync/mpsc/mod.rs:1313
No locals.
#11 test::run_tests () at src/libtest/lib.rs:1125
No locals.
#12 test::run_tests_console () at src/libtest/lib.rs:957
No locals.
#13 0x000056161434f992 in test::test_main () at src/libtest/lib.rs:295
No locals.
#14 0x00005616143501ad in test::test_main_static () at src/libtest/lib.rs:329
No locals.
#15 0x000056161432eb13 in std::rt::lang_start::{{closure}} ()
No symbol table info available.
#16 0x0000561614385cb3 in std::rt::lang_start_internal::{{closure}} () at src/libstd/rt.rs:49
No locals.
#17 std::panicking::try::do_call () at src/libstd/panicking.rs:292
No locals.
#18 0x0000561614388a4a in __rust_maybe_catch_panic () at src/libpanic_unwind/lib.rs:80
No locals.
#19 0x000056161438677d in std::panicking::try () at src/libstd/panicking.rs:271
No locals.
#20 std::panic::catch_unwind () at src/libstd/panic.rs:394
No locals.
#21 std::rt::lang_start_internal () at src/libstd/rt.rs:48
No locals.
#22 0x0000561614329ad2 in main ()
No symbol table info available.

Hopefully something in that mess is relevant.

[tmiasko]

This issue happens to be easily reproducible on my system. The deadlocked situation is as follows:

• T1 is executing let o9 = rx3.recv().unwrap();. Waits for T3 to send a sender through tx3 / rx3.
• T2 is executing block_on_stream(rx.buffer_unordered(N)). It has obtained 8 receivers from rx already; 4 of them have been used, other 4 are buffered and pending. Waits for T1 to send items through any of 4 receivers. Note that T2 isn't polling from rx at this point.
• T3 is in the last iteration of the second loop, executing block_on(tx.send(myrx)).unwrap();. In fact it has already sent all 9 items through tx, but is blocked while flushing the sink, waiting for free capacity in the channel.

The T3 should be able to proceed successfully, since channel has capacity 1 and additional 1 for the sender. Unfortunatelly the implementation of per sender capacity is racy and it is possible for a sender to be incorrectly parked even though free capacity is there. If the receiver is being polled from, the situation eventually corrects itself. But in the case here rx is not being polled from so the problem persist.

Execution that leads to a mpsc bounded sender being incorrectly parked:

• sender: increase the number of messages
• sender: check if it needs to park itself: yes
• receiver: pop message from the queue
• receiver: check if any sender needs to be unparked: no
• receiver: decrement the number of messages
• sender: park and send the handle through parked_queue
• sender: send the message and wake the receiver

The receiver is woken up afterwards, but it isn't being polled from so that doesn't help.