tampon.hpp
Go to the documentation of this file.
/*********************************************************************/
// libthreadar - is a library providing several C++ classes to work with threads
// Copyright (C) 2014-2020 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_TAMPON_H
#define LIBTHREADAR_TAMPON_H
#include "config.h"
// C system headers
extern "C"
{
}
// C++ standard headers
// libthreadar headers
#include "mutex.hpp"
#include "exceptions.hpp"
namespace libthreadar
{
template <class T> class tampon
{
public:
tampon(unsigned int max_block, unsigned int block_size);
tampon(const tampon & ref) = delete;
tampon(tampon && ref) noexcept = default;
tampon & operator = (const tampon & ref) = delete;
tampon & operator = (tampon && ref) noexcept = default;
~tampon();
void get_block_to_feed(T * & ptr, unsigned int & num);
void feed(T* ptr, unsigned int written);
void feed_cancel_get_block(T *ptr);
void fetch(T* & ptr, unsigned int & num);
void fetch_recycle(T* ptr);
void fetch_push_back(T *ptr, unsigned int new_num);
void fetch_push_back_and_skip(T *ptr, unsigned int new_num);
void fetch_skip_back();
bool has_readable_block_next() const;
bool is_empty() const;
bool is_not_empty() const { return !is_empty(); };
bool is_full() const { return full; }; // no need to acquire mutex "modif"
bool is_not_full() const { return !is_full(); };
bool is_quiet_full() const { unsigned int tmp = next_feed; shift_by_one(tmp); return tmp == fetch_head; };
unsigned int size() const { return table_size; };
unsigned int block_size() const { return alloc_size; };
unsigned int load() const { return fetch_head <= next_feed ? next_feed - fetch_head : table_size - (fetch_head - next_feed); };
void reset();
private:
struct atom
{
T* mem;
unsigned int data_size;
atom() { mem = nullptr; data_size = 0; };
};
mutex modif; //< to make critical section when non atomic action requires a status has not changed between a test and following action
atom *table; //< datastructure holding data in transit between two threads
unsigned int table_size; //< size of table, i.e. number of struct atom it holds
unsigned int alloc_size; //< size of allocated memory for each atom in table
unsigned int next_feed; //< index in table of the next atom to use for feeding table
unsigned int next_fetch; //< index in table of the next atom to use for fetch table
unsigned int fetch_head; //< the oldest object to be fetched
bool fetch_outside; //< if set to true, table's index pointed to by next_fetch is used by the fetcher
bool feed_outside; //< if set to true, table's index pointed to by next_feed is used by the feeder
mutex waiting_feeder; //< feeder thread may be stuck waiting on that semaphore if table is full
mutex waiting_fetcher; //< fetcher thread may be stuck waiting on that semaphore if table is empty
bool full; //< set when tampon is full
bool feeder_go_lock; //< true to inform fetcher than feeder is about to or has already acquire lock on waiting_feeder
bool feeder_lock_track; //< only used by feeder to lock on waiting_feeder once outside of critical section
bool fetcher_go_lock; //< true to inform feeder than fetcher is about to or has already acquire lock on waiting_fetcher
bool fetcher_lock_track; //< only used by fetcher to lock on waiting_fetcher once outside of critical section
bool is_empty_no_lock() const { return next_feed == fetch_head && !full; };
bool has_readable_block_next_no_lock() const { return next_feed != next_fetch || full; }
void shift_by_one(unsigned int & x) const;
void shift_back_by_one(unsigned int & x) const;
void shift_by_one_data_in_range(unsigned int begin, unsigned int end);
};
template <class T> tampon<T>::tampon(unsigned int max_block, unsigned int block_size)
{
table_size = max_block;
table = new atom[table_size];
if(table == nullptr)
throw exception_memory();
try
{
alloc_size = block_size;
try
{
for(unsigned int i = 0 ; i < table_size ; ++i)
{
table[i].mem = new T[alloc_size];
if(table[i].mem == nullptr)
throw exception_memory();
table[i].data_size = 0;
}
reset();
}
catch(...)
{
for(unsigned int i = 0; i < table_size ; ++i)
{
if(table[i].mem != nullptr)
delete [] table[i].mem;
}
throw;
}
}
catch(...)
{
if(table != nullptr)
delete [] table;
throw;
}
}
template <class T> tampon<T>::~tampon()
{
if(table != nullptr)
{
for(unsigned int i = 0 ; i < table_size ; ++i)
{
if(table[i].mem != nullptr)
delete [] table[i].mem;
}
delete [] table;
}
}
template <class T> void tampon<T>::get_block_to_feed(T * & ptr, unsigned int & num)
{
if(feed_outside)
throw exception_range("feed already out!");
modif.lock(); // --- critical section START
if(is_full())
{
feeder_go_lock = true; // inform fetcher that we will suspend in waiting_feeder
feeder_lock_track = true;// to suspend on waiting_feeder once we will be out of the critical section
}
modif.unlock(); // --- critical section END
if(feeder_lock_track)
{
feeder_lock_track = false;
waiting_feeder.lock(); // cannot lock inside a critical section ...
}
if(is_full())
throw THREADAR_BUG; // still full!
feed_outside = true;
ptr = table[next_feed].mem;
num = alloc_size;
}
template <class T> void tampon<T>::feed(T *ptr, unsigned int num)
{
if(!feed_outside)
throw exception_range("fetch not outside!");
feed_outside = false;
if(ptr != table[next_feed].mem)
throw exception_range("returned ptr is not the one given earlier for feeding");
table[next_feed].data_size = num;
modif.lock(); // --- critical section START
shift_by_one(next_feed);
if(next_feed == fetch_head)
full = true;
if(fetcher_go_lock)
{
fetcher_go_lock = false;
waiting_fetcher.unlock();
}
modif.unlock(); // --- critical section END
}
template <class T> void tampon<T>::feed_cancel_get_block(T *ptr)
{
if(!feed_outside)
throw exception_range("feed not outside!");
feed_outside = false;
if(ptr != table[next_feed].mem)
throw exception_range("returned ptr is not the one given earlier for feeding");
}
template <class T> void tampon<T>::fetch(T* & ptr, unsigned int & num)
{
if(fetch_outside)
throw exception_range("already fetched block outside");
modif.lock(); // --- critical section START
if(!has_readable_block_next_no_lock())
{
fetcher_go_lock = true; // to inform feeder that we will suspend on waiting_fetcher
fetcher_lock_track = true; // to suspend on waiting_fetcher once we will be out of the critical section
}
modif.unlock(); // --- critical section END
if(fetcher_lock_track)
{
fetcher_lock_track = false;
waiting_fetcher.lock(); // cannot lock inside a critical section ...
}
if(is_empty())
throw THREADAR_BUG;
fetch_outside = true;
ptr = table[next_fetch].mem;
num = table[next_fetch].data_size;
}
template <class T> void tampon<T>::fetch_recycle(T* ptr)
{
if(!fetch_outside)
throw exception_range("no block outside for fetching");
fetch_outside = false;
if(ptr != table[next_fetch].mem)
throw exception_range("returned ptr is no the one given earlier for fetching");
modif.lock(); // --- critical section START
if(next_fetch == fetch_head)
{
// no block were skipped
shift_by_one(fetch_head);
next_fetch = fetch_head;
full = false;
}
else
{
unsigned int tmp = next_fetch;
atom tmp_tom;
shift_by_one(tmp);
shift_by_one_data_in_range(tmp, next_feed);
// we also take into account the situation
// where a block has been given for feeding
// so the next call to feed() will match the
// expected address of the returned block
tmp = next_feed; // recording old position of next_feed
shift_back_by_one(next_feed);
// swapping contents between old next_feed position
// and new one:
tmp_tom = table[next_feed];
table[next_feed] = table[tmp];
table[tmp] = tmp_tom;
// done!
full = false;
}
if(feeder_go_lock)
{
feeder_go_lock = false;
waiting_feeder.unlock();
}
modif.unlock(); // --- critical section END
}
template <class T> void tampon<T>::fetch_push_back(T* ptr, unsigned int new_num)
{
if(!fetch_outside)
throw exception_range("no block outside for fetching");
fetch_outside = false;
if(ptr != table[next_fetch].mem)
throw exception_range("returned ptr is not the one given earlier for fetching");
table[next_fetch].data_size = new_num;
}
template <class T> void tampon<T>::fetch_push_back_and_skip(T *ptr,
unsigned int new_num)
{
fetch_push_back(ptr, new_num);
modif.lock(); // --- critical section START
if(full && next_fetch == next_feed) // reach last block feed, cannot skip it
throw exception_range("cannot skip the last fed block when the tampon is full");
shift_by_one(next_fetch);
modif.unlock(); // --- critical section END
}
template <class T> void tampon<T>::fetch_skip_back()
{
if(fetch_outside)
throw exception_range("cannot skip back fetching while a block is being fetched");
next_fetch = fetch_head;
}
template <class T> bool tampon<T>::has_readable_block_next() const
{
bool ret;
tampon<T> *me = const_cast<tampon<T> *>(this);
if(me == nullptr)
throw THREADAR_BUG;
me->modif.lock();
ret = has_readable_block_next_no_lock();
me->modif.unlock();
return ret;
}
template <class T> bool tampon<T>::is_empty() const
{
bool ret;
tampon<T> * me = const_cast<tampon<T> *>(this);
if(me == nullptr)
throw THREADAR_BUG;
me->modif.lock();
ret = is_empty_no_lock();
me->modif.unlock();
return ret;
}
template <class T> void tampon<T>::reset()
{
next_feed = 0;
next_fetch = 0;
fetch_head = 0;
fetch_outside = false;
feed_outside = false;
full = false;
feeder_go_lock = false;
feeder_lock_track = false;
fetcher_go_lock = false;
fetcher_lock_track = false;
(void)waiting_feeder.try_lock();
(void)waiting_fetcher.try_lock();
}
template <class T> void tampon<T>::shift_by_one(unsigned int & x) const
{
++x;
if(x >= table_size)
x = 0;
}
template <class T> void tampon<T>::shift_back_by_one(unsigned int & x) const
{
if(x == 0)
x = table_size - 1;
else
--x;
}
template <class T> void tampon<T>::shift_by_one_data_in_range(unsigned int begin, unsigned int end)
{
if(begin != end)
{
unsigned int prev = begin;
shift_back_by_one(prev);
T* not_squeezed = table[prev].mem; // we will erase the address pointed to by mem so we keep it in memory here
while(begin != end)
{
table[prev] = table[begin]; // this copies both mem (the value of the pointer, not the pointed to) and data_size,
prev = begin;
shift_by_one(begin);
}
table[prev].mem = not_squeezed;
table[prev].data_size = 0; // by precaution
}
}
} // end of namespace
#endif
libthreadar::exception_memory
Exception used to report memory allocation failures.
Definition: exceptions.hpp:152
libthreadar::exception_range
Exception used to report out or range value or argument.
Definition: exceptions.hpp:207
libthreadar::mutex::unlock
void unlock()
unlock the mutex
libthreadar::mutex::lock
void lock()
lock the mutex
libthreadar::tampon
DEPRECATED see fast_tampon instead!
Definition: tampon.hpp:102
libthreadar::tampon::size
unsigned int size() const
returns the size of the tampon in maximum number of block it can contain
Definition: tampon.hpp:205
libthreadar::tampon::is_full
bool is_full() const
for feeder to know whether the next call to get_block_to_feed() will be blocking
Definition: tampon.hpp:194
libthreadar::tampon::fetch_recycle
void fetch_recycle(T *ptr)
fetcher call - step 2
Definition: tampon.hpp:402
libthreadar::tampon::is_not_full
bool is_not_full() const
to know whether the tampon is not full
Definition: tampon.hpp:197
libthreadar::tampon::load
unsigned int load() const
returns the current number of blocks currently used in the tampon (fed but not fetched)
Definition: tampon.hpp:213
libthreadar::tampon::operator=
tampon & operator=(const tampon &ref)=delete
no assignment operator
libthreadar::tampon::tampon
tampon(const tampon &ref)=delete
no copy constructor
libthreadar::tampon::block_size
unsigned int block_size() const
returns the allocation size of each block
Definition: tampon.hpp:210
libthreadar::tampon::feed_cancel_get_block
void feed_cancel_get_block(T *ptr)
feeder call - step 2 alternative
Definition: tampon.hpp:366
libthreadar::tampon::has_readable_block_next
bool has_readable_block_next() const
to known whether next fetch will be blocking (not skipped blocks)
Definition: tampon.hpp:482
libthreadar::tampon::tampon
tampon(unsigned int max_block, unsigned int block_size)
constructor
Definition: tampon.hpp:264
libthreadar::tampon::is_quiet_full
bool is_quiet_full() const
returns true if only one slot is available before filling the tampon
Definition: tampon.hpp:200
libthreadar::tampon::get_block_to_feed
void get_block_to_feed(T *&ptr, unsigned int &num)
feeder call - step 1
Definition: tampon.hpp:317
libthreadar::tampon::is_not_empty
bool is_not_empty() const
to know whether the tampon is not empty
Definition: tampon.hpp:191
libthreadar::tampon::~tampon
~tampon()
Definition: tampon.hpp:304
libthreadar::tampon::tampon
tampon(tampon &&ref) noexcept=default
no move constructor
libthreadar::tampon::fetch
void fetch(T *&ptr, unsigned int &num)
fetcher call - step 1
Definition: tampon.hpp:375
libthreadar::tampon::feed
void feed(T *ptr, unsigned int written)
feeder call - step 2
Definition: tampon.hpp:344
libthreadar::tampon::fetch_push_back_and_skip
void fetch_push_back_and_skip(T *ptr, unsigned int new_num)
put back the fetched block and skip to next block for the next fetch()
Definition: tampon.hpp:463
libthreadar::tampon::is_empty
bool is_empty() const
to know whether the tampon has objects (readable or skipped)
Definition: tampon.hpp:497
libthreadar::tampon::fetch_push_back
void fetch_push_back(T *ptr, unsigned int new_num)
fetcher call - step 2 alternative
Definition: tampon.hpp:452
libthreadar::tampon::fetch_skip_back
void fetch_skip_back()
reactivate all skipped blocks, next fetch() will be the oldest available block
Definition: tampon.hpp:474
libthreadar::tampon::reset
void reset()
reset the object fields and mutex as if the object was just created
Definition: tampon.hpp:511
exceptions.hpp
defines a set of exceptions that are used by libthreadar to report error situations
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