Loading...
Searching...
No Matches
ratelier_gather.hpp
Go to the documentation of this file.
/*********************************************************************/
// libthreadar - is a library providing several C++ classes to work with threads
// Copyright (C) 2014-2024 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_GATHER_HPP
#define LIBTHREADAR_RATELIER_GATHER_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_gather
{
public:
ratelier_gather(unsigned int size, signed int flag = 0);
ratelier_gather(const ratelier_gather & ref) = delete;
ratelier_gather(ratelier_gather && ref) = default;
ratelier_gather & operator = (const ratelier_gather & ref) = delete;
ratelier_gather & operator = (ratelier_gather && ref) noexcept = default;
virtual ~ratelier_gather() = default;
void worker_push_one(unsigned int slot, std::unique_ptr<T> & one, signed int flag = 0);
void gather(std::deque<std::unique_ptr<T> > & ones, std::deque<signed int> & flag);
void reset();
private:
static const unsigned int cond_pending_data = 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;
std::vector<slot> table;
std::map<unsigned int, unsigned int> corres;
std::deque<unsigned int> empty_slot;
libthreadar::condition verrou;
};
template <class T> ratelier_gather<T>::ratelier_gather(unsigned int size, signed int flag):
table(size, slot(flag)),
verrou(2)
{
next_index = 0;
for(unsigned int i = 0; i < size; ++i)
empty_slot.push_back(i);
}
template <class T> void ratelier_gather<T>::worker_push_one(unsigned int slot, std::unique_ptr<T> & one, signed int flag)
{
verrou.lock();
try
{
while(empty_slot.empty() // no free slot available
|| ((empty_slot.size() == 1 && slot != next_index) // one slot available and we do not provide the lowest expecting slot num
&& corres.begin() != corres.end() && (corres.begin())->first != next_index)) // and lowest slot is still not received
verrou.wait(cond_full);
std::map<unsigned int, unsigned int>::iterator it = corres.find(slot);
unsigned int index;
if(it != corres.end())
throw exception_range("the ratelier_gather index to fill is already used");
index = empty_slot.back();
// sanity checks
if(index >= table.size())
throw THREADAR_BUG;
if( ! table[index].empty)
throw THREADAR_BUG;
// recording the change
corres[slot] = index;
table[index].obj = std::move(one);
table[index].empty = false;
table[index].index = slot;
table[index].flag = flag;
empty_slot.pop_back();
if(verrou.get_waiting_thread_count(cond_pending_data) > 0)
if(corres.find(next_index) != corres.end()) // some data can be gathered
verrou.signal(cond_pending_data); // awaking the gathering thread
}
catch(...)
{
verrou.unlock(); // unlock first, as broadcast/signal may be the cause of the exception
verrou.broadcast(cond_pending_data);
verrou.broadcast(cond_full);
throw;
}
verrou.unlock();
}
template <class T> void ratelier_gather<T>::gather(std::deque<std::unique_ptr<T> > & ones, std::deque<signed int> & flag)
{
ones.clear();
flag.clear();
verrou.lock();
try
{
std::map<unsigned int, unsigned int>::iterator it;
std::map<unsigned int, unsigned int>::iterator tmp;
do
{
it = corres.begin();
while(it != corres.end())
{
if(it->first > next_index) // not continuous sequence
break; // exiting the inner while loop
if(it->first == next_index)
{
// sanity checks
if(it->second >= table.size())
throw THREADAR_BUG;
if(table[it->second].index != next_index)
throw THREADAR_BUG;
if(table[it->second].empty)
throw THREADAR_BUG;
if( ! table[it->second].obj)
throw THREADAR_BUG;
// recording the change
ones.push_back(std::move(table[it->second].obj));
flag.push_back(table[it->second].flag);
table[it->second].empty = true;
empty_slot.push_back(it->second);
tmp = it;
++it;
corres.erase(tmp);
++next_index;
}
else // integer overload occured for the index
++it; // skipping this entry
}
if(ones.empty())
verrou.wait(cond_pending_data);
}
while(ones.empty());
if(verrou.get_waiting_thread_count(cond_full) > 0)
verrou.broadcast(cond_full); // awake all pending workers
}
catch(...)
{
verrou.unlock(); // unlock first, as broadcast() may be the cause of the exception
verrou.broadcast(cond_pending_data);
verrou.broadcast(cond_full);
throw;
}
verrou.unlock();
if(ones.size() != flag.size())
throw THREADAR_BUG;
}
template <class T> void ratelier_gather<T>::reset()
{
unsigned int size = table.size();
next_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_pending_data);
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:46
libthreadar::exception_range
Exception used to report out or range value or argument.
Definition exceptions.hpp:207
libthreadar::ratelier_gather
the class ratelier_gather has a fixed length range of slots of arbitrary defined object type
Definition ratelier_gather.hpp:98
libthreadar::ratelier_gather::worker_push_one
void worker_push_one(unsigned int slot, std::unique_ptr< T > &one, signed int flag=0)
provides to a worker thread a mean to given data with its associated index to a gathering thread
Definition ratelier_gather.hpp:160
libthreadar::ratelier_gather::gather
void gather(std::deque< std::unique_ptr< T > > &ones, std::deque< signed int > &flag)
obtain the lowest continuous filled slots of the ratelier_gather and free them
Definition ratelier_gather.hpp:210
libthreadar::ratelier_gather::reset
void reset()
reset the object in its prestine state
Definition ratelier_gather.hpp:281
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