Skip to content

Commit

Permalink
pw_transfer: Version 2 opening handshake in C++
Browse files Browse the repository at this point in the history 
This implements the opening handshake of pw_transfer's version 2
protocol within the C++ transfer client and server. The handshake
consists of protocol version negotiation and ephemeral session ID
assignment.

The protocol version used for a transfer session is controlled by the
client when it starts a new transfer. By default, this still remains the
legacy protocol for now, though the client API is extended to allow
specifying the desired version.

If the START chunk a transfer service receives is configured for version
2, the service will assign a transfer session ID and proceed with the
handshake. Otherwise, it will fall back to the legacy protocol.

The version 2 START chunk sent by a client retains all of the chunk
proto fields set by the legacy protocol, allowing it to be understood
by a server which is not version 2 aware. In such a case, the server
will process the chunk per the legacy protocol and send a non-handshake
response. The client will recognize the legacy response and revert to
running the legacy protocol.

As a result of this, version 2 capable transfer clients and servers
remain fully backwards-compatible with older code that only runs the
legacy protocol.

Change-Id: Ie0a295509e754b963d3a78593ba1c43bbe13c977
Reviewed-on: https://pigweed-review.googlesource.com/c/pigweed/pigweed/+/99500
Reviewed-by: Wyatt Hepler <hepler@google.com>
Commit-Queue: Alexei Frolov <frolv@google.com>
  • Loading branch information
@frolv
frolv authored and CQ Bot Account committed Aug 12, 2022
1 parent 400c0f3 commit 64a88a4
Show file tree
Hide file tree
Showing 20 changed files with 2,344 additions and 357 deletions.
3 changes: 3 additions & 0 deletions pw_thread/docs.rst
View file
Original file line number Diff line number Diff line change
Expand Up @@ -145,6 +145,9 @@ Example
const pw::thread::Id my_id = pw::this_thread::get_id();
}
.. _module-pw_thread-thread-creation:

---------------
Thread Creation
---------------
Expand Down
99 changes: 77 additions & 22 deletions pw_transfer/chunk.cc
View file
Original file line number Diff line number Diff line change
Expand Up @@ -24,30 +24,45 @@ namespace pw::transfer::internal {

namespace ProtoChunk = transfer::Chunk;

Result<uint32_t> Chunk::ExtractSessionId(ConstByteSpan message) {
Result<uint32_t> Chunk::ExtractIdentifier(ConstByteSpan message) {
protobuf::Decoder decoder(message);

uint32_t session_id = 0;
uint32_t resource_id = 0;

// During the initial handshake, a START_ACK chunk sent from the server
// to a client identifies its transfer context using a resource_id, as
// the client does not yet know its session_id.
bool should_use_resource_id = false;

while (decoder.Next().ok()) {
ProtoChunk::Fields field =
static_cast<ProtoChunk::Fields>(decoder.FieldNumber());

if (field == ProtoChunk::Fields::TRANSFER_ID) {
// Interpret a legacy transfer_id field as a session ID, but don't
// return immediately. Instead, check to see if the message also
// contains a newer session_id field.
PW_TRY(decoder.ReadUint32(&session_id));

// Interpret a legacy transfer_id field as a session ID if an explicit
// session_id field has not already been seen.
if (session_id == 0) {
PW_TRY(decoder.ReadUint32(&session_id));
}
} else if (field == ProtoChunk::Fields::SESSION_ID) {
// A session_id field always takes precedence over transfer_id, so
// return it immediately when encountered.
// A session_id field always takes precedence over transfer_id.
PW_TRY(decoder.ReadUint32(&session_id));
return session_id;
} else if (field == ProtoChunk::Fields::TYPE) {
// Check if the chunk is a START_ACK.
uint32_t type;
PW_TRY(decoder.ReadUint32(&type));
should_use_resource_id = static_cast<Type>(type) == Type::kStartAck;
} else if (field == ProtoChunk::Fields::RESOURCE_ID) {
PW_TRY(decoder.ReadUint32(&resource_id));
}
}

if (session_id != 0) {
if (should_use_resource_id) {
if (resource_id != 0) {
return resource_id;
}
} else if (session_id != 0) {
return session_id;
}

Expand All @@ -61,9 +76,9 @@ Result<Chunk> Chunk::Parse(ConstByteSpan message) {

Chunk chunk;

// Assume the legacy protocol by default. Field presence in the serialized
// message may change this.
chunk.protocol_version_ = ProtocolVersion::kLegacy;
// Determine the protocol version of the chunk depending on field presence in
// the serialized message.
chunk.protocol_version_ = ProtocolVersion::kUnknown;

// Some older versions of the protocol set the deprecated pending_bytes field
// in their chunks. The newer transfer handling code does not process this
Expand All @@ -88,8 +103,12 @@ Result<Chunk> Chunk::Parse(ConstByteSpan message) {

case ProtoChunk::Fields::SESSION_ID:
// The existence of a session_id field indicates that a newer protocol
// is running.
chunk.protocol_version_ = ProtocolVersion::kVersionTwo;
// is running. Update the deduced protocol unless it was explicitly
// specified.
if (chunk.protocol_version_ == ProtocolVersion::kUnknown) {
chunk.protocol_version_ = ProtocolVersion::kVersionTwo;
}

PW_TRY(decoder.ReadUint32(&chunk.session_id_));
break;

Expand Down Expand Up @@ -141,16 +160,29 @@ Result<Chunk> Chunk::Parse(ConstByteSpan message) {
case ProtoChunk::Fields::RESOURCE_ID:
PW_TRY(decoder.ReadUint32(&value));
chunk.set_resource_id(value);
break;

// The existence of a resource_id field indicates that a newer protocol
// is running.
chunk.protocol_version_ = ProtocolVersion::kVersionTwo;
case ProtoChunk::Fields::PROTOCOL_VERSION:
// The protocol_version field is added as part of the initial handshake
// starting from version 2. If provided, it should override any deduced
// protocol version.
PW_TRY(decoder.ReadUint32(&value));
if (!ValidProtocolVersion(value)) {
return Status::DataLoss();
}
chunk.protocol_version_ = static_cast<ProtocolVersion>(value);
break;

// Silently ignore any unrecognized fields.
}
}

if (chunk.protocol_version_ == ProtocolVersion::kUnknown) {
// If no fields in the chunk specified its protocol version, assume it is a
// legacy chunk.
chunk.protocol_version_ = ProtocolVersion::kLegacy;
}

if (pending_bytes != 0) {
// Compute window_end_offset if it isn't explicitly provided (in older
// protocol versions).
Expand Down Expand Up @@ -186,6 +218,13 @@ Result<ConstByteSpan> Chunk::Encode(ByteSpan buffer) const {
}
}

// During the initial handshake, the chunk's configured protocol version is
// explicitly serialized to the wire.
if (IsInitialHandshakeChunk()) {
encoder.WriteProtocolVersion(static_cast<uint32_t>(protocol_version_))
.IgnoreError();
}

if (type_.has_value()) {
encoder.WriteType(static_cast<ProtoChunk::Type>(type_.value()))
.IgnoreError();
Expand All @@ -197,8 +236,13 @@ Result<ConstByteSpan> Chunk::Encode(ByteSpan buffer) const {

// Encode additional fields from the legacy protocol.
if (ShouldEncodeLegacyFields()) {
// The legacy protocol uses the transfer_id field instead of session_id.
encoder.WriteTransferId(session_id_).IgnoreError();
// The legacy protocol uses the transfer_id field instead of session_id or
// resource_id.
if (resource_id_.has_value()) {
encoder.WriteTransferId(resource_id_.value()).IgnoreError();
} else {
encoder.WriteTransferId(session_id_).IgnoreError();
}

// In the legacy protocol, the pending_bytes field must be set alongside
// window_end_offset, as some transfer implementations require it.
Expand Down Expand Up @@ -241,11 +285,22 @@ size_t Chunk::EncodedSize() const {
}

if (ShouldEncodeLegacyFields()) {
size += protobuf::SizeOfVarintField(ProtoChunk::Fields::TRANSFER_ID,
session_id_);
if (resource_id_.has_value()) {
size += protobuf::SizeOfVarintField(ProtoChunk::Fields::TRANSFER_ID,
resource_id_.value());
} else {
size += protobuf::SizeOfVarintField(ProtoChunk::Fields::TRANSFER_ID,
session_id_);
}
}
}

if (IsInitialHandshakeChunk()) {
size +=
protobuf::SizeOfVarintField(ProtoChunk::Fields::PROTOCOL_VERSION,
static_cast<uint32_t>(protocol_version_));
}

if (protocol_version_ >= ProtocolVersion::kVersionTwo) {
if (resource_id_.has_value()) {
size += protobuf::SizeOfVarintField(ProtoChunk::Fields::RESOURCE_ID,
Expand Down
22 changes: 12 additions & 10 deletions pw_transfer/client.cc
View file
Original file line number Diff line number Diff line change
Expand Up @@ -23,8 +23,10 @@ namespace pw::transfer {
Status Client::Read(uint32_t resource_id,
stream::Writer& output,
CompletionFunc&& on_completion,
chrono::SystemClock::duration timeout) {
if (on_completion == nullptr) {
chrono::SystemClock::duration timeout,
ProtocolVersion protocol_version) {
if (on_completion == nullptr ||
protocol_version == ProtocolVersion::kUnknown) {
return Status::InvalidArgument();
}

Expand All @@ -40,10 +42,9 @@ Status Client::Read(uint32_t resource_id,
has_read_stream_ = true;
}

// TODO(frolv): Only send the resource ID. The server should assign a session.
transfer_thread_.StartClientTransfer(internal::TransferType::kReceive,
/*session_id=*/resource_id,
/*resource_id=*/resource_id,
protocol_version,
resource_id,
&output,
max_parameters_,
std::move(on_completion),
Expand All @@ -55,8 +56,10 @@ Status Client::Read(uint32_t resource_id,
Status Client::Write(uint32_t resource_id,
stream::Reader& input,
CompletionFunc&& on_completion,
chrono::SystemClock::duration timeout) {
if (on_completion == nullptr) {
chrono::SystemClock::duration timeout,
ProtocolVersion protocol_version) {
if (on_completion == nullptr ||
protocol_version == ProtocolVersion::kUnknown) {
return Status::InvalidArgument();
}

Expand All @@ -72,10 +75,9 @@ Status Client::Write(uint32_t resource_id,
has_write_stream_ = true;
}

// TODO(frolv): Only send the resource ID. The server should assign a session.
transfer_thread_.StartClientTransfer(internal::TransferType::kTransmit,
/*session_id=*/resource_id,
/*resource_id=*/resource_id,
protocol_version,
resource_id,
&input,
max_parameters_,
std::move(on_completion),
Expand Down
Loading

0 comments on commit 64a88a4

Please sign in to comment.