pw_transfer/public/pw_transfer/internal/context.h - pigweed/pigweed - Git at Google

 // 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 <cinttypes>
 #include <cstddef>
 #include <limits>

 #include "pw_assert/assert.h"
 #include "pw_chrono/system_timer.h"
 #include "pw_rpc/writer.h"
 #include "pw_status/status.h"
 #include "pw_stream/stream.h"
 #include "pw_sync/interrupt_spin_lock.h"
 #include "pw_sync/lock_annotations.h"
 #include "pw_transfer/internal/chunk.h"
 #include "pw_transfer/internal/chunk_data_buffer.h"
 #include "pw_transfer/internal/config.h"
 #include "pw_work_queue/work_queue.h"

 namespace pw::transfer::internal {

 class TransferParameters {
  public:
   constexpr TransferParameters(uint32_t pending_bytes,
                                uint32_t max_chunk_size_bytes)
       : pending_bytes_(pending_bytes),
         max_chunk_size_bytes_(max_chunk_size_bytes) {
     PW_ASSERT(pending_bytes > 0);
     PW_ASSERT(max_chunk_size_bytes > 0);
   }

   uint32_t pending_bytes() const { return pending_bytes_; }
   uint32_t max_chunk_size_bytes() const { return max_chunk_size_bytes_; }

   // The fractional position within a window at which a receive transfer should
   // extend its window size to minimize the amount of time the transmitter
   // spends blocked.
   //
   // For example, a divisor of 2 will extend the window when half of the
   // requested data has been received, a divisor of three will extend at a third
   // of the window, and so on.
   //
   // TODO(frolv): Find a good threshold for this; maybe make it configurable?
   static constexpr uint32_t kExtendWindowDivisor = 2;
   static_assert(kExtendWindowDivisor > 1);

  private:
   uint32_t pending_bytes_;
   uint32_t max_chunk_size_bytes_;
 };

 // Information about a single transfer.
 class Context {
  public:
   enum Type : bool { kTransmit, kReceive };

   Context(const Context&) = delete;
   Context(Context&&) = delete;
   Context& operator=(const Context&) = delete;
   Context& operator=(Context&&) = delete;

   constexpr uint32_t transfer_id() const { return transfer_id_; }

   // True if the context has been used for a transfer (it has an ID).
   bool initialized() {
     state_lock_.lock();
     bool initialized = transfer_state_ != TransferState::kInactive;
     state_lock_.unlock();
     return initialized;
   }

   // True if the transfer is active.
   bool active() {
     state_lock_.lock();
     bool active = transfer_state_ >= TransferState::kData;
     state_lock_.unlock();
     return active;
   }

   // Starts a new transfer from an initialized context by sending the initial
   // transfer chunk. This is generally called from within a transfer client, as
   // it is unusual for a server to initiate a transfer.
   Status InitiateTransfer(const TransferParameters& max_parameters);

   // Extracts data from the provided chunk into the transfer context. This is
   // intended to be the immediate part of the transfer, run directly from within
   // the RPC message handler.
   //
   // Returns true if there is any deferred work required for this chunk (i.e.
   // ProcessChunk should be called to complete the operation).
   bool ReadChunkData(ChunkDataBuffer& buffer,
                      const TransferParameters& max_parameters,
                      const Chunk& chunk);

   // Handles the chunk from the previous invocation of ReadChunkData(). This
   // operation is intended to be deferred, running from a different context than
   // the RPC callback in which the chunk was received.
   void ProcessChunk(ChunkDataBuffer& buffer,
                     const TransferParameters& max_parameters);

  protected:
   using CompletionFunction = Status (*)(Context&, Status);

   Context(CompletionFunction on_completion)
       : transfer_id_(0),
         flags_(0),
         transfer_state_(TransferState::kInactive),
         retries_(0),
         max_retries_(0),
         stream_(nullptr),
         rpc_writer_(nullptr),
         offset_(0),
         window_size_(0),
         window_end_offset_(0),
         pending_bytes_(0),
         max_chunk_size_bytes_(std::numeric_limits<uint32_t>::max()),
         last_chunk_offset_(0),
         timer_([this](chrono::SystemClock::time_point) { this->OnTimeout(); }),
         chunk_timeout_(chrono::SystemClock::duration::zero()),
         work_queue_(nullptr),
         on_completion_(on_completion) {}

   enum class TransferState : uint8_t {
     // This ServerContext has never been used for a transfer. It is available
     // for use for a transfer.
     kInactive,
     // A transfer completed and the final status chunk was sent. The Context is
     // available for use for a new transfer. A receive transfer uses this state
     // to allow a transmitter to retry its last chunk if the final status chunk
     // was dropped.
     kCompleted,
     // Sending or receiving data for an active transfer.
     kData,
     // Recovering after one or more chunks was dropped in an active transfer.
     kRecovery,
     // Hit a timeout and waiting for the timeout handler to run.
     kTimedOut,
   };

   constexpr Type type() const { return static_cast<Type>(flags_ & kFlagsType); }

   void set_transfer_state(TransferState state) {
     state_lock_.lock();
     transfer_state_ = state;
     state_lock_.unlock();
   }

   // Begins a new transmit transfer from this context.
   // Precondition: context is not active.
   void InitializeForTransmit(
       uint32_t transfer_id,
       work_queue::WorkQueue& work_queue,
       rpc::Writer& rpc_writer,
       stream::Reader& reader,
       chrono::SystemClock::duration chunk_timeout = cfg::kDefaultChunkTimeout,
       uint8_t max_retries = cfg::kDefaultMaxRetries) {
     Initialize(kTransmit,
                transfer_id,
                work_queue,
                rpc_writer,
                reader,
                chunk_timeout,
                max_retries);
   }

   // Begins a new receive transfer from this context.
   // Precondition: context is not active.
   void InitializeForReceive(
       uint32_t transfer_id,
       work_queue::WorkQueue& work_queue,
       rpc::Writer& rpc_writer,
       stream::Writer& writer,
       chrono::SystemClock::duration chunk_timeout = cfg::kDefaultChunkTimeout,
       uint8_t max_retries = cfg::kDefaultMaxRetries) {
     Initialize(kReceive,
                transfer_id,
                work_queue,
                rpc_writer,
                writer,
                chunk_timeout,
                max_retries);
   }

   // Calculates the maximum size of actual data that can be sent within a single
   // client write transfer chunk, accounting for the overhead of the transfer
   // protocol and RPC system.
   //
   // Note: This function relies on RPC protocol internals. This is generally a
   // *bad* idea, but is necessary here due to limitations of the RPC system and
   // its asymmetric ingress and egress paths.
   //
   // TODO(frolv): This should be investigated further and perhaps addressed
   // within the RPC system, at the least through a helper function.
   uint32_t MaxWriteChunkSize(uint32_t max_chunk_size_bytes,
                              uint32_t channel_id) const;

  private:
   enum TransmitAction : bool { kExtend, kRetransmit };

   void Initialize(Type type,
                   uint32_t transfer_id,
                   work_queue::WorkQueue& work_queue,
                   rpc::Writer& rpc_writer,
                   stream::Stream& stream,
                   chrono::SystemClock::duration chunk_timeout,
                   uint8_t max_retries);

   stream::Reader& reader() {
     PW_DASSERT(active() && type() == kTransmit);
     return static_cast<stream::Reader&>(*stream_);
   }

   stream::Writer& writer() {
     PW_DASSERT(active() && type() == kReceive);
     return static_cast<stream::Writer&>(*stream_);
   }

   // Sends the first chunk in a transmit transfer.
   Status SendInitialTransmitChunk();

   // Functions which extract relevant data from a chunk into the context.
   bool ReadTransmitChunk(const TransferParameters& max_parameters,
                          const Chunk& chunk);
   bool ReadReceiveChunk(ChunkDataBuffer& buffer,
                         const TransferParameters& max_parameters,
                         const Chunk& chunk);

   // Functions which handle the last received chunk.
   void ProcessTransmitChunk();
   void ProcessReceiveChunk(ChunkDataBuffer& buffer,
                            const TransferParameters& max_parameters);

   // In a transmit transfer, sends the next data chunk from the local stream.
   // Returns status indicating what to do next:
   //
   //    OK - continue
   //    OUT_OF_RANGE - done for now
   //    other errors - abort transfer with this error
   //
   Status SendNextDataChunk();

   // In a receive transfer, processes the fields from a data chunk and stages
   // the data for a deferred write. Returns true if there is a deferred
   // operation to complete.
   bool HandleDataChunk(ChunkDataBuffer& buffer,
                        const TransferParameters& max_parameters,
                        const Chunk& chunk);

   // In a receive transfer, sends a parameters chunk telling the transmitter how
   // much data they can send.
   Status SendTransferParameters(TransmitAction action);

   // Updates the current receive transfer parameters from the provided object,
   // then sends them.
   Status UpdateAndSendTransferParameters(
       const TransferParameters& max_parameters, TransmitAction action);

   void SendStatusChunk(Status status);
   void FinishAndSendStatus(Status status);

   void CancelTimer() {
     timer_.Cancel();
     retries_ = 0;
   }

   // Timeout function invoked from the timer context. This may occur in an
   // interrupt, so no real work can be done. Instead, sets state to timed out
   // and adds a job to run the timeout handler.
   void OnTimeout();

   // The acutal timeout handler, invoked from within the work queue.
   void HandleTimeout();

   static constexpr uint8_t kFlagsType = 1 << 0;
   static constexpr uint8_t kFlagsDataSent = 1 << 1;

   // TODO(frolv): Make this value configurable per transfer.
   static constexpr uint32_t kDefaultChunkDelayMicroseconds = 2000;

   uint32_t transfer_id_;
   uint8_t flags_;
   TransferState transfer_state_ PW_GUARDED_BY(state_lock_);
   uint8_t retries_;
   uint8_t max_retries_;

   sync::InterruptSpinLock state_lock_;

   stream::Stream* stream_;
   rpc::Writer* rpc_writer_;

   uint32_t offset_;
   uint32_t window_size_;
   uint32_t window_end_offset_;
   // TODO(pwbug/584): Remove pending_bytes in favor of window_end_offset.
   uint32_t pending_bytes_;
   uint32_t max_chunk_size_bytes_;

   union {
     Status status_;               // Used when state is kCompleted.
     uint32_t last_chunk_offset_;  // Used in states kData and kRecovery.
   };

   // Timer used to handle timeouts waiting for chunks.
   chrono::SystemTimer timer_;
   chrono::SystemClock::duration chunk_timeout_;
   work_queue::WorkQueue* work_queue_;

   CompletionFunction on_completion_;
 };

 }  // namespace pw::transfer::internal
	// 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 <cinttypes>
	#include <cstddef>
	#include <limits>

	#include "pw_assert/assert.h"
	#include "pw_chrono/system_timer.h"
	#include "pw_rpc/writer.h"
	#include "pw_status/status.h"
	#include "pw_stream/stream.h"
	#include "pw_sync/interrupt_spin_lock.h"
	#include "pw_sync/lock_annotations.h"
	#include "pw_transfer/internal/chunk.h"
	#include "pw_transfer/internal/chunk_data_buffer.h"
	#include "pw_transfer/internal/config.h"
	#include "pw_work_queue/work_queue.h"

	namespace pw::transfer::internal {

	class TransferParameters {
	public:
	constexpr TransferParameters(uint32_t pending_bytes,
	uint32_t max_chunk_size_bytes)
	: pending_bytes_(pending_bytes),
	max_chunk_size_bytes_(max_chunk_size_bytes) {
	PW_ASSERT(pending_bytes > 0);
	PW_ASSERT(max_chunk_size_bytes > 0);
	}

	uint32_t pending_bytes() const { return pending_bytes_; }
	uint32_t max_chunk_size_bytes() const { return max_chunk_size_bytes_; }

	// The fractional position within a window at which a receive transfer should
	// extend its window size to minimize the amount of time the transmitter
	// spends blocked.
	//
	// For example, a divisor of 2 will extend the window when half of the
	// requested data has been received, a divisor of three will extend at a third
	// of the window, and so on.
	//
	// TODO(frolv): Find a good threshold for this; maybe make it configurable?
	static constexpr uint32_t kExtendWindowDivisor = 2;
	static_assert(kExtendWindowDivisor > 1);

	private:
	uint32_t pending_bytes_;
	uint32_t max_chunk_size_bytes_;
	};

	// Information about a single transfer.
	class Context {
	public:
	enum Type : bool { kTransmit, kReceive };

	Context(const Context&) = delete;
	Context(Context&&) = delete;
	Context& operator=(const Context&) = delete;
	Context& operator=(Context&&) = delete;

	constexpr uint32_t transfer_id() const { return transfer_id_; }

	// True if the context has been used for a transfer (it has an ID).
	bool initialized() {
	state_lock_.lock();
	bool initialized = transfer_state_ != TransferState::kInactive;
	state_lock_.unlock();
	return initialized;
	}

	// True if the transfer is active.
	bool active() {
	state_lock_.lock();
	bool active = transfer_state_ >= TransferState::kData;
	state_lock_.unlock();
	return active;
	}

	// Starts a new transfer from an initialized context by sending the initial
	// transfer chunk. This is generally called from within a transfer client, as
	// it is unusual for a server to initiate a transfer.
	Status InitiateTransfer(const TransferParameters& max_parameters);

	// Extracts data from the provided chunk into the transfer context. This is
	// intended to be the immediate part of the transfer, run directly from within
	// the RPC message handler.
	//
	// Returns true if there is any deferred work required for this chunk (i.e.
	// ProcessChunk should be called to complete the operation).
	bool ReadChunkData(ChunkDataBuffer& buffer,
	const TransferParameters& max_parameters,
	const Chunk& chunk);

	// Handles the chunk from the previous invocation of ReadChunkData(). This
	// operation is intended to be deferred, running from a different context than
	// the RPC callback in which the chunk was received.
	void ProcessChunk(ChunkDataBuffer& buffer,
	const TransferParameters& max_parameters);

	protected:
	using CompletionFunction = Status (*)(Context&, Status);

	Context(CompletionFunction on_completion)
	: transfer_id_(0),
	flags_(0),
	transfer_state_(TransferState::kInactive),
	retries_(0),
	max_retries_(0),
	stream_(nullptr),
	rpc_writer_(nullptr),
	offset_(0),
	window_size_(0),
	window_end_offset_(0),
	pending_bytes_(0),
	max_chunk_size_bytes_(std::numeric_limits<uint32_t>::max()),
	last_chunk_offset_(0),
	timer_([this](chrono::SystemClock::time_point) { this->OnTimeout(); }),
	chunk_timeout_(chrono::SystemClock::duration::zero()),
	work_queue_(nullptr),
	on_completion_(on_completion) {}

	enum class TransferState : uint8_t {
	// This ServerContext has never been used for a transfer. It is available
	// for use for a transfer.
	kInactive,
	// A transfer completed and the final status chunk was sent. The Context is
	// available for use for a new transfer. A receive transfer uses this state
	// to allow a transmitter to retry its last chunk if the final status chunk
	// was dropped.
	kCompleted,
	// Sending or receiving data for an active transfer.
	kData,
	// Recovering after one or more chunks was dropped in an active transfer.
	kRecovery,
	// Hit a timeout and waiting for the timeout handler to run.
	kTimedOut,
	};

	constexpr Type type() const { return static_cast<Type>(flags_ & kFlagsType); }

	void set_transfer_state(TransferState state) {
	state_lock_.lock();
	transfer_state_ = state;
	state_lock_.unlock();
	}

	// Begins a new transmit transfer from this context.
	// Precondition: context is not active.
	void InitializeForTransmit(
	uint32_t transfer_id,
	work_queue::WorkQueue& work_queue,
	rpc::Writer& rpc_writer,
	stream::Reader& reader,
	chrono::SystemClock::duration chunk_timeout = cfg::kDefaultChunkTimeout,
	uint8_t max_retries = cfg::kDefaultMaxRetries) {
	Initialize(kTransmit,
	transfer_id,
	work_queue,
	rpc_writer,
	reader,
	chunk_timeout,
	max_retries);
	}

	// Begins a new receive transfer from this context.
	// Precondition: context is not active.
	void InitializeForReceive(
	uint32_t transfer_id,
	work_queue::WorkQueue& work_queue,
	rpc::Writer& rpc_writer,
	stream::Writer& writer,
	chrono::SystemClock::duration chunk_timeout = cfg::kDefaultChunkTimeout,
	uint8_t max_retries = cfg::kDefaultMaxRetries) {
	Initialize(kReceive,
	transfer_id,
	work_queue,
	rpc_writer,
	writer,
	chunk_timeout,
	max_retries);
	}

	// Calculates the maximum size of actual data that can be sent within a single
	// client write transfer chunk, accounting for the overhead of the transfer
	// protocol and RPC system.
	//
	// Note: This function relies on RPC protocol internals. This is generally a
	// bad idea, but is necessary here due to limitations of the RPC system and
	// its asymmetric ingress and egress paths.
	//
	// TODO(frolv): This should be investigated further and perhaps addressed
	// within the RPC system, at the least through a helper function.
	uint32_t MaxWriteChunkSize(uint32_t max_chunk_size_bytes,
	uint32_t channel_id) const;

	private:
	enum TransmitAction : bool { kExtend, kRetransmit };

	void Initialize(Type type,
	uint32_t transfer_id,
	work_queue::WorkQueue& work_queue,
	rpc::Writer& rpc_writer,
	stream::Stream& stream,
	chrono::SystemClock::duration chunk_timeout,
	uint8_t max_retries);

	stream::Reader& reader() {
	PW_DASSERT(active() && type() == kTransmit);
	return static_cast<stream::Reader&>(*stream_);
	}

	stream::Writer& writer() {
	PW_DASSERT(active() && type() == kReceive);
	return static_cast<stream::Writer&>(*stream_);
	}

	// Sends the first chunk in a transmit transfer.
	Status SendInitialTransmitChunk();

	// Functions which extract relevant data from a chunk into the context.
	bool ReadTransmitChunk(const TransferParameters& max_parameters,
	const Chunk& chunk);
	bool ReadReceiveChunk(ChunkDataBuffer& buffer,
	const TransferParameters& max_parameters,
	const Chunk& chunk);

	// Functions which handle the last received chunk.
	void ProcessTransmitChunk();
	void ProcessReceiveChunk(ChunkDataBuffer& buffer,
	const TransferParameters& max_parameters);

	// In a transmit transfer, sends the next data chunk from the local stream.
	// Returns status indicating what to do next:
	//
	// OK - continue
	// OUT_OF_RANGE - done for now
	// other errors - abort transfer with this error
	//
	Status SendNextDataChunk();

	// In a receive transfer, processes the fields from a data chunk and stages
	// the data for a deferred write. Returns true if there is a deferred
	// operation to complete.
	bool HandleDataChunk(ChunkDataBuffer& buffer,
	const TransferParameters& max_parameters,
	const Chunk& chunk);

	// In a receive transfer, sends a parameters chunk telling the transmitter how
	// much data they can send.
	Status SendTransferParameters(TransmitAction action);

	// Updates the current receive transfer parameters from the provided object,
	// then sends them.
	Status UpdateAndSendTransferParameters(
	const TransferParameters& max_parameters, TransmitAction action);

	void SendStatusChunk(Status status);
	void FinishAndSendStatus(Status status);

	void CancelTimer() {
	timer_.Cancel();
	retries_ = 0;
	}

	// Timeout function invoked from the timer context. This may occur in an
	// interrupt, so no real work can be done. Instead, sets state to timed out
	// and adds a job to run the timeout handler.
	void OnTimeout();

	// The acutal timeout handler, invoked from within the work queue.
	void HandleTimeout();

	static constexpr uint8_t kFlagsType = 1 << 0;
	static constexpr uint8_t kFlagsDataSent = 1 << 1;

	// TODO(frolv): Make this value configurable per transfer.
	static constexpr uint32_t kDefaultChunkDelayMicroseconds = 2000;

	uint32_t transfer_id_;
	uint8_t flags_;
	TransferState transfer_state_ PW_GUARDED_BY(state_lock_);
	uint8_t retries_;
	uint8_t max_retries_;

	sync::InterruptSpinLock state_lock_;

	stream::Stream* stream_;
	rpc::Writer* rpc_writer_;

	uint32_t offset_;
	uint32_t window_size_;
	uint32_t window_end_offset_;
	// TODO(pwbug/584): Remove pending_bytes in favor of window_end_offset.
	uint32_t pending_bytes_;
	uint32_t max_chunk_size_bytes_;

	union {
	Status status_; // Used when state is kCompleted.
	uint32_t last_chunk_offset_; // Used in states kData and kRecovery.
	};

	// Timer used to handle timeouts waiting for chunks.
	chrono::SystemTimer timer_;
	chrono::SystemClock::duration chunk_timeout_;
	work_queue::WorkQueue* work_queue_;

	CompletionFunction on_completion_;
	};

	} // namespace pw::transfer::internal