title: Deterministic Monitor Pulse/ Wait and implementing timeout in producer-consumer collection
body: <p>I'm trying to implement a concurrent producer-consumer collection (multiple producers and consumers) that supports timeouts for consumers.</p>
  
  <p>Now the actual collection is pretty complicated (nothing in System.Collections.Concurrent that does the job unfortunately), but I have a minimal sample here that demonstrates my problem (looks a bit like <code>BlockingCollection&lt;T&gt;</code>).</p>
  
  <pre><code>public sealed class ProducerConsumerQueueDraft&lt;T&gt;
  {
      private readonly Queue&lt;T&gt; queue = new Queue&lt;T&gt;();
      private readonly object locker = new object();
  
      public void Enqueue(T item)
      {
          lock (locker)
          {
              queue.Enqueue(item);
  
              /* This "optimization" is broken, as Nicholas Butler points out.
              if(queue.Count == 1) // Optimization
              */
                  Monitor.Pulse(locker); // Notify any waiting consumer threads.
          }
      }
  
      public T Dequeue(T item)
      {
          lock (locker)
          {
              // Surprisingly, this needs to be a *while* and not an *if*
              // which is the core of my problem.
              while (queue.Count == 0)
                  Monitor.Wait(locker);
  
              return queue.Dequeue();
          }
      }
  
      // This isn't thread-safe, but is how I want TryDequeue to look.
      public bool TryDequeueDesired(out T item, TimeSpan timeout)
      {
          lock (locker)
          {
              if (queue.Count == 0 &amp;&amp; !Monitor.Wait(locker, timeout))
              {
                  item = default(T);
                  return false;
              }
  
              // This is wrong! The queue may be empty even though we were pulsed!
              item = queue.Dequeue();
              return true;
          }
      }
  
      // Has nasty timing-gymnastics I want to avoid.
      public bool TryDequeueThatWorks(out T item, TimeSpan timeout)
      {
          lock (locker)
          {
              var watch = Stopwatch.StartNew();
              while (queue.Count == 0)
              {
                  var remaining = timeout - watch.Elapsed;
  
                  if (!Monitor.Wait(locker, remaining &lt; TimeSpan.Zero ? TimeSpan.Zero : remaining))
                  {
                      item = default(T);
                      return false;
                  }
              }
              item = queue.Dequeue();
              return true;
          }
      }
  }
  </code></pre>
  
  <p>The idea is straightforward: consumers who find an empty queue wait to be signaled, and producers <code>Pulse</code> (note: not <code>PulseAll</code>, which would be inefficient) them to notify them of a waiting item. </p>
  
  <p>My problem is this property of  <a href="http://msdn.microsoft.com/en-us/library/system.threading.monitor.pulse" rel="nofollow"><code>Monitor.Pulse</code></a>:</p>
  
  <blockquote>
    <p>When the thread that invoked Pulse releases the lock, the next
    thread in the ready queue <strong>(which is not necessarily the thread that
    was pulsed)</strong> acquires the lock.</p>
  </blockquote>
  
  <p>What this means is that consumer-thread C1 could be woken up by a producer-thread to consume an item, but <em>another</em> consumer-thread C2 could acquire the lock before C1 has a chance to reacquire it, and consume the item, leaving C1 with an empty queue when it is given control. </p>
  
  <p>This means I have to defensively check in the consumer code on every pulse if the queue is indeed non-empty, and go back and wait empty-handed if this not the case.</p>
  
  <p>My primary issue with this is that it inefficient - threads may be woken up to do work and then promptly sent back to wait again. A related consequence of this is that implementing a <code>TryDequeue</code> with a timeout is unnecessarily difficult and inefficient (see <code>TryDequeueThatWorks</code>) when it should be elegant (see <code>TryDequeueDesired</code>). </p>
  
  <p><strong>How can I twist <code>Monitor.Pulse</code> to do what I want? Alternatively, is there another synchronization primitive that does? Is there a more efficient and/or elegant way to implement a <code>TryDequeue</code> timeout than what I have done?</strong></p>
  
  <p>FYI, here's a test that demonstrates the issues with my desired solution:</p>
  
  <pre><code>var queue = new ProducerConsumerQueueDraft&lt;int&gt;();
  
  for (int consumer = 0; consumer &lt; 3; consumer++)
      new Thread(() =&gt;
      {
          while (true)
          {
              int item;
  
              // This call should occasionally throw an exception.
              // Switching to queue.TryDequeueThatWorks should make
              // the problem go away.
              if (queue.TryDequeueDesired(out item, TimeSpan.FromSeconds(1)))
              {
                  // Do nothing.
              }
          }
  
      }).Start();
  
  Thread.Sleep(1000); // Let consumers get up and running
  
  for (int itemIndex = 0; itemIndex &lt; 50000000; itemIndex++)
  {
      queue.Enqueue(0);
  }
  </code></pre>