disk_buffer.h

Go to the documentation of this file.

// TODO validate capacity of disk

#pragma once

#include "memory.h"
#include <string.h>
#include <mutex>
#include "fifo_buffer_base.h"
#include <fstream>
#include "risingdelay.h"

#if __has_include(<filesystem>)
#include <filesystem>
#else
#include <experimental/filesystem>
#endif

BEGIN_NAMESPACE_MEMORY

template <typename T>
class FileHandler
{
public:
    explicit FileHandler(const std::string& filename, bool deleteFileAfterClose = true)
        : filename(filename)
        , deleteFileAfterClose(deleteFileAfterClose)
    {
    }

    virtual ~FileHandler()
    {
        close();
    }

    void open()
    {
        open("w");
    }

    void open(char const* mode)
    {
        if (strcmp(mode, "w") == 0)
        {
            file.open(filename.c_str(), std::fstream::app | std::fstream::out);
            file.close();
            file.open(filename.c_str(), std::fstream::binary | std::fstream::in | std::fstream::out);
        }
        else if (strcmp(mode, "r") == 0)
        {
            file = std::fstream(filename.c_str(), std::fstream::binary | std::fstream::in);
        }
        else
            throw std::invalid_argument("Opening disk buffer file with argument: " + std::string(mode));

        if (!file)
            throw std::runtime_error("Error opening disk buffer file");
    }

    bool close()
    {
        if (!file || !file.is_open())
            return true;
        file.close();
        if (file.fail())
            throw std::runtime_error("File closing error");
        assert(!file.is_open());
        if (deleteFileAfterClose)
            removeFile();
        return true;
    }

    void removeFile()
    {
        const size_t MAX_RETRIES = 5;

        auto start = std::chrono::steady_clock::now();
    
        Threading::RisingDelay risingDelay{100, 1000};
        size_t removeResult;
        size_t it = 0;

        do
        {
            removeResult = remove(filename.c_str());
            if (removeResult == 0)
                return;

            risingDelay.wait();
        } while (++it < MAX_RETRIES);

        auto end = std::chrono::steady_clock::now();
        auto durationInMs = std::chrono::duration_cast<std::chrono::milliseconds>(end - start).count();
        std::stringstream ss;
        ss << "File closing error; file:" << this->filename << "; error code:" << removeResult << "; retries: " << MAX_RETRIES
           << "; elapsed[ms]:" << durationInMs;
        throw std::runtime_error(ss.str().c_str());
    }

    void rewind()
    {
        file.clear();
        file.seekg(0, std::fstream::beg);
    }

    size_t write(const T* data, size_t count)
    {
        return write(data, sizeof(T), count);
    }

    size_t writeAt(const T* data, size_t count, size_t pos)
    {
        seekPos(pos);
        return write(data, count);
    }

    size_t read(T* data, size_t count)
    {
        return read(data, sizeof(T), count);
    }

    size_t readAt(T* data, size_t count, size_t pos)
    {
        seekPos(pos);
        return read(data, count);
    }

    void seekPos(size_t offset)
    {
        file.seekg(offset * sizeof(T));
        if (file.fail())
            throw std::runtime_error("Failed to change read position in file");
    }

    size_t getSizeInBytes()
    {
        file.seekg(0, std::fstream::end);
        if (file.fail())
            throw std::runtime_error("Failed to determine size of a file in bytes");
        size_t size = size_t(file.tellg());
        return size;
    }

    size_t getSize()
    {
        return getSizeInBytes() / sizeof(T);
    }

    void reset()
    {
        close();
        open();
    }

protected:
    std::fstream file;
    std::string filename;
    bool deleteFileAfterClose;

    size_t write(const void* data, size_t size, size_t count)
    {
        file.write(static_cast<const char*>(data), size * count);
        if (file.fail())
            throw std::runtime_error("Error writing data to file");
        return count;
    }

    size_t read(void* data, size_t size, size_t count)
    {
        file.read(static_cast<char*>(data), size * count);
        if (file.fail())
            throw std::runtime_error("Error reading data from file");
        return count;
    }
};

template <typename T>
class DiskBuffer : public FifoBufferBase<T>
{
public:
    DiskBuffer(size_t capacity, const std::string& fileName, bool deleteFileAfterClose = true);
    virtual ~DiskBuffer() = default;

    void reset() override;

    FileHandler<T>& getHandler();

    size_t write(const T* samples, size_t samplesCount, size_t pos) override;

    size_t read(T* samples, size_t samplesCount, const size_t pos) override;

    T* getData() const override;

    bool validateReadPos(size_t readPos);

private:
    FileHandler<T> fileHandler;
};

template <typename T>
inline DiskBuffer<T>::DiskBuffer(size_t capacity, const std::string& fileName, bool deleteFileAfterClose)
    : fileHandler(fileName, deleteFileAfterClose)
{
    DiskBuffer<T>::setCapacity(capacity);
}

template <typename T>
inline void DiskBuffer<T>::reset()
{
    std::lock_guard<std::recursive_mutex> guard(this->getLock());
    fileHandler.reset();
    FifoBufferBase<T>::reset();
}

template <typename T>
inline FileHandler<T>& DiskBuffer<T>::getHandler()
{
    return fileHandler;
}

template <typename T>
size_t DiskBuffer<T>::write(const T* samples, size_t samplesCount, size_t pos)
{
    return fileHandler.writeAt(samples, samplesCount, pos);
}

template <typename T>
size_t Dewesoft::Utils::Memory::DiskBuffer<T>::read(T* samples, size_t samplesCount, const size_t pos)
{
    return fileHandler.readAt(samples, samplesCount, pos);
}

template <typename T>
T* DiskBuffer<T>::getData() const
{
    throw std::runtime_error("Not supported");
}

template <typename T>
bool DiskBuffer<T>::validateReadPos(size_t readPos)
{
    std::lock_guard<std::recursive_mutex> guard(this->getLock());

    auto lower = this->getReadPos();
    auto upper = this->getWritePos();
    auto capacity = this->getCapacity();

    return (readPos < capacity) &&
           ((lower < upper && readPos < upper && readPos >= lower) || (upper < lower && !(readPos >= upper && readPos < lower)));
}

template <typename T>
size_t transferBuffer(IDirectArraySourceBuffer<T>& source,
                      DiskBuffer<T>& target,
                      bool incrementPosition = true,
                      size_t samplesCount = (std::numeric_limits<size_t>::max)())
{
    return transferBuffer(
        static_cast<IDirectArraySourceBuffer<T>&>(source), static_cast<ITargetBuffer<T>&>(target), incrementPosition, samplesCount, true);
}

template <typename T>
size_t transferBuffer(DiskBuffer<T>& source,
                      IDirectArrayTargetBuffer<T>& target,
                      bool incrementPosition = true,
                      size_t samplesCount = (std::numeric_limits<size_t>::max)())
{
    return transferBuffer(
        static_cast<ISourceBuffer<T>&>(source), static_cast<IDirectArrayTargetBuffer<T>&>(target), incrementPosition, samplesCount);
}

END_NAMESPACE_MEMORY