ProcessingQueue.h

Go to the documentation of this file.

#ifndef OSMSCOUT_UTIL_PROCESSINGQUEUE_H
#define OSMSCOUT_UTIL_PROCESSINGQUEUE_H
 
/*
  This source is part of the libosmscout library
  Copyright (C) 2020 Tim Teulings
 
  This library is free software; you can redistribute it and/or
  modify it under the terms of the GNU Lesser General Public
  License as published by the Free Software Foundation; either
  version 2.1 of the License, or (at your option) any later version.
 
  This library is distributed in the hope that it will be useful,
  but WITHOUT ANY WARRANTY; without even the implied warranty of
  MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
  Lesser General Public License for more details.
 
  You should have received a copy of the GNU Lesser General Public
  License along with this library; if not, write to the Free Software
  Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307  USA
*/
 
#include <condition_variable>
#include <deque>
#include <limits>
#include <memory>
#include <mutex>
#include <optional>
 
#include <osmscout/lib/CoreImportExport.h>
 
namespace osmscout {

  template<typename T>
  class ProcessingQueue
  {
  private:
    std::mutex              mutex;
    std::condition_variable pushCondition;
    std::condition_variable popCondition;
    std::deque<T>           tasks;
    size_t                  queueLimit=std::numeric_limits<size_t>::max();
    bool                    running{true};
 
  public:
    ProcessingQueue();
    explicit ProcessingQueue(size_t queueLimit);
 
    ProcessingQueue(const ProcessingQueue&) = delete;
    ProcessingQueue &operator=(const ProcessingQueue&) = delete;
 
    virtual ~ProcessingQueue() = default;
 
    void PushTask(const T& task);
    void PushTask(T&& task);
    std::optional<T> PopTask();
 
    void Stop();
  };

  template<class T>
  ProcessingQueue<T>::ProcessingQueue() = default;

  template<class T>
  ProcessingQueue<T>::ProcessingQueue(size_t queueLimit)
    : queueLimit(queueLimit)
  {
    // no code
  }

  template<class T>
  void ProcessingQueue<T>::PushTask(const T& task)
  {
    std::unique_lock lock(mutex);
 
    pushCondition.wait(lock,[this]{return tasks.size()<queueLimit;});
 
    tasks.push_back(task);
 
    lock.unlock();
 
    popCondition.notify_one();
  }

  template<class T>
  void ProcessingQueue<T>::PushTask(T&& task)
  {
    std::unique_lock lock(mutex);
 
    pushCondition.wait(lock,[this]{return tasks.size()<queueLimit;});
 
    tasks.push_back(std::forward<T>(task));
 
    lock.unlock();
 
    popCondition.notify_one();
  }

  template<class T>
  std::optional<T> ProcessingQueue<T>::PopTask()
  {
    std::unique_lock lock(mutex);
 
    popCondition.wait(lock,[this]{return !tasks.empty() || !running;});
 
    if (!running &&
        tasks.empty()) {
      return std::nullopt;
    }
 
    std::optional<T> task=std::move(tasks.front());
    tasks.pop_front();
 
    lock.unlock();
 
    pushCondition.notify_one();
 
    return task;
  }

 
  template<class R>
  void ProcessingQueue<R>::Stop()
  {
    std::unique_lock lock(mutex);
 
    running=false;
 
    lock.unlock();
 
    popCondition.notify_all();
  }
}
 
#endif