SyncQueue.hpp

// Copyright © 2015 GEOSYSTEMS SRL
// All Rights Reserved.

#ifndef FIRE_THREADING_SYNCQUEUE_HPP_INCLUDED
#define FIRE_THREADING_SYNCQUEUE_HPP_INCLUDED

#include <boost/thread/mutex.hpp>
#include <boost/thread/condition_variable.hpp>

#include <list>
#include <exception>

#include "fire/gear/threading/types.hpp"

namespace fire {
namespace gear {
namespace threading {

template < class T >
class SyncQueue {
 public:
  SyncQueue(): _max(0) {
  }

  explicit SyncQueue(int const & max): _max(max) {
  }

  void push(const T& item) {
        // acquisisco il lock
        auto_lock l(_mtx);
        // verifico se posso caricare elementi
        while (_isFullNoWait()) {
            // attendo che ci sia spazio nel contenitore per la registrazione di item
            _notFull.wait(l);
            // verifico se esiste spazio per l'archiviazione (un altro thread potrebbe essere stato svegliato)
        }

        // archivio l'elemento in testa
        _items.push_front(item);
        // notifico ai consumatori la presenza di elmenti da consumare
        _notEmpty.notify_all();
  }

  T pop() {
      // acquisisco il lock
      auto_lock l(_mtx);
      // verifico se posso estrarre elementi
      while (_isEmptyNoWait()) {
          // attendo che ci siano elementi da leggere
          _notEmpty.wait(l);
          // verifico se ci sono element disponibili
      }

      // leggo l'elemento dal fondo della lista
      T item = _items.back();
      _items.pop_back();
      // notifico la presenza di slot vuoti per l'archiviazione
      _notFull.notify_all();

      return item;
  }

  size_t size() const {
    auto_lock l(_mtx);
    return _items.size();
  }

  size_t getMax() const {
    auto_lock l(_mtx);
    return _max;
  }

  bool isEmpty() {
      auto_lock l(_mtx);
      return _isEmptyNoWait();
  }

  bool _isFull() {
      auto_lock l(_mtx);
      return _isFullNoWait();
  }

 protected:
  bool _isEmptyNoWait() {
      return _items.size() == 0;
  }

  bool _isFullNoWait() {
      if (_max < 1)
          return false;
      else
          return _items.size() >= _max;
  }

 private:
  std::list< T > _items;
  size_t _max;

  mutable boost::mutex _mtx;
  mutable boost::condition_variable _notFull;
  mutable boost::condition_variable _notEmpty;
};

}  // namespace threading
}  // namespace gear
}  // namespace fire

#endif