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