fast_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_FAST_TAMPON_H
#define LIBTHREADAR_FAST_TAMPON_H
 
 
#include "config.h"
 
    // C system headers
extern "C"
{
 
}
    // C++ standard headers
 
 
    // libthreadar headers
#include "condition.hpp"
#include "exceptions.hpp"
 
namespace libthreadar
{
 
 
 
 
    template <class T> class fast_tampon
    {
    public:
 
        fast_tampon(unsigned int max_block, unsigned int block_size);
 
        fast_tampon(const fast_tampon & ref) = delete;
 
        fast_tampon(fast_tampon && ref) = default;
 
        fast_tampon & operator = (const fast_tampon & ref) = delete;
 
        fast_tampon & operator = (fast_tampon && ref) noexcept = default;
 
        ~fast_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);
 
        bool is_empty() const { return next_feed == next_fetch; };
 
        bool is_not_empty() const { return !is_empty(); };
 
        bool is_full() const { unsigned int tmp = next_feed; shift_by_one(tmp); return tmp == next_fetch; };
 
        bool is_not_full() const { return !is_full(); };
 
        unsigned int size() const { return table_size; };
 
        unsigned int block_size() const { return alloc_size; };
 
        void reset();
 
    private:
 
        struct atom
        {
            T* mem;
            unsigned int data_size;
 
            atom() { mem = nullptr; data_size = 0; };
        };
 
        static const unsigned int cond_full = 0;
        static const unsigned int cond_empty = 0;
 
        condition modif;
        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 the table
        unsigned int next_fetch;  //< index in table of the next atom to fetch from table
        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
 
        void shift_by_one(unsigned int & x) const;
 
    };
 
    template <class T> fast_tampon<T>::fast_tampon(unsigned int max_block, unsigned int block_size): modif(2)
    {
        if(max_block < 2)
            throw exception_range("max_block for fast_tampon should be strictly greater than 1");
        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> fast_tampon<T>::~fast_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 fast_tampon<T>::get_block_to_feed(T * & ptr, unsigned int & num)
    {
        if(feed_outside)
            throw exception_range("feed already out!");
 
        if(is_full())
        {
            modif.lock();  // --- critical section START
            try
            {
                if(is_full())
                {
                    modif.wait(cond_full);
 
                    if(is_full())
                        throw THREADAR_BUG; // still full!
                }
                    // *else*
                    // full condition was transitional
                    // only the feeder (this is us) can make it happen again
                    // so the full condition should not occur before we return
                    // the block we are about to fetch
            }
            catch(...)
            {
                modif.unlock();
                throw;
            }
            modif.unlock(); // --- critical section END
        }
 
        feed_outside = true;
        ptr = table[next_feed].mem;
        num = alloc_size;
    }
 
    template <class T> void fast_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
        try
        {
            shift_by_one(next_feed);
            if(modif.get_waiting_thread_count(cond_empty) > 0)
                modif.signal(cond_empty);  // releasing the fetcher when unlock() will complete
        }
        catch(...)
        {
            modif.unlock();
            throw;
        }
        modif.unlock(); // --- critical section END
    }
 
    template <class T> void fast_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 fast_tampon<T>::fetch(T* & ptr, unsigned int & num)
    {
        if(fetch_outside)
            throw exception_range("already fetched block outside");
 
        if(is_empty())
        {
            modif.lock();   // --- critical section START
            try
            {
                if(is_empty())
                {
                    modif.wait(cond_empty);
 
                    if(is_empty())
                        throw THREADAR_BUG;
                }
                    // *else*
                    // the emptiness condition was transitional
                    // only the fetcher (this is us) can make it happen again
                    // so the empty condition should not occur before we return
                    // the block we are about to fetch
            }
            catch(...)
            {
                modif.unlock();
                throw;
            }
            modif.unlock(); // --- critical section END
        }
 
        fetch_outside = true;
        ptr = table[next_fetch].mem;
        num = table[next_fetch].data_size;
    }
 
    template <class T> void fast_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
        try
        {
            shift_by_one(next_fetch);
            if(modif.get_waiting_thread_count(cond_full) > 0)
                modif.signal(cond_full);  // releasing the fetcher when unlock() will complete
        }
        catch(...)
        {
            modif.unlock();
            throw;
        }
        modif.unlock(); // --- critical section END
    }
 
    template <class T> void fast_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 fast_tampon<T>::reset()
    {
        modif.lock(); // --- critical section START
        try
        {
            if(modif.get_waiting_thread_count(cond_full) > 0
               || modif.get_waiting_thread_count(cond_empty) > 0)
            {
                modif.broadcast(cond_full);
                modif.broadcast(cond_empty);
                throw exception_range("reseting fast_tampon while some thread were waiting on it");
            }
 
            next_feed = 0;
            next_fetch = 0;
            fetch_outside = false;
            feed_outside = false;
        }
        catch(...)
        {
            modif.unlock();
            throw;
        }
        modif.unlock(); // --- critical section END
    }
 
    template <class T> void fast_tampon<T>::shift_by_one(unsigned int & x) const
    {
        ++x;
        if(x >= table_size)
            x = 0;
    }
 
 
} // end of namespace
 
#endif