InputPort.hpp

Go to the documentation of this file.

#pragma once
#include <Pothos/Config.hpp>
#include <Pothos/Object/Object.hpp>
#include <Pothos/Framework/DType.hpp>
#include <Pothos/Framework/Label.hpp>
#include <Pothos/Framework/BufferChunk.hpp>
#include <Pothos/Framework/BufferAccumulator.hpp>
#include <Pothos/Util/RingDeque.hpp>
#include <Pothos/Util/SpinLock.hpp>
#include <string>

namespace Pothos {

class WorkerActor;
class OutputPort;

class POTHOS_API InputPort
{
public:

    ~InputPort(void);

    int index(void) const;

    const std::string &name(void) const;

    const std::string &alias(void) const;

    void setAlias(const std::string &alias);

    const DType &dtype(void) const;

    const std::string &domain(void) const;

    const BufferChunk &buffer(void) const;

    size_t elements(void) const;

    unsigned long long totalElements(void) const;

    unsigned long long totalBuffers(void) const;

    unsigned long long totalLabels(void) const;

    unsigned long long totalMessages(void) const;

    bool hasMessage(void);

    const LabelIteratorRange &labels(void) const;

    void removeLabel(const Label &label);

    void consume(const size_t numElements);

    BufferChunk takeBuffer(void);

    Object popMessage(void);

    Object peekMessage(void);

    void setReserve(const size_t numElements);

    bool isSlot(void) const;

    void pushBuffer(const BufferChunk &buffer);

    void pushLabel(const Label &label);

    void pushMessage(const Object &message);

    void clear(void);

private:
    WorkerActor *_actor;

    //port configuration
    bool _isSlot;
    int _index;
    std::string _name;
    std::string _alias;
    DType _dtype;
    std::string _domain;

    //state set in pre-work
    BufferChunk _buffer;
    size_t _elements;
    LabelIteratorRange _labelIter;

    //port stats
    unsigned long long _totalElements;
    unsigned long long _totalBuffers;
    unsigned long long _totalLabels;
    unsigned long long _totalMessages;

    //state changes from work
    size_t _pendingElements;
    size_t _reserveElements;

    //counts work actions which we will use to establish activity
    size_t _workEvents;

    Util::SpinLock _asyncMessagesLock;
    Util::RingDeque<std::pair<Object, BufferChunk>> _asyncMessages;

    Util::SpinLock _slotCallsLock;
    Util::RingDeque<std::pair<Object, BufferChunk>> _slotCalls;

    std::vector<Label> _inlineMessages; //user api structure
    Util::RingDeque<Label> _inputInlineMessages; //shared structure

    Util::SpinLock _bufferAccumulatorLock;
    BufferAccumulator _bufferAccumulator;

    std::vector<OutputPort *> _subscribers;

    void asyncMessagesPush(const Object &message, const BufferChunk &token = BufferChunk::null());
    bool asyncMessagesEmpty(void);
    Object asyncMessagesPop(void);
    Object asyncMessagesPeek(void);
    void asyncMessagesClear(void);

    void slotCallsPush(const Object &args, const BufferChunk &token);
    bool slotCallsEmpty(void);
    Object slotCallsPop(void);
    void slotCallsClear(void);

    void inlineMessagesPush(const Label &label);
    void inlineMessagesClear(void);

    void bufferAccumulatorFront(BufferChunk &);
    void bufferAccumulatorPush(const BufferChunk &buffer);
    void bufferAccumulatorPushNoLock(BufferChunk &&buffer);
    void bufferAccumulatorPop(const size_t numBytes);
    void bufferAccumulatorRequire(const size_t numBytes);
    void bufferAccumulatorClear(void);

    void bufferLabelPush(
        const bool enableMove,
        std::vector<Label> &postedLabels,
        Util::RingDeque<BufferChunk> &postedBuffers);

    InputPort(void);
    InputPort(const InputPort &) = delete; // non construction-copyable
    InputPort &operator=(const InputPort &) = delete; // non copyable
    friend class WorkerActor;
    friend class OutputPort;
};

} //namespace Pothos