Deadlock

[najamelan]

I had a quick look at it. It does have an unsafe without comment clarifying why it's guaranteed to be safe, which is unfortunate. On performance, it seems to be in the same league as async_std::task::block_on.

Now for the bad news, this deadlocks, and it doesn't on async_std::task::block_on, nor on futures::executor::block_on:

use
{
   std   :: { thread, sync::atomic::{ AtomicUsize, Ordering::SeqCst } } ,
   tokio :: { sync::mpsc                                              } ,
};


// queue size for the bounded channel, 16 is same as actix default size.
//
const BOUNDED : usize = 16;
const MESSAGES: usize = 100_000;


fn main()
{
   let (mut a_tx, mut a_rx) = mpsc::channel( BOUNDED );
   let (mut b_tx, mut b_rx) = mpsc::channel( BOUNDED );


   let thread_a = thread::spawn( move ||
   {
      pollster::block_on( async
      {
         while let Some( msg ) = a_rx.recv().await
         {
            b_tx.send( msg ).await.expect( "send on b" );
         }
      });
   });


   let thread_b = thread::spawn( move ||
   {
      pollster::block_on( async move
      {
         for _ in 0..MESSAGES
         {
            a_tx.send( () ).await.expect( "Send on a" );
         }
      });
   });


   pollster::block_on( async move
   {
      let sum = AtomicUsize::new(0);

      while sum.fetch_add( 1, SeqCst ) < MESSAGES
      {
         b_rx.recv().await;
      }

      assert_eq!( sum.load( SeqCst), MESSAGES + 1 );
   });


   thread_a.join().expect( "join thread_a" );
   thread_b.join().expect( "join thread_b" );
}

Some proper testing is in order I think. Good luck.

[zesterer]

Thanks for making me aware of this. In hindsight, it's fairly clear to me why this happens. I'll address it now.

[zesterer]

I've now fixed this deadlock and have added your code as a test. Somewhat frustrated with myself for not seeing this in advance! This has now been released as 0.2.0.