tap_proxy - bad network simulation for linux

Three files:
shadow.sh - setup a tap network which shadows eth1
tap_proxy - forward traffic between two tap devices with delay/drops
netload.py - create TCP network traffic

See shadow.sh for more detailed information.

Review URL: https://codereview.chromium.org/415403009

Cr-Commit-Position: refs/heads/master@{#288737}
git-svn-id: svn://svn.chromium.org/chrome/trunk/src@288737 0039d316-1c4b-4281-b951-d872f2087c98

6 files changed, 619 insertions, 29 deletions

diff --git a/media/cast/cast_testing.gypi b/media/cast/cast_testing.gypi
index 7cfc50c..d0c0f04 100644
--- a/media/cast/cast_testing.gypi
+++ b/media/cast/cast_testing.gypi
@@ -318,6 +318,29 @@
       'sources': [
         'test/utility/udp_proxy_main.cc',
       ],
-    }
+    },
+  ], # targets
+
+  'conditions': [
+    ['OS=="linux"',
+      { 'targets': [
+          {
+            'target_name': 'tap_proxy',
+            'type': 'executable',
+            'include_dirs': [
+              '<(DEPTH)/',
+            ],
+            'dependencies': [
+              'cast_test_utility',
+              '<(DEPTH)/base/base.gyp:base',
+              '<(DEPTH)/media/media.gyp:media',
+            ],
+            'sources': [
+              'test/utility/tap_proxy.cc',
+            ],
+          }
+        ]
+      }
+    ]
   ], # targets
 }
diff --git a/media/cast/test/utility/netload.py b/media/cast/test/utility/netload.py
new file mode 100755
index 0000000..248eca2
--- /dev/null
+++ b/media/cast/test/utility/netload.py
@@ -0,0 +1,100 @@
+#!/usr/bin/env python
+# 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.
+#
+# Simple client/server script for generating an unlimited TCP stream.
+# see shadow.sh for how it's intended to be used.
+
+import socket
+import sys
+import thread
+import time
+
+sent = 0
+received = 0
+
+def Sink(socket):
+  global received
+  while True:
+    tmp = socket.recv(4096)
+    received += len(tmp)
+    if not tmp:
+      break;
+
+def Spew(socket):
+  global sent
+  data = " " * 4096
+  while True:
+    tmp = socket.send(data)
+    if tmp <= 0:
+      break
+    sent += tmp;
+
+def PrintStats():
+  global sent
+  global received
+  last_report = time.time()
+  last_sent = 0
+  last_received = 0
+  while True:
+    time.sleep(5)
+    now = time.time();
+    sent_now = sent
+    received_now = received
+    delta = now - last_report
+    sent_mbps = ((sent_now - last_sent) * 8.0 / 1000000) / delta
+    received_mbps = ((received_now - last_received) * 8.0 / 1000000) / delta
+    print "Sent: %5.2f mbps  Received: %5.2f mbps" % (sent_mbps, received_mbps)
+    last_report = now
+    last_sent = sent_now
+    last_received = received_now
+
+def Serve(socket, upload=True, download=True):
+  while True:
+    (s, addr) = socket.accept()
+    if upload:
+      thread.start_new_thread(Spew, (s,))
+    if download:
+      thread.start_new_thread(Sink, (s,))
+
+def Receiver(port, upload=True, download=True):
+  s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
+  s.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
+  s.bind(('', port))
+  s.listen(5)
+  thread.start_new_thread(Serve, (s, upload, download))
+
+
+def Connect(to_hostport, upload=True, download=False):
+  s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
+  s.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
+  s.connect(to_hostport)
+  if upload:
+    thread.start_new_thread(Spew, (s,))
+  if download:
+    thread.start_new_thread(Sink, (s,))
+
+
+def Usage():
+  print "One of:"
+  print "%s listen <port>" % sys.arv[0]
+  print "%s upload <host> <port>" % sys.arv[0]
+  print "%s download <host> <port>" % sys.arv[0]
+  print "%s updown <host> <port>" % sys.arv[0]
+  sys.exit(1)
+
+if len(sys.argv) < 2:
+  Usage()
+if sys.argv[1] == "listen":
+  Receiver(int(sys.argv[2]))
+elif sys.argv[1] == "download":
+  Connect( (sys.argv[2], int(sys.argv[3])), upload=False, download=True)
+elif sys.argv[1] == "upload":
+  Connect( (sys.argv[2], int(sys.argv[3])), upload=True, download=False)
+elif sys.argv[1] == "updown":
+  Connect( (sys.argv[2], int(sys.argv[3])), upload=True, download=True)
+else:
+  Usage()
+
+PrintStats()
diff --git a/media/cast/test/utility/shadow.sh b/media/cast/test/utility/shadow.sh
new file mode 100755
index 0000000..4163fd0
--- /dev/null
+++ b/media/cast/test/utility/shadow.sh
@@ -0,0 +1,121 @@
+#!/bin/sh
+# 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.
+#
+# The purpose of this script is to set up all the neccessary magic to
+# pipe network traffic through a user-space process. That user-space
+# process can then delay, reorder and drop packets as it pleases to
+# emulate various network environments.
+#
+# The script currently assumes that you communicate with your cast streaming
+# receiver through eth1. After running "shadow.sh start", your network will
+# look something like this:
+#
+#              +--------------------------------------------------+
+#              |            Your linux machine                    |
+#              | +---------------+                                |
+# cast         | |shadowbr bridge|               +-------------+  |
+# streaming <--+-+---> eth1      |               |routing table|  |
+# receiver     | |     tap2  <---+-> tap_proxy <-+-> tap1      |  |
+#              | |  +->veth      |               |   eth0 <----+--+->internet
+#              | +--+------------+               |   lo        |  |
+#              |    |                            +-------------+  |
+#              |    |      +------------------+       ^           |
+#              |    |      |shadow container  |       |           |
+#              |    +------+-->veth           |     chrome        |
+#              |           | netload.py server|  netload.py client|
+#              |           +------------------+                   |
+#              +--------------------------------------------------+
+#
+# The result should be that all traffic to/from the cast streaming receiver
+# will go through tap_proxy. All traffic to/from the shadow container
+# will also go through the tap_proxy. (A container is kind of like a
+# virtual machine, but more lightweight.) Running "shadow.sh start" does
+# not start the tap_proxy, so you'll have to start it manually with
+# the command "tap_proxy tap1 tap2 <network_profile>" where
+# <network_profile> is one of "perfect", "good", "wifi", "bad" or "evil".
+#
+# While testing mirroring, we can now generate TCP traffic through
+# the tap proxy by talking to the netload server inside the "shadow"
+# container by using the following command:
+#
+# $ netload.py upload IP PORT
+#
+# The IP and PORT are printed out by this script when you run
+# "shadow.sh start", but will generally be the *.*.*.253 address
+# of the eth1 network, so hopefully that's not already taken...
+
+set -x
+
+DEV=eth1
+TAP1=tap1
+TAP2=tap2
+
+IP="$(ifconfig $DEV | sed -n 's@.*inet addr:\([^ ]*\).*@\1@gp')"
+MASK="$(ifconfig $DEV | sed -n 's@.*Mask:\([^ ]*\).*@\1@gp')"
+BCAST="$(ifconfig $DEV | sed -n 's@.*Bcast:\([^ ]*\).*@\1@gp')"
+NET=$(route -n | grep $DEV | head -1 | awk '{print $1}')
+DIR=$(dirname "$0")
+
+case "$MASK" in
+  255.255.255.0) MASK_BITS=24 ;;
+  255.255.0.0) MASK_BITS=16 ;;
+  255.0.0.0) MASK_BITS=8 ;;
+  *)
+    echo "Unknown network mask"
+    exit 1
+  ;;
+esac
+
+SHADOWIP="$(echo $IP | sed 's@[^.]*$@@g')253"
+SHADOWCONF="/tmp/shadowconf.$$"
+cat <<EOF >$SHADOWCONF
+lxc.utsname = shadow
+lxc.network.type = veth
+lxc.network.link = shadowbr
+lxc.network.flags = up
+lxc.network.ipv4 = $SHADOWIP/$MASK_BITS
+lxc.network.ipv4.gateway = $IP
+lxc.kmsg = 0
+EOF
+
+trap "rm $SHADOWCONF" SIGINT SIGTERM EXIT
+LXC_COMMON="-n shadow -f $SHADOWCONF"
+
+case "$1" in
+  start)
+    openvpn --mktun --dev $TAP1
+    openvpn --mktun --dev $TAP2
+    ifconfig $TAP1 $IP netmask $MASK broadcast $BCAST up
+    ifconfig $TAP2 up
+    route add -net $NET netmask $MASK $TAP1
+    brctl addbr shadowbr
+    brctl addif shadowbr $TAP2 $DEV
+    ifconfig shadowbr up
+    lxc-create $LXC_COMMON
+    lxc-execute $LXC_COMMON -- \
+       "$DIRNAME/netload.py listen 9999" >/dev/null </dev/null 2>&1 &
+    echo "Now run: tap_proxy $TAP1 $TAP2 wifi"
+    echo "Data sink/source is available on $SHADOWIP 9999"
+  ;;
+
+  stop)
+    lxc-kill -n shadow
+    sleep 1
+    lxc-destroy $LXC_COMMON
+    ifconfig $TAP1 down
+    ifconfig $TAP2 down
+    ifconfig shadowbr down
+    brctl delbr shadowbr
+    openvpn --rmtun --dev $TAP1
+    openvpn --rmtun --dev $TAP2
+  ;;
+
+  *)
+    echo "$0 start/stop"
+    echo "Read $0 for more information."
+  ;;
+esac
+
+
diff --git a/media/cast/test/utility/tap_proxy.cc b/media/cast/test/utility/tap_proxy.cc
new file mode 100644
index 0000000..7827bf9
--- /dev/null
+++ b/media/cast/test/utility/tap_proxy.cc
@@ -0,0 +1,318 @@
+// 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 <fcntl.h>
+#include <linux/if_tun.h>
+#include <linux/types.h>
+#include <math.h>
+#include <net/if.h>
+#include <netinet/in.h>
+#include <stdio.h>
+#include <stdlib.h>
+#include <sys/ioctl.h>
+#include <sys/stat.h>
+#include <sys/types.h>
+#include <unistd.h>
+
+#include <deque>
+#include <map>
+
+#include "base/at_exit.h"
+#include "base/bind.h"
+#include "base/command_line.h"
+#include "base/logging.h"
+#include "base/rand_util.h"
+#include "base/synchronization/waitable_event.h"
+#include "base/threading/thread.h"
+#include "base/time/default_tick_clock.h"
+#include "media/cast/test/utility/udp_proxy.h"
+#include "net/base/io_buffer.h"
+#include "net/base/net_errors.h"
+#include "net/udp/udp_socket.h"
+
+namespace media {
+namespace cast {
+namespace test {
+
+const size_t kMaxPacketSize = 4096;
+
+class SendToFDPipe : public PacketPipe {
+ public:
+  explicit SendToFDPipe(int fd) : fd_(fd) {
+  }
+  virtual void Send(scoped_ptr<Packet> packet) OVERRIDE {
+    while (1) {
+      int written = write(
+          fd_,
+          reinterpret_cast<char*>(&packet->front()),
+          packet->size());
+      if (written < 0) {
+        if (errno == EINTR) continue;
+        perror("write");
+        exit(1);
+      }
+      if (written != static_cast<int>(packet->size())) {
+        fprintf(stderr, "Truncated write!\n");
+        exit(1);
+      }
+      break;
+    }
+  }
+ private:
+  int fd_;
+};
+
+class QueueManager : public base::MessageLoopForIO::Watcher {
+ public:
+  QueueManager(int input_fd,
+               int output_fd,
+               scoped_ptr<PacketPipe> pipe) :
+      input_fd_(input_fd),
+      packet_pipe_(pipe.Pass()) {
+
+    CHECK(base::MessageLoopForIO::current()->WatchFileDescriptor(
+        input_fd_, true, base::MessageLoopForIO::WATCH_READ,
+        &read_socket_watcher_, this));
+
+    scoped_ptr<PacketPipe> tmp(new SendToFDPipe(output_fd));
+    if (packet_pipe_) {
+      packet_pipe_->AppendToPipe(tmp.Pass());
+    } else {
+      packet_pipe_ = tmp.Pass();
+    }
+    packet_pipe_->InitOnIOThread(base::MessageLoopProxy::current(),
+                                 &tick_clock_);
+  }
+
+  virtual ~QueueManager() {
+  }
+
+  // MessageLoopForIO::Watcher methods
+  virtual void OnFileCanReadWithoutBlocking(int fd) OVERRIDE {
+    scoped_ptr<Packet> packet(new Packet(kMaxPacketSize));
+    int nread = read(input_fd_,
+                     reinterpret_cast<char*>(&packet->front()),
+                     kMaxPacketSize);
+    if (nread < 0) {
+      if (errno == EINTR) return;
+      perror("read");
+      exit(1);
+    }
+    if (nread == 0) return;
+    packet->resize(nread);
+    packet_pipe_->Send(packet.Pass());
+  }
+  virtual void OnFileCanWriteWithoutBlocking(int fd) OVERRIDE {
+    NOTREACHED();
+  }
+
+ private:
+  int input_fd_;
+  scoped_ptr<PacketPipe> packet_pipe_;
+  base::MessageLoopForIO::FileDescriptorWatcher read_socket_watcher_;
+  base::DefaultTickClock tick_clock_;
+};
+
+}  // namespace test
+}  // namespace cast
+}  // namespace media
+
+base::TimeTicks last_printout;
+
+class ByteCounter {
+ public:
+  ByteCounter() : bytes_(0), packets_(0) {
+    push(base::TimeTicks::Now());
+  }
+
+  base::TimeDelta time_range() {
+    return time_data_.back() - time_data_.front();
+  }
+
+  void push(base::TimeTicks now) {
+    byte_data_.push_back(bytes_);
+    packet_data_.push_back(packets_);
+    time_data_.push_back(now);
+    while (time_range().InSeconds() > 10) {
+      byte_data_.pop_front();
+      packet_data_.pop_front();
+      time_data_.pop_front();
+    }
+  }
+
+  double megabits_per_second() {
+    double megabits = (byte_data_.back() - byte_data_.front()) * 8 / 1E6;
+    return megabits / time_range().InSecondsF();
+  }
+
+  double packets_per_second() {
+    double packets = packet_data_.back()- packet_data_.front();
+    return packets / time_range().InSecondsF();
+  }
+
+  void Increment(uint64 x) {
+    bytes_ += x;
+    packets_ ++;
+  }
+
+ private:
+  uint64 bytes_;
+  uint64 packets_;
+  std::deque<uint64> byte_data_;
+  std::deque<uint64> packet_data_;
+  std::deque<base::TimeTicks> time_data_;
+};
+
+ByteCounter in_pipe_input_counter;
+ByteCounter in_pipe_output_counter;
+ByteCounter out_pipe_input_counter;
+ByteCounter out_pipe_output_counter;
+
+class ByteCounterPipe : public media::cast::test::PacketPipe {
+ public:
+  ByteCounterPipe(ByteCounter* counter) : counter_(counter) {}
+  virtual void Send(scoped_ptr<media::cast::Packet> packet)
+      OVERRIDE {
+    counter_->Increment(packet->size());
+    pipe_->Send(packet.Pass());
+  }
+ private:
+  ByteCounter* counter_;
+};
+
+void SetupByteCounters(scoped_ptr<media::cast::test::PacketPipe>* pipe,
+                       ByteCounter* pipe_input_counter,
+                       ByteCounter* pipe_output_counter) {
+  media::cast::test::PacketPipe* new_pipe =
+      new ByteCounterPipe(pipe_input_counter);
+  new_pipe->AppendToPipe(pipe->Pass());
+  new_pipe->AppendToPipe(
+      scoped_ptr<media::cast::test::PacketPipe>(
+          new ByteCounterPipe(pipe_output_counter)).Pass());
+  pipe->reset(new_pipe);
+}
+
+void CheckByteCounters() {
+  base::TimeTicks now = base::TimeTicks::Now();
+  in_pipe_input_counter.push(now);
+  in_pipe_output_counter.push(now);
+  out_pipe_input_counter.push(now);
+  out_pipe_output_counter.push(now);
+  if ((now - last_printout).InSeconds() >= 5) {
+    fprintf(stderr, "Sending  : %5.2f / %5.2f mbps %6.2f / %6.2f packets / s\n",
+            in_pipe_output_counter.megabits_per_second(),
+            in_pipe_input_counter.megabits_per_second(),
+            in_pipe_output_counter.packets_per_second(),
+            in_pipe_input_counter.packets_per_second());
+    fprintf(stderr, "Receiving: %5.2f / %5.2f mbps %6.2f / %6.2f packets / s\n",
+            out_pipe_output_counter.megabits_per_second(),
+            out_pipe_input_counter.megabits_per_second(),
+            out_pipe_output_counter.packets_per_second(),
+            out_pipe_input_counter.packets_per_second());
+
+    last_printout = now;
+  }
+  base::MessageLoopProxy::current()->PostDelayedTask(
+      FROM_HERE,
+      base::Bind(&CheckByteCounters),
+      base::TimeDelta::FromMilliseconds(100));
+}
+
+int tun_alloc(char *dev, int flags) {
+  struct ifreq ifr;
+  int fd, err;
+  const char *clonedev = "/dev/net/tun";
+
+  /* Arguments taken by the function:
+   *
+   * char *dev: the name of an interface (or '\0'). MUST have enough
+   *   space to hold the interface name if '\0' is passed
+   * int flags: interface flags (eg, IFF_TUN etc.)
+   */
+
+  /* open the clone device */
+  if( (fd = open(clonedev, O_RDWR)) < 0 ) {
+    return fd;
+  }
+
+  /* preparation of the struct ifr, of type "struct ifreq" */
+  memset(&ifr, 0, sizeof(ifr));
+
+  ifr.ifr_flags = flags;   /* IFF_TUN or IFF_TAP, plus maybe IFF_NO_PI */
+
+  if (*dev) {
+    /* if a device name was specified, put it in the structure; otherwise,
+     * the kernel will try to allocate the "next" device of the
+     * specified type */
+    strncpy(ifr.ifr_name, dev, IFNAMSIZ);
+  }
+
+  /* try to create the device */
+  if( (err = ioctl(fd, TUNSETIFF, (void *) &ifr)) < 0 ) {
+    close(fd);
+    return err;
+  }
+
+  if (!*dev) {
+    /* if the operation was successful, write back the name of the
+     * interface to the variable "dev", so the caller can know
+     * it. Note that the caller MUST reserve space in *dev (see calling
+     * code below) */
+    strcpy(dev, ifr.ifr_name);
+  }
+
+  /* this is the special file descriptor that the caller will use to talk
+   * with the virtual interface */
+  return fd;
+}
+
+
+int main(int argc, char **argv) {
+  base::AtExitManager exit_manager;
+  CommandLine::Init(argc, argv);
+  InitLogging(logging::LoggingSettings());
+
+  if (argc < 4) {
+    fprintf(stderr, "Usage: tap_proxy tap1 tap2 type\n");
+    fprintf(stderr,
+            "Where 'type' is one of perfect, good, wifi, bad or evil\n");
+    exit(1);
+  }
+
+  scoped_ptr<media::cast::test::PacketPipe> in_pipe, out_pipe;
+  std::string network_type = argv[3];
+  if (network_type == "perfect") {
+    // No action needed.
+  } else if (network_type == "good") {
+    in_pipe = media::cast::test::GoodNetwork().Pass();
+    out_pipe = media::cast::test::GoodNetwork().Pass();
+  } else if (network_type == "wifi") {
+    in_pipe = media::cast::test::WifiNetwork().Pass();
+    out_pipe = media::cast::test::WifiNetwork().Pass();
+  } else if (network_type == "bad") {
+    in_pipe = media::cast::test::BadNetwork().Pass();
+    out_pipe = media::cast::test::BadNetwork().Pass();
+  } else if (network_type == "evil") {
+    in_pipe = media::cast::test::EvilNetwork().Pass();
+    out_pipe = media::cast::test::EvilNetwork().Pass();
+  } else {
+    fprintf(stderr, "Unknown network type.\n");
+    exit(1);
+  }
+
+  SetupByteCounters(&in_pipe, &in_pipe_input_counter, &in_pipe_output_counter);
+  SetupByteCounters(
+      &out_pipe, &out_pipe_input_counter, &out_pipe_output_counter);
+
+  int fd1 = tun_alloc(argv[1], IFF_TAP);
+  int fd2 = tun_alloc(argv[2], IFF_TAP);
+
+  base::MessageLoopForIO message_loop;
+  last_printout = base::TimeTicks::Now();
+  media::cast::test::QueueManager qm1(fd1, fd2, in_pipe.Pass());
+  media::cast::test::QueueManager qm2(fd2, fd1, out_pipe.Pass());
+  CheckByteCounters();
+  printf("Press Ctrl-C when done.\n");
+  message_loop.Run();
+}
diff --git a/media/cast/test/utility/udp_proxy.cc b/media/cast/test/utility/udp_proxy.cc
index 4714b7e..e71678c 100644
--- a/media/cast/test/utility/udp_proxy.cc
+++ b/media/cast/test/utility/udp_proxy.cc
@@ -68,6 +68,7 @@ class Buffer : public PacketPipe {
 
  private:
   void Schedule() {
+    last_schedule_ = clock_->NowTicks();
     double megabits = buffer_.front()->size() * 8 / 1000000.0;
     double seconds = megabits / max_megabits_per_second_;
     int64 microseconds = static_cast<int64>(seconds * 1E6);
@@ -78,17 +79,28 @@ class Buffer : public PacketPipe {
   }
 
   void ProcessBuffer() {
-    CHECK(!buffer_.empty());
-    scoped_ptr<Packet> packet(buffer_.front().release());
-    buffer_size_ -= packet->size();
-    buffer_.pop_front();
-    pipe_->Send(packet.Pass());
+    int64 bytes_to_send = static_cast<int64>(
+        (clock_->NowTicks() - last_schedule_).InSecondsF() *
+        max_megabits_per_second_ * 1E6 / 8);
+    if (bytes_to_send < static_cast<int64>(buffer_.front()->size())) {
+      bytes_to_send = buffer_.front()->size();
+    }
+    while (!buffer_.empty() &&
+           static_cast<int64>(buffer_.front()->size()) <= bytes_to_send) {
+      CHECK(!buffer_.empty());
+      scoped_ptr<Packet> packet(buffer_.front().release());
+      bytes_to_send -= packet->size();
+      buffer_size_ -= packet->size();
+      buffer_.pop_front();
+      pipe_->Send(packet.Pass());
+    }
     if (!buffer_.empty()) {
       Schedule();
     }
   }
 
   std::deque<linked_ptr<Packet> > buffer_;
+  base::TimeTicks last_schedule_;
   size_t buffer_size_;
   size_t max_buffer_size_;
   double max_megabits_per_second_;  // megabits per second
@@ -188,7 +200,11 @@ class RandomSortedDelay : public PacketPipe {
   virtual void Send(scoped_ptr<Packet> packet) OVERRIDE {
     buffer_.push_back(linked_ptr<Packet>(packet.release()));
     if (buffer_.size() == 1) {
-      Schedule();
+      next_send_ = std::max(
+          clock_->NowTicks() +
+          base::TimeDelta::FromSecondsD(base::RandDouble() * random_delay_),
+          next_send_);
+      ProcessBuffer();
     }
   }
   virtual void InitOnIOThread(
@@ -212,38 +228,34 @@ class RandomSortedDelay : public PacketPipe {
   }
 
   void CauseExtraDelay() {
-    block_until_ = clock_->NowTicks() +
+    next_send_ = std::max<base::TimeTicks>(
+        clock_->NowTicks() +
         base::TimeDelta::FromMicroseconds(
-            static_cast<int64>(extra_delay_ * 1E6));
+            static_cast<int64>(extra_delay_ * 1E6)),
+        next_send_);
     // An extra delay just happened, wait up to seconds_between_extra_delay_*2
     // before scheduling another one to make the average equal to
     // seconds_between_extra_delay_.
     ScheduleExtraDelay(2.0);
   }
 
-  void Schedule() {
-    double seconds = base::RandDouble() * random_delay_;
-    base::TimeDelta block_time = block_until_ - base::TimeTicks::Now();
-    base::TimeDelta delay_time =
-        base::TimeDelta::FromMicroseconds(
-            static_cast<int64>(seconds * 1E6));
-    if (block_time > delay_time) {
-      block_time = delay_time;
-    }
+  void ProcessBuffer() {
+    base::TimeTicks now = clock_->NowTicks();
+    while (!buffer_.empty() && next_send_ <= now) {
+      scoped_ptr<Packet> packet(buffer_.front().release());
+      pipe_->Send(packet.Pass());
+      buffer_.pop_front();
 
-    task_runner_->PostDelayedTask(FROM_HERE,
-                                  base::Bind(&RandomSortedDelay::ProcessBuffer,
-                                             weak_factory_.GetWeakPtr()),
-                                  delay_time);
-  }
+      next_send_ += base::TimeDelta::FromSecondsD(
+          base::RandDouble() * random_delay_);
+    }
 
-  void ProcessBuffer() {
-    CHECK(!buffer_.empty());
-    scoped_ptr<Packet> packet(buffer_.front().release());
-    pipe_->Send(packet.Pass());
-    buffer_.pop_front();
     if (!buffer_.empty()) {
-      Schedule();
+      task_runner_->PostDelayedTask(
+          FROM_HERE,
+          base::Bind(&RandomSortedDelay::ProcessBuffer,
+                     weak_factory_.GetWeakPtr()),
+          next_send_ - now);
     }
   }
 
@@ -253,6 +265,7 @@ class RandomSortedDelay : public PacketPipe {
   double extra_delay_;
   double seconds_between_extra_delay_;
   base::WeakPtrFactory<RandomSortedDelay> weak_factory_;
+  base::TimeTicks next_send_;
 };
 
 scoped_ptr<PacketPipe> NewRandomSortedDelay(
@@ -536,6 +549,17 @@ void BuildPipe(scoped_ptr<PacketPipe>* pipe, PacketPipe* next) {
 }
 }  // namespace
 
+scoped_ptr<PacketPipe> GoodNetwork() {
+  // This represents the buffer on the sender.
+  scoped_ptr<PacketPipe> pipe;
+  BuildPipe(&pipe, new Buffer(2 << 20, 50));
+  BuildPipe(&pipe, new ConstantDelay(1E-3));
+  BuildPipe(&pipe, new RandomSortedDelay(1E-3, 2E-3, 3));
+  // This represents the buffer on the receiving device.
+  BuildPipe(&pipe, new Buffer(2 << 20, 50));
+  return pipe.Pass();
+}
+
 scoped_ptr<PacketPipe> WifiNetwork() {
   // This represents the buffer on the sender.
   scoped_ptr<PacketPipe> pipe;
diff --git a/media/cast/test/utility/udp_proxy.h b/media/cast/test/utility/udp_proxy.h
index ea50a2c..6c72fb6 100644
--- a/media/cast/test/utility/udp_proxy.h
+++ b/media/cast/test/utility/udp_proxy.h
@@ -156,6 +156,10 @@ scoped_ptr<PacketPipe> NewNetworkGlitchPipe(double average_work_time,
                                             double average_outage_time);
 
 // This method builds a stack of PacketPipes to emulate a reasonably
+// good network. ~50mbit, ~3ms latency, no packet loss unless saturated.
+scoped_ptr<PacketPipe> GoodNetwork();
+
+// This method builds a stack of PacketPipes to emulate a reasonably
 // good wifi network. ~20mbit, 1% packet loss, ~3ms latency.
 scoped_ptr<PacketPipe> WifiNetwork();