Reader.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_READER_H_
#define ALEXA_CLIENT_SDK_AVSCOMMON_UTILS_INCLUDE_AVSCOMMON_UTILS_SDS_READER_H_

#include <cstdint>
#include <cstddef>
#include <vector>
#include <mutex>
#include <limits>
#include <cstring>

#include "AVSCommon/Utils/Logger/LoggerUtils.h"
#include "AVSCommon/Utils/PlatformDefinitions.h"
#include "SharedDataStream.h"
#include "ReaderPolicy.h"

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

template <typename T>
class SharedDataStream<T>::Reader {
public:
    using Policy = ReaderPolicy;

    enum class Reference {
        AFTER_READER,
        BEFORE_READER,
        BEFORE_WRITER,
        ABSOLUTE
    };

    struct Error {
        enum {
            CLOSED = 0,
            OVERRUN = -1,
            WOULDBLOCK = -2,
            TIMEDOUT = -3,
            INVALID = -4
        };
    };

    Reader(Policy policy, std::shared_ptr<BufferLayout> bufferLayout, size_t id);

    ~Reader();

    using OutputFunction = std::function<bool(void* buf, size_t nWords)>;

    ssize_t read(
        OutputFunction function,
        size_t maxWordsCanRead,
        std::chrono::milliseconds timeout = std::chrono::milliseconds(0));

    ssize_t read(void* buf, size_t nWords, std::chrono::milliseconds timeout = std::chrono::milliseconds(0));

    bool seek(Index offset, Reference reference = Reference::ABSOLUTE);

    Index tell(Reference reference = Reference::ABSOLUTE) const;

    void close(Index offset = 0, Reference reference = Reference::AFTER_READER);

    size_t getId() const;

    size_t getWordSize() const;

    static std::string errorToString(Error error);

private:
    static const std::string TAG;

    Policy m_policy;

    std::shared_ptr<BufferLayout> m_bufferLayout;

    size_t m_id;

    AtomicIndex* m_readerCursor;

    AtomicIndex* m_readerCloseIndex;
};

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

template <typename T>
SharedDataStream<T>::Reader::Reader(Policy policy, std::shared_ptr<BufferLayout> bufferLayout, size_t id) :
        m_policy{policy},
        m_bufferLayout{bufferLayout},
        m_id{id},
        m_readerCursor{&m_bufferLayout->getReaderCursorArray()[m_id]},
        m_readerCloseIndex{&m_bufferLayout->getReaderCloseIndexArray()[m_id]} {
    // Note - SharedDataStream::createReader() holds readerEnableMutex while calling this function.
    // Read new data only.
    // Note: It is important that new readers start with their cursor at the writer.  This allows
    // updateOldestUnconsumedCursor() to be thread-safe without holding readerEnableMutex.  See
    // updateOldestUnconsumedCursor() comments for further explanation.
    *m_readerCursor = m_bufferLayout->getHeader()->writeStartCursor.load();

    // Read indefinitely.
    *m_readerCloseIndex = std::numeric_limits<Index>::max();

    m_bufferLayout->enableReaderLocked(m_id);
}

template <typename T>
SharedDataStream<T>::Reader::~Reader() {
    // Note: We can't leave a reader with its cursor in the future; doing so can introduce a race condition in
    // updateOldestUnconsumedCursor().  See updateOldestUnconsumedCursor() comments for further explanation.
    seek(0, Reference::BEFORE_WRITER);

    std::lock_guard<Mutex> lock(m_bufferLayout->getHeader()->readerEnableMutex);
    m_bufferLayout->disableReaderLocked(m_id);
    m_bufferLayout->updateOldestUnconsumedCursor();
}

template <typename T>
ssize_t SharedDataStream<T>::Reader::read(void* buf, size_t nWords, std::chrono::milliseconds timeout) {
    if (nullptr == buf) {
        logger::acsdkError(logger::LogEntry(TAG, "readFailed").d("reason", "nullBuffer"));
        return Error::INVALID;
    }

    auto buf8 = static_cast<uint8_t*>(buf);
    auto wordSize = getWordSize();
    return read(
        [&buf8, wordSize](void* readBuf, size_t wordsToRead) {
            memcpy(buf8, readBuf, wordsToRead * wordSize);
            buf8 += wordsToRead * wordSize;
            return true;
        },
        nWords,
        timeout);
}

template <typename T>
ssize_t SharedDataStream<T>::Reader::read(OutputFunction function, size_t nWords, std::chrono::milliseconds timeout) {
    if (nullptr == function) {
        logger::acsdkError(logger::LogEntry(TAG, "readFailed").d("reason", "nullFunction"));
        return Error::INVALID;
    }

    if (0 == nWords) {
        logger::acsdkError(logger::LogEntry(TAG, "readFailed").d("reason", "invalidNumWords").d("numWords", nWords));
        return Error::INVALID;
    }

    // Check if closed.
    auto readerCloseIndex = m_readerCloseIndex->load();
    if (*m_readerCursor >= readerCloseIndex) {
        return Error::CLOSED;
    }

    // Initial check for overrun.
    auto header = m_bufferLayout->getHeader();
    if ((header->writeEndCursor >= *m_readerCursor) &&
        (header->writeEndCursor - *m_readerCursor) > m_bufferLayout->getDataSize()) {
        return Error::OVERRUN;
    }

    std::unique_lock<Mutex> lock(header->dataAvailableMutex, std::defer_lock);
    if (Policy::BLOCKING == m_policy) {
        lock.lock();
    }

    // Figure out how much we can actually copy.
    size_t wordsAvailable = tell(Reference::BEFORE_WRITER);
    if (0 == wordsAvailable) {
        if (header->writeEndCursor > 0 && !header->isWriterEnabled) {
            return Error::CLOSED;
        } else if (Policy::NONBLOCKING == m_policy) {
            return Error::WOULDBLOCK;
        } else if (Policy::BLOCKING == m_policy) {
            // Condition for returning from read: the Writer has been closed or there is data to read
            auto predicate = [this, header] {
                return header->hasWriterBeenClosed || tell(Reference::BEFORE_WRITER) > 0;
            };

            if (std::chrono::milliseconds::zero() == timeout) {
                header->dataAvailableConditionVariable.wait(lock, predicate);
            } else if (!header->dataAvailableConditionVariable.wait_for(lock, timeout, predicate)) {
                return Error::TIMEDOUT;
            }
        }
        wordsAvailable = tell(Reference::BEFORE_WRITER);

        // If there is still no data, the writer has closed in the interim
        if (0 == wordsAvailable) {
            return Error::CLOSED;
        }
    }

    if (Policy::BLOCKING == m_policy) {
        lock.unlock();
    }
    if (nWords > wordsAvailable) {
        nWords = wordsAvailable;
    }

    // Don't read beyond our close index.
    if ((*m_readerCursor + nWords) > readerCloseIndex) {
        nWords = readerCloseIndex - *m_readerCursor;
    }

    // Split it across the wrap.
    size_t beforeWrap = m_bufferLayout->wordsUntilWrap(*m_readerCursor);
    if (beforeWrap > nWords) {
        beforeWrap = nWords;
    }
    size_t afterWrap = nWords - beforeWrap;

    // Copy the two segments.
    if (!function(m_bufferLayout->getData(*m_readerCursor), beforeWrap)) {
        // We haven't changed the read pointer yet, just error out.
        return Error::INVALID;
    }

    if (afterWrap > 0 && !function(m_bufferLayout->getData(*m_readerCursor + beforeWrap), afterWrap)) {
        // By API if either of these calls fail the entire read fails and no data is consumed.
        return Error::INVALID;
    }

    // Advance the read cursor.
    *m_readerCursor += nWords;

    // Final check for overrun (do this before the updateOldestUnconsumedCursor() call below for improved accuracy).
    bool overrun = ((header->writeEndCursor - *m_readerCursor) > m_bufferLayout->getDataSize());

    // Move the unconsumed cursor before returning.
    m_bufferLayout->updateOldestUnconsumedCursor();

    // Now we can safely error out if there was an overrun.
    if (overrun) {
        return Error::OVERRUN;
    }

    return nWords;
}

template <typename T>
bool SharedDataStream<T>::Reader::seek(Index offset, Reference reference) {
    auto header = m_bufferLayout->getHeader();
    auto writeStartCursor = &header->writeStartCursor;
    auto writeEndCursor = &header->writeEndCursor;
    Index absolute = std::numeric_limits<Index>::max();

    switch (reference) {
        case Reference::AFTER_READER:
            absolute = *m_readerCursor + offset;
            break;
        case Reference::BEFORE_READER:
            if (offset > *m_readerCursor) {
                logger::acsdkError(logger::LogEntry(TAG, "seekFailed")
                                       .d("reason", "seekBeforeStreamStart")
                                       .d("reference", "BEFORE_READER")
                                       .d("seekOffset", offset)
                                       .d("readerCursor", m_readerCursor->load()));
                return false;
            }
            absolute = *m_readerCursor - offset;
            break;
        case Reference::BEFORE_WRITER:
            if (offset > *writeStartCursor) {
                logger::acsdkError(logger::LogEntry(TAG, "seekFailed")
                                       .d("reason", "seekBeforeStreamStart")
                                       .d("reference", "BEFORE_WRITER")
                                       .d("seekOffset", offset)
                                       .d("writeStartCursor", writeStartCursor->load()));
                return false;
            }
            absolute = *writeStartCursor - offset;
            break;
        case Reference::ABSOLUTE:
            absolute = offset;
    }

    // Don't seek beyond the close index.
    if (absolute > *m_readerCloseIndex) {
        logger::acsdkError(logger::LogEntry(TAG, "seekFailed")
                               .d("reason", "seekBeyondCloseIndex")
                               .d("position", absolute)
                               .d("readerCloseIndex", m_readerCloseIndex->load()));
        return false;
    }

    // Per documentation of updateOldestUnconsumedCursor(), don't try to seek backwards while oldestConsumedCursor is
    // being updated.
    bool backward = absolute < *m_readerCursor;
    std::unique_lock<Mutex> lock(header->backwardSeekMutex, std::defer_lock);
    if (backward) {
        lock.lock();
    }

    // Don't seek to past data which has been (or soon will be) overwritten.
    // Note: If this is a backward seek, it is important that this check is performed while holding the
    // backwardSeekMutex to prevent a writer from starting to overwrite us between here and the m_readerCursor update
    // below.  If this is not a backward seek, then the mutex is not held.
    if (*writeEndCursor >= absolute && *writeEndCursor - absolute > m_bufferLayout->getDataSize()) {
        logger::acsdkError(logger::LogEntry(TAG, "seekFailed").d("reason", "seekOverwrittenData"));
        return false;
    }

    *m_readerCursor = absolute;

    if (backward) {
        header->oldestUnconsumedCursor = 0;
        m_bufferLayout->updateOldestUnconsumedCursorLocked();
        lock.unlock();
    } else {
        m_bufferLayout->updateOldestUnconsumedCursor();
    }

    return true;
}

template <typename T>
typename SharedDataStream<T>::Index SharedDataStream<T>::Reader::tell(Reference reference) const {
    auto writeStartCursor = &m_bufferLayout->getHeader()->writeStartCursor;
    switch (reference) {
        case Reference::AFTER_READER:
            return 0;
        case Reference::BEFORE_READER:
            return 0;
        case Reference::BEFORE_WRITER:
            return (*writeStartCursor >= *m_readerCursor) ? *writeStartCursor - *m_readerCursor : 0;
        case Reference::ABSOLUTE:
            return *m_readerCursor;
    }
    logger::acsdkError(logger::LogEntry(TAG, "tellFailed").d("reason", "invalidReference"));
    return std::numeric_limits<Index>::max();
}

template <typename T>
void SharedDataStream<T>::Reader::close(Index offset, Reference reference) {
    auto writeStartCursor = &m_bufferLayout->getHeader()->writeStartCursor;
    Index absolute = 0;
    bool validReference = false;
    switch (reference) {
        case Reference::AFTER_READER:
            absolute = *m_readerCursor + offset;
            validReference = true;
            break;
        case Reference::BEFORE_READER:
            absolute = *m_readerCursor;
            validReference = true;
            break;
        case Reference::BEFORE_WRITER:
            if (*writeStartCursor < offset) {
                logger::acsdkError(logger::LogEntry(TAG, "closeFailed")
                                       .d("reason", "invalidIndex")
                                       .d("reference", "BEFORE_WRITER")
                                       .d("offset", offset)
                                       .d("writeStartCursor", writeStartCursor->load()));
            } else {
                absolute = *writeStartCursor - offset;
            }
            validReference = true;
            break;
        case Reference::ABSOLUTE:
            absolute = offset;
            validReference = true;
            break;
    }
    if (!validReference) {
        logger::acsdkError(logger::LogEntry(TAG, "closeFailed").d("reason", "invalidReference"));
    }

    *m_readerCloseIndex = absolute;
}

template <typename T>
size_t SharedDataStream<T>::Reader::getId() const {
    return m_id;
}

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

template <typename T>
std::string SharedDataStream<T>::Reader::errorToString(Error error) {
    switch (error) {
        case Error::CLOSED:
            return "CLOSED";
        case Error::OVERRUN:
            return "OVERRUN";
        case Error::WOULDBLOCK:
            return "WOULDBLOCK";
        case Error::TIMEDOUT:
            return "TIMEDOUT";
        case Error::INVALID:
            return "INVALID";
    }
    logger::acsdkError(logger::LogEntry(TAG, "errorToStringFailed").d("reason", "invalidError").d("error", error));
    return "(unknown error " + to_string(error) + ")";
}

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

#endif  // ALEXA_CLIENT_SDK_AVSCOMMON_UTILS_INCLUDE_AVSCOMMON_UTILS_SDS_READER_H_