peak_data_stream.hpp

Go to the documentation of this file.

#pragma once


#include <peak/backend/peak_backend.h>
#include <peak/buffer/peak_buffer.hpp>
#include <peak/common/peak_common_structs.hpp>
#include <peak/common/peak_event_supporting_module.hpp>
#include <peak/dll_interface/peak_dll_interface_util.hpp>

#include <unordered_map>
#include <algorithm>
#include <cstddef>
#include <cstdint>
#include <functional>
#include <iterator>
#include <memory>
#include <mutex>
#include <string>
#include <vector>


namespace peak
{
namespace core
{

enum class DataStreamFlushMode
{
    InputPoolToOutputQueue,
    DiscardOutputQueue,
    AllToInputPool,
    UnqueuedToInputPool,
    DiscardAll,

    Custom = 1000
};

enum class AcquisitionStartMode
{
    Default,

    Custom = 1000
};

enum class AcquisitionStopMode
{
    Default,
    Kill,

    Custom = 1000
};


class Device;

class DataStream
    : public EventSupportingModule
    , public std::enable_shared_from_this<DataStream>
{
public:
    using BufferRevocationCallback = std::function<void(void* buffer, void* userPtr)>;
    static const uint64_t INFINITE_NUMBER = PEAK_INFINITE_NUMBER;

    DataStream() = delete;
    ~DataStream() override;
    DataStream(const DataStream& other) = delete;
    DataStream& operator=(const DataStream& other) = delete;
    DataStream(DataStream&& other) = delete;
    DataStream& operator=(DataStream&& other) = delete;

    std::string Key() const;

    RawInformation Info(int32_t infoCommand) const;
    std::string ID() const;
    std::string TLType() const;
    size_t NumBuffersAnnouncedMinRequired() const;
    size_t NumBuffersAnnounced() const;
    size_t NumBuffersQueued() const;
    size_t NumBuffersAwaitDelivery() const;
    uint64_t NumBuffersDelivered() const;
    uint64_t NumBuffersStarted() const;
    uint64_t NumUnderruns() const;
    size_t NumChunksPerBufferMax() const;
    size_t BufferAlignment() const;
    size_t PayloadSize() const;
    bool DefinesPayloadSize() const;
    bool IsGrabbing() const;


    std::shared_ptr<Buffer> AnnounceBuffer(
        void* buffer, size_t size, void* userPtr, const BufferRevocationCallback& revocationCallback);
    std::shared_ptr<Buffer> AllocAndAnnounceBuffer(size_t size, void* userPtr);
    void QueueBuffer(const std::shared_ptr<Buffer>& buffer);
    void RevokeBuffer(const std::shared_ptr<Buffer>& buffer);
    void Flush(DataStreamFlushMode mode);
    std::vector<std::shared_ptr<Buffer>> AnnouncedBuffers() const;


    void StartAcquisition(
        AcquisitionStartMode mode = AcquisitionStartMode::Default, uint64_t numToAcquire = INFINITE_NUMBER);
    void StopAcquisition(AcquisitionStopMode mode = AcquisitionStopMode::Default);


    std::shared_ptr<Buffer> WaitForFinishedBuffer(Timeout timeout_ms);
    void KillWait();

    std::shared_ptr<Device> ParentDevice() const;

private:
    static void PEAK_CALL_CONV BufferRevocationCallbackCWrapper(void* buffer, void* userPtr, void* context);

    PEAK_MODULE_HANDLE ModuleHandle() const override;
    PEAK_EVENT_SUPPORTING_MODULE_HANDLE EventSupportingModuleHandle() const override;

    std::shared_ptr<Buffer> GetAnnouncedBuffer(PEAK_BUFFER_HANDLE bufferHandle) const;
    void AddAnnouncedBuffer(const std::shared_ptr<Buffer>& buffer);
    void RemoveAnnouncedBuffer(const std::shared_ptr<Buffer>& buffer);
    void RevokeAnnouncedBuffers();

    void Shutdown();

    friend ClassCreator<DataStream>;
    DataStream(PEAK_DATA_STREAM_HANDLE dataStreamHandle, const std::weak_ptr<Device>& parentDevice);
    PEAK_DATA_STREAM_HANDLE m_backendHandle;

    std::weak_ptr<Device> m_parentDevice;

    std::vector<std::shared_ptr<Buffer>> m_announcedBuffers;
    std::unordered_map<PEAK_BUFFER_HANDLE, std::shared_ptr<Buffer>> m_announcedBuffersByHandle;
    std::unordered_map<PEAK_BUFFER_HANDLE, std::unique_ptr<BufferRevocationCallback>>
        m_bufferRevocationCallbacksByBufferHandle;

    mutable std::mutex m_announcedBuffersMutex;
    mutable std::mutex m_bufferRevocationCallbacksByBufferHandleMutex;

    std::string m_key;
};

} /* namespace core */
} /* namespace peak */

/* Implementation */
namespace peak
{
namespace core
{

inline std::string ToString(DataStreamFlushMode entry)
{
    std::string entryString;

    if (entry == DataStreamFlushMode::InputPoolToOutputQueue)
    {
        entryString = "InputPoolToOutputQueue";
    }
    else if (entry == DataStreamFlushMode::DiscardOutputQueue)
    {
        entryString = "DiscardOutputQueue";
    }
    else if (entry == DataStreamFlushMode::AllToInputPool)
    {
        entryString = "AllToInputPool";
    }
    else if (entry == DataStreamFlushMode::UnqueuedToInputPool)
    {
        entryString = "UnqueuedToInputPool";
    }
    else if (entry == DataStreamFlushMode::DiscardAll)
    {
        entryString = "DiscardAll";
    }
    else if (entry >= DataStreamFlushMode::Custom)
    {
        entryString = "Custom";
    }

    return entryString;
}

inline std::string ToString(AcquisitionStartMode entry)
{
    std::string entryString;

    if (entry == AcquisitionStartMode::Default)
    {
        entryString = "Default";
    }
    else if (entry >= AcquisitionStartMode::Custom)
    {
        entryString = "Custom";
    }

    return entryString;
}

inline std::string ToString(AcquisitionStopMode entry)
{
    std::string entryString;

    if (entry == AcquisitionStopMode::Default)
    {
        entryString = "Default";
    }
    else if (entry == AcquisitionStopMode::Kill)
    {
        entryString = "Kill";
    }
    else if (entry >= AcquisitionStopMode::Custom)
    {
        entryString = "Custom";
    }

    return entryString;
}

inline DataStream::DataStream(PEAK_DATA_STREAM_HANDLE dataStreamHandle, const std::weak_ptr<Device>& parentDevice)
    : m_backendHandle(dataStreamHandle)
    , m_parentDevice(parentDevice)
    , m_announcedBuffers()
    , m_announcedBuffersByHandle()
    , m_announcedBuffersMutex()
    , m_key(QueryStringFromCInterfaceFunction([&](char* key, size_t* keySize) {
        return PEAK_C_ABI_PREFIX PEAK_DataStream_GetKey(dataStreamHandle, key, keySize);
    }))
{}

inline DataStream::~DataStream()
{
    try
    {
        Shutdown();
    }
    catch (const Exception&)
    {}

    (void)PEAK_C_ABI_PREFIX PEAK_DataStream_Destruct(m_backendHandle);
}

inline std::string DataStream::Key() const
{
    return m_key;
}

inline RawInformation DataStream::Info(int32_t infoCommand) const
{
    return QueryRawInformationFromCInterfaceFunction([&](int32_t* dataType, uint8_t* buffer, size_t* bufferSize) {
        return PEAK_C_ABI_PREFIX PEAK_DataStream_GetInfo(
            m_backendHandle, infoCommand, dataType, buffer, bufferSize);
    });
}

inline std::string DataStream::ID() const
{
    return QueryStringFromCInterfaceFunction([&](char* id, size_t* idSize) {
        return PEAK_C_ABI_PREFIX PEAK_DataStream_GetID(m_backendHandle, id, idSize);
    });
}

inline std::string DataStream::TLType() const
{
    return QueryStringFromCInterfaceFunction([&](char* tlType, size_t* tlTypeSize) {
        return PEAK_C_ABI_PREFIX PEAK_DataStream_GetTLType(m_backendHandle, tlType, tlTypeSize);
    });
}

inline size_t DataStream::NumBuffersAnnouncedMinRequired() const
{
    return QueryNumericFromCInterfaceFunction<size_t>([&](size_t* numBuffersAnnouncedMinRequired) {
        return PEAK_C_ABI_PREFIX PEAK_DataStream_GetNumBuffersAnnouncedMinRequired(
            m_backendHandle, numBuffersAnnouncedMinRequired);
    });
}

inline size_t DataStream::NumBuffersAnnounced() const
{
    return QueryNumericFromCInterfaceFunction<size_t>([&](size_t* numBuffersAnnounced) {
        return PEAK_C_ABI_PREFIX PEAK_DataStream_GetNumBuffersAnnounced(m_backendHandle, numBuffersAnnounced);
    });
}

inline size_t DataStream::NumBuffersQueued() const
{
    return QueryNumericFromCInterfaceFunction<size_t>([&](size_t* numBuffersQueued) {
        return PEAK_C_ABI_PREFIX PEAK_DataStream_GetNumBuffersQueued(m_backendHandle, numBuffersQueued);
    });
}

inline size_t DataStream::NumBuffersAwaitDelivery() const
{
    return QueryNumericFromCInterfaceFunction<size_t>([&](size_t* numBuffersAwaitDelivery) {
        return PEAK_C_ABI_PREFIX PEAK_DataStream_GetNumBuffersAwaitDelivery(
            m_backendHandle, numBuffersAwaitDelivery);
    });
}

inline uint64_t DataStream::NumBuffersDelivered() const
{
    return QueryNumericFromCInterfaceFunction<uint64_t>([&](uint64_t* numBuffersDelivered) {
        return PEAK_C_ABI_PREFIX PEAK_DataStream_GetNumBuffersDelivered(m_backendHandle, numBuffersDelivered);
    });
}

inline uint64_t DataStream::NumBuffersStarted() const
{
    return QueryNumericFromCInterfaceFunction<uint64_t>([&](uint64_t* numBuffersStarted) {
        return PEAK_C_ABI_PREFIX PEAK_DataStream_GetNumBuffersStarted(m_backendHandle, numBuffersStarted);
    });
}

inline uint64_t DataStream::NumUnderruns() const
{
    return QueryNumericFromCInterfaceFunction<uint64_t>([&](uint64_t* numUnderruns) {
        return PEAK_C_ABI_PREFIX PEAK_DataStream_GetNumUnderruns(m_backendHandle, numUnderruns);
    });
}

inline size_t DataStream::NumChunksPerBufferMax() const
{
    return QueryNumericFromCInterfaceFunction<size_t>([&](size_t* numChunksPerBufferMax) {
        return PEAK_C_ABI_PREFIX PEAK_DataStream_GetNumChunksPerBufferMax(m_backendHandle, numChunksPerBufferMax);
    });
}

inline size_t DataStream::BufferAlignment() const
{
    return QueryNumericFromCInterfaceFunction<size_t>([&](size_t* bufferAlignment) {
        return PEAK_C_ABI_PREFIX PEAK_DataStream_GetBufferAlignment(m_backendHandle, bufferAlignment);
    });
}

inline size_t DataStream::PayloadSize() const
{
    return QueryNumericFromCInterfaceFunction<size_t>([&](size_t* payloadSize) {
        return PEAK_C_ABI_PREFIX PEAK_DataStream_GetPayloadSize(m_backendHandle, payloadSize);
    });
}

inline bool DataStream::DefinesPayloadSize() const
{
    return QueryNumericFromCInterfaceFunction<PEAK_BOOL8>([&](PEAK_BOOL8* definesPayloadSize) {
        return PEAK_C_ABI_PREFIX PEAK_DataStream_GetDefinesPayloadSize(m_backendHandle, definesPayloadSize);
    }) > 0;
}

inline bool DataStream::IsGrabbing() const
{
    return QueryNumericFromCInterfaceFunction<PEAK_BOOL8>([&](PEAK_BOOL8* isGrabbing) {
        return PEAK_C_ABI_PREFIX PEAK_DataStream_GetIsGrabbing(m_backendHandle, isGrabbing);
    }) > 0;
}

inline std::shared_ptr<Buffer> DataStream::AnnounceBuffer(
    void* buffer, size_t size, void* userPtr, const DataStream::BufferRevocationCallback& revocationCallback)
{
    auto _callback = std::make_unique<BufferRevocationCallback>(revocationCallback);

    auto callbackContext = _callback.get();

    auto bufferHandle = QueryNumericFromCInterfaceFunction<PEAK_BUFFER_HANDLE>(
        [&](PEAK_BUFFER_HANDLE* _bufferHandle) {
            return PEAK_C_ABI_PREFIX PEAK_DataStream_AnnounceBuffer(m_backendHandle, buffer, size, userPtr,
                BufferRevocationCallbackCWrapper, callbackContext, _bufferHandle);
        });

    auto _buffer = std::make_shared<ClassCreator<Buffer>>(bufferHandle, shared_from_this());
    AddAnnouncedBuffer(_buffer);

    {
        std::lock_guard<std::mutex> lock(m_bufferRevocationCallbacksByBufferHandleMutex);

        m_bufferRevocationCallbacksByBufferHandle.emplace(bufferHandle, std::move(_callback));
    }

    return _buffer;
}

inline std::shared_ptr<Buffer> DataStream::AllocAndAnnounceBuffer(size_t size, void* userPtr)
{
    auto bufferHandle = QueryNumericFromCInterfaceFunction<PEAK_BUFFER_HANDLE>(
        [&](PEAK_BUFFER_HANDLE* _bufferHandle) {
            return PEAK_C_ABI_PREFIX PEAK_DataStream_AllocAndAnnounceBuffer(
                m_backendHandle, size, userPtr, _bufferHandle);
        });

    auto buffer = std::make_shared<ClassCreator<Buffer>>(bufferHandle, shared_from_this());
    AddAnnouncedBuffer(buffer);

    return buffer;
}

inline void DataStream::QueueBuffer(const std::shared_ptr<Buffer>& buffer)
{
    if (buffer == nullptr)
    {
        throw InvalidArgumentException("The given buffer is a nullptr!");
    }

    auto bufferHandle = buffer->m_backendHandle;

    CallAndCheckCInterfaceFunction(
        [&] { return PEAK_C_ABI_PREFIX PEAK_DataStream_QueueBuffer(m_backendHandle, bufferHandle); });
}

inline void DataStream::RevokeBuffer(const std::shared_ptr<Buffer>& buffer)
{
    if (buffer == nullptr)
    {
        throw InvalidArgumentException("The given buffer is a nullptr!");
    }

    auto bufferHandle = buffer->m_backendHandle;

    CallAndCheckCInterfaceFunction(
        [&] { return PEAK_C_ABI_PREFIX PEAK_DataStream_RevokeBuffer(m_backendHandle, bufferHandle); });

    RemoveAnnouncedBuffer(buffer);

    {
        std::lock_guard<std::mutex> lock(m_bufferRevocationCallbacksByBufferHandleMutex);

        m_bufferRevocationCallbacksByBufferHandle.erase(bufferHandle);
    }

    buffer->Revoke();
}

inline std::shared_ptr<Buffer> DataStream::WaitForFinishedBuffer(Timeout timeout_ms)
{
    auto bufferHandle = QueryNumericFromCInterfaceFunction<PEAK_BUFFER_HANDLE>(
        [&](PEAK_BUFFER_HANDLE* _bufferHandle) {
            return PEAK_C_ABI_PREFIX PEAK_DataStream_WaitForFinishedBuffer(
                m_backendHandle, timeout_ms, _bufferHandle);
        });

    return GetAnnouncedBuffer(bufferHandle);
}

inline void DataStream::KillWait()
{
    CallAndCheckCInterfaceFunction([&] { return PEAK_C_ABI_PREFIX PEAK_DataStream_KillWait(m_backendHandle); });
}

inline void DataStream::Flush(DataStreamFlushMode mode)
{
    CallAndCheckCInterfaceFunction([&] {
        return PEAK_C_ABI_PREFIX PEAK_DataStream_Flush(
            m_backendHandle, static_cast<PEAK_DATA_STREAM_FLUSH_MODE>(mode));
    });
}

inline std::vector<std::shared_ptr<Buffer>> DataStream::AnnouncedBuffers() const
{
    std::lock_guard<std::mutex> lock(m_announcedBuffersMutex);

    return m_announcedBuffers;
}

inline void DataStream::StartAcquisition(AcquisitionStartMode mode, uint64_t numToAcquire /*= INFINITE_NUMBER*/)
{
    CallAndCheckCInterfaceFunction([&] {
        return PEAK_C_ABI_PREFIX PEAK_DataStream_StartAcquisition(
            m_backendHandle, static_cast<PEAK_ACQUISITION_START_MODE>(mode), numToAcquire);
    });
}

inline void DataStream::StopAcquisition(AcquisitionStopMode mode)
{
    CallAndCheckCInterfaceFunction([&] {
        return PEAK_C_ABI_PREFIX PEAK_DataStream_StopAcquisition(
            m_backendHandle, static_cast<PEAK_ACQUISITION_STOP_MODE>(mode));
    });
}

inline std::shared_ptr<Device> DataStream::ParentDevice() const
{
    return LockOrThrow(m_parentDevice);
}

inline void PEAK_CALL_CONV DataStream::BufferRevocationCallbackCWrapper(void* buffer, void* userPtr, void* context)
{
    auto callback = static_cast<DataStream::BufferRevocationCallback*>(context);
    if (*callback)
    {
        callback->operator()(buffer, userPtr);
    }
}

inline PEAK_MODULE_HANDLE DataStream::ModuleHandle() const
{
    auto moduleHandle = QueryNumericFromCInterfaceFunction<PEAK_MODULE_HANDLE>(
        [&](PEAK_MODULE_HANDLE* _moduleHandle) {
            return PEAK_C_ABI_PREFIX PEAK_DataStream_ToModule(m_backendHandle, _moduleHandle);
        });

    return moduleHandle;
}

inline PEAK_EVENT_SUPPORTING_MODULE_HANDLE DataStream::EventSupportingModuleHandle() const
{
    auto eventSupportingModuleHandle = QueryNumericFromCInterfaceFunction<PEAK_EVENT_SUPPORTING_MODULE_HANDLE>(
        [&](PEAK_EVENT_SUPPORTING_MODULE_HANDLE* _eventSupportingModuleHandle) {
            return PEAK_C_ABI_PREFIX PEAK_DataStream_ToEventSupportingModule(
                m_backendHandle, _eventSupportingModuleHandle);
        });

    return eventSupportingModuleHandle;
}

inline std::shared_ptr<Buffer> DataStream::GetAnnouncedBuffer(PEAK_BUFFER_HANDLE bufferHandle) const
{
    std::lock_guard<std::mutex> lock(m_announcedBuffersMutex);

    return m_announcedBuffersByHandle.at(bufferHandle);
}

inline void DataStream::AddAnnouncedBuffer(const std::shared_ptr<Buffer>& buffer)
{
    std::lock_guard<std::mutex> lock(m_announcedBuffersMutex);

    m_announcedBuffers.emplace_back(buffer);
    m_announcedBuffersByHandle.emplace(buffer->m_backendHandle, buffer);
}

inline void DataStream::RemoveAnnouncedBuffer(const std::shared_ptr<Buffer>& buffer)
{
    std::lock_guard<std::mutex> lock(m_announcedBuffersMutex);

    m_announcedBuffers.erase(std::remove(std::begin(m_announcedBuffers), std::end(m_announcedBuffers), buffer));
    m_announcedBuffersByHandle.erase(buffer->m_backendHandle);
}

inline void DataStream::RevokeAnnouncedBuffers()
{
    for (const auto& buffer : AnnouncedBuffers())
    {
        RevokeBuffer(buffer);
    }
}

inline void DataStream::Shutdown()
{
    if (IsGrabbing())
    {
        StopAcquisition(AcquisitionStopMode::Default);
    }
    Flush(DataStreamFlushMode::DiscardAll);
    RevokeAnnouncedBuffers();
}

} /* namespace core */
} /* namespace peak */