Writer.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_WRITER_H_
#define ALEXA_CLIENT_SDK_AVSCOMMON_UTILS_INCLUDE_AVSCOMMON_UTILS_SDS_WRITER_H_

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

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

#include "SharedDataStream.h"
#include "WriterPolicy.h"

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

template <typename T>
class SharedDataStream<T>::Writer {
public:
    using Policy = WriterPolicy;

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

    Writer(Policy policy, std::shared_ptr<BufferLayout> bufferLayout);

    ~Writer();

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

    Index tell() const;

    void close();

    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;

    bool m_closed;
};

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

template <typename T>
SharedDataStream<T>::Writer::Writer(Policy policy, std::shared_ptr<BufferLayout> bufferLayout) :
        m_policy{policy},
        m_bufferLayout{bufferLayout},
        m_closed{false} {
    // Note - SharedDataStream::createWriter() holds writerEnableMutex while calling this function.
    auto header = m_bufferLayout->getHeader();
    header->isWriterEnabled = true;
    header->writeEndCursor = header->writeStartCursor.load();
}

template <typename T>
SharedDataStream<T>::Writer::~Writer() {
    close();
}

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

    auto header = m_bufferLayout->getHeader();
    if (!header->isWriterEnabled) {
        logger::acsdkError(logger::LogEntry(TAG, "writeFailed").d("reason", "writerDisabled"));
        return Error::CLOSED;
    }

    auto wordsToCopy = nWords;
    auto buf8 = static_cast<const uint8_t*>(buf);
    std::unique_lock<Mutex> backwardSeekLock(header->backwardSeekMutex, std::defer_lock);
    Index writeEnd = header->writeStartCursor + nWords;

    switch (m_policy) {
        case Policy::NONBLOCKABLE:
            // For NONBLOCKABLE, we can truncate the write if it won't fit in the buffer.
            if (nWords > m_bufferLayout->getDataSize()) {
                wordsToCopy = nWords = m_bufferLayout->getDataSize();
                writeEnd = header->writeStartCursor + nWords;
            }
            break;
        case Policy::ALL_OR_NOTHING:
            // For ALL_OR_NOTHING, we can't overwrite readers, and we can't truncate, but we might be able to discard
            // bytes that overflow if oldestUnconsumedCursor is in the future (e.g. there are readers waiting for future
            // data which has not been written yet).

            // Note - this check must be performed while locked to prevent a reader from backwards-seeking into the
            // write region between here and the writeEndCursor update below.
            backwardSeekLock.lock();
            if ((writeEnd >= header->oldestUnconsumedCursor) &&
                ((writeEnd - header->oldestUnconsumedCursor) > m_bufferLayout->getDataSize())) {
                return Error::WOULDBLOCK;
            }
            break;
        case Policy::BLOCKING:
            // For BLOCKING, we need to wait until there is room for at least one word.

            // Condition for returning from write: there is space for a write.
            auto predicate = [this, header] {
                return (header->writeStartCursor < header->oldestUnconsumedCursor) ||
                       (header->writeStartCursor - header->oldestUnconsumedCursor) < m_bufferLayout->getDataSize();
            };

            // Note - this check must be performed while locked to prevent a reader from backwards-seeking into the
            // write region between here and the writeEndCursor update below.
            backwardSeekLock.lock();

            // Wait for space to become available.
            if (std::chrono::milliseconds::zero() == timeout) {
                header->spaceAvailableConditionVariable.wait(backwardSeekLock, predicate);
            } else if (!header->spaceAvailableConditionVariable.wait_for(backwardSeekLock, timeout, predicate)) {
                return Error::TIMEDOUT;
            }

            // Figure out how much space we have.
            auto spaceAvailable = m_bufferLayout->getDataSize();
            if (header->writeStartCursor >= header->oldestUnconsumedCursor) {
                auto wordsToOverrun =
                    m_bufferLayout->getDataSize() - (header->writeStartCursor - header->oldestUnconsumedCursor);
                if (wordsToOverrun < spaceAvailable) {
                    spaceAvailable = wordsToOverrun;
                }
            }

            // For BLOCKING, we can truncate the write if it won't fit in the buffer.
            if (spaceAvailable < nWords) {
                wordsToCopy = nWords = spaceAvailable;
                writeEnd = header->writeStartCursor + nWords;
            }

            break;
    }

    header->writeEndCursor = writeEnd;

    // We've updated our end cursor, so we no longer need to hold off backward seeks.
    if (backwardSeekLock) {
        backwardSeekLock.unlock();
    }

    if (Policy::ALL_OR_NOTHING == m_policy) {
        // If we have more data than the SDS can hold and we're not going to be overwriting oldestUnconsumedCursor, we
        // can safely discard the initial data and just leave the trailing data in the buffer.
        if (wordsToCopy > m_bufferLayout->getDataSize()) {
            wordsToCopy = m_bufferLayout->getDataSize();
            buf8 += (nWords - wordsToCopy) * getWordSize();
        }
    }

    // Split it across the wrap.
    size_t beforeWrap = m_bufferLayout->wordsUntilWrap(header->writeStartCursor);
    if (beforeWrap > wordsToCopy) {
        beforeWrap = wordsToCopy;
    }
    size_t afterWrap = wordsToCopy - beforeWrap;

    // Copy the two segments.
    memcpy(m_bufferLayout->getData(header->writeStartCursor), buf8, beforeWrap * getWordSize());
    if (afterWrap > 0) {
        memcpy(
            m_bufferLayout->getData(header->writeStartCursor + beforeWrap),
            buf8 + beforeWrap * getWordSize(),
            afterWrap * getWordSize());
    }

    // Advance the write cursor.
    // Note: To prevent a race condition and ensure that readers which block on dataAvailableConditionVariable don't
    // miss a notify, we should always lock the dataAvailableConditionVariable mutex while moving writeStartCursor.  As
    // an optimization, we skip that lock for NONBLOCKABLE writers under the assumption that they will be writing
    // continuously, so a missed notification is not significant.
    // Note: As a further optimization, the lock could be omitted if no blocking readers are in use (ACSDK-251).
    std::unique_lock<Mutex> dataAvailableLock(header->dataAvailableMutex, std::defer_lock);
    if (Policy::NONBLOCKABLE != m_policy) {
        dataAvailableLock.lock();
    }
    header->writeStartCursor = header->writeEndCursor.load();
    if (Policy::NONBLOCKABLE != m_policy) {
        dataAvailableLock.unlock();
    }

    // Notify the reader(s).
    // Note: as an optimization, we could skip this if there are no blocking readers (ACSDK-251).
    header->dataAvailableConditionVariable.notify_all();

    return nWords;
}

template <typename T>
typename SharedDataStream<T>::Index SharedDataStream<T>::Writer::tell() const {
    return m_bufferLayout->getHeader()->writeStartCursor;
}

template <typename T>
void SharedDataStream<T>::Writer::close() {
    auto header = m_bufferLayout->getHeader();
    std::lock_guard<Mutex> lock(header->writerEnableMutex);
    if (m_closed) {
        return;
    }
    if (header->isWriterEnabled) {
        header->isWriterEnabled = false;

        std::unique_lock<Mutex> dataAvailableLock(header->dataAvailableMutex);

        header->hasWriterBeenClosed = true;

        header->dataAvailableConditionVariable.notify_all();
    }
    m_closed = true;
}

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

template <typename T>
std::string SharedDataStream<T>::Writer::errorToString(Error error) {
    switch (error) {
        case Error::CLOSED:
            return "CLOSED";
        case Error::WOULDBLOCK:
            return "WOULDBLOCK";
        case Error::INVALID:
            return "INVALID";
        case Error::TIMEDOUT:
            return "TIMEDOUT";
    }
    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_WRITER_H_