media/cast/test/utility/udp_proxy.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 #include <math.h>
   6 #include <stdlib.h>
   7 #include <vector>
   8
   9 #include "media/cast/test/utility/udp_proxy.h"
  10
  11 #include "base/logging.h"
  12 #include "base/rand_util.h"
  13 #include "base/synchronization/waitable_event.h"
  14 #include "base/threading/thread.h"
  15 #include "base/time/default_tick_clock.h"
  16 #include "net/base/io_buffer.h"
  17 #include "net/base/net_errors.h"
  18 #include "net/udp/udp_socket.h"
  19
  20 namespace media {
  21 namespace cast {
  22 namespace test {
  23
  24 const size_t kMaxPacketSize = 65536;
  25
  26 PacketPipe::PacketPipe() {}
  27 PacketPipe::~PacketPipe() {}
  28 void PacketPipe::InitOnIOThread(
  29     const scoped_refptr<base::SingleThreadTaskRunner>& task_runner,
  30     base::TickClock* clock) {
  31   task_runner_ = task_runner;
  32   clock_ = clock;
  33   if (pipe_) {
  34     pipe_->InitOnIOThread(task_runner, clock);
  35   }
  36 }
  37 void PacketPipe::AppendToPipe(scoped_ptr<PacketPipe> pipe) {
  38   if (pipe_) {
  39     pipe_->AppendToPipe(pipe.Pass());
  40   } else {
  41     pipe_ = pipe.Pass();
  42   }
  43 }
  44
  45 // Roughly emulates a buffer inside a device.
  46 // If the buffer is full, packets are dropped.
  47 // Packets are output at a maximum bandwidth.
  48 class Buffer : public PacketPipe {
  49  public:
  50   Buffer(size_t buffer_size, double max_megabits_per_second)
  51       : buffer_size_(0),
  52         max_buffer_size_(buffer_size),
  53         max_megabits_per_second_(max_megabits_per_second),
  54         weak_factory_(this) {
  55     CHECK_GT(max_buffer_size_, 0UL);
  56     CHECK_GT(max_megabits_per_second, 0);
  57   }
  58
  59   virtual void Send(scoped_ptr<Packet> packet) OVERRIDE {
  60     if (packet->size() + buffer_size_ <= max_buffer_size_) {
  61       buffer_size_ += packet->size();
  62       buffer_.push_back(linked_ptr<Packet>(packet.release()));
  63       if (buffer_.size() == 1) {
  64         Schedule();
  65       }
  66     }
  67   }
  68
  69  private:
  70   void Schedule() {
  71     last_schedule_ = clock_->NowTicks();
  72     double megabits = buffer_.front()->size() * 8 / 1000000.0;
  73     double seconds = megabits / max_megabits_per_second_;
  74     int64 microseconds = static_cast<int64>(seconds * 1E6);
  75     task_runner_->PostDelayedTask(
  76         FROM_HERE,
  77         base::Bind(&Buffer::ProcessBuffer, weak_factory_.GetWeakPtr()),
  78         base::TimeDelta::FromMicroseconds(microseconds));
  79   }
  80
  81   void ProcessBuffer() {
  82     int64 bytes_to_send = static_cast<int64>(
  83         (clock_->NowTicks() - last_schedule_).InSecondsF() *
  84         max_megabits_per_second_ * 1E6 / 8);
  85     if (bytes_to_send < static_cast<int64>(buffer_.front()->size())) {
  86       bytes_to_send = buffer_.front()->size();
  87     }
  88     while (!buffer_.empty() &&
  89            static_cast<int64>(buffer_.front()->size()) <= bytes_to_send) {
  90       CHECK(!buffer_.empty());
  91       scoped_ptr<Packet> packet(buffer_.front().release());
  92       bytes_to_send -= packet->size();
  93       buffer_size_ -= packet->size();
  94       buffer_.pop_front();
  95       pipe_->Send(packet.Pass());
  96     }
  97     if (!buffer_.empty()) {
  98       Schedule();
  99     }
 100   }
 101
 102   std::deque<linked_ptr<Packet> > buffer_;
 103   base::TimeTicks last_schedule_;
 104   size_t buffer_size_;
 105   size_t max_buffer_size_;
 106   double max_megabits_per_second_;  // megabits per second
 107   base::WeakPtrFactory<Buffer> weak_factory_;
 108 };
 109
 110 scoped_ptr<PacketPipe> NewBuffer(size_t buffer_size, double bandwidth) {
 111   return scoped_ptr<PacketPipe>(new Buffer(buffer_size, bandwidth)).Pass();
 112 }
 113
 114 class RandomDrop : public PacketPipe {
 115  public:
 116   RandomDrop(double drop_fraction)
 117       : drop_fraction_(static_cast<int>(drop_fraction * RAND_MAX)) {}
 118
 119   virtual void Send(scoped_ptr<Packet> packet) OVERRIDE {
 120     if (rand() > drop_fraction_) {
 121       pipe_->Send(packet.Pass());
 122     }
 123   }
 124
 125  private:
 126   int drop_fraction_;
 127 };
 128
 129 scoped_ptr<PacketPipe> NewRandomDrop(double drop_fraction) {
 130   return scoped_ptr<PacketPipe>(new RandomDrop(drop_fraction)).Pass();
 131 }
 132
 133 class SimpleDelayBase : public PacketPipe {
 134  public:
 135   SimpleDelayBase() : weak_factory_(this) {}
 136   virtual ~SimpleDelayBase() {}
 137
 138   virtual void Send(scoped_ptr<Packet> packet) OVERRIDE {
 139     double seconds = GetDelay();
 140     task_runner_->PostDelayedTask(
 141         FROM_HERE,
 142         base::Bind(&SimpleDelayBase::SendInternal,
 143                    weak_factory_.GetWeakPtr(),
 144                    base::Passed(&packet)),
 145         base::TimeDelta::FromMicroseconds(static_cast<int64>(seconds * 1E6)));
 146   }
 147  protected:
 148   virtual double GetDelay() = 0;
 149
 150  private:
 151   virtual void SendInternal(scoped_ptr<Packet> packet) {
 152     pipe_->Send(packet.Pass());
 153   }
 154
 155   base::WeakPtrFactory<SimpleDelayBase> weak_factory_;
 156 };
 157
 158 class ConstantDelay : public SimpleDelayBase {
 159  public:
 160   ConstantDelay(double delay_seconds) : delay_seconds_(delay_seconds) {}
 161   virtual double GetDelay() OVERRIDE {
 162     return delay_seconds_;
 163   }
 164
 165  private:
 166   double delay_seconds_;
 167 };
 168
 169 scoped_ptr<PacketPipe> NewConstantDelay(double delay_seconds) {
 170   return scoped_ptr<PacketPipe>(new ConstantDelay(delay_seconds)).Pass();
 171 }
 172
 173 class RandomUnsortedDelay : public SimpleDelayBase {
 174  public:
 175   RandomUnsortedDelay(double random_delay) : random_delay_(random_delay) {}
 176
 177   virtual double GetDelay() OVERRIDE {
 178     return random_delay_ * base::RandDouble();
 179   }
 180
 181  private:
 182   double random_delay_;
 183 };
 184
 185 scoped_ptr<PacketPipe> NewRandomUnsortedDelay(double random_delay) {
 186   return scoped_ptr<PacketPipe>(new RandomUnsortedDelay(random_delay)).Pass();
 187 }
 188
 189 class DuplicateAndDelay : public RandomUnsortedDelay {
 190  public:
 191   DuplicateAndDelay(double delay_min,
 192                     double random_delay) :
 193       RandomUnsortedDelay(random_delay),
 194       delay_min_(delay_min) {
 195   }
 196   virtual void Send(scoped_ptr<Packet> packet) OVERRIDE {
 197     pipe_->Send(scoped_ptr<Packet>(new Packet(*packet.get())));
 198     RandomUnsortedDelay::Send(packet.Pass());
 199   }
 200   virtual double GetDelay() OVERRIDE {
 201     return RandomUnsortedDelay::GetDelay() + delay_min_;
 202   }
 203  private:
 204   double delay_min_;
 205 };
 206
 207 scoped_ptr<PacketPipe> NewDuplicateAndDelay(double delay_min,
 208                                             double random_delay) {
 209   return scoped_ptr<PacketPipe>(
 210       new DuplicateAndDelay(delay_min, random_delay)).Pass();
 211 }
 212
 213 class RandomSortedDelay : public PacketPipe {
 214  public:
 215   RandomSortedDelay(double random_delay,
 216                     double extra_delay,
 217                     double seconds_between_extra_delay)
 218       : random_delay_(random_delay),
 219         extra_delay_(extra_delay),
 220         seconds_between_extra_delay_(seconds_between_extra_delay),
 221         weak_factory_(this) {}
 222
 223   virtual void Send(scoped_ptr<Packet> packet) OVERRIDE {
 224     buffer_.push_back(linked_ptr<Packet>(packet.release()));
 225     if (buffer_.size() == 1) {
 226       next_send_ = std::max(
 227           clock_->NowTicks() +
 228           base::TimeDelta::FromSecondsD(base::RandDouble() * random_delay_),
 229           next_send_);
 230       ProcessBuffer();
 231     }
 232   }
 233   virtual void InitOnIOThread(
 234       const scoped_refptr<base::SingleThreadTaskRunner>& task_runner,
 235       base::TickClock* clock) OVERRIDE {
 236     PacketPipe::InitOnIOThread(task_runner, clock);
 237     // As we start the stream, assume that we are in a random
 238     // place between two extra delays, thus multiplier = 1.0;
 239     ScheduleExtraDelay(1.0);
 240   }
 241
 242  private:
 243   void ScheduleExtraDelay(double mult) {
 244     double seconds = seconds_between_extra_delay_ * mult * base::RandDouble();
 245     int64 microseconds = static_cast<int64>(seconds * 1E6);
 246     task_runner_->PostDelayedTask(
 247         FROM_HERE,
 248         base::Bind(&RandomSortedDelay::CauseExtraDelay,
 249                    weak_factory_.GetWeakPtr()),
 250         base::TimeDelta::FromMicroseconds(microseconds));
 251   }
 252
 253   void CauseExtraDelay() {
 254     next_send_ = std::max<base::TimeTicks>(
 255         clock_->NowTicks() +
 256         base::TimeDelta::FromMicroseconds(
 257             static_cast<int64>(extra_delay_ * 1E6)),
 258         next_send_);
 259     // An extra delay just happened, wait up to seconds_between_extra_delay_*2
 260     // before scheduling another one to make the average equal to
 261     // seconds_between_extra_delay_.
 262     ScheduleExtraDelay(2.0);
 263   }
 264
 265   void ProcessBuffer() {
 266     base::TimeTicks now = clock_->NowTicks();
 267     while (!buffer_.empty() && next_send_ <= now) {
 268       scoped_ptr<Packet> packet(buffer_.front().release());
 269       pipe_->Send(packet.Pass());
 270       buffer_.pop_front();
 271
 272       next_send_ += base::TimeDelta::FromSecondsD(
 273           base::RandDouble() * random_delay_);
 274     }
 275
 276     if (!buffer_.empty()) {
 277       task_runner_->PostDelayedTask(
 278           FROM_HERE,
 279           base::Bind(&RandomSortedDelay::ProcessBuffer,
 280                      weak_factory_.GetWeakPtr()),
 281           next_send_ - now);
 282     }
 283   }
 284
 285   base::TimeTicks block_until_;
 286   std::deque<linked_ptr<Packet> > buffer_;
 287   double random_delay_;
 288   double extra_delay_;
 289   double seconds_between_extra_delay_;
 290   base::WeakPtrFactory<RandomSortedDelay> weak_factory_;
 291   base::TimeTicks next_send_;
 292 };
 293
 294 scoped_ptr<PacketPipe> NewRandomSortedDelay(
 295     double random_delay,
 296     double extra_delay,
 297     double seconds_between_extra_delay) {
 298   return scoped_ptr<PacketPipe>(
 299              new RandomSortedDelay(
 300                  random_delay, extra_delay, seconds_between_extra_delay))
 301       .Pass();
 302 }
 303
 304 class NetworkGlitchPipe : public PacketPipe {
 305  public:
 306   NetworkGlitchPipe(double average_work_time, double average_outage_time)
 307       : works_(false),
 308         max_work_time_(average_work_time * 2),
 309         max_outage_time_(average_outage_time * 2),
 310         weak_factory_(this) {}
 311
 312   virtual void InitOnIOThread(
 313       const scoped_refptr<base::SingleThreadTaskRunner>& task_runner,
 314       base::TickClock* clock) OVERRIDE {
 315     PacketPipe::InitOnIOThread(task_runner, clock);
 316     Flip();
 317   }
 318
 319   virtual void Send(scoped_ptr<Packet> packet) OVERRIDE {
 320     if (works_) {
 321       pipe_->Send(packet.Pass());
 322     }
 323   }
 324
 325  private:
 326   void Flip() {
 327     works_ = !works_;
 328     double seconds = base::RandDouble() *
 329         (works_ ? max_work_time_ : max_outage_time_);
 330     int64 microseconds = static_cast<int64>(seconds * 1E6);
 331     task_runner_->PostDelayedTask(
 332         FROM_HERE,
 333         base::Bind(&NetworkGlitchPipe::Flip, weak_factory_.GetWeakPtr()),
 334         base::TimeDelta::FromMicroseconds(microseconds));
 335   }
 336
 337   bool works_;
 338   double max_work_time_;
 339   double max_outage_time_;
 340   base::WeakPtrFactory<NetworkGlitchPipe> weak_factory_;
 341 };
 342
 343 scoped_ptr<PacketPipe> NewNetworkGlitchPipe(double average_work_time,
 344                                             double average_outage_time) {
 345   return scoped_ptr<PacketPipe>(
 346              new NetworkGlitchPipe(average_work_time, average_outage_time))
 347       .Pass();
 348 }
 349
 350
 351 // Internal buffer object for a client of the IPP model.
 352 class InterruptedPoissonProcess::InternalBuffer : public PacketPipe {
 353  public:
 354   InternalBuffer(base::WeakPtr<InterruptedPoissonProcess> ipp,
 355                  size_t size)
 356       : ipp_(ipp),
 357         stored_size_(0),
 358         stored_limit_(size),
 359         clock_(NULL),
 360         weak_factory_(this) {
 361   }
 362
 363   virtual void Send(scoped_ptr<Packet> packet) OVERRIDE {
 364     // Drop if buffer is full.
 365     if (stored_size_ >= stored_limit_)
 366       return;
 367     stored_size_ += packet->size();
 368     buffer_.push_back(linked_ptr<Packet>(packet.release()));
 369     buffer_time_.push_back(clock_->NowTicks());
 370     DCHECK(buffer_.size() == buffer_time_.size());
 371   }
 372
 373   virtual void InitOnIOThread(
 374       const scoped_refptr<base::SingleThreadTaskRunner>& task_runner,
 375       base::TickClock* clock) OVERRIDE {
 376     clock_ = clock;
 377     if (ipp_)
 378       ipp_->InitOnIOThread(task_runner, clock);
 379     PacketPipe::InitOnIOThread(task_runner, clock);
 380   }
 381
 382   void SendOnePacket() {
 383     scoped_ptr<Packet> packet(buffer_.front().release());
 384     stored_size_ -= packet->size();
 385     buffer_.pop_front();
 386     buffer_time_.pop_front();
 387     pipe_->Send(packet.Pass());
 388     DCHECK(buffer_.size() == buffer_time_.size());
 389   }
 390
 391   bool Empty() const {
 392     return buffer_.empty();
 393   }
 394
 395   base::TimeTicks FirstPacketTime() const {
 396     DCHECK(!buffer_time_.empty());
 397     return buffer_time_.front();
 398   }
 399
 400   base::WeakPtr<InternalBuffer> GetWeakPtr() {
 401     return weak_factory_.GetWeakPtr();
 402
 403   }
 404
 405  private:
 406   const base::WeakPtr<InterruptedPoissonProcess> ipp_;
 407   size_t stored_size_;
 408   const size_t stored_limit_;
 409   std::deque<linked_ptr<Packet> > buffer_;
 410   std::deque<base::TimeTicks> buffer_time_;
 411   base::TickClock* clock_;
 412   base::WeakPtrFactory<InternalBuffer> weak_factory_;
 413
 414   DISALLOW_COPY_AND_ASSIGN(InternalBuffer);
 415 };
 416
 417 InterruptedPoissonProcess::InterruptedPoissonProcess(
 418     const std::vector<double>& average_rates,
 419     double coef_burstiness,
 420     double coef_variance,
 421     uint32 rand_seed)
 422     : clock_(NULL),
 423       average_rates_(average_rates),
 424       coef_burstiness_(coef_burstiness),
 425       coef_variance_(coef_variance),
 426       rate_index_(0),
 427       on_state_(true),
 428       weak_factory_(this) {
 429   mt_rand_.init_genrand(rand_seed);
 430   DCHECK(!average_rates.empty());
 431   ComputeRates();
 432 }
 433
 434 InterruptedPoissonProcess::~InterruptedPoissonProcess() {
 435 }
 436
 437 void InterruptedPoissonProcess::InitOnIOThread(
 438     const scoped_refptr<base::SingleThreadTaskRunner>& task_runner,
 439     base::TickClock* clock) {
 440   // Already initialized and started.
 441   if (task_runner_.get() && clock_)
 442     return;
 443   task_runner_ = task_runner;
 444   clock_ = clock;
 445   UpdateRates();
 446   SwitchOn();
 447   SendPacket();
 448 }
 449
 450 scoped_ptr<PacketPipe> InterruptedPoissonProcess::NewBuffer(size_t size) {
 451   scoped_ptr<InternalBuffer> buffer(
 452       new InternalBuffer(weak_factory_.GetWeakPtr(), size));
 453   send_buffers_.push_back(buffer->GetWeakPtr());
 454   return buffer.PassAs<PacketPipe>();
 455 }
 456
 457 base::TimeDelta InterruptedPoissonProcess::NextEvent(double rate) {
 458   // Rate is per milliseconds.
 459   // The time until next event is exponentially distributed to the
 460   // inverse of |rate|.
 461   return base::TimeDelta::FromMillisecondsD(
 462       fabs(-log(1.0 - RandDouble()) / rate));
 463 }
 464
 465 double InterruptedPoissonProcess::RandDouble() {
 466   // Generate a 64-bits random number from MT19937 and then convert
 467   // it to double.
 468   uint64 rand = mt_rand_.genrand_int32();
 469   rand <<= 32;
 470   rand |= mt_rand_.genrand_int32();
 471   return base::BitsToOpenEndedUnitInterval(rand);
 472 }
 473
 474 void InterruptedPoissonProcess::ComputeRates() {
 475   double avg_rate = average_rates_[rate_index_];
 476
 477   send_rate_ = avg_rate / coef_burstiness_;
 478   switch_off_rate_ =
 479       2 * avg_rate * (1 - coef_burstiness_) * (1 - coef_burstiness_) /
 480       coef_burstiness_ / (coef_variance_ - 1);
 481   switch_on_rate_ =
 482       2 * avg_rate * (1 - coef_burstiness_) / (coef_variance_ - 1);
 483 }
 484
 485 void InterruptedPoissonProcess::UpdateRates() {
 486   ComputeRates();
 487
 488   // Rates are updated once per second.
 489   rate_index_ = (rate_index_ + 1) % average_rates_.size();
 490   task_runner_->PostDelayedTask(
 491       FROM_HERE,
 492       base::Bind(&InterruptedPoissonProcess::UpdateRates,
 493                  weak_factory_.GetWeakPtr()),
 494       base::TimeDelta::FromSeconds(1));
 495 }
 496
 497 void InterruptedPoissonProcess::SwitchOff() {
 498   on_state_ = false;
 499   task_runner_->PostDelayedTask(
 500       FROM_HERE,
 501       base::Bind(&InterruptedPoissonProcess::SwitchOn,
 502                  weak_factory_.GetWeakPtr()),
 503       NextEvent(switch_on_rate_));
 504 }
 505
 506 void InterruptedPoissonProcess::SwitchOn() {
 507   on_state_ = true;
 508   task_runner_->PostDelayedTask(
 509       FROM_HERE,
 510       base::Bind(&InterruptedPoissonProcess::SwitchOff,
 511                  weak_factory_.GetWeakPtr()),
 512       NextEvent(switch_off_rate_));
 513 }
 514
 515 void InterruptedPoissonProcess::SendPacket() {
 516   task_runner_->PostDelayedTask(
 517       FROM_HERE,
 518       base::Bind(&InterruptedPoissonProcess::SendPacket,
 519                  weak_factory_.GetWeakPtr()),
 520       NextEvent(send_rate_));
 521
 522   // If OFF then don't send.
 523   if (!on_state_)
 524     return;
 525
 526   // Find the earliest packet to send.
 527   base::TimeTicks earliest_time;
 528   for (size_t i = 0; i < send_buffers_.size(); ++i) {
 529     if (!send_buffers_[i])
 530       continue;
 531     if (send_buffers_[i]->Empty())
 532       continue;
 533     if (earliest_time.is_null() ||
 534         send_buffers_[i]->FirstPacketTime() < earliest_time)
 535       earliest_time = send_buffers_[i]->FirstPacketTime();
 536   }
 537   for (size_t i = 0; i < send_buffers_.size(); ++i) {
 538     if (!send_buffers_[i])
 539       continue;
 540     if (send_buffers_[i]->Empty())
 541       continue;
 542     if (send_buffers_[i]->FirstPacketTime() != earliest_time)
 543       continue;
 544     send_buffers_[i]->SendOnePacket();
 545     break;
 546   }
 547 }
 548
 549 class UDPProxyImpl;
 550
 551 class PacketSender : public PacketPipe {
 552  public:
 553   PacketSender(UDPProxyImpl* udp_proxy, const net::IPEndPoint* destination)
 554       : udp_proxy_(udp_proxy), destination_(destination) {}
 555   virtual void Send(scoped_ptr<Packet> packet) OVERRIDE;
 556   virtual void AppendToPipe(scoped_ptr<PacketPipe> pipe) OVERRIDE {
 557     NOTREACHED();
 558   }
 559
 560  private:
 561   UDPProxyImpl* udp_proxy_;
 562   const net::IPEndPoint* destination_;  // not owned
 563 };
 564
 565 namespace {
 566 void BuildPipe(scoped_ptr<PacketPipe>* pipe, PacketPipe* next) {
 567   if (*pipe) {
 568     (*pipe)->AppendToPipe(scoped_ptr<PacketPipe>(next).Pass());
 569   } else {
 570     pipe->reset(next);
 571   }
 572 }
 573 }  // namespace
 574
 575 scoped_ptr<PacketPipe> GoodNetwork() {
 576   // This represents the buffer on the sender.
 577   scoped_ptr<PacketPipe> pipe;
 578   BuildPipe(&pipe, new Buffer(2 << 20, 50));
 579   BuildPipe(&pipe, new ConstantDelay(1E-3));
 580   BuildPipe(&pipe, new RandomSortedDelay(1E-3, 2E-3, 3));
 581   // This represents the buffer on the receiving device.
 582   BuildPipe(&pipe, new Buffer(2 << 20, 50));
 583   return pipe.Pass();
 584 }
 585
 586 scoped_ptr<PacketPipe> WifiNetwork() {
 587   // This represents the buffer on the sender.
 588   scoped_ptr<PacketPipe> pipe;
 589   BuildPipe(&pipe, new Buffer(256 << 10, 20));
 590   BuildPipe(&pipe, new RandomDrop(0.005));
 591   // This represents the buffer on the router.
 592   BuildPipe(&pipe, new ConstantDelay(1E-3));
 593   BuildPipe(&pipe, new RandomSortedDelay(1E-3, 20E-3, 3));
 594   BuildPipe(&pipe, new Buffer(256 << 10, 20));
 595   BuildPipe(&pipe, new ConstantDelay(1E-3));
 596   BuildPipe(&pipe, new RandomSortedDelay(1E-3, 20E-3, 3));
 597   BuildPipe(&pipe, new RandomDrop(0.005));
 598   // This represents the buffer on the receiving device.
 599   BuildPipe(&pipe, new Buffer(256 << 10, 20));
 600   return pipe.Pass();
 601 }
 602
 603 scoped_ptr<PacketPipe> BadNetwork() {
 604   scoped_ptr<PacketPipe> pipe;
 605   // This represents the buffer on the sender.
 606   BuildPipe(&pipe, new Buffer(64 << 10, 5)); // 64 kb buf, 5mbit/s
 607   BuildPipe(&pipe, new RandomDrop(0.05));  // 5% packet drop
 608   BuildPipe(&pipe, new RandomSortedDelay(2E-3, 20E-3, 1));
 609   // This represents the buffer on the router.
 610   BuildPipe(&pipe, new Buffer(64 << 10, 5));  // 64 kb buf, 4mbit/s
 611   BuildPipe(&pipe, new ConstantDelay(1E-3));
 612   // Random 40ms every other second
 613   //  BuildPipe(&pipe, new NetworkGlitchPipe(2, 40E-1));
 614   BuildPipe(&pipe, new RandomUnsortedDelay(5E-3));
 615   // This represents the buffer on the receiving device.
 616   BuildPipe(&pipe, new Buffer(64 << 10, 5));  // 64 kb buf, 5mbit/s
 617   return pipe.Pass();
 618 }
 619
 620
 621 scoped_ptr<PacketPipe> EvilNetwork() {
 622   // This represents the buffer on the sender.
 623   scoped_ptr<PacketPipe> pipe;
 624   BuildPipe(&pipe, new Buffer(4 << 10, 5));  // 4 kb buf, 2mbit/s
 625   // This represents the buffer on the router.
 626   BuildPipe(&pipe, new RandomDrop(0.1));  // 10% packet drop
 627   BuildPipe(&pipe, new RandomSortedDelay(20E-3, 60E-3, 1));
 628   BuildPipe(&pipe, new Buffer(4 << 10, 2));  // 4 kb buf, 2mbit/s
 629   BuildPipe(&pipe, new RandomDrop(0.1));  // 10% packet drop
 630   BuildPipe(&pipe, new ConstantDelay(1E-3));
 631   BuildPipe(&pipe, new NetworkGlitchPipe(2.0, 0.3));
 632   BuildPipe(&pipe, new RandomUnsortedDelay(20E-3));
 633   // This represents the buffer on the receiving device.
 634   BuildPipe(&pipe, new Buffer(4 << 10, 2));  // 4 kb buf, 2mbit/s
 635   return pipe.Pass();
 636 }
 637
 638 class UDPProxyImpl : public UDPProxy {
 639  public:
 640   UDPProxyImpl(const net::IPEndPoint& local_port,
 641                const net::IPEndPoint& destination,
 642                scoped_ptr<PacketPipe> to_dest_pipe,
 643                scoped_ptr<PacketPipe> from_dest_pipe,
 644                net::NetLog* net_log)
 645       : local_port_(local_port),
 646         destination_(destination),
 647         destination_is_mutable_(destination.address().empty()),
 648         proxy_thread_("media::cast::test::UdpProxy Thread"),
 649         to_dest_pipe_(to_dest_pipe.Pass()),
 650         from_dest_pipe_(from_dest_pipe.Pass()),
 651         blocked_(false),
 652         weak_factory_(this) {
 653     proxy_thread_.StartWithOptions(
 654         base::Thread::Options(base::MessageLoop::TYPE_IO, 0));
 655     base::WaitableEvent start_event(false, false);
 656     proxy_thread_.message_loop_proxy()->PostTask(
 657         FROM_HERE,
 658         base::Bind(&UDPProxyImpl::Start,
 659                    base::Unretained(this),
 660                    base::Unretained(&start_event),
 661                    net_log));
 662     start_event.Wait();
 663   }
 664
 665   virtual ~UDPProxyImpl() {
 666     base::WaitableEvent stop_event(false, false);
 667     proxy_thread_.message_loop_proxy()->PostTask(
 668         FROM_HERE,
 669         base::Bind(&UDPProxyImpl::Stop,
 670                    base::Unretained(this),
 671                    base::Unretained(&stop_event)));
 672     stop_event.Wait();
 673     proxy_thread_.Stop();
 674   }
 675
 676   void Send(scoped_ptr<Packet> packet,
 677             const net::IPEndPoint& destination) {
 678     if (blocked_) {
 679       LOG(ERROR) << "Cannot write packet right now: blocked";
 680       return;
 681     }
 682
 683     VLOG(1) << "Sending packet, len = " << packet->size();
 684     // We ignore all problems, callbacks and errors.
 685     // If it didn't work we just drop the packet at and call it a day.
 686     scoped_refptr<net::IOBuffer> buf =
 687         new net::WrappedIOBuffer(reinterpret_cast<char*>(&packet->front()));
 688     size_t buf_size = packet->size();
 689     int result;
 690     if (destination.address().empty()) {
 691       VLOG(1) << "Destination has not been set yet.";
 692       result = net::ERR_INVALID_ARGUMENT;
 693     } else {
 694       VLOG(1) << "Destination:" << destination.ToString();
 695       result = socket_->SendTo(buf.get(),
 696                                static_cast<int>(buf_size),
 697                                destination,
 698                                base::Bind(&UDPProxyImpl::AllowWrite,
 699                                           weak_factory_.GetWeakPtr(),
 700                                           buf,
 701                                           base::Passed(&packet)));
 702     }
 703     if (result == net::ERR_IO_PENDING) {
 704       blocked_ = true;
 705     } else if (result < 0) {
 706       LOG(ERROR) << "Failed to write packet.";
 707     }
 708   }
 709
 710  private:
 711   void Start(base::WaitableEvent* start_event,
 712              net::NetLog* net_log) {
 713     socket_.reset(new net::UDPSocket(net::DatagramSocket::DEFAULT_BIND,
 714                                      net::RandIntCallback(),
 715                                      net_log,
 716                                      net::NetLog::Source()));
 717     BuildPipe(&to_dest_pipe_, new PacketSender(this, &destination_));
 718     BuildPipe(&from_dest_pipe_, new PacketSender(this, &return_address_));
 719     to_dest_pipe_->InitOnIOThread(base::MessageLoopProxy::current(),
 720                                   &tick_clock_);
 721     from_dest_pipe_->InitOnIOThread(base::MessageLoopProxy::current(),
 722                                     &tick_clock_);
 723
 724     VLOG(0) << "From:" << local_port_.ToString();
 725     if (!destination_is_mutable_)
 726       VLOG(0) << "To:" << destination_.ToString();
 727
 728     CHECK_GE(socket_->Bind(local_port_), 0);
 729
 730     start_event->Signal();
 731     PollRead();
 732   }
 733
 734   void Stop(base::WaitableEvent* stop_event) {
 735     to_dest_pipe_.reset(NULL);
 736     from_dest_pipe_.reset(NULL);
 737     socket_.reset(NULL);
 738     stop_event->Signal();
 739   }
 740
 741   void ProcessPacket(scoped_refptr<net::IOBuffer> recv_buf, int len) {
 742     DCHECK_NE(len, net::ERR_IO_PENDING);
 743     VLOG(1) << "Got packet, len = " << len;
 744     if (len < 0) {
 745       LOG(WARNING) << "Socket read error: " << len;
 746       return;
 747     }
 748     packet_->resize(len);
 749     if (destination_is_mutable_ && set_destination_next_ &&
 750         !(recv_address_ == return_address_) &&
 751         !(recv_address_ == destination_)) {
 752       destination_ = recv_address_;
 753     }
 754     if (recv_address_ == destination_) {
 755       set_destination_next_ = false;
 756       from_dest_pipe_->Send(packet_.Pass());
 757     } else {
 758       set_destination_next_ = true;
 759       VLOG(1) << "Return address = " << recv_address_.ToString();
 760       return_address_ = recv_address_;
 761       to_dest_pipe_->Send(packet_.Pass());
 762     }
 763   }
 764
 765   void ReadCallback(scoped_refptr<net::IOBuffer> recv_buf, int len) {
 766     ProcessPacket(recv_buf, len);
 767     PollRead();
 768   }
 769
 770   void PollRead() {
 771     while (true) {
 772       packet_.reset(new Packet(kMaxPacketSize));
 773       scoped_refptr<net::IOBuffer> recv_buf =
 774           new net::WrappedIOBuffer(reinterpret_cast<char*>(&packet_->front()));
 775       int len = socket_->RecvFrom(
 776           recv_buf.get(),
 777           kMaxPacketSize,
 778           &recv_address_,
 779           base::Bind(
 780               &UDPProxyImpl::ReadCallback, base::Unretained(this), recv_buf));
 781       if (len == net::ERR_IO_PENDING)
 782         break;
 783       ProcessPacket(recv_buf, len);
 784     }
 785   }
 786
 787   void AllowWrite(scoped_refptr<net::IOBuffer> buf,
 788                   scoped_ptr<Packet> packet,
 789                   int unused_len) {
 790     DCHECK(blocked_);
 791     blocked_ = false;
 792   }
 793
 794   // Input
 795   net::IPEndPoint local_port_;
 796
 797   net::IPEndPoint destination_;
 798   bool destination_is_mutable_;
 799
 800   net::IPEndPoint return_address_;
 801   bool set_destination_next_;
 802
 803   base::DefaultTickClock tick_clock_;
 804   base::Thread proxy_thread_;
 805   scoped_ptr<net::UDPSocket> socket_;
 806   scoped_ptr<PacketPipe> to_dest_pipe_;
 807   scoped_ptr<PacketPipe> from_dest_pipe_;
 808
 809   // For receiving.
 810   net::IPEndPoint recv_address_;
 811   scoped_ptr<Packet> packet_;
 812
 813   // For sending.
 814   bool blocked_;
 815
 816   base::WeakPtrFactory<UDPProxyImpl> weak_factory_;
 817 };
 818
 819 void PacketSender::Send(scoped_ptr<Packet> packet) {
 820   udp_proxy_->Send(packet.Pass(), *destination_);
 821 }
 822
 823 scoped_ptr<UDPProxy> UDPProxy::Create(
 824     const net::IPEndPoint& local_port,
 825     const net::IPEndPoint& destination,
 826     scoped_ptr<PacketPipe> to_dest_pipe,
 827     scoped_ptr<PacketPipe> from_dest_pipe,
 828     net::NetLog* net_log) {
 829   scoped_ptr<UDPProxy> ret(new UDPProxyImpl(local_port,
 830                                             destination,
 831                                             to_dest_pipe.Pass(),
 832                                             from_dest_pipe.Pass(),
 833                                             net_log));
 834   return ret.Pass();
 835 }
 836
 837 }  // namespace test
 838 }  // namespace cast
 839 }  // namespace media