src/stages/concentrator.hh

/* -*- mode: c++; c-basic-offset: 4; -*- */
#ifndef SENDS_TEMPLATE_STAGE_HH
#define SENDS_TEMPLATE_STAGE_HH

#include "seda_stage.hh"
#include <vector>
#include <gmutex.hh>

namespace sends {


    class concentrator : public seda_stage {
    public:
        concentrator(int nThread, int nTrans);

        virtual ~concentrator(void);

        void add_transaction(transaction* trans, double timeout);

        void filter(transaction* trans);

        void init_thread(void);

        const char* stage_name(void) const;

    private:
        void lock(void);
        void unlock(void);
        bool try_collect(void);
        void release_collect(void);

    private:
        struct hold {
            hold(transaction* t, const timespec& t);
            timespec     timeout;
            transaction* trans;
            int          socket;
            bool         collected;
        };

        typedef std::vector<hold> hold_vect;
        typedef hold_vect::iterator hold_iter;

    private:
        double        mTimeout;

        thread::mutex gate;
        volatile bool collector;
        hold_vect     trans_vect;
    };
}

#endif // !defined(SENDS_CONCENTRATOR_HH)

---