SharedDataStream.h

Go to the documentation of this file.

/*
 * Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
 *
 * Licensed under the Apache License, Version 2.0 (the "License").
 * You may not use this file except in compliance with the License.
 * A copy of the License is located at
 *
 *     http://aws.amazon.com/apache2.0/
 *
 * or in the "license" file accompanying this file. This file is distributed
 * on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either
 * express or implied. See the License for the specific language governing
 * permissions and limitations under the License.
 */

#ifndef ALEXA_CLIENT_SDK_AVSCOMMON_UTILS_INCLUDE_AVSCOMMON_UTILS_SDS_SHAREDDATASTREAM_H_
#define ALEXA_CLIENT_SDK_AVSCOMMON_UTILS_INCLUDE_AVSCOMMON_UTILS_SDS_SHAREDDATASTREAM_H_

#include <cstdint>
#include <cstddef>
#include <memory>

#include "AVSCommon/Utils/Logger/LoggerUtils.h"

namespace alexaClientSDK {
namespace avsCommon {
namespace utils {
namespace sds {

template <typename T>
class SharedDataStream {
private:
    // Forward declare the nested @c BufferLayout structure (full declaration is in @c BufferLayout.h).
    class BufferLayout;

public:
    using Index = uint64_t;

    using AtomicIndex = typename T::AtomicIndex;

    using AtomicBool = typename T::AtomicBool;

    using Buffer = typename T::Buffer;

    using Mutex = typename T::Mutex;

    using ConditionVariable = typename T::ConditionVariable;

    // Forward declare the nested @c Reader class (full declaration is in @c Reader.h).
    class Reader;

    // Forward declare the nested @c Writer class (full declaraion is in @c Writer.h).
    class Writer;

    static size_t calculateBufferSize(size_t nWords, size_t wordSize = 1, size_t maxReaders = 1);

    static std::unique_ptr<SharedDataStream> create(
        std::shared_ptr<Buffer> buffer,
        size_t wordSize = 1,
        size_t maxReaders = 1);

    static std::unique_ptr<SharedDataStream> create(
        std::shared_ptr<Buffer> buffer,
        size_t wordSize,
        size_t maxReaders,
        size_t maxEphemeralReaders);

    static std::unique_ptr<SharedDataStream> open(std::shared_ptr<Buffer> buffer);

    size_t getMaxReaders() const;

    Index getDataSize() const;

    size_t getWordSize() const;

    std::unique_ptr<Writer> createWriter(typename Writer::Policy policy, bool forceReplacement = false);

    std::unique_ptr<Reader> createReader(typename Reader::Policy policy, bool startWithNewData = false);

    std::unique_ptr<Reader> createReader(
        size_t id,
        typename Reader::Policy policy,
        bool startWithNewData = false,
        bool forceReplacement = false);

private:
    SharedDataStream(std::shared_ptr<typename T::Buffer> buffer);

    std::unique_ptr<Reader> createReaderLocked(
        size_t id,
        typename Reader::Policy policy,
        bool startWithNewData,
        bool forceReplacement,
        std::unique_lock<Mutex>* lock);

    static const std::string TAG;

    static const int MAX_READER_CREATION_RETRIES;

    std::shared_ptr<BufferLayout> m_bufferLayout;
};

template <typename T>
const int SharedDataStream<T>::MAX_READER_CREATION_RETRIES = 3;

template <typename T>
const std::string SharedDataStream<T>::TAG = "SharedDataStream";

template <typename T>
size_t SharedDataStream<T>::calculateBufferSize(size_t nWords, size_t wordSize, size_t maxReaders) {
    if (0 == nWords) {
        logger::acsdkError(logger::LogEntry(TAG, "calculateBufferSizeFailed").d("reason", "numWordsZero"));
        return 0;
    } else if (0 == wordSize) {
        logger::acsdkError(logger::LogEntry(TAG, "calculateBufferSizeFailed").d("reason", "wordSizeZero"));
        return 0;
    }
    size_t overhead = BufferLayout::calculateDataOffset(wordSize, maxReaders);
    size_t dataSize = nWords * wordSize;
    return overhead + dataSize;
}

template <typename T>
std::unique_ptr<SharedDataStream<T>> SharedDataStream<T>::create(
    std::shared_ptr<Buffer> buffer,
    size_t wordSize,
    size_t maxReaders) {
    return create(buffer, wordSize, maxReaders, maxReaders);
}

template <typename T>
std::unique_ptr<SharedDataStream<T>> SharedDataStream<T>::create(
    std::shared_ptr<Buffer> buffer,
    size_t wordSize,
    size_t maxReaders,
    size_t maxEphemeralReaders) {
    size_t expectedSize = calculateBufferSize(1, wordSize, maxReaders);
    if (0 == expectedSize) {
        // Logged in calcutlateBuffersize().
        return nullptr;
    } else if (nullptr == buffer) {
        logger::acsdkError(logger::LogEntry(TAG, "createFailed").d("reason", "nullBuffer"));
        return nullptr;
    } else if (expectedSize > buffer->size()) {
        logger::acsdkError(logger::LogEntry(TAG, "createFailed")
                               .d("reason", "bufferSizeTooSmall")
                               .d("bufferSize", buffer->size())
                               .d("expectedSize", expectedSize));
        return nullptr;
    } else if (maxEphemeralReaders > maxReaders) {
        logger::acsdkError(logger::LogEntry(TAG, "createFailed").d("reason", "maxEphemeralReaders > maxReaders"));
        return nullptr;
    }

    std::unique_ptr<SharedDataStream<T>> sds(new SharedDataStream<T>(buffer));
    if (!sds->m_bufferLayout->init(wordSize, maxReaders, maxEphemeralReaders)) {
        // Logged in init().
        return nullptr;
    }
    return sds;
}

template <typename T>
std::unique_ptr<SharedDataStream<T>> SharedDataStream<T>::open(std::shared_ptr<Buffer> buffer) {
    std::unique_ptr<SharedDataStream<T>> sds(new SharedDataStream<T>(buffer));
    if (!sds->m_bufferLayout->attach()) {
        // Logged in attach().
        return nullptr;
    } else {
        return sds;
    }
}

template <typename T>
size_t SharedDataStream<T>::getMaxReaders() const {
    return m_bufferLayout->getHeader()->maxReaders;
}

template <typename T>
typename SharedDataStream<T>::Index SharedDataStream<T>::getDataSize() const {
    return m_bufferLayout->getDataSize();
}

template <typename T>
size_t SharedDataStream<T>::getWordSize() const {
    return m_bufferLayout->getHeader()->wordSize;
}

template <typename T>
std::unique_ptr<typename SharedDataStream<T>::Writer> SharedDataStream<T>::createWriter(
    typename Writer::Policy policy,
    bool forceReplacement) {
    auto header = m_bufferLayout->getHeader();
    std::lock_guard<Mutex> lock(header->writerEnableMutex);
    if (header->isWriterEnabled && !forceReplacement) {
        logger::acsdkError(logger::LogEntry(TAG, "createWriterFailed")
                               .d("reason", "existingWriterAttached")
                               .d("forceReplacement", "false"));
        return nullptr;
    } else {
        return std::unique_ptr<Writer>(new Writer(policy, m_bufferLayout));
    }
}

template <typename T>
std::unique_ptr<typename SharedDataStream<T>::Reader> SharedDataStream<T>::createReader(
    typename Reader::Policy policy,
    bool startWithNewData) {
    std::unique_lock<Mutex> lock(m_bufferLayout->getHeader()->readerEnableMutex);
    for (size_t id = 0; id < m_bufferLayout->getHeader()->maxEphemeralReaders; ++id) {
        if (!m_bufferLayout->isReaderEnabled(id)) {
            return createReaderLocked(id, policy, startWithNewData, false, &lock);
        }
    }
    logger::acsdkError(logger::LogEntry(TAG, "createWriterFailed").d("reason", "noAvailableReaders"));
    return nullptr;
}

template <typename T>
std::unique_ptr<typename SharedDataStream<T>::Reader> SharedDataStream<T>::createReader(
    size_t id,
    typename Reader::Policy policy,
    bool startWithNewData,
    bool forceReplacement) {
    std::unique_lock<Mutex> lock(m_bufferLayout->getHeader()->readerEnableMutex);
    return createReaderLocked(id, policy, startWithNewData, forceReplacement, &lock);
}

template <typename T>
SharedDataStream<T>::SharedDataStream(std::shared_ptr<Buffer> buffer) :
        m_bufferLayout{std::make_shared<BufferLayout>(buffer)} {
}

template <typename T>
std::unique_ptr<typename SharedDataStream<T>::Reader> SharedDataStream<T>::createReaderLocked(
    size_t id,
    typename Reader::Policy policy,
    bool startWithNewData,
    bool forceReplacement,
    std::unique_lock<Mutex>* lock) {
    if (m_bufferLayout->isReaderEnabled(id) && !forceReplacement) {
        logger::acsdkError(logger::LogEntry(TAG, "createReaderLockedFailed")
                               .d("reason", "readerAlreadyAttached")
                               .d("readerId", id)
                               .d("forceReplacement", "false"));
        return nullptr;
    } else {
        // Note: Reader constructor does not call updateUnconsumedCursor() automatically, because we may be seeking to
        // a blocked writer's cursor below (if !startWithNewData), and we don't want the writer to start moving before
        // we seek.
        auto reader = std::unique_ptr<Reader>(new Reader(policy, m_bufferLayout, id));
        lock->unlock();

        if (startWithNewData) {
            // We're not moving the cursor again, so call updateUnconsumedCursor() now.
            m_bufferLayout->updateOldestUnconsumedCursor();
            return reader;
        }

        auto headerPtr = m_bufferLayout->getHeader();
        for (int i = 0; i < MAX_READER_CREATION_RETRIES; i++) {
            auto offset = m_bufferLayout->getDataSize();
            auto writeStartCursor = headerPtr->writeStartCursor.load();
            auto reference = Reader::Reference::BEFORE_WRITER;
            if (writeStartCursor < offset) {
                // For SDS without buffer overwritten, seek the very beginning of the stream using ABSOLUTE reference
                // ABSOLUTE reference prevents a race condition between blocking writer and reader seek when writer
                // writes more data to buffer, and writing cursor moves forward.
                offset = 0;
                reference = Reader::Reference::ABSOLUTE;
            } else {
                auto writeEndCursor = headerPtr->writeEndCursor.load();
                if (writeEndCursor < writeStartCursor) {
                    logger::acsdkError(logger::LogEntry(TAG, "createReaderLockedError")
                                           .d("reason", "writeCursorBeyondEndCursor?")
                                           .d("readerId", id));
                    continue;
                }
                auto wordsBeingWritten = writeEndCursor - writeStartCursor;
                if (offset < wordsBeingWritten) {
                    logger::acsdkWarn(logger::LogEntry(TAG, "createReaderLockedWarning")
                                          .d("reason", "detectedWriterOverflow")
                                          .d("readerId", id));
                    continue;
                }
                offset -= wordsBeingWritten;
            }

            if (reader->seek(offset, reference)) {
                // Note: seek() will call updateUnconsumedCursor() if it returns true.
                return reader;
            }
        }

        logger::acsdkError(
            logger::LogEntry(TAG, "createReaderLockedFailed").d("reason", "seekRetriesExhausted").d("readerId", id));
        return nullptr;
    }
}

}  // namespace sds
}  // namespace utils
}  // namespace avsCommon
}  // namespace alexaClientSDK

#include "BufferLayout.h"
#include "Reader.h"
#include "Writer.h"

#endif  // ALEXA_CLIENT_SDK_AVSCOMMON_UTILS_INCLUDE_AVSCOMMON_UTILS_SDS_SHAREDDATASTREAM_H_