ojph_threads.cpp

Go to the documentation of this file.

//***************************************************************************/
// This software is released under the 2-Clause BSD license, included
// below.
//
// Copyright (c) 2024, Aous Naman
// Copyright (c) 2024, Kakadu Software Pty Ltd, Australia
// Copyright (c) 2024, The University of New South Wales, Australia
//
// Redistribution and use in source and binary forms, with or without
// modification, are permitted provided that the following conditions are
// met:
//
// 1. Redistributions of source code must retain the above copyright
// notice, this list of conditions and the following disclaimer.
//
// 2. Redistributions in binary form must reproduce the above copyright
// notice, this list of conditions and the following disclaimer in the
// documentation and/or other materials provided with the distribution.
//
// THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS
// IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED
// TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A
// PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT
// HOLDER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
// SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED
// TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR
// PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF
// LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING
// NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF THIS
// SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
//***************************************************************************/
// This file is part of the OpenJPH software implementation.
// File: ojph_threads.h
// Author: Aous Naman
// Date: 22 April 2024
//***************************************************************************/
 
#include "ojph_threads.h"
 
namespace ojph
{
namespace thds
{

//
//
//
//
//

thread_pool::~thread_pool()
{
  stop.store(true, std::memory_order_release);
  condition.notify_all();
  for (size_t i = 0; i < threads.size(); ++i)
    threads[i].join();
}

void thread_pool::init(size_t num_threads)
{
  if (threads.size() < num_threads)
    threads.resize(num_threads);
 
  for (size_t i = 0; i < num_threads; ++i) 
    threads[i] = std::thread(start_thread, this);
}

void thread_pool::add_task(worker_thread_base* task)
{
  mutex.lock();
  tasks.push_back(task);
  condition.notify_one();
  mutex.unlock();
}

void thread_pool::start_thread(thread_pool* tp)
{
  while (1)
  {
    // setup the condition variable
    std::unique_lock<std::mutex> lock(tp->mutex);
    // wait releases the mutex, blocks until notified (or spuriously), 
    // and acquire the mutex
    tp->condition.wait(lock);
  
    if(tp->stop.load(std::memory_order_acquire))
      return;
    
    worker_thread_base* task = NULL;
    if (!tp->tasks.empty())
    {
      task = tp->tasks.front();
      tp->tasks.pop_front();
    }
    lock.unlock();
    if (task)
      task->execute();
  }
}
 
} // !thds namespace 
} // !ojph namespace