OutputPort.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/BufferPool.hpp>
#include <Pothos/Framework/BufferChunk.hpp>
#include <Pothos/Framework/BufferManager.hpp>
#include <Pothos/Util/RingDeque.hpp>
#include <Pothos/Util/SpinLock.hpp>
#include <string>

namespace Pothos {

class WorkerActor;
class InputPort;

class POTHOS_API OutputPort
{
public:

    ~OutputPort(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;

    void produce(const size_t numElements);

    void popElements(const size_t numElements);

    BufferChunk getBuffer(const size_t numElements);

    BufferChunk getBuffer(const DType &dtype, const size_t numElements);

    template <typename... ValueType>
    void postLabel(ValueType&&... label);

    template <typename ValueType>
    void postMessage(ValueType &&message);

    template <typename T, typename... Args>
    void emplaceMessage(Args&&... args);

    template <typename ValueType>
    void postBuffer(ValueType &&buffer);

    void setReserve(const size_t numElements);

    bool isSignal(void) const;

    void setReadBeforeWrite(InputPort *port);

private:
    WorkerActor *_actor;

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

    //state set in pre-work
    std::string _domain;
    BufferChunk _buffer;
    size_t _elements;

    //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;
    std::vector<Label> _postedLabels;
    Util::RingDeque<BufferChunk> _postedBuffers;

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

    Util::SpinLock _bufferManagerLock;
    BufferManager::Sptr _bufferManager;

    Util::SpinLock _tokenManagerLock;
    BufferManager::Sptr _tokenManager; //used for message backpressure

    void bufferManagerSetup(const BufferManager::Sptr &manager);
    bool bufferManagerEmpty(void);
    void bufferManagerFront(BufferChunk &);
    void bufferManagerPop(const size_t numBytes);
    void bufferManagerPush(Pothos::Util::SpinLock *mutex, const ManagedBuffer &buff);

    void tokenManagerInit(void);
    bool tokenManagerEmpty(void);
    BufferChunk tokenManagerPop(void);
    void tokenManagerPop(const size_t numBytes);

    std::vector<InputPort *> _subscribers;
    InputPort *_readBeforeWritePort;
    bool _bufferFromManager;
    BufferPool _bufferPool;

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

} //namespace Pothos