xref: /external/pigweed/pw_multisink/public/pw_multisink/multisink.h

// Copyright 2021 The Pigweed Authors
//
// Licensed under the Apache License, Version 2.0 (the "License"); you may not
// use this file except in compliance with the License. You may obtain a copy of
// the License at
//
//     https://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License 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.
#pragma once

#include <limits>
#include <mutex>

#include "pw_assert/assert.h"
#include "pw_bytes/span.h"
#include "pw_function/function.h"
#include "pw_multisink/config.h"
#include "pw_result/result.h"
#include "pw_ring_buffer/prefixed_entry_ring_buffer.h"
#include "pw_status/status.h"
#include "pw_sync/lock_annotations.h"

namespace pw {
namespace multisink {

// An asynchronous single-writer multi-reader queue that ensures readers can
// poll for dropped message counts, which is useful for logging or similar
// scenarios where readers need to be aware of the input message sequence.
//
// This class is thread-safe but NOT IRQ-safe when
// PW_MULTISINK_LOCK_INTERRUPT_SAFE is disabled.
class MultiSink {
 public:
  // An asynchronous reader which is attached to a MultiSink via AttachDrain.
  // Each Drain holds a PrefixedEntryRingBufferMulti::Reader and abstracts away
  // entry sequence information for clients when popping.
  class Drain {
   public:
    // Holds the context for a peeked entry, tha the user may pass to `PopEntry`
    // to advance the drain.
    class PeekedEntry {
     public:
      // Provides access to the peeked entry's data.
      ConstByteSpan entry() const { return entry_; }

     private:
      friend MultiSink;
      friend MultiSink::Drain;

      constexpr PeekedEntry(ConstByteSpan entry, uint32_t sequence_id)
          : entry_(entry), sequence_id_(sequence_id) {}

      uint32_t sequence_id() const { return sequence_id_; }

      const ConstByteSpan entry_;
      const uint32_t sequence_id_;
    };

    constexpr Drain()
        : last_handled_sequence_id_(0),
          last_peek_sequence_id_(0),
          last_handled_ingress_drop_count_(0),
          multisink_(nullptr) {}

    // Returns the next available entry if it exists and acquires the latest
    // drop count in parallel.
    //
    // If the read operation was successful or returned OutOfRange (i.e. no
    // entries to read) then the `drain_drop_count_out` is set to the number of
    // entries that were dropped since the last call to PopEntry due to
    // advancing the drain, and `ingress_drop_count_out` is set to the number of
    // logs that were dropped before being added to the MultiSink. Otherwise,
    // the drop counts are set to zero, so should always be processed.
    //
    // Drop counts are internally maintained with a 32-bit counter. If
    // UINT32_MAX entries have been handled by the attached multisink between
    // subsequent calls to PopEntry, the drop count will overflow and will
    // report a lower count erroneously. Users should ensure that sinks call
    // PopEntry at least once every UINT32_MAX entries.
    //
    // Example Usage:
    //
    // void ProcessEntriesFromDrain(Drain& drain) {
    //   std::array<std::byte, kEntryBufferSize> buffer;
    //   uint32_t drop_count = 0;
    //
    //   // Example#1: Request the drain for a new entry.
    //   {
    //     const Result<ConstByteSpan> result = drain.PopEntry(buffer,
    //                                                         drop_count);
    //
    //     // If a non-zero drop count is received, process them.
    //     if (drop_count > 0) {
    //       ProcessDropCount(drop_count);
    //     }
    //
    //     // If the call was successful, process the entry that was received.
    //     if (result.ok()) {
    //       ProcessEntry(result.value());
    //     }
    //   }
    //
    //   // Example#2: Drain out all messages.
    //   {
    //     Result<ConstByteSpan> result = Status::OutOfRange();
    //     do {
    //       result = drain.PopEntry(buffer, drop_count);
    //
    //       if (drop_count > 0) {
    //         ProcessDropCount(drop_count);
    //       }
    //
    //       if (result.ok()) {
    //         ProcessEntry(result.value());
    //       }
    //
    //       // Keep trying until we hit OutOfRange. Note that a new entry may
    //       // have arrived after the PopEntry call.
    //     } while (!result.IsOutOfRange());
    //   }
    // }
    // Precondition: the buffer data must not be corrupt, otherwise there will
    // be a crash.
    //
    // Return values:
    // OK - An entry was successfully read from the multisink.
    // OUT_OF_RANGE - No entries were available.
    // FAILED_PRECONDITION - The drain must be attached to a sink.
    // RESOURCE_EXHAUSTED - The provided buffer was not large enough to store
    // the next available entry, which was discarded.
    Result<ConstByteSpan> PopEntry(ByteSpan buffer,
                                   uint32_t& drain_drop_count_out,
                                   uint32_t& ingress_drop_count_out)
        PW_LOCKS_EXCLUDED(multisink_->lock_);
    // Overload that combines drop counts.
    // TODO(cachinchilla): remove when downstream projects migrated to new API.
    [[deprecated("Use PopEntry with different drop count outputs")]] Result<
        ConstByteSpan>
    PopEntry(ByteSpan buffer, uint32_t& drop_count_out)
        PW_LOCKS_EXCLUDED(multisink_->lock_) {
      uint32_t ingress_drop_count = 0;
      Result<ConstByteSpan> result =
          PopEntry(buffer, drop_count_out, ingress_drop_count);
      drop_count_out += ingress_drop_count;
      return result;
    }

    // Removes the previously peeked entry from the multisink.
    //
    // Example Usage:
    //
    //  // Peek entry to send it, and remove entry from multisink on success.
    //  uint32_t drop_count;
    //  const Result<PeekedEntry> peek_result =
    //      PeekEntry(out_buffer, drop_count);
    //  if (!peek_result.ok()) {
    //    return peek_result.status();
    //  }
    //  Status send_status = UserSendFunction(peek_result.value().entry())
    //  if (!send_status.ok()) {
    //    return send_status;
    //  }
    //  PW_CHECK_OK(PopEntry(peek_result.value());
    //
    // Precondition: the buffer data must not be corrupt, otherwise there will
    // be a crash.
    //
    // Return values:
    // OK - the entry or entries were removed from the multisink successfully.
    // FAILED_PRECONDITION - The drain must be attached to a sink.
    Status PopEntry(const PeekedEntry& entry)
        PW_LOCKS_EXCLUDED(multisink_->lock_);

    // Returns a copy of the next available entry if it exists and acquires the
    // latest drop count if the drain was advanced, and the latest ingress drop
    // count, without moving the drain forward, except if there is a
    // RESOURCE_EXHAUSTED error when peeking, in which case the drain is
    // automatically advanced.
    // The `drain_drop_count_out` follows the same logic as `PopEntry`. The user
    // must call `PopEntry` once the data in peek was used successfully.
    //
    // Precondition: the buffer data must not be corrupt, otherwise there will
    // be a crash.
    //
    // Return values:
    // OK - An entry was successfully read from the multisink.
    // OUT_OF_RANGE - No entries were available.
    // FAILED_PRECONDITION - The drain must be attached to a sink.
    // RESOURCE_EXHAUSTED - The provided buffer was not large enough to store
    // the next available entry, which was discarded.
    Result<PeekedEntry> PeekEntry(ByteSpan buffer,
                                  uint32_t& drain_drop_count_out,
                                  uint32_t& ingress_drop_count_out)
        PW_LOCKS_EXCLUDED(multisink_->lock_);

    // Drains are not copyable or movable.
    Drain(const Drain&) = delete;
    Drain& operator=(const Drain&) = delete;
    Drain(Drain&&) = delete;
    Drain& operator=(Drain&&) = delete;

    // Returns size of unread entries in the sink for this drain. This is a
    // thread-safe version and must not be used inside a
    // Listener::OnNewEntryAvailable() to avoid deadlocks. Use
    // UnsafeGetUnreadEntriesSize() instead in such case.
    size_t GetUnreadEntriesSize() const PW_LOCKS_EXCLUDED(multisink_->lock_) {
      std::lock_guard lock(multisink_->lock_);
      return UnsafeGetUnreadEntriesSize();
    }

    // Returns size of unread entries in the sink for this drain.
    // Marked unsafe because it requires external synchronization.
    size_t UnsafeGetUnreadEntriesSize() const PW_NO_LOCK_SAFETY_ANALYSIS {
      return reader_.EntriesSize();
    }

   protected:
    friend MultiSink;

    // The `reader_` and `last_handled_sequence_id_` are managed by attached
    // multisink and are guarded by `multisink_->lock_` when used.
    ring_buffer::PrefixedEntryRingBufferMulti::Reader reader_;
    uint32_t last_handled_sequence_id_;
    uint32_t last_peek_sequence_id_;
    uint32_t last_handled_ingress_drop_count_;
    MultiSink* multisink_;
  };

  // A pure-virtual listener of a MultiSink, attached via AttachListener.
  // MultiSink's invoke listeners when new data arrives, allowing them to
  // schedule the draining of messages out of the MultiSink.
  class Listener : public IntrusiveList<Listener>::Item {
   public:
    constexpr Listener() {}
    virtual ~Listener() = default;

    // Listeners are not copyable or movable.
    Listener(const Listener&) = delete;
    Listener& operator=(const Drain&) = delete;
    Listener(Listener&&) = delete;
    Listener& operator=(Drain&&) = delete;

   protected:
    friend MultiSink;

    // Invoked by the attached multisink when a new entry or drop count is
    // available. The multisink lock is held during this call, so neither the
    // multisink nor it's drains can be used during this callback.
    virtual void OnNewEntryAvailable() = 0;
  };

  class iterator {
   public:
    iterator& operator++() {
      it_++;
      return *this;
    }
    iterator operator++(int) {
      iterator original = *this;
      ++*this;
      return original;
    }

    ConstByteSpan& operator*() {
      entry_ = (*it_).buffer;
      return entry_;
    }
    ConstByteSpan* operator->() { return &operator*(); }

    constexpr bool operator==(const iterator& rhs) const {
      return it_ == rhs.it_;
    }

    constexpr bool operator!=(const iterator& rhs) const {
      return it_ != rhs.it_;
    }

    // Returns the status of the last iteration operation. If the iterator
    // fails to read an entry, it will move to iterator::end() and indicate
    // the failure reason here.
    //
    // Return values:
    // OK - iteration is successful and iterator points to the next entry.
    // DATA_LOSS - Failed to read the metadata at this location.
    Status status() const { return it_.status(); }

   private:
    friend class MultiSink;

    iterator(ring_buffer::PrefixedEntryRingBufferMulti::Reader& reader)
        : it_(reader) {}
    iterator() {}

    ring_buffer::PrefixedEntryRingBufferMulti::iterator it_;
    ConstByteSpan entry_;
  };

  class UnsafeIterationWrapper {
   public:
    using element_type = ConstByteSpan;
    using value_type = std::remove_cv_t<ConstByteSpan>;
    using pointer = ConstByteSpan*;
    using reference = ConstByteSpan&;
    using const_iterator = iterator;  // Standard alias for iterable types.

    iterator begin() const { return iterator(*reader_); }
    iterator end() const { return iterator(); }
    const_iterator cbegin() const { return begin(); }
    const_iterator cend() const { return end(); }

   private:
    friend class MultiSink;
    UnsafeIterationWrapper(
        ring_buffer::PrefixedEntryRingBufferMulti::Reader& reader)
        : reader_(&reader) {}
    ring_buffer::PrefixedEntryRingBufferMulti::Reader* reader_;
  };

  UnsafeIterationWrapper UnsafeIteration() PW_NO_LOCK_SAFETY_ANALYSIS {
    return UnsafeIterationWrapper(oldest_entry_drain_.reader_);
  }

  // Constructs a multisink using a ring buffer backed by the provided buffer.
  // If we're using a virtual lock, then the lock needs to be passed in.
#if PW_MULTISINK_CONFIG_LOCK_TYPE == PW_MULTISINK_VIRTUAL_LOCK
  MultiSink(ByteSpan buffer, LockType lock)
      : lock_(lock),
#else
  MultiSink(ByteSpan buffer)
      :
#endif
        ring_buffer_(true),
        sequence_id_(0),
        total_ingress_drops_(0) {
    PW_ASSERT(ring_buffer_.SetBuffer(buffer).ok());
    AttachDrain(oldest_entry_drain_);
  }

  // Write an entry to the multisink. If available space is less than the
  // size of the entry, the internal ring buffer will push the oldest entries
  // out to make space, so long as the entry is not larger than the buffer.
  // The sequence ID of the multisink will always increment as a result of
  // calling HandleEntry, regardless of whether pushing the entry succeeds.
  //
  // Precondition: If PW_MULTISINK_LOCK_INTERRUPT_SAFE is disabled, this
  // function must not be called from an interrupt context.
  // Precondition: entry.size() <= `ring_buffer_` size
  void HandleEntry(ConstByteSpan entry) PW_LOCKS_EXCLUDED(lock_);

  // Notifies the multisink of messages dropped before ingress. The writer
  // may use this to signal to readers that an entry (or entries) failed
  // before being sent to the multisink (e.g. the writer failed to encode
  // the message). This API increments the sequence ID of the multisink by
  // the provided `drop_count`.
  void HandleDropped(uint32_t drop_count = 1) PW_LOCKS_EXCLUDED(lock_);

  // Attach a drain to the multisink. Drains may not be associated with more
  // than one multisink at a time. Drains can consume entries pushed before
  // the drain was attached, so long as they have not yet been evicted from
  // the underlying ring buffer.
  //
  // Precondition: The drain must not be attached to a multisink.
  void AttachDrain(Drain& drain) PW_LOCKS_EXCLUDED(lock_);

  // Detaches a drain from the multisink. Drains may only be detached if they
  // were previously attached to this multisink.
  //
  // Precondition: The drain must be attached to this multisink.
  void DetachDrain(Drain& drain) PW_LOCKS_EXCLUDED(lock_);

  // Attach a listener to the multisink. The listener will be notified
  // immediately when attached, to allow late drain users to consume existing
  // entries. If draining in response to the notification, ensure that the drain
  // is attached prior to registering the listener; attempting to drain when
  // unattached will crash. Once attached, listeners are invoked on all new
  // messages.
  //
  // Precondition: The listener must not be attached to a multisink.
  void AttachListener(Listener& listener) PW_LOCKS_EXCLUDED(lock_);

  // Detaches a listener from the multisink.
  //
  // Precondition: The listener must be attached to this multisink.
  void DetachListener(Listener& listener) PW_LOCKS_EXCLUDED(lock_);

  // Removes all data from the internal buffer. The multisink's sequence ID is
  // not modified, so readers may interpret this event as droppping entries.
  void Clear() PW_LOCKS_EXCLUDED(lock_);

  // Uses MultiSink's unsafe iteration to dump the contents to a user-provided
  // callback. max_num_entries can be used to limit the dump to the N most
  // recent entries.
  //
  // Returns:
  //   OK - Successfully dumped entire multisink.
  //   DATA_LOSS - Corruption detected, some entries may have been lost.
  Status UnsafeForEachEntry(
      const Function<void(ConstByteSpan)>& callback,
      size_t max_num_entries = std::numeric_limits<size_t>::max());

 protected:
  friend Drain;

  enum class Request { kPop, kPeek };
  // Removes the previously peeked entry from the front of the multisink.
  Status PopEntry(Drain& drain, const Drain::PeekedEntry& entry)
      PW_LOCKS_EXCLUDED(lock_);

  // Gets a copy of the entry from the provided drain and unpacks sequence ID
  // information. The entry is removed from the multisink when `request` is set
  // to `Request::kPop`. Drains use this API to strip away sequence ID
  // information for drop calculation.
  //
  // Precondition: the buffer data must not be corrupt, otherwise there will
  // be a crash.
  //
  // Returns:
  // OK - An entry was successfully read from the multisink. The
  // `drain_drop_count_out` is set to the difference between the current
  // sequence ID and the last handled ID.
  // FAILED_PRECONDITION - The drain is not attached to
  // a multisink.
  // RESOURCE_EXHAUSTED - The provided buffer was not large enough to store
  // the next available entry, which was discarded.
  Result<ConstByteSpan> PeekOrPopEntry(Drain& drain,
                                       ByteSpan buffer,
                                       Request request,
                                       uint32_t& drain_drop_count_out,
                                       uint32_t& ingress_drop_count_out,
                                       uint32_t& entry_sequence_id_out)
      PW_LOCKS_EXCLUDED(lock_);

 private:
  // Notifies attached listeners of new entries or an updated drop count.
  void NotifyListeners() PW_EXCLUSIVE_LOCKS_REQUIRED(lock_);

  LockType lock_;
  IntrusiveList<Listener> listeners_ PW_GUARDED_BY(lock_);
  ring_buffer::PrefixedEntryRingBufferMulti ring_buffer_ PW_GUARDED_BY(lock_);
  Drain oldest_entry_drain_ PW_GUARDED_BY(lock_);
  uint32_t sequence_id_ PW_GUARDED_BY(lock_);
  uint32_t total_ingress_drops_ PW_GUARDED_BY(lock_);
};

}  // namespace multisink
}  // namespace pw