mpsc_stream.h

Go to the documentation of this file.

// Copyright 2023 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 <cstddef>
 
#include "pw_bytes/span.h"
#include "pw_chrono/system_clock.h"
#include "pw_containers/intrusive_list.h"
#include "pw_function/function.h"
#include "pw_status/status.h"
#include "pw_status/status_with_size.h"
#include "pw_stream/stream.h"
#include "pw_sync/lock_annotations.h"
#include "pw_sync/mutex.h"
#include "pw_sync/timed_thread_notification.h"
 
namespace pw::stream {
 
// Forward declaration.
class MpscReader;
class MpscWriter;
 
 
void CreateMpscStream(MpscReader& reader, MpscWriter& writer);
 
 
 
class MpscWriter : public NonSeekableWriter,
                   public IntrusiveList<MpscWriter>::Item {
 public:
  using duration = std::optional<chrono::SystemClock::duration>;
 
  struct Request : public IntrusiveList<Request>::Item {
    sync::TimedThreadNotification notification;
    using IntrusiveList<Request>::Item::unlisted;
  };
 
  MpscWriter() = default;
  MpscWriter(const MpscWriter& other);
  MpscWriter& operator=(const MpscWriter& other);
  MpscWriter(MpscWriter&& other);
  MpscWriter& operator=(MpscWriter&& other);
  ~MpscWriter() override;
 
  bool connected() const PW_LOCKS_EXCLUDED(mutex_);
 
  size_t last_write() const PW_LOCKS_EXCLUDED(mutex_);
 
  const duration& timeout() const PW_LOCKS_EXCLUDED(mutex_);
 
  void SetTimeout(const duration& timeout) PW_LOCKS_EXCLUDED(mutex_);
 
  void SetLimit(size_t limit) PW_LOCKS_EXCLUDED(mutex_);
 
  void Close() PW_LOCKS_EXCLUDED(mutex_);
 
 private:
  // The factory method is allowed to directly modify a writer to connect it
  // to the reader.
  friend void CreateMpscStream(MpscReader&, MpscWriter&);
 
  size_t ConservativeLimit(LimitType type) const override;
 
  Status DoWrite(ConstByteSpan data) override;
 
  void CloseLocked() PW_EXCLUSIVE_LOCKS_REQUIRED(mutex_);
 
  mutable sync::Mutex mutex_;
  MpscReader* reader_ PW_GUARDED_BY(mutex_) = nullptr;
  size_t limit_ PW_GUARDED_BY(mutex_) = kUnlimited;
  Request write_request_;
  duration timeout_ PW_GUARDED_BY(mutex_);
  size_t last_write_ PW_GUARDED_BY(mutex_) = 0;
};
 
 
 
class MpscReader : public NonSeekableReader {
 public:
  using duration = std::optional<chrono::SystemClock::duration>;
 
  MpscReader();
  ~MpscReader() override;
 
  bool connected() const PW_LOCKS_EXCLUDED(mutex_);
 
  void SetTimeout(const duration& timeout) PW_LOCKS_EXCLUDED(mutex_);
 
  void SetBuffer(ByteSpan buffer) PW_LOCKS_EXCLUDED(mutex_);
 
  using ReadAllCallback = Function<Status(ConstByteSpan data)>;
  Status ReadAll(ReadAllCallback callback) PW_LOCKS_EXCLUDED(mutex_);
 
  void Close() PW_LOCKS_EXCLUDED(mutex_);
 
 private:
  // The factory method is allowed to directly modify the reader to connect it
  // to a writer.
  friend void CreateMpscStream(MpscReader&, MpscWriter&);
 
  // The writer is allowed to call directly into the reader to:
  //  * Add/remove itself to the reader's list of writer.
  //  * Request space to write data, and to write to that space.
  friend class MpscWriter;
 
  void IncreaseLimit(size_t delta) PW_LOCKS_EXCLUDED(mutex_);
  void IncreaseLimitLocked(size_t delta) PW_EXCLUSIVE_LOCKS_REQUIRED(mutex_);
 
  void DecreaseLimit(size_t delta) PW_LOCKS_EXCLUDED(mutex_);
  void DecreaseLimitLocked(size_t delta) PW_EXCLUSIVE_LOCKS_REQUIRED(mutex_);
 
  size_t ConservativeLimit(Stream::LimitType type) const override
      PW_LOCKS_EXCLUDED(mutex_);
 
  void RequestWrite(MpscWriter::Request& write_request)
      PW_LOCKS_EXCLUDED(mutex_);
 
  void CheckWriteableLocked() PW_EXCLUSIVE_LOCKS_REQUIRED(mutex_);
 
  StatusWithSize WriteData(ConstByteSpan data, size_t limit)
      PW_LOCKS_EXCLUDED(mutex_);
 
  void CompleteWrite(MpscWriter::Request& write_request)
      PW_LOCKS_EXCLUDED(mutex_);
  void CompleteWriteLocked(MpscWriter::Request& write_request)
      PW_EXCLUSIVE_LOCKS_REQUIRED(mutex_);
 
  StatusWithSize DoRead(ByteSpan destination) override
      PW_LOCKS_EXCLUDED(mutex_);
 
  // Locked implementations.
 
  mutable sync::Mutex mutex_;
  IntrusiveList<MpscWriter> writers_ PW_GUARDED_BY(mutex_);
  IntrusiveList<MpscWriter::Request> write_requests_ PW_GUARDED_BY(mutex_);
  IntrusiveList<MpscWriter::Request>::iterator last_request_
      PW_GUARDED_BY(mutex_);
 
  size_t num_unlimited_ PW_GUARDED_BY(mutex_) = 0;
  size_t limit_ PW_GUARDED_BY(mutex_) = 0;
 
  bool reading_ PW_GUARDED_BY(mutex_) = false;
  sync::TimedThreadNotification readable_;
  sync::ThreadNotification closeable_;
  duration timeout_ PW_GUARDED_BY(mutex_);
 
  ByteSpan destination_ PW_GUARDED_BY(mutex_);
  size_t written_ PW_GUARDED_BY(mutex_) = 0;
 
  ByteSpan buffer_ PW_GUARDED_BY(mutex_);
  size_t offset_ PW_GUARDED_BY(mutex_) = 0;
  size_t length_ PW_GUARDED_BY(mutex_) = 0;
};
 
template <size_t kCapacity>
class BufferedMpscReader : public MpscReader {
 public:
  BufferedMpscReader() { SetBuffer(buffer_); }
 
 private:
  std::array<std::byte, kCapacity> buffer_;
};
 
 
}  // namespace pw::stream