Implement network performance simulation for remoting perf tests.

remoting/test/fake_socket_factory.cc

+// Copyright 2014 The Chromium Authors. All rights reserved.
+// Use of this source code is governed by a BSD-style license that can be
+// found in the LICENSE file.
+
+#include "remoting/test/fake_socket_factory.h"
+
+#include "base/bind.h"
+#include "base/callback.h"
+#include "base/location.h"
+#include "base/rand_util.h"
+#include "base/single_thread_task_runner.h"
+#include "base/thread_task_runner_handle.h"
+#include "net/base/io_buffer.h"
+#include "remoting/test/leaky_bucket.h"
+#include "third_party/libjingle/source/talk/base/asyncpacketsocket.h"
+
+namespace remoting {
+
+namespace {
+
+double GetNormalRandom(double average, double variance) {
+  // Based on Box-Muller transform, see
+  // http://en.wikipedia.org/wiki/Box_Muller_transform .
+  return average +
+         variance * sqrt(-2.0 * log(1.0 - base::RandDouble())) *

[rmsousa, 2014/08/12 20:42:10]

This should be named stddev. variance is stddev^2 (that is, it would go inside
the sqrt, above, and would be a terribly inconvenient API, since its unit would
be TimeDelta^2).

[Sergey Ulanov, 2014/08/16 00:03:34]

Done.

+             cos(base::RandDouble() * 2.0 * M_PI);
+}
+
+class FakeUdpSocket : public talk_base::AsyncPacketSocket {
+ public:
+  FakeUdpSocket(FakePacketSocketFactory* factory,
+                scoped_refptr<FakeNetworkDispatcher> dispatcher,
+                const talk_base::SocketAddress& local_address);
+  virtual ~FakeUdpSocket();
+
+  void ReceivePacket(const talk_base::SocketAddress& from,
+                     const talk_base::SocketAddress& to,
+                     const scoped_refptr<net::IOBuffer>& data,
+                     int data_size);
+
+  // talk_base::AsyncPacketSocket interface.
+  virtual talk_base::SocketAddress GetLocalAddress() const OVERRIDE;
+  virtual talk_base::SocketAddress GetRemoteAddress() const OVERRIDE;
+  virtual int Send(const void* data, size_t data_size,
+                   const talk_base::PacketOptions& options) OVERRIDE;
+  virtual int SendTo(const void* data, size_t data_size,
+                     const talk_base::SocketAddress& address,
+                     const talk_base::PacketOptions& options) OVERRIDE;
+  virtual int Close() OVERRIDE;
+  virtual State GetState() const OVERRIDE;
+  virtual int GetOption(talk_base::Socket::Option option, int* value) OVERRIDE;
+  virtual int SetOption(talk_base::Socket::Option option, int value) OVERRIDE;
+  virtual int GetError() const OVERRIDE;
+  virtual void SetError(int error) OVERRIDE;
+
+ private:
+  FakePacketSocketFactory* factory_;
+  scoped_refptr<FakeNetworkDispatcher> dispatcher_;
+  talk_base::SocketAddress local_address_;
+  State state_;
+
+  DISALLOW_COPY_AND_ASSIGN(FakeUdpSocket);
+};
+
+FakeUdpSocket::FakeUdpSocket(FakePacketSocketFactory* factory,
+                             scoped_refptr<FakeNetworkDispatcher> dispatcher,
+                             const talk_base::SocketAddress& local_address)
+    : factory_(factory),
+      dispatcher_(dispatcher),
+      local_address_(local_address),
+      state_(STATE_BOUND) {
+}
+
+FakeUdpSocket::~FakeUdpSocket() {
+  factory_->OnSocketDestroyed(local_address_.port());
+}
+
+void FakeUdpSocket::ReceivePacket(const talk_base::SocketAddress& from,
+                                  const talk_base::SocketAddress& to,
+                                  const scoped_refptr<net::IOBuffer>& data,
+                                  int data_size) {
+  SignalReadPacket(
+      this, data->data(), data_size, from, talk_base::CreatePacketTime(0));
+}
+
+talk_base::SocketAddress FakeUdpSocket::GetLocalAddress() const {
+  return local_address_;
+}
+
+talk_base::SocketAddress FakeUdpSocket::GetRemoteAddress() const {
+  NOTREACHED();
+  return talk_base::SocketAddress();
+}
+
+int FakeUdpSocket::Send(const void* data, size_t data_size,
+                        const talk_base::PacketOptions& options) {
+  NOTREACHED();
+  return EINVAL;
+}
+
+int FakeUdpSocket::SendTo(const void* data, size_t data_size,
+                          const talk_base::SocketAddress& address,
+                          const talk_base::PacketOptions& options) {
+  scoped_refptr<net::IOBuffer> buffer = new net::IOBuffer(data_size);
+  memcpy(buffer->data(), data, data_size);
+  dispatcher_->DeliverPacket(local_address_, address, buffer, data_size);
+  return data_size;
+}
+
+int FakeUdpSocket::Close() {
+  state_ = STATE_CLOSED;
+  return 0;
+}
+
+talk_base::AsyncPacketSocket::State FakeUdpSocket::GetState() const {
+  return state_;
+}
+
+int FakeUdpSocket::GetOption(talk_base::Socket::Option option, int* value) {
+  NOTIMPLEMENTED();
+  return -1;
+}
+
+int FakeUdpSocket::SetOption(talk_base::Socket::Option option, int value) {
+  NOTIMPLEMENTED();
+  return -1;
+}
+
+int FakeUdpSocket::GetError() const {
+  return 0;
+}
+
+void FakeUdpSocket::SetError(int error) {
+  NOTREACHED();
+}
+
+}  // namespace
+
+FakePacketSocketFactory::PendingPacket::PendingPacket()
+    : data_size(0) {
+}
+
+FakePacketSocketFactory::PendingPacket::PendingPacket(
+    const talk_base::SocketAddress& from,
+    const talk_base::SocketAddress& to,
+    const scoped_refptr<net::IOBuffer>& data,
+    int data_size)
+    : from(from), to(to), data(data), data_size(data_size) {
+}
+
+FakePacketSocketFactory::PendingPacket::~PendingPacket() {
+}
+
+FakePacketSocketFactory::FakePacketSocketFactory(
+    FakeNetworkDispatcher* dispatcher)
+    : task_runner_(base::ThreadTaskRunnerHandle::Get()),
+      dispatcher_(dispatcher),
+      address_(dispatcher_->AllocateAddress()),
+      out_of_order_rate_(0.0),
+      next_port_(1000),

[rmsousa, 2014/08/12 20:42:10]

nit: 1024 (if the intention is to emulate unix "high" (non-root) ports).

[Sergey Ulanov, 2014/08/16 00:03:34]

Done.

+      weak_factory_(this) {
+  dispatcher_->AddNode(this);
+}
+
+FakePacketSocketFactory::~FakePacketSocketFactory() {
+  CHECK(udp_sockets_.empty());
+  dispatcher_->RemoveNode(this);
+}
+
+void FakePacketSocketFactory::OnSocketDestroyed(int port) {
+  DCHECK(task_runner_->BelongsToCurrentThread());
+  udp_sockets_.erase(port);
+}
+
+void FakePacketSocketFactory::SetBandwidth(int bandwidth, int max_buffer) {
+  DCHECK(task_runner_->BelongsToCurrentThread());
+  if (bandwidth <= 0) {
+    leaky_bucket_.reset();
+  } else {
+    leaky_bucket_.reset(new LeakyBucket(max_buffer, bandwidth));
+  }
+}
+
+void FakePacketSocketFactory::SetLatency(base::TimeDelta average,
+                                         base::TimeDelta variance) {
+  DCHECK(task_runner_->BelongsToCurrentThread());
+  latency_average_ = average;
+  latency_variance_ = variance;
+}
+
+talk_base::AsyncPacketSocket* FakePacketSocketFactory::CreateUdpSocket(
+    const talk_base::SocketAddress& local_address,
+    int min_port, int max_port) {
+  DCHECK(task_runner_->BelongsToCurrentThread());
+
+  int port = -1;
+  if (min_port > 0 && max_port > 0) {
+    for (int i = min_port; i <= max_port; ++i) {
+      if (udp_sockets_.find(i) == udp_sockets_.end()) {
+        port = i;
+        break;
+      }
+    }
+    if (port < 0)
+      return NULL;
+  } else {
+    do {
+      port = next_port_;
+      next_port_ = (next_port_ == 65535) ? 1000 : next_port_ + 1;

[rmsousa, 2014/08/12 20:42:10]

nit: 1024 here as well (probably should define a constant for the high port
boundary)

[Sergey Ulanov, 2014/08/16 00:03:34]

Done.

+    } while (udp_sockets_.find(port) != udp_sockets_.end());
+  }
+
+  CHECK(local_address.ipaddr() == address_);
+
+  FakeUdpSocket* result =
+      new FakeUdpSocket(this, dispatcher_,
+                        talk_base::SocketAddress(local_address.ipaddr(), port));
+
+  udp_sockets_[port] =
+      base::Bind(&FakeUdpSocket::ReceivePacket, base::Unretained(result));
+
+  return result;
+}
+
+talk_base::AsyncPacketSocket* FakePacketSocketFactory::CreateServerTcpSocket(
+    const talk_base::SocketAddress& local_address,
+    int min_port, int max_port,
+    int opts) {
+  return NULL;
+}
+
+talk_base::AsyncPacketSocket* FakePacketSocketFactory::CreateClientTcpSocket(
+    const talk_base::SocketAddress& local_address,
+    const talk_base::SocketAddress& remote_address,
+    const talk_base::ProxyInfo& proxy_info,
+    const std::string& user_agent,
+    int opts) {
+  return NULL;
+}
+
+talk_base::AsyncResolverInterface*
+FakePacketSocketFactory::CreateAsyncResolver() {
+  return NULL;
+}
+
+const scoped_refptr<base::SingleThreadTaskRunner>&
+FakePacketSocketFactory::GetThread() const {
+  return task_runner_;
+}
+
+const talk_base::IPAddress& FakePacketSocketFactory::GetAddress() const {
+  return address_;
+}
+
+void FakePacketSocketFactory::ReceivePacket(
+    const talk_base::SocketAddress& from,
+    const talk_base::SocketAddress& to,
+    const scoped_refptr<net::IOBuffer>& data,
+    int data_size) {
+  DCHECK(task_runner_->BelongsToCurrentThread());
+  DCHECK(to.ipaddr() == address_);
+
+  base::TimeDelta delay;
+
+  if (leaky_bucket_) {
+    delay = leaky_bucket_->AddPacket(data_size);
+    if (delay.is_max()) {
+      // Drop the packet.
+      return;
+    }
+  }
+
+  if (latency_average_ > base::TimeDelta()) {
+    delay += base::TimeDelta::FromMillisecondsD(
+        GetNormalRandom(latency_average_.InMillisecondsF(),
+                        latency_variance_.InMillisecondsF()));
+  }
+  if (delay < base::TimeDelta())
+    delay = base::TimeDelta();
+
+  PendingPacket packet(from, to, data, data_size);
+  pending_packets_.push_back(packet);
+  task_runner_->PostDelayedTask(
+      FROM_HERE,
+      base::Bind(&FakePacketSocketFactory::DoReceivePacket,

[rmsousa, 2014/08/12 20:42:10]

This is a bit weird - since delay has a random component, above, one could post
packet1 with 100ms delay, then packet2 with 80ms delay, then we'd dequeue
packet1 at 80ms then packet2 at 100ms (because we always dequeue from the front)

I guess in the end it's probably best this way (otherwise we would get an
out-of-order rate much higher than any realistic network), but it's worth
explaining this behavior in a comment.

[Sergey Ulanov, 2014/08/16 00:03:34]

Yes, that's by design. I actually implemented it without the queue first. That
unintentionally introduced out-of-order delivery which affects PseudoTCP
performance very badly. So I added the queue to decouple out-of-order packets
from latency. Let me know if you have ideas for a cleaner implementation.

Added more comments now.

+                 weak_factory_.GetWeakPtr()),
+      delay);
+}
+
+void FakePacketSocketFactory::DoReceivePacket() {
+  DCHECK(task_runner_->BelongsToCurrentThread());
+
+  PendingPacket packet;
+  if (pending_packets_.size() > 1 && base::RandDouble() < out_of_order_rate_) {
+    std::list<PendingPacket>::iterator it = pending_packets_.begin();
+    ++it;
+    packet = *it;
+    pending_packets_.erase(it);
+  } else {
+    packet = pending_packets_.front();
+    pending_packets_.pop_front();
+  }
+
+  UdpSocketsMap::iterator iter = udp_sockets_.find(packet.to.port());
+  if (iter == udp_sockets_.end()) {
+    // Invalid port number.
+    return;
+  }
+
+  iter->second.Run(packet.from, packet.to, packet.data, packet.data_size);
+}
+
+}  // namespace remoting