Loading...
Searching...
No Matches
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
ojph::thds::thread_pool::threads
std::vector< std::thread > threads
Definition ojph_threads.h:140
ojph::thds::thread_pool::stop
std::atomic_bool stop
Definition ojph_threads.h:144
ojph::thds::thread_pool::start_thread
static void start_thread(thread_pool *tp)
A static function to start a thread.
Definition ojph_threads.cpp:82
ojph::thds::thread_pool::mutex
std::mutex mutex
Definition ojph_threads.h:142
ojph::thds::thread_pool::condition
std::condition_variable condition
Definition ojph_threads.h:143
ojph::thds::thread_pool::~thread_pool
~thread_pool()
default destructor
Definition ojph_threads.cpp:54
ojph::thds::thread_pool::tasks
std::deque< worker_thread_base * > tasks
Definition ojph_threads.h:141
ojph::thds::thread_pool::add_task
void add_task(worker_thread_base *task)
Adds a task to the thread pool.
Definition ojph_threads.cpp:73
ojph::thds::thread_pool::init
void init(size_t num_threads)
Initializes the thread pool.
Definition ojph_threads.cpp:63
ojph::thds::thread_pool::thread_pool
thread_pool()
default constructor
Definition ojph_threads.h:103
ojph::thds::worker_thread_base
A base object for queuing tasks in the thread_pool.
Definition ojph_threads.h:69
ojph::thds::worker_thread_base::execute
virtual void execute()=0
Derived functions must define this function to execute its work.
ojph::thds
Definition ojph_threads.h:51
ojph
Definition ojph_img_io.h:52
ojph_threads.h