pw_transfer/integration_test/server.cc

// Copyright 2022 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.

// Simple RPC server with the transfer service registered. Reads HDLC frames
// with RPC packets through a socket. This server has a single resource ID that
// is available, and data must be written to the server before data can be read
// from the resource ID.
//
// Usage:
//
//   integration_test_server 3300 <<< "resource_id: 12 file: '/tmp/gotbytes'"

#include <sys/socket.h>

#include <chrono>
#include <cstddef>
#include <cstdlib>
#include <deque>
#include <map>
#include <memory>
#include <string>
#include <thread>
#include <utility>
#include <variant>
#include <vector>

#include "google/protobuf/text_format.h"
#include "pw_assert/check.h"
#include "pw_chrono/system_clock.h"
#include "pw_log/log.h"
#include "pw_rpc_system_server/rpc_server.h"
#include "pw_rpc_system_server/socket.h"
#include "pw_stream/std_file_stream.h"
#include "pw_thread/thread.h"
#include "pw_thread_stl/options.h"
#include "pw_transfer/integration_test/config.pb.h"
#include "pw_transfer/transfer.h"

namespace pw::transfer {
namespace {

using stream::MemoryReader;
using stream::MemoryWriter;

// This is the maximum size of the socket send buffers. Ideally, this is set
// to the lowest allowed value to minimize buffering between the proxy and
// clients so rate limiting causes the client to block and wait for the
// integration test proxy to drain rather than allowing OS buffers to backlog
// large quantities of data.
//
// Note that the OS may chose to not strictly follow this requested buffer size.
// Still, setting this value to be as small as possible does reduce bufer sizes
// significantly enough to better reflect typical inter-device communication.
//
// For this to be effective, servers should also configure their sockets to a
// smaller receive buffer size.
constexpr int kMaxSocketSendBufferSize = 1;

class FileTransferHandler final : public ReadWriteHandler {
 public:
  FileTransferHandler(uint32_t resource_id,
                      std::deque<std::string>&& sources,
                      std::deque<std::string>&& destinations)
      : ReadWriteHandler(resource_id),
        sources_(sources),
        destinations_(destinations) {}

  ~FileTransferHandler() = default;

  Status PrepareRead() final {
    if (sources_.empty()) {
      PW_LOG_ERROR("Source paths exhausted");
      return Status::ResourceExhausted();
    }

    auto path = sources_.front();
    PW_LOG_DEBUG("Preparing read for file %s", path.c_str());
    set_reader(stream_.emplace<stream::StdFileReader>(path.c_str()));

    sources_.pop_front();
    return OkStatus();
  }

  void FinalizeRead(Status) final {
    std::get<stream::StdFileReader>(stream_).Close();
  }

  Status PrepareWrite() final {
    if (destinations_.empty()) {
      PW_LOG_ERROR("Destination paths exhausted");
      return Status::ResourceExhausted();
    }

    auto path = destinations_.front();
    PW_LOG_DEBUG("Preparing write for file %s", path.c_str());
    set_writer(stream_.emplace<stream::StdFileWriter>(path.c_str()));

    destinations_.pop_front();
    return OkStatus();
  }

  Status FinalizeWrite(Status) final {
    std::get<stream::StdFileWriter>(stream_).Close();
    return OkStatus();
  }

 private:
  std::deque<std::string> sources_;
  std::deque<std::string> destinations_;
  std::variant<std::monostate, stream::StdFileReader, stream::StdFileWriter>
      stream_;
};

void RunServer(int socket_port, ServerConfig config) {
  std::vector<std::byte> chunk_buffer(config.chunk_size_bytes());
  std::vector<std::byte> encode_buffer(config.chunk_size_bytes());
  transfer::Thread<4, 4> transfer_thread(chunk_buffer, encode_buffer);
  TransferService transfer_service(
      transfer_thread,
      config.pending_bytes(),
      std::chrono::seconds(config.chunk_timeout_seconds()),
      config.transfer_service_retries(),
      config.extend_window_divisor());

  rpc::system_server::set_socket_port(socket_port);

  rpc::system_server::Init();
  rpc::system_server::Server().RegisterService(transfer_service);

  // Start transfer thread.
  thread::Thread transfer_thread_handle =
      thread::Thread(thread::stl::Options(), transfer_thread);

  int retval = setsockopt(rpc::system_server::GetServerSocketFd(),
                          SOL_SOCKET,
                          SO_SNDBUF,
                          &kMaxSocketSendBufferSize,
                          sizeof(kMaxSocketSendBufferSize));
  PW_CHECK_INT_EQ(retval,
                  0,
                  "Failed to configure socket send buffer size with errno=%d",
                  errno);

  std::vector<std::unique_ptr<FileTransferHandler>> handlers;
  for (const auto& resource : config.resources()) {
    uint32_t id = resource.first;

    std::deque<std::string> source_paths(resource.second.source_paths().begin(),
                                         resource.second.source_paths().end());
    std::deque<std::string> destination_paths(
        resource.second.destination_paths().begin(),
        resource.second.destination_paths().end());

    auto handler = std::make_unique<FileTransferHandler>(
        id, std::move(source_paths), std::move(destination_paths));

    transfer_service.RegisterHandler(*handler);
    handlers.push_back(std::move(handler));
  }

  PW_LOG_INFO("Starting pw_rpc server");
  PW_CHECK_OK(rpc::system_server::Start());

  // Unregister transfer handler before cleaning up the thread since doing so
  // requires the transfer thread to be running.
  for (auto& handler : handlers) {
    transfer_service.UnregisterHandler(*handler);
  }

  // End transfer thread.
  transfer_thread.Terminate();
  transfer_thread_handle.join();
}

}  // namespace
}  // namespace pw::transfer

int main(int argc, char* argv[]) {
  if (argc != 2) {
    PW_LOG_INFO("Usage: %s PORT <<< config textproto", argv[0]);
    return 1;
  }

  int port = std::atoi(argv[1]);
  PW_CHECK_UINT_GT(port, 0, "Invalid port!");

  std::string config_string;
  std::string line;
  while (std::getline(std::cin, line)) {
    config_string = config_string + line + '\n';
  }
  pw::transfer::ServerConfig config;

  bool ok =
      google::protobuf::TextFormat::ParseFromString(config_string, &config);
  if (!ok) {
    PW_LOG_INFO("Failed to parse config: %s", config_string.c_str());
    PW_LOG_INFO("Usage: %s PORT <<< config textproto", argv[0]);
    return 1;
  } else {
    PW_LOG_INFO("Server loaded config:\n%s", config.DebugString().c_str());
  }

  pw::transfer::RunServer(port, config);
  return 0;
}