ThreadScheduler.h

Go to the documentation of this file.

// Mantid Repository : https://github.com/mantidproject/mantid
//
// Copyright © 2011 ISIS Rutherford Appleton Laboratory UKRI,
//   NScD Oak Ridge National Laboratory, European Spallation Source,
//   Institut Laue - Langevin & CSNS, Institute of High Energy Physics, CAS
// SPDX - License - Identifier: GPL - 3.0 +
#pragma once
 
#include "MantidKernel/DllConfig.h"
#include "MantidKernel/SingletonHolder.h"
#include "MantidKernel/Task.h"
#include <atomic>
#include <deque>
#include <map>
#include <memory>
#include <vector>
 
namespace Mantid {
 
namespace Kernel {
 
//===========================================================================
//===========================================================================
//===========================================================================
class MANTID_KERNEL_DLL ThreadScheduler {
public:
  ThreadScheduler() : m_cost(0), m_costExecuted(0), m_abortException(""), m_aborted(false) {}
 
  virtual ~ThreadScheduler() = default;
 
  //-----------------------------------------------------------------------------------
  virtual void push(std::shared_ptr<Task> newTask) = 0;
 
  //-----------------------------------------------------------------------------------
  virtual std::shared_ptr<Task> pop(size_t threadnum) = 0;
 
  //-----------------------------------------------------------------------------------
  virtual void finished(Task *task, size_t threadnum) {
    UNUSED_ARG(task);
    UNUSED_ARG(threadnum);
  }
 
  //-----------------------------------------------------------------------------------
  virtual void abort(std::runtime_error exception) {
    // Save the exception for re-throwing
    m_abortException = exception;
    m_aborted = true;
    // Clear (and delete) the queue
    clear();
  }
 
  //-----------------------------------------------------------------------------------
  virtual size_t size() = 0;
 
  virtual bool empty() = 0;
 
  virtual void clear() = 0;
 
  //-------------------------------------------------------------------------------
  double totalCost() { return m_cost; }
 
  //-------------------------------------------------------------------------------
  double totalCostExecuted() { return m_costExecuted; }
 
  //-------------------------------------------------------------------------------
  std::runtime_error getAbortException() { return m_abortException; }
  //-------------------------------------------------------------------------------
  bool getAborted() { return m_aborted; }
 
  //-------------------------------------------------------------------------------
  void markBusy() { ++m_numBusyThreads; }
 
  void markIdle() { --m_numBusyThreads; }
 
  bool hasBusyThreads() const { return m_numBusyThreads.load() > 0; }
 
protected:
  double m_cost;
  double m_costExecuted;
  std::mutex m_queueLock;
  std::runtime_error m_abortException;
  bool m_aborted;
  std::atomic<size_t> m_numBusyThreads{0};
};
 
//===========================================================================
//===========================================================================
//===========================================================================
class MANTID_KERNEL_DLL ThreadSchedulerFIFO : public ThreadScheduler {
public:
  ThreadSchedulerFIFO() : ThreadScheduler() {}
 
  ~ThreadSchedulerFIFO() override { clear(); }
 
  //-------------------------------------------------------------------------------
  bool empty() override {
    std::lock_guard<std::mutex> _lock(m_queueLock);
    return m_queue.empty();
  }
 
  //-------------------------------------------------------------------------------
  void push(std::shared_ptr<Task> newTask) override {
    // Cache the total cost
    m_queueLock.lock();
    m_cost += newTask->cost();
    m_queue.emplace_back(newTask);
    m_queueLock.unlock();
  }
 
  //-------------------------------------------------------------------------------
  std::shared_ptr<Task> pop(size_t threadnum) override {
    UNUSED_ARG(threadnum);
    std::shared_ptr<Task> temp = nullptr;
    m_queueLock.lock();
    // Check the size within the same locking block; otherwise the size may
    // change before you get the next item.
    if (!m_queue.empty()) {
      // TODO: Would a try/catch block be smart here?
      temp = m_queue.front();
      m_queue.pop_front();
    }
    m_queueLock.unlock();
    return temp;
  }
 
  //-------------------------------------------------------------------------------
  size_t size() override {
    m_queueLock.lock();
    size_t temp = m_queue.size();
    m_queueLock.unlock();
    return temp;
  }
 
  //-------------------------------------------------------------------------------
  void clear() override final {
    m_queueLock.lock();
    // Empty out the queue
    m_queue.clear();
    m_cost = 0;
    m_costExecuted = 0;
    m_queueLock.unlock();
  }
 
protected:
  std::deque<std::shared_ptr<Task>> m_queue;
};
 
//===========================================================================
//===========================================================================
//===========================================================================
class MANTID_KERNEL_DLL ThreadSchedulerLIFO : public ThreadSchedulerFIFO {
 
  //-------------------------------------------------------------------------------
  std::shared_ptr<Task> pop(size_t threadnum) override {
    UNUSED_ARG(threadnum);
    std::shared_ptr<Task> temp = nullptr;
    m_queueLock.lock();
    // Check the size within the same locking block; otherwise the size may
    // change before you get the next item.
    if (!m_queue.empty()) {
      // TODO: Would a try/catch block be smart here?
      temp = m_queue.back();
      m_queue.pop_back();
    }
    m_queueLock.unlock();
    return temp;
  }
};
 
//===========================================================================
//===========================================================================
//===========================================================================
class MANTID_KERNEL_DLL ThreadSchedulerLargestCost : public ThreadScheduler {
public:
  ThreadSchedulerLargestCost() : ThreadScheduler() {}
 
  ~ThreadSchedulerLargestCost() override { clear(); }
 
  //-------------------------------------------------------------------------------
  bool empty() override {
    std::lock_guard<std::mutex> _lock(m_queueLock);
    return m_map.empty();
  }
 
  //-------------------------------------------------------------------------------
  void push(std::shared_ptr<Task> newTask) override {
    // Cache the total cost
    m_queueLock.lock();
    m_cost += newTask->cost();
    m_map.emplace(newTask->cost(), newTask);
    m_queueLock.unlock();
  }
 
  //-------------------------------------------------------------------------------
  std::shared_ptr<Task> pop(size_t threadnum) override {
    UNUSED_ARG(threadnum);
    std::shared_ptr<Task> temp = nullptr;
    m_queueLock.lock();
    // Check the size within the same locking block; otherwise the size may
    // change before you get the next item.
    if (!m_map.empty()) {
      // Since the map is sorted by cost, we want the LAST item.
      auto it = m_map.end();
      it--;
      temp = it->second;
      m_map.erase(it);
    }
    m_queueLock.unlock();
    return temp;
  }
 
  //-------------------------------------------------------------------------------
  size_t size() override {
    m_queueLock.lock();
    size_t temp = m_map.size();
    m_queueLock.unlock();
    return temp;
  }
 
  //-------------------------------------------------------------------------------
  void clear() override final {
    m_queueLock.lock();
    // Empty out the queue and delete the pointers!
    m_map.clear();
    m_cost = 0;
    m_costExecuted = 0;
    m_queueLock.unlock();
  }
 
protected:
  std::multimap<double, std::shared_ptr<Task>> m_map;
};
 
} // namespace Kernel
} // namespace Mantid