syncque.h

Go to the documentation of this file.

/*
  Copyright (c) Mircea Neacsu (2014-2025) Licensed under MIT License.
  This file is part of MLIB project. See LICENSE file for full license terms.
*/

 
#pragma once
 
#if __has_include("defs.h")
#include "defs.h"
#endif
 
#include <queue>
#include "semaphore.h"
#include "critsect.h"
#include "stopwatch.h"
 
namespace mlib {
template <class M>
class sync_queue
{
public:
  sync_queue ()
    : message (nullptr)
  {}

  void produce (const M& obj)
  {
    update.enter ();
    while (message)
    {
      update.leave ();
      prod_sema.wait ();
      update.enter ();
    }
    cons_sema.signal ();
    message = &obj;
    update.leave ();
  }
 
  void consume (M& obj)
  {
    update.enter ();
    while (!message)
    {
      update.leave ();
      cons_sema.wait ();
      update.enter ();
    }
    prod_sema.signal ();
    obj = *message;
    message = nullptr;
    update.leave ();
  }
 
private:
  criticalsection update; 
  semaphore prod_sema;    
  semaphore cons_sema;    
  const M* message;
};

template <class M, class C = std::deque<M>>
class async_queue : protected std::queue<M, C>
{
public:
  async_queue (size_t limit_ = INFINITE)
    : limit (limit_)
  {
    if (limit > 0 && limit < INFINITE)
      prod_sema.signal ((int)limit);
  }

  virtual bool produce (const M& obj, DWORD timeout = INFINITE)
  {
    if (limit < INFINITE)
    {
      // Bounded queue. See if there is enough space to produce
      stopwatch t; // need a stopwatch to count cumulated wait time
      if (timeout != INFINITE)
        t.start ();
      update.enter ();
      while (std::queue<M, C>::size () >= limit)
      {
        update.leave ();
        if (timeout != INFINITE)
        {
          int t_wait = timeout - (int)t.msecLap ();
          if (t_wait < 0 || prod_sema.wait (t_wait) == WAIT_TIMEOUT)
            return false;
        }
        else
          prod_sema.wait ();
        update.enter ();
      }
      std::queue<M, C>::push (obj);
      cons_sema.signal ();
      update.leave ();
    }
    else
    {
      // Unbounded queue. Producer will always be successful
      lock l (update);              // take control of the queue
      std::queue<M, C>::push (obj); // put a copy of the object at the end
      cons_sema.signal ();          // and signal the semaphore
    }
    return true;
  }

  virtual bool consume (M& result, int timeout = INFINITE)
  {
    if (timeout != INFINITE)
    {
      stopwatch t;
      int t_wait;
      t.start ();
      update.enter ();
      while (std::queue<M, C>::empty ())
      {
        update.leave ();
        t_wait = timeout - (int)t.msecLap ();
        // give up or wait for a producer
        if (t_wait <= 0 || cons_sema.wait (t_wait) == WAIT_TIMEOUT)
          return false;
        update.enter ();
      }
    }
    else
    {
      update.enter ();
      if (std::queue<M, C>::empty ())
      {
        while (1)
        {
          update.leave ();
          cons_sema.wait (); // wait for a producer
          update.enter ();
          if (!std::queue<M, C>::empty ())
            break;
        }
      }
    }
    result = std::queue<M, C>::front (); // get the message
    std::queue<M, C>::pop ();
    if (limit != INFINITE)
      prod_sema.signal (); // signal producers there is space available
 
    update.leave ();
    return true;
  }

  bool empty ()
  {
    lock l (update);
    return std::queue<M, C>::empty ();
  }

  bool full ()
  {
    lock l (this->update);
    return (std::queue<M, C>::size () == limit);
  }

  size_t size ()
  {
    lock l (update);
    return std::queue<M, C>::size ();
  }
 
protected:
  size_t limit;           
  semaphore prod_sema;    
  semaphore cons_sema;    
  criticalsection update; 
};
 
} // namespace mlib