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-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_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