AsyncNotifiableBlock.hxx

Go to the documentation of this file.

 
#ifndef _EXECUTOR_ASYNCNOTIFIABLEBLOCK_HXX_
#define _EXECUTOR_ASYNCNOTIFIABLEBLOCK_HXX_
 
#include <memory>
 
#include "executor/Notifiable.hxx"
#include "utils/Queue.hxx"
#include "utils/logging.h"
 
#include "utils/Buffer.hxx"
 
class AsyncNotifiableBlock : private Notifiable, public QAsync
{
private:
    class QueuedBarrier : public BarrierNotifiable, public QMember
    {
    public:
        void notify() override
        {
            AtomicHolder h(this);
            if (count_ == 1)
            {
                LOG(VERBOSE, "block notifiable %p returned pool size %u",
                    (BarrierNotifiable *)this,
                    (unsigned)mainBufferPool->total_size());
                auto *tgt = static_cast<AsyncNotifiableBlock *>(done_);
                tgt->insert(this);
            }
            else
            {
                --count_;
            }
        }
 
        void check_one_count()
        {
            HASSERT(count_ == 1);
        }
    };
 
public:
    AsyncNotifiableBlock(unsigned num_parallelism)
        : count_(num_parallelism)
        , barriers_(new QueuedBarrier[num_parallelism])
    {
        for (unsigned i = 0; i < num_parallelism; ++i)
        {
            barriers_[i].reset(this);
            this->insert(&barriers_[i]);
        }
    }
 
    ~AsyncNotifiableBlock();
 
    BarrierNotifiable *initialize(QMember *entry)
    {
        QueuedBarrier *b = static_cast<QueuedBarrier *>(entry);
        // We must be owning this entry.
        HASSERT(barriers_.get() <= b);
        HASSERT(b <= (barriers_.get() + count_));
        b->check_one_count();
        return b;
    }
 
    void notify() override
    {
        DIE("Should not receive this notification");
    }
 
private:
    unsigned count_;
    std::unique_ptr<QueuedBarrier[]> barriers_;
 
    DISALLOW_COPY_AND_ASSIGN(AsyncNotifiableBlock);
};
 
#endif // _EXECUTOR_ASYNCNOTIFIABLEBLOCK_HXX_