Implement an event-based simulator for QuicConnection

net/quic/core/congestion_control/simulation/quic_endpoint.cc

+#include "net/quic/core/congestion_control/simulation/quic_endpoint.h"
+#include "net/quic/core/congestion_control/simulation/simulator.h"
+#include "net/quic/core/crypto/crypto_handshake_message.h"
+#include "net/quic/core/crypto/crypto_protocol.h"
+#include "util/gtl/ptr_util.h"
+#include "util/sig/sha.h"
+
+namespace net {
+namespace simulation {
+
+const QuicStreamId kDataStream = 3;
+const QuicByteCount kWriteChunkSize = 128 * 1024;
+
+// Takes a SHA-1 hash of the name and converts it into five 32-bit integers.
+static std::vector<uint32_t> HashNameIntoFive32BitIntegers(std::string name) {
+  const std::string hash = Sha1_Hash(name);
+
+  std::vector<uint32_t> output;
+  uint32_t current_number = 0;
+  for (size_t i = 0; i < hash.size(); i++) {
+    current_number = (current_number << 8) + hash[i];
+    if (i % 4 == 3) {
+      output.push_back(i);
+      current_number = 0;
+    }
+  }
+
+  return output;
+}
+
+IPEndPoint GetAddressFromName(std::string name) {
+  const std::vector<uint32_t> hash = HashNameIntoFive32BitIntegers(name);
+
+  // Generate a random port between 1025 and 65535.
+  const uint16_t port = 1025 + hash[0] % (65535 - 1025 + 1);
+
+  // Generate a random 10.x.x.x address, where x is between 1 and 254.
+  std::string ip_address({10, 0, 0, 0});
+  for (size_t i = 1; i <= 4; i++) {
+    ip_address[i] = 1 + hash[i] % 254;
+  }
+
+  return IPEndPoint(PackedStringToIPAddressOrDie(ip_address), port);
+}
+
+QuicEndpoint::QuicEndpoint(Simulator* simulator,
+                           std::string name,
+                           std::string peer_name,
+                           Perspective perspective,
+                           QuicConnectionId connection_id)
+    : Endpoint(simulator, name),
+      peer_name_(peer_name),
+      writer_(this),
+      nic_tx_queue_(simulator,
+                    StringPrintf("%s (TX Queue)", name.c_str()),
+                    kMaxPacketSize * kTxQueueSize),
+      connection_(connection_id,
+                  GetAddressFromName(peer_name),
+                  simulator,
+                  simulator->GetAlarmFactory(),
+                  &writer_,
+                  false,
+                  perspective,
+                  CurrentSupportedVersions()),
+      bytes_to_transfer_(0),
+      bytes_transferred_(0),
+      wrong_data_received_(false),
+      transmission_buffer_(new char[kWriteChunkSize]) {
+  connection_.SetSelfAddress(GetAddressFromName(name));
+  connection_.set_visitor(this);
+  connection_.SetEncrypter(ENCRYPTION_FORWARD_SECURE, new NullEncrypter());
+  connection_.SetDecrypter(ENCRYPTION_FORWARD_SECURE, new NullDecrypter());
+  connection_.SetDefaultEncryptionLevel(ENCRYPTION_FORWARD_SECURE);
+
+  // Configure the connection as if it received a handshake.  This is important
+  // primarily because
+  //  - this enables pacing, and
+  //  - this sets the non-handshake timeouts.
+  string error;
+  CryptoHandshakeMessage peer_hello;
+  peer_hello.SetValue(kICSL,
+                      static_cast<uint32_t>(kMaximumIdleTimeoutSecs - 1));
+  QuicConfig config;
+  QuicErrorCode error_code = config.ProcessPeerHello(
+      peer_hello, perspective == Perspective::IS_CLIENT ? SERVER : CLIENT,
+      &error);
+  DCHECK_EQ(error_code, QUIC_NO_ERROR) << "Configuration failed: " << error;
+  connection_.SetFromConfig(config);
+}
+
+void QuicEndpoint::AddBytesToTransfer(QuicByteCount bytes) {
+  if (bytes_to_transfer_ > 0) {
+    Schedule(clock_->Now());
+  }
+
+  bytes_to_transfer_ += bytes;
+  WriteStreamData();
+}
+
+void QuicEndpoint::AcceptPacket(std::unique_ptr<Packet> packet) {
+  if (packet->destination != name_) {
+    return;
+  }
+
+  QuicReceivedPacket received_packet(packet->contents.data(),
+                                     packet->contents.size(), clock_->Now());
+  connection_.ProcessUdpPacket(connection_.self_address(),
+                               connection_.peer_address(), received_packet);
+}
+
+UnconstrainedPortInterface* QuicEndpoint::GetRxPort() {
+  return this;
+}
+
+void QuicEndpoint::SetTxPort(ConstrainedPortInterface* port) {
+  // Any egress done by the endpoint is actually handled by a queue on an NIC.
+  nic_tx_queue_.set_tx_port(port);
+}
+
+// Return the data that |kDataStream| is supposed to have at offset |offset|.
+inline static uint8_t DataAtOffset(QuicStreamOffset offset) {
+  return offset % 256;
+}
+
+void QuicEndpoint::OnPacketDequeued() {
+  if (writer_.IsWriteBlocked() &&
+      (nic_tx_queue_.capacity() - nic_tx_queue_.bytes_queued()) >=
+          kMaxPacketSize) {
+    writer_.SetWritable();
+    connection_.OnCanWrite();
+  }
+}
+
+void QuicEndpoint::OnStreamFrame(const QuicStreamFrame& frame) {
+  // Verify that the data received always matches the output of DataAtOffset().
+  DCHECK(frame.stream_id == kDataStream);
+  for (size_t i = 0; i < frame.data_length; i++) {
+    const QuicStreamOffset absolute_offset = frame.offset + i;
+    if (frame.data_buffer[i] != DataAtOffset(absolute_offset)) {
+      wrong_data_received_ = true;
+    }
+  }
+}
+void QuicEndpoint::OnCanWrite() {
+  WriteStreamData();
+}
+bool QuicEndpoint::WillingAndAbleToWrite() const {
+  return bytes_to_transfer_ != 0;
+}
+bool QuicEndpoint::HasPendingHandshake() const {
+  return false;
+}
+bool QuicEndpoint::HasOpenDynamicStreams() const {
+  return true;
+}
+
+QuicEndpoint::Writer::Writer(QuicEndpoint* endpoint)
+    : QuicDefaultPacketWriter(0), endpoint_(endpoint) {}
+
+QuicEndpoint::Writer::~Writer() {}
+
+WriteResult QuicEndpoint::Writer::WritePacket(const char* buffer,
+                                              size_t buf_len,
+                                              const IPAddress& self_address,
+                                              const IPEndPoint& peer_address,
+                                              PerPacketOptions* options) {
+  DCHECK(!IsWriteBlocked());
+  DCHECK(options == nullptr);
+  DCHECK(buf_len <= kMaxPacketSize);
+
+  DCHECK(self_address == GetAddressFromName(endpoint_->name()).host());
+  DCHECK(peer_address == GetAddressFromName(endpoint_->peer_name_));
+
+  // Instead of losing a packet, become write-blocked when the egress queue is
+  // full.
+  if (endpoint_->nic_tx_queue_.packets_queued() > kTxQueueSize) {
+    set_write_blocked(true);
+    return WriteResult(WRITE_STATUS_BLOCKED, 0);
+  }
+
+  auto packet = gtl::MakeUnique<Packet>();
+  packet->source = endpoint_->name();
+  packet->destination = endpoint_->peer_name_;
+  packet->tx_timestamp = endpoint_->clock_->Now();
+
+  packet->contents = std::string(buffer, buf_len);
+  packet->size = buf_len;
+
+  endpoint_->nic_tx_queue_.AcceptPacket(std::move(packet));
+
+  return WriteResult(WRITE_STATUS_OK, buf_len);
+}
+
+void QuicEndpoint::WriteStreamData() {
+  // Instantiate a bundler which would normally be here due to QuicSession.
+  QuicConnection::ScopedPacketBundler packet_bundler(&connection_,
+                                                     QuicConnection::SEND_ACK);
+
+  while (bytes_to_transfer_ > 0) {
+    // Transfer data in chunks of size at most |kWriteChunkSize|.
+    const size_t transmission_size =
+        std::min(kWriteChunkSize, bytes_to_transfer_);
+    for (size_t i = 0; i < transmission_size; i++) {
+      const QuicStreamOffset offset = bytes_transferred_ + i;
+      transmission_buffer_[i] = DataAtOffset(offset);
+    }
+
+    iovec iov;
+    iov.iov_base = transmission_buffer_.get();
+    iov.iov_len = transmission_size;
+
+    QuicIOVector io_vector(&iov, 1, transmission_size);
+    QuicConsumedData consumed_data = connection_.SendStreamData(
+        kDataStream, io_vector, bytes_transferred_, false, nullptr);
+
+    DCHECK(consumed_data.bytes_consumed <= transmission_size);
+    bytes_transferred_ += consumed_data.bytes_consumed;
+    bytes_to_transfer_ -= consumed_data.bytes_consumed;
+    if (consumed_data.bytes_consumed != transmission_size) {
+      return;
+    }
+  }
+}
+
+QuicEndpointMultiplexer::QuicEndpointMultiplexer(
+    std::string name,
+    std::initializer_list<QuicEndpoint*> endpoints)
+    : Endpoint((*endpoints.begin())->simulator(), name) {
+  for (QuicEndpoint* endpoint : endpoints) {
+    mapping_.emplace(endpoint->name(), endpoint);
+  }
+}
+
+void QuicEndpointMultiplexer::AcceptPacket(std::unique_ptr<Packet> packet) {
+  auto key_value_pair_it = mapping_.find(packet->destination);
+  if (key_value_pair_it == mapping_.end()) {
+    return;
+  }
+
+  key_value_pair_it->second->GetRxPort()->AcceptPacket(std::move(packet));
+}
+UnconstrainedPortInterface* QuicEndpointMultiplexer::GetRxPort() {
+  return this;
+}
+void QuicEndpointMultiplexer::SetTxPort(ConstrainedPortInterface* port) {
+  for (auto& key_value_pair : mapping_) {
+    key_value_pair.second->SetTxPort(port);
+  }
+}
+
+}  // namespace simulation
+}  // namespace net