Loading...
Searching...
No Matches
ratelier_scatter.hpp
Go to the documentation of this file.
/*********************************************************************/
// libthreadar - is a library providing several C++ classes to work with threads
// Copyright (C) 2014-2025 Denis Corbin
//
// This file is part of libthreadar
//
// libthreadar 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 3 of the License, or
// (at your option) any later version.
//
// libhtreadar 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 libthreadar. If not, see <http://www.gnu.org/licenses/>
//
//----
// to contact the author: dar.linux@free.fr
/*********************************************************************/
#ifndef LIBTHREADAR_RATELIER_SCATTER_HPP
#define LIBTHREADAR_RATELIER_SCATTER_HPP
#include "config.h"
// C system headers
extern "C"
{
}
// C++ standard headers
#include <vector>
#include <map>
#include <deque>
#include <memory>
// libthreadar headers
#include "mutex.hpp"
namespace libthreadar
{
template <class T> class ratelier_scatter
{
public:
ratelier_scatter(unsigned int size, signed int flag = 0);
ratelier_scatter(const ratelier_scatter & ref) = delete;
ratelier_scatter(ratelier_scatter && ref) = default;
ratelier_scatter & operator = (const ratelier_scatter & ref) = delete;
ratelier_scatter & operator = (ratelier_scatter && ref) noexcept = default;
virtual ~ratelier_scatter() = default;
void scatter(std::unique_ptr<T> & one, signed int flag = 0);
std::unique_ptr<T> worker_get_one(unsigned int & slot, signed int & flag);
void reset();
private:
static const unsigned int cond_empty = 0;
static const unsigned int cond_full = 1;
struct slot
{
std::unique_ptr<T> obj;
bool empty;
unsigned int index;
signed int flag;
slot(signed int val) { empty = true; flag = val; };
slot(const slot & ref) { obj.reset(); empty = ref.empty; index = ref.index; flag = ref.flag; };
};
unsigned int next_index;
unsigned int lowest_index;
std::vector<slot> table;
std::map<unsigned int, unsigned int> corres;
std::deque<unsigned int> empty_slot;
libthreadar::condition verrou;
};
template <class T> ratelier_scatter<T>::ratelier_scatter(unsigned int size, signed int flag):
table(size, slot(flag)),
verrou(2)
{
next_index = 0;
lowest_index = 0;
for(unsigned int i = 0; i < size; ++i)
empty_slot.push_back(i);
}
template <class T> void ratelier_scatter<T>::scatter(std::unique_ptr<T> & one, signed int flag)
{
unsigned int tableindex;
verrou.lock();
try
{
while(empty_slot.empty()) // ratelier_scatter is full
verrou.wait(cond_full);
tableindex = empty_slot.back();
// sanity checks
if(tableindex >= table.size())
throw THREADAR_BUG;
if( ! table[tableindex].empty)
throw THREADAR_BUG;
// recording the change
table[tableindex].empty = false;
table[tableindex].obj = std::move(one);
table[tableindex].index = next_index;
table[tableindex].flag = flag;
corres[next_index] = tableindex;
++next_index;
empty_slot.pop_back();
if(verrou.get_waiting_thread_count(cond_empty) > 0)
verrou.signal(cond_empty);
}
catch(...)
{
verrou.unlock();
verrou.broadcast(cond_empty);
verrou.broadcast(cond_full);
throw;
}
verrou.unlock();
}
template <class T> std::unique_ptr<T> ratelier_scatter<T>::worker_get_one(unsigned int & slot, signed int & flag)
{
std::unique_ptr<T> ret;
verrou.lock();
try
{
std::map<unsigned int, unsigned int>::iterator it = corres.begin();
// using sequential reading provides sorted scanning
// of the map, looking first for the lowest index available (oldest entries)
do
{
if(it != corres.end())
{
if(it->first < lowest_index) // overflooding occured
++it; // ignoring this slot
else
{
// sanity checks
if(it->second >= table.size())
throw THREADAR_BUG;
if(table[it->second].empty)
throw THREADAR_BUG;
if( ! table[it->second].obj)
throw THREADAR_BUG;
// recording the change
ret = std::move(table[it->second].obj);
slot = table[it->second].index;
flag = table[it->second].flag;
table[it->second].empty = true;
if(lowest_index != slot)
throw THREADAR_BUG;
++lowest_index;
// reusing quicker the last block used
// as the back() be used first
empty_slot.push_back(it->second);
corres.erase(it); // removing the correspondance
if(verrou.get_waiting_thread_count(cond_full) > 0)
verrou.signal(cond_full);
}
}
else
{
// ratelier_scatter is empty
verrou.wait(cond_empty);
it = corres.begin();
}
}
while( ! ret);
}
catch(...)
{
verrou.unlock();
verrou.broadcast(cond_empty);
verrou.broadcast(cond_full);
throw;
}
verrou.unlock();
return ret;
}
template <class T> void ratelier_scatter<T>::reset()
{
unsigned int size = table.size();
next_index = 0;
lowest_index = 0;
corres.clear();
empty_slot.clear();
for(unsigned int i = 0; i < size; ++i)
{
table[i].obj.reset();
table[i].empty = true;
empty_slot.push_back(i);
}
verrou.lock();
verrou.broadcast(cond_empty);
verrou.broadcast(cond_full);
verrou.unlock();
}
} // end of namespace
#endif
libthreadar::condition
Wrapper around the Posix pthread_cond_t object and its associated mutex.
Definition condition.hpp:50
libthreadar::ratelier_scatter
The class ratelier_scatter's purpose it to scatter an ordered set of data to many worker threads.
Definition ratelier_scatter.hpp:65
libthreadar::ratelier_scatter::worker_get_one
std::unique_ptr< T > worker_get_one(unsigned int &slot, signed int &flag)
For a worker thread to obtain an object in the lowest slot available.
Definition ratelier_scatter.hpp:179
libthreadar::ratelier_scatter::scatter
void scatter(std::unique_ptr< T > &one, signed int flag=0)
For the non-worker thread to provide data to the ratelier_scatter.
Definition ratelier_scatter.hpp:136
libthreadar::ratelier_scatter::reset
void reset()
reset the object in its prestine state
Definition ratelier_scatter.hpp:250
THREADAR_BUG
#define THREADAR_BUG
Macro used to throw an exception_bug when execution reach that statement.
Definition exceptions.hpp:164
mutex.hpp
defines the mutex C++ class
libthreadar
This is the only namespace used in libthreadar and all symbols provided by libthreadar are member of ...
Definition barrier.hpp:46