cbthreadpool.h

Go to the documentation of this file.

/*
 * This file is part of the Code::Blocks IDE and licensed under the GNU Lesser General Public License, version 3
 * http://www.gnu.org/licenses/lgpl-3.0.html
 */

#ifndef CBTHREADPOOL_H
#define CBTHREADPOOL_H

#include <wx/thread.h>
#include <wx/event.h>
#include <vector>
#include <list>

#include "cbthreadedtask.h"
#include "settings.h"
#include "prep.h"

class DLLIMPORT cbThreadPool
{
  public:
    cbThreadPool(wxEvtHandler *owner, int id = -1, int concurrentThreads = -1, unsigned int stackSize = 0);

    ~cbThreadPool();

    void SetConcurrentThreads(int concurrentThreads);

    int GetConcurrentThreads() const;

    int GetId() const { return m_ID; }

    void AddTask(cbThreadedTask *task, bool autodelete = true);

    void AbortAllTasks();

    bool Done() const;

    void BatchBegin();

    void BatchEnd();

  private:

    /* class for counted reference semantics
     * - deletes the object to which it refers when the last CountedPtr
     *   that refers to it is destroyed
     */
    template <typename T>
    class CountedPtr
    {
      private:
        T *ptr;         // pointer to the value
        long *count;    // shared number of owners

      public:
        // initialize pointer with existing pointer
        // - requires that the pointer p is a return value of new
        explicit CountedPtr(T *p = nullptr) : ptr(p), count(new long(1)) {}
        // copy pointer (one more owner)
        CountedPtr(const CountedPtr<T> &p) : ptr(p.ptr), count(p.count)
        {
          ++*count;
        }

        // destructor (delete value if this was the last owner)
        ~CountedPtr()
        {
            dispose();
        }
        CountedPtr<T> &operator = (const CountedPtr<T> &p)
        {
            if (this != &p)
            {
                dispose();
                ptr = p.ptr;
                count = p.count;
                ++*count;
            }

            return *this;
        }

        T &operator * () const
        {
            return *ptr;
        }

        T *operator -> () const
        {
            return ptr;
        }

      private:
        void dispose()
        {
            if (--*count == 0)
            {
                delete count;
                delete ptr;
            }
        }
    };

    class cbWorkerThread : public wxThread
    {
      public:
        cbWorkerThread(cbThreadPool *pool, CountedPtr<wxSemaphore> &semaphore);

        ExitCode Entry() override;

        void Abort();

        bool Aborted() const;

        void AbortTask();

      private:
        bool m_abort;

        cbThreadPool *m_pPool;

        CountedPtr<wxSemaphore> m_semaphore;

        cbThreadedTask *m_pTask;

        wxMutex m_taskMutex;
    };

    typedef std::vector<cbWorkerThread *> WorkerThreadsArray;

    struct cbThreadedTaskElement
    {
      cbThreadedTaskElement(cbThreadedTask *_task = nullptr, bool _autodelete = false)
      : task(_task),
        autodelete(_autodelete)
      {
        // empty
      }

      void Delete()
      {
        if (autodelete)
        {
          delete task;
          task = nullptr; // better safe than sorry
        }
      }

      cbThreadedTask *task;
      bool autodelete;
    };

    typedef std::list<cbThreadedTaskElement> TasksQueue;

    wxEvtHandler *m_pOwner; // events notification will send to this guy
    int m_ID;           // id used to fill the ID field of the event
    bool m_batching;    // whether in batch mode of adding tasks

    // current number of concurrent threads, this is the maximum value of the m_workingThreads
    // this variable should always be positive, 0 and -1 is not allowed.
    int m_concurrentThreads;
    unsigned int m_stackSize; // stack size for every threads

    // if we cannot apply the new value of concurrent threads, keep it here, usually the time to
    // apply a scheduled value is when all the tasks is done.
    int m_concurrentThreadsSchedule;

    // the total threads(cbWorkerThread) are stored here, this contains all the threads either is
    // currently running or in idle() mode.
    WorkerThreadsArray m_threads;

    // the pending tasks (cbThreadedTaskElement), usually we have many tasks to run in the pool, but
    // we have limited number of threads to run those tasks, so tasks which don't have the chance to
    // run will be put in the queue. Once a thread finishes a task, it will fetch a new task from
    // this task queue.
    TasksQueue m_tasksQueue;

    // true if any task added, reset to false if all the tasks is done
    bool m_taskAdded;

    int m_workingThreads;

    mutable wxMutex m_Mutex; // we better be safe, protect the change of member variables

    // used to synchronize the Worker Threads, the counted value is that how many threads are
    // sharing this semaphore. The semaphore's initial value is the thread number we can used to
    // run the tasks.
    // initial counted value = m_concurrentThreads
    // the value of semaphore = the number of threads in Idle mode
    CountedPtr<wxSemaphore> m_semaphore;

    void _SetConcurrentThreads(int concurrentThreads); // like SetConcurrentThreads, but non-thread safe

    // awakes all threads, so they will leave from the Idle mode to working mode
    // this is used when we are going to abort all the threads, there are two
    // cases we need to call Broadcast(), one is the destructor, the other is the user need to
    // change the concurrent thread numbers, so we abort all the threads, and re-create them again.
    void Broadcast();

    // awakes only a few threads, this usually happens when we add some tasks, and there are some
    // threads which is currently in idle mode, so we can awake these idle threads to run tasks.
    void AwakeNeeded();


  protected:
    friend class cbWorkerThread;

    cbThreadedTaskElement GetNextTask();

    void WorkingThread();

    bool WaitingThread();

    void TaskDone(cbWorkerThread *thread);
};

/* ************************************************ */
/* **************** INLINE MEMBERS **************** */
/* ************************************************ */

inline cbThreadPool::cbThreadPool(wxEvtHandler *owner, int id, int concurrentThreads, unsigned int stackSize)
: m_pOwner(owner),
  m_ID(id),
  m_batching(false),
  m_concurrentThreads(-1),
  m_stackSize(stackSize),
  m_concurrentThreadsSchedule(0),
  m_taskAdded(false),
  m_workingThreads(0),
  m_semaphore(new wxSemaphore)
{
  // m_concurrentThreads will be set to a positive integer value.
  SetConcurrentThreads(concurrentThreads);
}

inline int cbThreadPool::GetConcurrentThreads() const
{
  wxMutexLocker lock(m_Mutex);
  return m_concurrentThreadsSchedule ? m_concurrentThreadsSchedule : m_concurrentThreads;
}

inline bool cbThreadPool::Done() const
{
  wxMutexLocker lock(m_Mutex);
  return m_workingThreads == 0;
}

inline void cbThreadPool::BatchBegin()
{
  wxMutexLocker lock(m_Mutex);
  m_batching = true;
}

inline void cbThreadPool::Broadcast()
{
  // if m_concurrentThreads == -1, which means the pool is not initialized yet
  if (m_concurrentThreads == -1)
    return;
  // let the idle(pending) worker thread to execute tasks, those worker threads are waiting for semaphore
  for (std::size_t i = 0; i < static_cast<std::size_t>(m_concurrentThreads - m_workingThreads); ++i)
    m_semaphore->Post();
}

inline void cbThreadPool::AwakeNeeded()
{
  // if m_concurrentThreads == -1, which means the pool is not initialized yet
  if (m_concurrentThreads == -1)
    return;

  // the thread number to awake should be less than the idle thread number and the tasks queue's size
  std::size_t awakeThreadNumber = std::min<std::size_t>(m_tasksQueue.size(),
                                                        (m_concurrentThreads - m_workingThreads));
  for (std::size_t i = 0; i < awakeThreadNumber; ++i)
    m_semaphore->Post();
}

#endif  //CBTHREADPOOL_H