remoting/test/fake_socket_factory.cc

   1 // Copyright 2014 The Chromium Authors. All rights reserved.
   2 // Use of this source code is governed by a BSD-style license that can be
   3 // found in the LICENSE file.
   4
   5 // MSVC++ requires this to be set before any other includes to get M_PI.
   6 #define _USE_MATH_DEFINES
   7
   8 #include "remoting/test/fake_socket_factory.h"
   9
  10 #include <math.h>
  11
  12 #include "base/bind.h"
  13 #include "base/callback.h"
  14 #include "base/location.h"
  15 #include "base/rand_util.h"
  16 #include "base/single_thread_task_runner.h"
  17 #include "base/thread_task_runner_handle.h"
  18 #include "net/base/io_buffer.h"
  19 #include "remoting/test/leaky_bucket.h"
  20 #include "third_party/webrtc/base/asyncpacketsocket.h"
  21
  22 namespace remoting {
  23
  24 namespace {
  25
  26 const int kPortRangeStart = 1024;
  27 const int kPortRangeEnd = 65535;
  28
  29 double GetNormalRandom(double average, double stddev) {
  30   // Based on Box-Muller transform, see
  31   // http://en.wikipedia.org/wiki/Box_Muller_transform .
  32   return average +
  33          stddev * sqrt(-2.0 * log(1.0 - base::RandDouble())) *
  34              cos(base::RandDouble() * 2.0 * M_PI);
  35 }
  36
  37 class FakeUdpSocket : public rtc::AsyncPacketSocket {
  38  public:
  39   FakeUdpSocket(FakePacketSocketFactory* factory,
  40                 scoped_refptr<FakeNetworkDispatcher> dispatcher,
  41                 const rtc::SocketAddress& local_address);
  42   ~FakeUdpSocket() override;
  43
  44   void ReceivePacket(const rtc::SocketAddress& from,
  45                      const rtc::SocketAddress& to,
  46                      const scoped_refptr<net::IOBuffer>& data,
  47                      int data_size);
  48
  49   // rtc::AsyncPacketSocket interface.
  50   rtc::SocketAddress GetLocalAddress() const override;
  51   rtc::SocketAddress GetRemoteAddress() const override;
  52   int Send(const void* data,
  53            size_t data_size,
  54            const rtc::PacketOptions& options) override;
  55   int SendTo(const void* data,
  56              size_t data_size,
  57              const rtc::SocketAddress& address,
  58              const rtc::PacketOptions& options) override;
  59   int Close() override;
  60   State GetState() const override;
  61   int GetOption(rtc::Socket::Option option, int* value) override;
  62   int SetOption(rtc::Socket::Option option, int value) override;
  63   int GetError() const override;
  64   void SetError(int error) override;
  65
  66  private:
  67   FakePacketSocketFactory* factory_;
  68   scoped_refptr<FakeNetworkDispatcher> dispatcher_;
  69   rtc::SocketAddress local_address_;
  70   State state_;
  71
  72   DISALLOW_COPY_AND_ASSIGN(FakeUdpSocket);
  73 };
  74
  75 FakeUdpSocket::FakeUdpSocket(FakePacketSocketFactory* factory,
  76                              scoped_refptr<FakeNetworkDispatcher> dispatcher,
  77                              const rtc::SocketAddress& local_address)
  78     : factory_(factory),
  79       dispatcher_(dispatcher),
  80       local_address_(local_address),
  81       state_(STATE_BOUND) {
  82 }
  83
  84 FakeUdpSocket::~FakeUdpSocket() {
  85   factory_->OnSocketDestroyed(local_address_.port());
  86 }
  87
  88 void FakeUdpSocket::ReceivePacket(const rtc::SocketAddress& from,
  89                                   const rtc::SocketAddress& to,
  90                                   const scoped_refptr<net::IOBuffer>& data,
  91                                   int data_size) {
  92   SignalReadPacket(
  93       this, data->data(), data_size, from, rtc::CreatePacketTime(0));
  94 }
  95
  96 rtc::SocketAddress FakeUdpSocket::GetLocalAddress() const {
  97   return local_address_;
  98 }
  99
 100 rtc::SocketAddress FakeUdpSocket::GetRemoteAddress() const {
 101   NOTREACHED();
 102   return rtc::SocketAddress();
 103 }
 104
 105 int FakeUdpSocket::Send(const void* data, size_t data_size,
 106                         const rtc::PacketOptions& options) {
 107   NOTREACHED();
 108   return EINVAL;
 109 }
 110
 111 int FakeUdpSocket::SendTo(const void* data, size_t data_size,
 112                           const rtc::SocketAddress& address,
 113                           const rtc::PacketOptions& options) {
 114   scoped_refptr<net::IOBuffer> buffer = new net::IOBuffer(data_size);
 115   memcpy(buffer->data(), data, data_size);
 116   dispatcher_->DeliverPacket(local_address_, address, buffer, data_size);
 117   return data_size;
 118 }
 119
 120 int FakeUdpSocket::Close() {
 121   state_ = STATE_CLOSED;
 122   return 0;
 123 }
 124
 125 rtc::AsyncPacketSocket::State FakeUdpSocket::GetState() const {
 126   return state_;
 127 }
 128
 129 int FakeUdpSocket::GetOption(rtc::Socket::Option option, int* value) {
 130   NOTIMPLEMENTED();
 131   return -1;
 132 }
 133
 134 int FakeUdpSocket::SetOption(rtc::Socket::Option option, int value) {
 135   NOTIMPLEMENTED();
 136   return -1;
 137 }
 138
 139 int FakeUdpSocket::GetError() const {
 140   return 0;
 141 }
 142
 143 void FakeUdpSocket::SetError(int error) {
 144   NOTREACHED();
 145 }
 146
 147 }  // namespace
 148
 149 FakePacketSocketFactory::PendingPacket::PendingPacket()
 150     : data_size(0) {
 151 }
 152
 153 FakePacketSocketFactory::PendingPacket::PendingPacket(
 154     const rtc::SocketAddress& from,
 155     const rtc::SocketAddress& to,
 156     const scoped_refptr<net::IOBuffer>& data,
 157     int data_size)
 158     : from(from), to(to), data(data), data_size(data_size) {
 159 }
 160
 161 FakePacketSocketFactory::PendingPacket::~PendingPacket() {
 162 }
 163
 164 FakePacketSocketFactory::FakePacketSocketFactory(
 165     FakeNetworkDispatcher* dispatcher)
 166     : task_runner_(base::ThreadTaskRunnerHandle::Get()),
 167       dispatcher_(dispatcher),
 168       address_(dispatcher_->AllocateAddress()),
 169       out_of_order_rate_(0.0),
 170       next_port_(kPortRangeStart),
 171       weak_factory_(this) {
 172   dispatcher_->AddNode(this);
 173 }
 174
 175 FakePacketSocketFactory::~FakePacketSocketFactory() {
 176   CHECK(udp_sockets_.empty());
 177   dispatcher_->RemoveNode(this);
 178 }
 179
 180 void FakePacketSocketFactory::OnSocketDestroyed(int port) {
 181   DCHECK(task_runner_->BelongsToCurrentThread());
 182   udp_sockets_.erase(port);
 183 }
 184
 185 void FakePacketSocketFactory::SetBandwidth(int bandwidth, int max_buffer) {
 186   DCHECK(task_runner_->BelongsToCurrentThread());
 187   if (bandwidth <= 0) {
 188     leaky_bucket_.reset();
 189   } else {
 190     leaky_bucket_.reset(new LeakyBucket(max_buffer, bandwidth));
 191   }
 192 }
 193
 194 void FakePacketSocketFactory::SetLatency(base::TimeDelta average,
 195                                          base::TimeDelta stddev) {
 196   DCHECK(task_runner_->BelongsToCurrentThread());
 197   latency_average_ = average;
 198   latency_stddev_ = stddev;
 199 }
 200
 201 rtc::AsyncPacketSocket* FakePacketSocketFactory::CreateUdpSocket(
 202     const rtc::SocketAddress& local_address,
 203     int min_port, int max_port) {
 204   DCHECK(task_runner_->BelongsToCurrentThread());
 205
 206   int port = -1;
 207   if (min_port > 0 && max_port > 0) {
 208     for (int i = min_port; i <= max_port; ++i) {
 209       if (udp_sockets_.find(i) == udp_sockets_.end()) {
 210         port = i;
 211         break;
 212       }
 213     }
 214     if (port < 0)
 215       return NULL;
 216   } else {
 217     do {
 218       port = next_port_;
 219       next_port_ =
 220           (next_port_ >= kPortRangeEnd) ? kPortRangeStart : (next_port_ + 1);
 221     } while (udp_sockets_.find(port) != udp_sockets_.end());
 222   }
 223
 224   CHECK(local_address.ipaddr() == address_);
 225
 226   FakeUdpSocket* result =
 227       new FakeUdpSocket(this, dispatcher_,
 228                         rtc::SocketAddress(local_address.ipaddr(), port));
 229
 230   udp_sockets_[port] =
 231       base::Bind(&FakeUdpSocket::ReceivePacket, base::Unretained(result));
 232
 233   return result;
 234 }
 235
 236 rtc::AsyncPacketSocket* FakePacketSocketFactory::CreateServerTcpSocket(
 237     const rtc::SocketAddress& local_address,
 238     int min_port, int max_port,
 239     int opts) {
 240   return NULL;
 241 }
 242
 243 rtc::AsyncPacketSocket* FakePacketSocketFactory::CreateClientTcpSocket(
 244     const rtc::SocketAddress& local_address,
 245     const rtc::SocketAddress& remote_address,
 246     const rtc::ProxyInfo& proxy_info,
 247     const std::string& user_agent,
 248     int opts) {
 249   return NULL;
 250 }
 251
 252 rtc::AsyncResolverInterface*
 253 FakePacketSocketFactory::CreateAsyncResolver() {
 254   return NULL;
 255 }
 256
 257 const scoped_refptr<base::SingleThreadTaskRunner>&
 258 FakePacketSocketFactory::GetThread() const {
 259   return task_runner_;
 260 }
 261
 262 const rtc::IPAddress& FakePacketSocketFactory::GetAddress() const {
 263   return address_;
 264 }
 265
 266 void FakePacketSocketFactory::ReceivePacket(
 267     const rtc::SocketAddress& from,
 268     const rtc::SocketAddress& to,
 269     const scoped_refptr<net::IOBuffer>& data,
 270     int data_size) {
 271   DCHECK(task_runner_->BelongsToCurrentThread());
 272   DCHECK(to.ipaddr() == address_);
 273
 274   base::TimeDelta delay;
 275
 276   if (leaky_bucket_) {
 277     delay = leaky_bucket_->AddPacket(data_size);
 278     if (delay.is_max()) {
 279       // Drop the packet.
 280       return;
 281     }
 282   }
 283
 284   if (latency_average_ > base::TimeDelta()) {
 285     delay += base::TimeDelta::FromMillisecondsD(
 286         GetNormalRandom(latency_average_.InMillisecondsF(),
 287                         latency_stddev_.InMillisecondsF()));
 288   }
 289   if (delay < base::TimeDelta())
 290     delay = base::TimeDelta();
 291
 292   // Put the packet to the |pending_packets_| and post a task for
 293   // DoReceivePackets(). Note that the DoReceivePackets() task posted here may
 294   // deliver a different packet, not the one added to the queue here. This
 295   // would happen if another task gets posted with a shorted delay or when
 296   // |out_of_order_rate_| is greater than 0. It's implemented this way to
 297   // decouple latency variability from out-of-order delivery.
 298   PendingPacket packet(from, to, data, data_size);
 299   pending_packets_.push_back(packet);
 300   task_runner_->PostDelayedTask(
 301       FROM_HERE,
 302       base::Bind(&FakePacketSocketFactory::DoReceivePacket,
 303                  weak_factory_.GetWeakPtr()),
 304       delay);
 305 }
 306
 307 void FakePacketSocketFactory::DoReceivePacket() {
 308   DCHECK(task_runner_->BelongsToCurrentThread());
 309
 310   PendingPacket packet;
 311   if (pending_packets_.size() > 1 && base::RandDouble() < out_of_order_rate_) {
 312     std::list<PendingPacket>::iterator it = pending_packets_.begin();
 313     ++it;
 314     packet = *it;
 315     pending_packets_.erase(it);
 316   } else {
 317     packet = pending_packets_.front();
 318     pending_packets_.pop_front();
 319   }
 320
 321   UdpSocketsMap::iterator iter = udp_sockets_.find(packet.to.port());
 322   if (iter == udp_sockets_.end()) {
 323     // Invalid port number.
 324     return;
 325   }
 326
 327   iter->second.Run(packet.from, packet.to, packet.data, packet.data_size);
 328 }
 329
 330 }  // namespace remoting