tampon.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_TAMPON_HPP
#define LIBTHREADAR_TAMPON_HPP
 
 
#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