From 1e1d841965c11f8720a863669af205ca60d0ce57 Mon Sep 17 00:00:00 2001 From: hai007 Date: Wed, 10 Apr 2024 10:40:29 -0700 Subject: [PATCH] Multiplex implementation - Create Virtual Output Stream PiperOrigin-RevId: 623540454 --- Package.swift | 1 + .../implementation/mediums/multiplex/BUILD | 4 + .../mediums/multiplex/multiplex_frames.cc | 16 +- .../mediums/multiplex/multiplex_frames.h | 16 +- .../multiplex/multiplex_frames_test.cc | 6 +- .../multiplex/multiplex_output_stream.cc | 374 ++++++++++++++++++ .../multiplex/multiplex_output_stream.h | 240 +++++++++++ .../multiplex/multiplex_output_stream_test.cc | 253 ++++++++++++ internal/platform/base64_utils.cc | 43 ++ internal/platform/base64_utils.h | 9 + internal/platform/feature_flags.h | 15 + 11 files changed, 960 insertions(+), 17 deletions(-) create mode 100644 connections/implementation/mediums/multiplex/multiplex_output_stream.cc create mode 100644 connections/implementation/mediums/multiplex/multiplex_output_stream.h create mode 100644 connections/implementation/mediums/multiplex/multiplex_output_stream_test.cc diff --git a/Package.swift b/Package.swift index 300e905c..fc207f10 100644 --- a/Package.swift +++ b/Package.swift @@ -461,6 +461,7 @@ let package = Package( "connections/implementation/mediums/ble_v2/discovered_peripheral_tracker_test.cc", "connections/implementation/mediums/ble_v2/instant_on_lost_advertisement_test.cc", "connections/implementation/mediums/multiplex/multiplex_frames_test.cc", + "connections/implementation/mediums/multiplex/multiplex_output_stream_test.cc", "connections/implementation/mediums/webrtc_peer_id_test.cc", "connections/implementation/mediums/wifi_lan_test.cc", "connections/implementation/mediums/bluetooth_classic_test.cc", diff --git a/connections/implementation/mediums/multiplex/BUILD b/connections/implementation/mediums/multiplex/BUILD index e134fc26..903fd433 100644 --- a/connections/implementation/mediums/multiplex/BUILD +++ b/connections/implementation/mediums/multiplex/BUILD @@ -17,9 +17,11 @@ cc_library( name = "multiplex", srcs = [ "multiplex_frames.cc", + "multiplex_output_stream.cc", ], hdrs = [ "multiplex_frames.h", + "multiplex_output_stream.h", ], copts = ["-DCORE_ADAPTER_DLL"], visibility = [ @@ -35,6 +37,7 @@ cc_library( "//internal/platform:util", "//internal/platform:uuid", "//internal/platform/implementation:comm", + "//internal/platform/implementation:types", "//proto/mediums:multiplex_frames_cc_proto", "@aappleby_smhasher//:libmurmur3", "@com_google_absl//absl/base:core_headers", @@ -55,6 +58,7 @@ cc_test( name = "multiplex_test", srcs = [ "multiplex_frames_test.cc", + "multiplex_output_stream_test.cc", ], deps = [ ":multiplex", diff --git a/connections/implementation/mediums/multiplex/multiplex_frames.cc b/connections/implementation/mediums/multiplex/multiplex_frames.cc index 028183d2..9ba8cf5d 100644 --- a/connections/implementation/mediums/multiplex/multiplex_frames.cc +++ b/connections/implementation/mediums/multiplex/multiplex_frames.cc @@ -66,8 +66,8 @@ ByteArray ToBytes(MultiplexFrame&& frame) { return bytes; } -ByteArray ForConnectionRequest(std::string service_id, - std::string service_id_hash_salt) { +ByteArray ForConnectionRequest(const std::string& service_id, + const std::string& service_id_hash_salt) { MultiplexFrame frame; frame.set_frame_type(MultiplexFrame::CONTROL_FRAME); @@ -84,7 +84,8 @@ ByteArray ForConnectionRequest(std::string service_id, } ByteArray ForConnectionResponse( - ByteArray& salted_service_id_hash, std::string service_id_hash_salt, + const ByteArray& salted_service_id_hash, + const std::string& service_id_hash_salt, ConnectionResponseFrame::ConnectionResponseCode response_code) { MultiplexFrame frame; @@ -103,8 +104,8 @@ ByteArray ForConnectionResponse( return ToBytes(std::move(frame)); } -ByteArray ForDisconnection(std::string service_id, - std::string service_id_hash_salt) { +ByteArray ForDisconnection(const std::string& service_id, + const std::string& service_id_hash_salt) { MultiplexFrame frame; frame.set_frame_type(MultiplexFrame::CONTROL_FRAME); @@ -120,8 +121,9 @@ ByteArray ForDisconnection(std::string service_id, return ToBytes(std::move(frame)); } -ByteArray ForData(std::string service_id, std::string service_id_hash_salt, - bool should_pass_salt, ByteArray& data) { +ByteArray ForData(const std::string& service_id, + const std::string& service_id_hash_salt, + bool should_pass_salt, const ByteArray& data) { MultiplexFrame frame; frame.set_frame_type(MultiplexFrame::DATA_FRAME); diff --git a/connections/implementation/mediums/multiplex/multiplex_frames.h b/connections/implementation/mediums/multiplex/multiplex_frames.h index 9f03f5b5..2483c382 100644 --- a/connections/implementation/mediums/multiplex/multiplex_frames.h +++ b/connections/implementation/mediums/multiplex/multiplex_frames.h @@ -62,31 +62,33 @@ std::string GenerateServiceIdHashKeyWithSalt(const std::string& service_id, // Build a MultiplexFrame Connection Request frame Bytes stream. // @param service_id The service ID of the connection. // @param service_id_hash_salt The salt used to generate the service ID hash. -ByteArray ForConnectionRequest(std::string service_id, - std::string service_id_hash_salt); +ByteArray ForConnectionRequest(const std::string& service_id, + const std::string& service_id_hash_salt); // Build a MultiplexFrame Connection Response frame Bytes stream. // @param salted_service_id_hash The salted service ID hash. // @param service_id_hash_salt The salt used to generate the service ID hash. // @param response_code The response code of the connection. ByteArray ForConnectionResponse( - ByteArray& salted_service_id_hash, std::string service_id_hash_salt, + const ByteArray& salted_service_id_hash, + const std::string& service_id_hash_salt, location::nearby::mediums::ConnectionResponseFrame::ConnectionResponseCode response_code); // Build a MultiplexFrame Disconnection frame Bytes stream. // @param service_id The service ID of the connection. // @param service_id_hash_salt The salt used to generate the service ID hash. -ByteArray ForDisconnection(std::string service_id, - std::string service_id_hash_salt); +ByteArray ForDisconnection(const std::string& service_id, + const std::string& service_id_hash_salt); // Build a MultiplexFrame Data frame Bytes stream. // @param service_id The service ID of the connection. // @param service_id_hash_salt The salt used to generate the service ID hash. // @param should_pass_salt Whether to pass the salt in the data frame. // @param data The data to send. -ByteArray ForData(std::string service_id, std::string service_id_hash_salt, - bool should_pass_salt, ByteArray& data); +ByteArray ForData(const std::string& service_id, + const std::string& service_id_hash_salt, + bool should_pass_salt, const ByteArray& data); ExceptionOr FromBytes( const ByteArray& multiplex_frame_bytes); diff --git a/connections/implementation/mediums/multiplex/multiplex_frames_test.cc b/connections/implementation/mediums/multiplex/multiplex_frames_test.cc index ad0b7665..238e02e8 100644 --- a/connections/implementation/mediums/multiplex/multiplex_frames_test.cc +++ b/connections/implementation/mediums/multiplex/multiplex_frames_test.cc @@ -108,9 +108,9 @@ TEST(MultiplexFrameTest, HashValidtion) { TEST(MultiplexFrameTest, CanGenerateConnectionRequest) { ByteArray bytes = ForConnectionRequest(std::string(kServiceId_1), "1234"); - auto response = FromBytes(bytes); - ASSERT_TRUE(response.ok()); - auto frame = response.result(); + auto request = FromBytes(bytes); + ASSERT_TRUE(request.ok()); + auto frame = request.result(); EXPECT_EQ(frame.control_frame().control_frame_type(), MultiplexControlFrame::CONNECTION_REQUEST); EXPECT_EQ(frame.header().salted_service_id_hash(), diff --git a/connections/implementation/mediums/multiplex/multiplex_output_stream.cc b/connections/implementation/mediums/multiplex/multiplex_output_stream.cc new file mode 100644 index 00000000..ba65ce5e --- /dev/null +++ b/connections/implementation/mediums/multiplex/multiplex_output_stream.cc @@ -0,0 +1,374 @@ +// Copyright 2024 Google LLC +// +// 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. + +#include "connections/implementation/mediums/multiplex/multiplex_output_stream.h" + +#include +#include +#include +#include + +#include "absl/strings/string_view.h" +#include "connections/implementation/mediums/multiplex/multiplex_frames.h" +#include "internal/platform/atomic_boolean.h" +#include "internal/platform/base64_utils.h" +#include "internal/platform/byte_array.h" +#include "internal/platform/exception.h" +#include "internal/platform/feature_flags.h" +#include "internal/platform/future.h" +#include "internal/platform/logging.h" +#include "internal/platform/mutex_lock.h" +#include "internal/platform/output_stream.h" + +namespace nearby { +namespace connections { +namespace mediums { +namespace multiplex { +namespace { +using ::location::nearby::mediums::ConnectionResponseFrame; + +constexpr absl::string_view TAG = "MultiplexOutputStream:"; +constexpr absl::string_view kFakeSalt = "RECEIVER_CONDIMENT"; +} // namespace + +// Implementation for class ArrayBlockingQueue +template +void ArrayBlockingQueue::Put(const T& value) { + MutexLock lock(&queue_mutex_); + if (queue_.size() >= capacity_) { + has_space_.Wait(); + } + queue_.push(value); + has_data_.Notify(); +} + +template +T ArrayBlockingQueue::Take() { + MutexLock lock(&queue_mutex_); + if (queue_.empty()) { + has_data_.Wait(); + } + T front = queue_.front(); + queue_.pop(); + has_space_.Notify(); + return front; +} + +template +bool ArrayBlockingQueue::TryPut(const T& value) { + MutexLock lock(&queue_mutex_); + if (queue_.size() < capacity_) { + queue_.push(value); + has_data_.Notify(); + return true; + } + return false; +} + +template +std::optional ArrayBlockingQueue::TryTake() { + MutexLock lock(&queue_mutex_); + if (!queue_.empty()) { + T front = queue_.front(); + queue_.pop(); + has_space_.Notify(); + return front; + } + return std::nullopt; +} + +// Implementation for class MultiplexOutputStream +MultiplexOutputStream::MultiplexOutputStream(OutputStream* physical_writer, + AtomicBoolean& is_enabled) + : is_enabled_(is_enabled), + physical_writer_(physical_writer), + multiplex_writer_{physical_writer} {} + +Exception MultiplexOutputStream::WaitForResult(const std::string& method_name, + Future* future) { + if (!future) { + NEARBY_LOGS(INFO) << TAG << "No future to wait for; return with error."; + return {Exception::kFailed}; + } + NEARBY_LOGS(INFO) << TAG << "Waiting for future to complete: " << method_name; + ExceptionOr result = + future->Get(FeatureFlags::GetInstance() + .GetFlags() + .mediums_frame_write_timeout_millis); + if (!result.ok()) { + NEARBY_LOGS(INFO) << TAG << "Future:[" << method_name + << "] completed with exception:" << result.exception(); + return {Exception::kFailed}; + } + if (result.result()) { + NEARBY_LOGS(INFO) << TAG << "Future:[" << method_name + << "] completed with success."; + return {Exception::kSuccess}; + } + NEARBY_LOGS(INFO) << TAG << "Future:[" << method_name + << "] completed with failure."; + return {Exception::kFailed}; +} + +bool MultiplexOutputStream::WriteConnectionRequestFrame( + const std::string& service_id, const std::string& service_id_hash_salt) { + if (!is_enabled_.Get()) { + return false; + } + Future future; + multiplex_writer_.EnqueueToSend( + &future, ForConnectionRequest(service_id, service_id_hash_salt), + "MultiplexFrame::CONNECTION_REQUEST"); + if (WaitForResult("MultiplexFrame::CONNECTION_REQUEST", &future).Ok()) + return true; + return false; +} + +bool MultiplexOutputStream::WriteConnectionResponseFrame( + const ByteArray& salted_service_id_hash, + const std::string& service_id_hash_salt, + ConnectionResponseFrame::ConnectionResponseCode response_code) { + if (!is_enabled_.Get()) { + return false; + } + Future future; + multiplex_writer_.EnqueueToSend( + &future, + ForConnectionResponse(salted_service_id_hash, service_id_hash_salt, + response_code), + "MultiplexFrame::CONNECTION_RESPONSE"); + if (WaitForResult("MultiplexFrame::CONNECTION_RESPONSE", &future).Ok()) + return true; + return false; +} + +bool MultiplexOutputStream::Close(const std::string& service_id) { + auto item = virtual_output_streams_.find(service_id); + if (item == virtual_output_streams_.end()) { + NEARBY_LOGS(WARNING) << TAG << "Failed to close VirtualOutputStream(" + << service_id << ") because it's already gone."; + return false; + } + + auto service_id_hash_salt = item->second->GetserviceIdHashSalt(); + item->second->Close(); + if (is_enabled_.Get()) { + Future future; + multiplex_writer_.EnqueueToSend( + &future, + ForDisconnection(service_id, item->second->GetserviceIdHashSalt()), + "MultiplexFrame::DISCONNECTION"); + WaitForResult("MultiplexFrame::DISCONNECTION", &future); + } + virtual_output_streams_.erase(service_id); + + if (virtual_output_streams_.empty()) { + physical_writer_->Close(); + multiplex_writer_.Close(); + } + return true; +} + +OutputStream* +MultiplexOutputStream::CreateVirtualOutputStreamForFirstVirtualSocket( + const std::string& service_id, const std::string& service_id_hash_salt) { + return virtual_output_streams_ + .emplace(service_id, + std::make_unique( + service_id, service_id_hash_salt, physical_writer_, + multiplex_writer_, + VirtualOutputStreamType::kFirstVirtualSocket, *this)) + .first->second.get(); +} + +OutputStream* MultiplexOutputStream::CreateVirtualOutputStream( + const std::string& service_id, const std::string& service_id_hash_salt) { + return virtual_output_streams_ + .emplace(service_id, + std::make_unique( + service_id, service_id_hash_salt, physical_writer_, + multiplex_writer_, + VirtualOutputStreamType::kNormalVirtualSocket, *this)) + .first->second.get(); +} + +std::string MultiplexOutputStream::GetserviceIdHashSalt( + const std::string& service_id) { + auto item = virtual_output_streams_.find(service_id); + if (item != virtual_output_streams_.end()) { + return item->second->GetserviceIdHashSalt(); + } + return {}; +} + +void MultiplexOutputStream::Shutdown() { + physical_writer_->Close(); + multiplex_writer_.Close(); +} + +// Implementation for class MultiplexOutputStream::MultiplexWriter +MultiplexOutputStream::MultiplexWriter::MultiplexWriter( + OutputStream* physical_writer) + : physical_writer_(physical_writer) {} + +MultiplexOutputStream::MultiplexWriter::~MultiplexWriter() { + Close(); + writer_thread_.Shutdown(); + physical_writer_ = nullptr; +} + +void MultiplexOutputStream::MultiplexWriter::EnqueueToSend( + Future* future, const ByteArray& data, + const std::string& frame_name) { + MutexLock lock(&mutex_); + data_queue_.Put(EnqueuedFrame(future, data)); + + if (is_writing_) { + return; + } + is_writing_ = true; + is_writing_cond_.Notify(); + if (!is_write_loop_running_) { + is_write_loop_running_ = true; + writer_thread_.Execute("Start writing", [this] { StartWriting(); }); + } +} + +void MultiplexOutputStream::MultiplexWriter::StartWriting() { + NEARBY_LOGS(INFO) << TAG << "Writing loop started."; + while (true) { + auto enqueued_frame = data_queue_.TryTake(); + if (enqueued_frame != std::nullopt) { + Write(enqueued_frame.value()); + continue; + } + + MutexLock lock(&mutex_); + if (data_queue_.Empty() && is_writing_) { + is_writing_ = false; + Exception wait_succeeded = is_writing_cond_.Wait(); + if (!wait_succeeded.Ok()) { + NEARBY_LOGS(WARNING) + << TAG << __func__ + << ": Failure waiting to wait: " << wait_succeeded.value; + return; + } + if (is_closed_) break; + } + } + NEARBY_LOGS(INFO) << TAG << "Writing loop stopped."; +} + +void MultiplexOutputStream::MultiplexWriter::Write( + EnqueuedFrame& enqueued_frame) { + MutexLock lock(&writer_mutex_); + if (!physical_writer_ + ->Write(Base64Utils::IntToBytes(enqueued_frame.data_.size())) + .Ok()) { + enqueued_frame.future_->SetException({Exception::kIo}); + return; + }; + if (!physical_writer_->Write(enqueued_frame.data_).Ok()) { + enqueued_frame.future_->SetException({Exception::kIo}); + return; + }; + if (!physical_writer_->Flush().Ok()) { + enqueued_frame.future_->SetException({Exception::kIo}); + return; + }; + enqueued_frame.future_->Set(true); +} + +void MultiplexOutputStream::MultiplexWriter::Close() { + MutexLock lock(&mutex_); + is_closed_ = true; + if (is_write_loop_running_) { + NEARBY_LOGS(INFO) << TAG << "Stop writing loop and Shutdown writer thread."; + is_write_loop_running_ = false; + is_writing_cond_.Notify(); + } +} + +MultiplexOutputStream::VirtualOutputStream::VirtualOutputStream( + std::string service_id, std::string service_id_hash_salt, + OutputStream* physical_writer, MultiplexWriter& multiplex_writer, + VirtualOutputStreamType virtual_output_stream_type, + MultiplexOutputStream& multiplex_output_stream) + : service_id_(service_id), + service_id_hash_salt_(service_id_hash_salt), + physical_writer_(physical_writer), + multiplex_writer_(multiplex_writer), + virtual_output_stream_type_(virtual_output_stream_type), + multiplex_output_stream_(multiplex_output_stream) {} + +Exception MultiplexOutputStream::VirtualOutputStream::Write( + const ByteArray& data) { + if (is_closed_) { + NEARBY_LOGS(WARNING) + << TAG << "Failed to write data because the VirtualOutputStream for " + << service_id_ << " closed"; + return {Exception::kIo}; + } + if (multiplex_output_stream_.is_enabled_.Get()) { + bool should_pass_salt = false; + if (IsFirstVirtualOutputStream()) { + if (!first_frame_sent_for_first_virtual_output_stream_) { + first_frame_sent_for_first_virtual_output_stream_ = true; + should_pass_salt = true; + } + // Fixes b/290724590, b/290983930 which can't get the correct socket + // from the virtualSockets map. NS receiver side will pass 2 + // DATA_FRAMEs continuously to the remote sender side but originally + // impl will only consider the 1st one. Add below fix to handle 2nd + // frame which the salt is still fake one and change shouldPassSalt to + // true to let the remote handle correctly. + if ((service_id_hash_salt_ == kFakeSalt) && !should_pass_salt) { + should_pass_salt = true; + NEARBY_LOGS(INFO) << TAG + << "service_idHashSalt is still a fake one and " + "not changed yet; continue to pass salt."; + } + } + ByteArray data_frame = + ForData(service_id_, service_id_hash_salt_, should_pass_salt, data); + Future future; + multiplex_writer_.EnqueueToSend(&future, data_frame, + "MultiplexFrame::DATA_FRAME"); + return multiplex_output_stream_.WaitForResult("MultiplexFrame::DATA_FRAME", + &future); + } else { + if (!physical_writer_->Write(data).Ok()) { + return {Exception::kIo}; + }; + if (!physical_writer_->Flush().Ok()) { + return {Exception::kIo}; + }; + } + + return {Exception::kSuccess}; +} + +Exception MultiplexOutputStream::VirtualOutputStream::Flush() { + return {Exception::kSuccess}; +} + +Exception MultiplexOutputStream::VirtualOutputStream::Close() { + is_closed_.Set(true); + return {Exception::kSuccess}; +} + +} // namespace multiplex +} // namespace mediums +} // namespace connections +} // namespace nearby diff --git a/connections/implementation/mediums/multiplex/multiplex_output_stream.h b/connections/implementation/mediums/multiplex/multiplex_output_stream.h new file mode 100644 index 00000000..f777af7a --- /dev/null +++ b/connections/implementation/mediums/multiplex/multiplex_output_stream.h @@ -0,0 +1,240 @@ +// Copyright 2024 Google LLC +// +// 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. + +#ifndef CORE_INTERNAL_MEDIUMS_MULTIPLEX_MULTIPLEX_OUTPUT_STREAM_H_ +#define CORE_INTERNAL_MEDIUMS_MULTIPLEX_MULTIPLEX_OUTPUT_STREAM_H_ + +#include +#include +#include +#include +#include + +#include "absl/base/thread_annotations.h" +#include "absl/container/flat_hash_map.h" +#include "internal/platform/atomic_boolean.h" +#include "internal/platform/byte_array.h" +#include "internal/platform/exception.h" +#include "internal/platform/feature_flags.h" +#include "internal/platform/future.h" +#include "internal/platform/mutex.h" +#include "internal/platform/output_stream.h" +#include "internal/platform/single_thread_executor.h" +#include "proto/mediums/multiplex_frames.pb.h" + +namespace nearby { +namespace connections { +namespace mediums { +namespace multiplex { + +/** + * Payload from different services/clients will be put into an + * ArrayBlockingQueue before sending to ensure each client has equal chance to + * send its data. Since C++ doesn't provide ArrayBlockingQueue as Java, we + * implement one here. + */ +template +class ArrayBlockingQueue { + public: + explicit ArrayBlockingQueue(size_t capacity) : capacity_(capacity) {} + void Put(const T& value); + T Take(); + bool TryPut(const T& value); + // Returns std::nullopt if the queue is empty. + std::optional TryTake(); + + size_t Size() const { + MutexLock lock(&queue_mutex_); + return queue_.size(); + } + bool Empty() const { + MutexLock lock(&queue_mutex_); + return queue_.empty(); + } + + private: + std::queue queue_; + mutable Mutex queue_mutex_; + ConditionVariable has_data_{&queue_mutex_}; + ConditionVariable has_space_{&queue_mutex_}; + const size_t capacity_; +}; + +/** + * A helper class to send out the {@code MultiplexControlFrame} and the outgoing + * data from clients. It schedules control and data frames with priority below + * + *

{@link MultiplexControlFrameType#CONNECTION_REQUEST} and {@link + * MultiplexControlFrameType#CONNECTION_RESPONSE} have the highest priority + * + *

All {@link MultiplexDataFrame} has the medium priority. If there's + * multiple clients send data at the same time, should poll every client's + * outgoing data in sequence. For example, client A and B send data at the same + * time, the outgoing data sequence should like A-Frame-1, B-Frame-1, A-Frame-2, + * B-Frame-2,... + * + *

{@link MultiplexControlFrameType#DISCONNECTION} has the same priority with + * {@link MultiplexDataFrame} because the disconnect should not make the already + * enqueued data failed to send out, so put it in the same priority queue with + * the MultiplexDataFrame. + */ +class MultiplexOutputStream { + public: + enum class VirtualOutputStreamType { + // The type of virtual socket established for the physical socket is + // created. + kFirstVirtualSocket = 0, + // The others except FIRST_VIRTUAL_SCOKET type. + kNormalVirtualSocket = 1, + }; + + MultiplexOutputStream(OutputStream* physical_writer, + AtomicBoolean& is_enabled); + ~MultiplexOutputStream() { Shutdown(); } + + // Writes the connection request frame to the physical output stream. + bool WriteConnectionRequestFrame(const std::string& service_id, + const std::string& service_id_hash_salt); + + // Writes the connection response frame to the physical output stream. + bool WriteConnectionResponseFrame( + const ByteArray& salted_service_id_hash, + const std::string& service_id_hash_salt, + ::location::nearby::mediums::ConnectionResponseFrame:: + ConnectionResponseCode response_code); + + // Closes the virtual output stream. + bool Close(const std::string& service_id); + + // Waits for the result of the future. + Exception WaitForResult(const std::string& method_name, Future* future); + + // Creates the virtual output stream for the first virtual socket. + OutputStream* CreateVirtualOutputStreamForFirstVirtualSocket( + const std::string& service_id, const std::string& service_id_hash_salt); + + // Creates the virtual output stream. + OutputStream* CreateVirtualOutputStream( + const std::string& service_id, const std::string& service_id_hash_salt); + + // Gets the service id hash salt. + std::string GetserviceIdHashSalt(const std::string& service_id); + + // Shuts down the multiplex output stream. + void Shutdown(); + + class EnqueuedFrame { + public: + EnqueuedFrame(Future* future, ByteArray data) + : future_(future), data_(data) {} + ~EnqueuedFrame() = default; + + Future* future_; + ByteArray data_; + }; + + class MultiplexWriter { + public: + explicit MultiplexWriter(OutputStream* physical_writer); + ~MultiplexWriter(); + + // Enqueues the frame to be sent out. + void EnqueueToSend(Future* future, const ByteArray& data, + const std::string& frame_name); + // Closes the writer. + void Close(); + + private: + // Starts the writer thread. + void StartWriting(); + + // Writes the enqueued frame. + void Write(EnqueuedFrame& enqueued_frame); + + Mutex writer_mutex_; + OutputStream* physical_writer_ ABSL_PT_GUARDED_BY(writer_mutex_); + + ArrayBlockingQueue data_queue_{ + FeatureFlags::GetInstance() + .GetFlags() + .multiplex_socket_middle_priority_queue_capacity}; + + mutable Mutex mutex_; + ConditionVariable is_writing_cond_{&mutex_}; + bool is_writing_ = false; + bool is_closed_ = false; + + // The single thread to write all enqueued frames. + SingleThreadExecutor writer_thread_; + bool is_write_loop_running_ = false; + }; + + class VirtualOutputStream : public OutputStream { + public: + VirtualOutputStream(std::string service_id, + std::string service_id_hash_salt, + OutputStream* physical_writer, + MultiplexWriter& multiplex_writer, + VirtualOutputStreamType virtual_output_stream_type, + MultiplexOutputStream& multiplex_output_stream); + ~VirtualOutputStream() override = default; + + // Returns true if the virtual output stream is the first virtual output + // stream. + bool IsFirstVirtualOutputStream() { + return virtual_output_stream_type_ == + VirtualOutputStreamType::kFirstVirtualSocket; + } + + // Returns the service id hash salt. + std::string GetserviceIdHashSalt() { return service_id_hash_salt_; } + + // Sets the service id hash salt. + void SetserviceIdHashSalt(std::string service_id_hash_salt) { + service_id_hash_salt_ = service_id_hash_salt; + } + + // Writes the data to the physical output stream. + Exception Write(const ByteArray& data) override; + // Flushes the physical output stream. + Exception Flush() override; + // Closes the virtual output stream. + Exception Close() override; + + private: + AtomicBoolean is_closed_{false}; + + std::string service_id_; + std::string service_id_hash_salt_; + OutputStream* physical_writer_; + MultiplexWriter& multiplex_writer_; + VirtualOutputStreamType virtual_output_stream_type_; + bool first_frame_sent_for_first_virtual_output_stream_ = false; + MultiplexOutputStream& multiplex_output_stream_; + }; + + private: + AtomicBoolean& is_enabled_; + OutputStream* physical_writer_; + absl::flat_hash_map> + virtual_output_streams_; + MultiplexWriter multiplex_writer_; +}; + +} // namespace multiplex +} // namespace mediums +} // namespace connections +} // namespace nearby + +#endif // CORE_INTERNAL_MEDIUMS_MULTIPLEX_MULTIPLEX_OUTPUT_STREAM_H_ diff --git a/connections/implementation/mediums/multiplex/multiplex_output_stream_test.cc b/connections/implementation/mediums/multiplex/multiplex_output_stream_test.cc new file mode 100644 index 00000000..af6bbda8 --- /dev/null +++ b/connections/implementation/mediums/multiplex/multiplex_output_stream_test.cc @@ -0,0 +1,253 @@ +// Copyright 2024 Google LLC +// +// 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. + +#include "connections/implementation/mediums/multiplex/multiplex_output_stream.h" + +#include +#include +#include +#include + +#include "gtest/gtest.h" +#include "absl/strings/string_view.h" +#include "absl/time/clock.h" +#include "absl/time/time.h" +#include "connections/implementation/mediums/multiplex/multiplex_frames.h" +#include "internal/platform/atomic_boolean.h" +#include "internal/platform/base64_utils.h" +#include "internal/platform/byte_array.h" +#include "internal/platform/count_down_latch.h" +#include "internal/platform/exception.h" +#include "internal/platform/input_stream.h" +#include "internal/platform/logging.h" +#include "internal/platform/multi_thread_executor.h" +#include "internal/platform/output_stream.h" +#include "internal/platform/pipe.h" +#include "proto/mediums/multiplex_frames.pb.h" + +namespace nearby { +namespace connections { +namespace mediums { +namespace multiplex { + +constexpr absl::string_view kServiceId_1 = "serviceId_1"; +constexpr absl::string_view kServiceId_2 = "serviceId_2"; +constexpr absl::string_view kNoSalt = ""; +constexpr absl::string_view kSalt_1 = "DNFG"; +constexpr absl::string_view kSalt_2 = "YFRT"; + +using ::location::nearby::mediums::ConnectionResponseFrame; +using ::location::nearby::mediums::MultiplexControlFrame; +using ::location::nearby::mediums::MultiplexFrame; + +class MultiplexOutputStreamTest : public ::testing::Test { + protected: + ExceptionOr ReadFrame() { + ExceptionOr read_int = Base64Utils::ReadInt(reader_.get()); + if (!read_int.ok()) return read_int.GetException(); + if (read_int.result() <= 0) return {Exception::kFailed}; + + ExceptionOr received_data = + reader_->ReadExactly(read_int.result()); + if (!received_data.ok()) return received_data.GetException(); + auto bytes = std::move(received_data.result()); + return FromBytes(bytes); + } + + AtomicBoolean enabled_{true}; + std::pair, std::unique_ptr> pipe_ = + CreatePipe(); + + std::unique_ptr reader_ = std::move(pipe_.first); + std::unique_ptr writer_ = std::move(pipe_.second); + std::unique_ptr multiplex_output_stream_; +}; + +TEST_F(MultiplexOutputStreamTest, SendConnectionRequestFrame) { + multiplex_output_stream_ = std::make_unique( + writer_.get(), enabled_); + EXPECT_TRUE(multiplex_output_stream_->WriteConnectionRequestFrame( + std::string(kServiceId_1), std::string(kNoSalt))); + + auto request = ReadFrame(); + ASSERT_TRUE(request.ok()); + auto frame = request.result(); + EXPECT_EQ(frame.control_frame().control_frame_type(), + MultiplexControlFrame::CONNECTION_REQUEST); + EXPECT_EQ(frame.header().salted_service_id_hash(), + std::string(GenerateServiceIdHashWithSalt(std::string(kServiceId_1), + std::string(kNoSalt)))); + + multiplex_output_stream_->Shutdown(); +} + +TEST_F(MultiplexOutputStreamTest, SendConnectionRequestFrameDisabled) { + enabled_.Set(false); + multiplex_output_stream_ = std::make_unique( + writer_.get(), enabled_); + EXPECT_FALSE(multiplex_output_stream_->WriteConnectionRequestFrame( + std::string(kServiceId_1), std::string(kNoSalt))); + + multiplex_output_stream_->Shutdown(); +} + +TEST_F(MultiplexOutputStreamTest, SendConnectionResponseFrame) { + multiplex_output_stream_ = std::make_unique( + writer_.get(), enabled_); + EXPECT_TRUE(multiplex_output_stream_->WriteConnectionResponseFrame( + GenerateServiceIdHash(std::string(kServiceId_1)), std::string(kNoSalt), + ConnectionResponseFrame::CONNECTION_ACCEPTED)); + + auto response = ReadFrame(); + ASSERT_TRUE(response.ok()); + auto frame = response.result(); + EXPECT_EQ(frame.control_frame().control_frame_type(), + MultiplexControlFrame::CONNECTION_RESPONSE); + EXPECT_EQ(frame.header().salted_service_id_hash(), + std::string(GenerateServiceIdHashWithSalt(std::string(kServiceId_1), + std::string(kNoSalt)))); + EXPECT_EQ(frame.control_frame() + .connection_response_frame() + .connection_response_code(), + ConnectionResponseFrame::CONNECTION_ACCEPTED); + + multiplex_output_stream_->Shutdown(); +} + +TEST_F(MultiplexOutputStreamTest, SendConnectionResponseFrameDisabled) { + enabled_.Set(false); + multiplex_output_stream_ = std::make_unique( + writer_.get(), enabled_); + EXPECT_FALSE(multiplex_output_stream_->WriteConnectionResponseFrame( + GenerateServiceIdHash(std::string(kServiceId_1)), std::string(kNoSalt), + ConnectionResponseFrame::CONNECTION_ACCEPTED)); + + multiplex_output_stream_->Shutdown(); +} + +TEST_F(MultiplexOutputStreamTest, CloseVirtualStreamFailed) { + multiplex_output_stream_ = std::make_unique( + writer_.get(), enabled_); + EXPECT_FALSE(multiplex_output_stream_->Close(std::string(kServiceId_1))); + + multiplex_output_stream_->Shutdown(); +} + +TEST_F(MultiplexOutputStreamTest, CloseVirtualStreamSuccess) { + multiplex_output_stream_ = std::make_unique( + writer_.get(), enabled_); + EXPECT_FALSE(multiplex_output_stream_->Close(std::string(kServiceId_1))); + + multiplex_output_stream_->CreateVirtualOutputStream(std::string(kServiceId_1), + std::string(kNoSalt)); + EXPECT_TRUE(multiplex_output_stream_->Close(std::string(kServiceId_1))); + + auto request = ReadFrame(); + ASSERT_TRUE(request.ok()); + auto frame = request.result(); + EXPECT_EQ(frame.control_frame().control_frame_type(), + MultiplexControlFrame::DISCONNECTION); + EXPECT_EQ(frame.header().salted_service_id_hash(), + std::string(GenerateServiceIdHashWithSalt(std::string(kServiceId_1), + std::string(kNoSalt)))); + + multiplex_output_stream_->Shutdown(); +} + +TEST_F(MultiplexOutputStreamTest, CreateVirtualStream_SendData) { + multiplex_output_stream_ = std::make_unique( + writer_.get(), enabled_); + + auto virtual_output_stream = + multiplex_output_stream_->CreateVirtualOutputStream( + std::string(kServiceId_1), std::string(kSalt_1)); + + const ByteArray data("abcdefghijklmnopqrstuvwxyz"); + virtual_output_stream->Write(data); + virtual_output_stream->Flush(); + auto frame_data = ReadFrame(); + ASSERT_TRUE(frame_data.ok()); + auto frame = frame_data.result(); + EXPECT_EQ(frame.frame_type(), MultiplexFrame::DATA_FRAME); + EXPECT_EQ(frame.header().salted_service_id_hash(), + std::string(GenerateServiceIdHashWithSalt(std::string(kServiceId_1), + std::string(kSalt_1)))); + EXPECT_EQ(frame.data_frame().data(), std::string(data)); + + multiplex_output_stream_->Shutdown(); +} + +TEST_F(MultiplexOutputStreamTest, CreateTwoVirtualStreams_SendData) { + multiplex_output_stream_ = std::make_unique( + writer_.get(), enabled_); + + auto virtual_output_stream_1 = + multiplex_output_stream_->CreateVirtualOutputStreamForFirstVirtualSocket( + std::string(kServiceId_1), std::string(kSalt_1)); + auto virtual_output_stream_2 = + multiplex_output_stream_->CreateVirtualOutputStreamForFirstVirtualSocket( + std::string(kServiceId_2), std::string(kSalt_2)); + + const ByteArray data_1("abcdefg"); + const ByteArray data_2("hijklmn"); + MultiThreadExecutor executor(2); + CountDownLatch latch(2); + executor.Execute([&virtual_output_stream_1, &latch, &data_1]() { + absl::SleepFor(absl::Milliseconds(500)); + virtual_output_stream_1->Write(data_1); + virtual_output_stream_1->Flush(); + latch.CountDown(); + }); + executor.Execute([&virtual_output_stream_2, &latch, &data_2]() { + virtual_output_stream_2->Write(data_2); + virtual_output_stream_2->Flush(); + latch.CountDown(); + }); + EXPECT_TRUE(latch.Await(absl::Milliseconds(2000)).result()); + + auto frame_data = ReadFrame(); + ASSERT_TRUE(frame_data.ok()); + auto frame = frame_data.result(); + EXPECT_EQ(frame.frame_type(), MultiplexFrame::DATA_FRAME); + bool first_frame_is_data_1 = true; + if (frame.header().salted_service_id_hash() == + std::string(GenerateServiceIdHashWithSalt(std::string(kServiceId_1), + std::string(kSalt_1)))) { + EXPECT_EQ(frame.data_frame().data(), std::string(data_1)); + NEARBY_LOGS(INFO) << "Read first virtual stream frame first."; + } else { + EXPECT_EQ(frame.header().salted_service_id_hash(), + std::string(GenerateServiceIdHashWithSalt(std::string(kServiceId_2), + std::string(kSalt_2)))); + EXPECT_EQ(frame.data_frame().data(), std::string(data_2)); + first_frame_is_data_1 = false; + NEARBY_LOGS(INFO) << "Read second virtual stream frame first."; + } + + frame_data = ReadFrame(); + ASSERT_TRUE(frame_data.ok()); + frame = frame_data.result(); + EXPECT_EQ(frame.frame_type(), MultiplexFrame::DATA_FRAME); + if (first_frame_is_data_1) { + EXPECT_EQ(frame.data_frame().data(), std::string(data_2)); + } else { + EXPECT_EQ(frame.data_frame().data(), std::string(data_1)); + } + multiplex_output_stream_->Shutdown(); +} + +} // namespace multiplex +} // namespace mediums +} // namespace connections +} // namespace nearby diff --git a/internal/platform/base64_utils.cc b/internal/platform/base64_utils.cc index 2ff77f41..b4834d0e 100644 --- a/internal/platform/base64_utils.cc +++ b/internal/platform/base64_utils.cc @@ -14,8 +14,16 @@ #include "internal/platform/base64_utils.h" +#include +#include +#include + #include "absl/strings/escaping.h" +#include "absl/strings/string_view.h" #include "internal/platform/byte_array.h" +#include "internal/platform/exception.h" +#include "internal/platform/input_stream.h" +#include "internal/platform/output_stream.h" namespace nearby { @@ -36,4 +44,39 @@ ByteArray Base64Utils::Decode(absl::string_view base64_string) { return ByteArray(decoded_string.data(), decoded_string.size()); } +std::int32_t Base64Utils::BytesToInt(const ByteArray& bytes) { + const char* int_bytes = bytes.data(); + + std::int32_t result = 0; + result |= (static_cast(int_bytes[0]) & 0x0FF) << 24; + result |= (static_cast(int_bytes[1]) & 0x0FF) << 16; + result |= (static_cast(int_bytes[2]) & 0x0FF) << 8; + result |= (static_cast(int_bytes[3]) & 0x0FF); + + return result; +} + +ByteArray Base64Utils::IntToBytes(std::int32_t value) { + char int_bytes[sizeof(std::int32_t)]; + int_bytes[0] = static_cast((value >> 24) & 0x0FF); + int_bytes[1] = static_cast((value >> 16) & 0x0FF); + int_bytes[2] = static_cast((value >> 8) & 0x0FF); + int_bytes[3] = static_cast((value) & 0x0FF); + + return ByteArray(int_bytes, sizeof(int_bytes)); +} + +ExceptionOr Base64Utils::ReadInt(InputStream* reader) { + ExceptionOr read_bytes = reader->ReadExactly(sizeof(std::int32_t)); + if (!read_bytes.ok()) { + return ExceptionOr(read_bytes.exception()); + } + return ExceptionOr( + BytesToInt(std::move(read_bytes.result()))); +} + +Exception Base64Utils::WriteInt(OutputStream* writer, std::int32_t value) { + return writer->Write(IntToBytes(value)); +} + } // namespace nearby diff --git a/internal/platform/base64_utils.h b/internal/platform/base64_utils.h index 3002890c..f673e8db 100644 --- a/internal/platform/base64_utils.h +++ b/internal/platform/base64_utils.h @@ -15,8 +15,13 @@ #ifndef PLATFORM_BASE_BASE64_UTILS_H_ #define PLATFORM_BASE_BASE64_UTILS_H_ +#include +#include #include "absl/strings/string_view.h" #include "internal/platform/byte_array.h" +#include "internal/platform/exception.h" +#include "internal/platform/input_stream.h" +#include "internal/platform/output_stream.h" namespace nearby { @@ -24,6 +29,10 @@ class Base64Utils { public: static std::string Encode(const ByteArray& bytes); static ByteArray Decode(absl::string_view base64_string); + static std::int32_t BytesToInt(const ByteArray& bytes); + static ByteArray IntToBytes(std::int32_t value); + static ExceptionOr ReadInt(InputStream* reader); + static Exception WriteInt(OutputStream* writer, std::int32_t value); }; } // namespace nearby diff --git a/internal/platform/feature_flags.h b/internal/platform/feature_flags.h index e4e69b96..1183db05 100644 --- a/internal/platform/feature_flags.h +++ b/internal/platform/feature_flags.h @@ -91,6 +91,21 @@ class FeatureFlags { // If the receiver doesn't ack with payload_received_ack frame in 1s, the // sender will timeout the waiting. absl::Duration wait_payload_received_ack_millis = absl::Milliseconds(1000); + + // Multiplex related flags + // Timeout value for read frame operation in endpoint channel. + absl::Duration mediums_frame_read_timeout_millis = + absl::Milliseconds(15000); + // Timeout value for write frame operation in endpoint channel. + absl::Duration mediums_frame_write_timeout_millis = + absl::Milliseconds(15000); + // The timeout for waiting on connection request response. + absl::Duration multiplex_socket_connection_response_timeout_millis = + absl::Milliseconds(3000); + // The capacity of the middle priority queue inner MultiplexOutputStream. + // The new outgoing frame with the middle priority will wait for space to + // become available if the queue is full.' + std::uint32_t multiplex_socket_middle_priority_queue_capacity = 50; }; static const FeatureFlags& GetInstance() {