Implement chutney performance testing

The following environmental variables affect chutney verify:
CHUTNEY_DATA_BYTES=n        sends n bytes per test connection (10 KBytes)
CHUTNEY_CONNECTIONS=n       makes n test connections per client (1)
CHUTNEY_HS_MULTI_CLIENT=1   makes each client connect to each HS (0)

When enough data is transmitted, chutney verify reports:
Single Stream Bandwidth: the speed of the slowest stream, end-to-end
Overall tor Bandwidth: the sum of the bandwidth across each tor instance
This approximates the CPU-bound tor performance on the current machine,
assuming everything is multithreaded and network performance is infinite.

These new features are all documented in the README.

README

@@ -7,7 +7,7 @@ It is supposed to be a good tool for:
   - Launching and monitoring a testing tor network
   - Running tests on a testing tor network
 
-Right now it only sorta does the first two.
+Right now it only sorta does these things.
 
 You will need, at the moment:
   - Tor installed somewhere in your path or the location of the 'tor' and
@@ -16,12 +16,49 @@ You will need, at the moment:
   - Python 2.7 or later
 
 Stuff to try:
+
+Standard Actions:
   ./chutney configure networks/basic
   ./chutney start networks/basic
   ./chutney status networks/basic
+  ./chutney verify networks/basic
   ./chutney hup networks/basic
   ./chutney stop networks/basic
 
+Bandwidth Tests:
+  ./chutney configure networks/basic-min
+  ./chutney start networks/basic-min
+  ./chutney status networks/basic-min
+  CHUTNEY_DATA_BYTES=104857600 ./chutney verify networks/basic-min
+  # Send 100MB of data per client connection
+  # verify produces performance figures for:
+  # Single Stream Bandwidth: the speed of the slowest stream, end-to-end
+  # Overall tor Bandwidth: the sum of the bandwidth across each tor instance
+  # This approximates the CPU-bound tor performance on the current machine,
+  # assuming everything is multithreaded and network performance is infinite.
+  ./chutney stop networks/basic-min
+
+Connection Tests:
+  ./chutney configure networks/basic-025
+  ./chutney start networks/basic-025
+  ./chutney status networks/basic-025
+  CHUTNEY_CONNECTIONS=5 ./chutney verify networks/basic-025
+  # Make 5 connections from each client through a random exit
+  ./chutney stop networks/basic-025
+
+Note: If you create 7 or more connections to a hidden service from a single
+client, you'll likely get a verification failure due to
+https://trac.torproject.org/projects/tor/ticket/15937
+
+HS Connection Tests:
+  ./chutney configure networks/hs-025
+  ./chutney start networks/hs-025
+  ./chutney status networks/hs-025
+  CHUTNEY_HS_MULTI_CLIENT=1 ./chutney verify networks/hs-025
+  # Make a connection from each client to each hs
+  # Default behavior is one client connects to each HS
+  ./chutney stop networks/hs-025
+
 The configuration files:
   networks/basic holds the configuration for the network you're configuring
   above.  It refers to some torrc template files in torrc_templates/.

lib/chutney/TorNet.py

@@ -690,6 +690,14 @@ DEFAULTS = {
      # Used when poll_launch_time is None, but RunAsDaemon is not set
      # Set low so that we don't interfere with the voting interval
     'poll_launch_time_default': 0.1,
+    # the number of bytes of random data we send on each connection
+    'data_bytes': int(os.environ.get('CHUTNEY_DATA_BYTES', 10 * 1024)),
+    # the number of times each client will connect
+    'connection_count': int(os.environ.get('CHUTNEY_CONNECTIONS', 1)),
+    # Do we want every client to connect to every HS, or one client
+    # to connect to each HS?
+    # (Clients choose an exit at random, so this doesn't apply to exits.)
+    'hs_multi_client': int(os.environ.get('CHUTNEY_HS_MULTI_CLIENT', 0)),
 }
 
 
@@ -908,46 +916,201 @@ class Network(object):
         # HSs must have a HiddenServiceDir with
         # "HiddenServicePort <HS_PORT> 127.0.0.1:<LISTEN_PORT>"
         HS_PORT = 5858
-        DATALEN = 10 * 1024               # Octets.
-        TIMEOUT = 3                     # Seconds.
-        with open('/dev/urandom', 'r') as randfp:
-            tmpdata = randfp.read(DATALEN)
+        # The amount of data to send between each source-sink pair,
+        # each time the source connects.
+        # We create a source-sink pair for each (bridge) client to an exit,
+        # and a source-sink pair for a (bridge) client to each hidden service
+        DATALEN = self._dfltEnv['data_bytes']
+        # Print a dot each time a sink verifies this much data
+        DOTDATALEN = 5 * 1024 * 1024 # Octets.
+        TIMEOUT = 3                  # Seconds.
+        # Calculate the amount of random data we should use
+        randomlen = self._calculate_randomlen(DATALEN)
+        reps = self._calculate_reps(DATALEN, randomlen)
+        # sanity check
+        if reps == 0:
+            DATALEN = 0
+        # Get the random data
+        if randomlen > 0:
+            # print a dot after every DOTDATALEN data is verified, rounding up
+            dot_reps = self._calculate_reps(DOTDATALEN, randomlen)
+            # make sure we get at least one dot per transmission
+            dot_reps = min(reps, dot_reps)
+            with open('/dev/urandom', 'r') as randfp:
+                tmpdata = randfp.read(randomlen)
+        else:
+            dot_reps = 0
+            tmpdata = {}
+        # now make the connections
         bind_to = ('127.0.0.1', LISTEN_PORT)
-        tt = chutney.Traffic.TrafficTester(bind_to, tmpdata, TIMEOUT)
+        tt = chutney.Traffic.TrafficTester(bind_to,
+                                           tmpdata,
+                                           TIMEOUT,
+                                           reps,
+                                           dot_reps)
         client_list = filter(lambda n:
                              n._env['tag'] == 'c' or n._env['tag'] == 'bc',
                              self._nodes)
+        exit_list = filter(lambda n:
+                           ('exit' in n._env.keys()) and n._env['exit'] == 1,
+                           self._nodes)
+        hs_list = filter(lambda n:
+                         n._env['tag'] == 'h',
+                         self._nodes)
         if len(client_list) == 0:
             print("  Unable to verify network: no client nodes available")
             return False
-        # Each client binds directly to 127.0.0.1:LISTEN_PORT via an Exit relay
-        for op in client_list:
-            print("  Exit to %s:%d via client %s:%s"
-                   % ('127.0.0.1', LISTEN_PORT,
-                      'localhost', op._env['socksport']))
-            tt.add(chutney.Traffic.Source(tt, bind_to, tmpdata,
-                                          ('localhost',
-                                           int(op._env['socksport']))))
-        # The HS redirects .onion connections made to hs_hostname:HS_PORT
-        # to the Traffic Tester's 127.0.0.1:LISTEN_PORT
-        # We must have at least one working client for the hs test to succeed
-        for hs in filter(lambda n:
-                         n._env['tag'] == 'h',
-                         self._nodes):
-            # Instead of binding directly to LISTEN_PORT via an Exit relay,
-            # we bind to hs_hostname:HS_PORT via a hidden service connection
-            # through the first available client
-            bind_to = (hs._env['hs_hostname'], HS_PORT)
-            # Just choose the first client
-            client = client_list[0]
-            print("  HS to %s:%d (%s:%d) via client %s:%s"
-                  % (hs._env['hs_hostname'], HS_PORT,
+        if len(exit_list) == 0 and len(hs_list) == 0:
+            print("  Unable to verify network: no exit/hs nodes available")
+            print("  Exit nodes must be declared 'relay=1, exit=1'")
+            print("  HS nodes must be declared 'tag=\"hs\"'")
+            return False
+        print("Connecting:")
+        # the number of tor nodes in paths which will send DATALEN data
+        # if a node is used in two paths, we count it twice
+        # this is a lower bound, as cannabilised circuits are one node longer
+        total_path_node_count = 0
+        total_path_node_count += self._configure_exits(tt, bind_to,
+                                                       tmpdata, reps,
+                                                       client_list, exit_list,
+                                                       LISTEN_PORT)
+        total_path_node_count += self._configure_hs(tt,
+                                                    tmpdata, reps,
+                                                    client_list, hs_list,
+                                                    HS_PORT,
+                                                    LISTEN_PORT)
+        print("Transmitting Data:")
+        start_time = time.clock()
+        status = tt.run()
+        end_time = time.clock()
+        # if we fail, don't report the bandwidth
+        if not status:
+            return status
+        # otherwise, report bandwidth used, if sufficient data was transmitted
+        self._report_bandwidth(DATALEN, total_path_node_count,
+                               start_time, end_time)
+        return status
+
+    # In order to performance test a tor network, we need to transmit
+    # several hundred megabytes of data or more. Passing around this
+    # much data in Python has its own performance impacts, so we provide
+    # a smaller amount of random data instead, and repeat it to DATALEN
+    def _calculate_randomlen(self, datalen):
+        MAX_RANDOMLEN = 128 * 1024   # Octets.
+        if datalen > MAX_RANDOMLEN:
+            return MAX_RANDOMLEN
+        else:
+            return datalen
+
+    def _calculate_reps(self, datalen, replen):
+        # sanity checks
+        if datalen == 0 or replen == 0:
+            return 0
+        # effectively rounds datalen up to the nearest replen
+        if replen < datalen:
+            return (datalen + replen - 1) / replen
+        else:
+            return 1
+
+    # if there are any exits, each client / bridge client transmits
+    # via 4 nodes (including the client) to an arbitrary exit
+    # Each client binds directly to 127.0.0.1:LISTEN_PORT via an Exit relay
+    def _configure_exits(self, tt, bind_to,
+                         tmpdata, reps,
+                         client_list, exit_list,
+                         LISTEN_PORT):
+        CLIENT_EXIT_PATH_NODES = 4
+        connection_count = self._dfltEnv['connection_count']
+        exit_path_node_count = 0
+        if len(exit_list) > 0:
+            exit_path_node_count += (len(client_list)
+                                     * CLIENT_EXIT_PATH_NODES
+                                     * connection_count)
+            for op in client_list:
+                print("  Exit to %s:%d via client %s:%s"
+                      % ('127.0.0.1', LISTEN_PORT,
+                         'localhost', op._env['socksport']))
+                for i in range(connection_count):
+                    tt.add(chutney.Traffic.Source(tt,
+                                                  bind_to,
+                                                  tmpdata,
+                                                  ('localhost',
+                                                   int(op._env['socksport'])),
+                                                  reps))
+        return exit_path_node_count
+
+    # The HS redirects .onion connections made to hs_hostname:HS_PORT
+    # to the Traffic Tester's 127.0.0.1:LISTEN_PORT
+    # an arbitrary client / bridge client transmits via 8 nodes
+    # (including the client and hs) to each hidden service
+    # Instead of binding directly to LISTEN_PORT via an Exit relay,
+    # we bind to hs_hostname:HS_PORT via a hidden service connection
+    def _configure_hs(self, tt,
+                      tmpdata, reps,
+                      client_list, hs_list,
+                      HS_PORT,
+                      LISTEN_PORT):
+        CLIENT_HS_PATH_NODES = 8
+        connection_count = self._dfltEnv['connection_count']
+        hs_path_node_count = (len(hs_list)
+                              * CLIENT_HS_PATH_NODES
+                              * connection_count)
+        # Each client in hs_client_list connects to each hs
+        if self._dfltEnv['hs_multi_client']:
+            hs_client_list = client_list
+            hs_path_node_count *= len(client_list)
+        else:
+            # only use the first client in the list
+            hs_client_list = client_list[:1]
+        # Setup the connections from each client in hs_client_list to each hs
+        for hs in hs_list:
+            hs_bind_to = (hs._env['hs_hostname'], HS_PORT)
+            for client in hs_client_list:
+                print("  HS to %s:%d (%s:%d) via client %s:%s"
+                      % (hs._env['hs_hostname'], HS_PORT,
                      '127.0.0.1', LISTEN_PORT,
                      'localhost', client._env['socksport']))
-            tt.add(chutney.Traffic.Source(tt, bind_to, tmpdata,
+                for i in range(connection_count):
+                    tt.add(chutney.Traffic.Source(tt,
+                                          hs_bind_to,
+                                          tmpdata,
                                           ('localhost',
-                                           int(client._env['socksport']))))
-        return tt.run()
+                                           int(client._env['socksport'])),
+                                          reps))
+        return hs_path_node_count
+
+    # calculate the single stream bandwidth and overall tor bandwidth
+    # the single stream bandwidth is the bandwidth of the
+    # slowest stream of all the simultaneously transmitted streams
+    # the overall bandwidth estimates the simultaneous bandwidth between
+    # all tor nodes over all simultaneous streams, assuming:
+    # * minimum path lengths (no cannibalized circuits)
+    # * unlimited network bandwidth (that is, localhost)
+    # * tor performance is CPU-limited
+    # This be used to estimate the bandwidth capacity of a CPU-bound
+    # tor relay running on this machine
+    def _report_bandwidth(self, data_length, total_path_node_count,
+                          start_time, end_time):
+        # otherwise, if we sent at least 5 MB cumulative total, and
+        # it took us at least a second to send, report bandwidth
+        MIN_BWDATA = 5 * 1024 * 1024 # Octets.
+        MIN_ELAPSED_TIME = 1.0       # Seconds.
+        cumulative_data_sent = total_path_node_count * data_length
+        elapsed_time = end_time - start_time
+        if (cumulative_data_sent >= MIN_BWDATA
+            and elapsed_time >= MIN_ELAPSED_TIME):
+            # Report megabytes per second
+            BWDIVISOR = 1024*1024
+            single_stream_bandwidth = (data_length
+                                       / elapsed_time
+                                       / BWDIVISOR)
+            overall_bandwidth = (cumulative_data_sent
+                                 / elapsed_time
+                                 / BWDIVISOR)
+            print("Single Stream Bandwidth: %.2f MBytes/s"
+                  % single_stream_bandwidth)
+            print("Overall tor Bandwidth: %.2f MBytes/s"
+                  % overall_bandwidth)
 
 
 def ConfigureNodes(nodelist):

lib/chutney/Traffic.py

@@ -141,6 +141,7 @@ class Sink(Peer):
     def __init__(self, tt, s):
         super(Sink, self).__init__(Peer.SINK, tt, s)
         self.inbuf = ''
+        self.repetitions = self.tt.repetitions
 
     def on_readable(self):
         """Invoked when the socket becomes readable.
@@ -151,13 +152,35 @@ class Sink(Peer):
         return self.verify(self.tt.data)
 
     def verify(self, data):
+        # shortcut read when we don't ever expect any data
+        if self.repetitions == 0 or len(self.tt.data) == 0:
+            debug("no verification required - no data")
+            return 0;
         self.inbuf += self.s.recv(len(data) - len(self.inbuf))
-        assert(len(self.inbuf) <= len(data))
-        if len(self.inbuf) == len(data):
-            if self.inbuf != data:
+        debug("successfully received (bytes=%d)" % len(self.inbuf))
+        while len(self.inbuf) >= len(data):
+            assert(len(self.inbuf) <= len(data) or self.repetitions > 1)
+            if self.inbuf[:len(data)] != data:
+                debug("receive comparison failed (bytes=%d)" % len(data))
                 return -1       # Failed verification.
+            # if we're not debugging, print a dot every dot_repetitions reps
+            elif (not debug_flag
+                  and self.tt.dot_repetitions > 0
+                  and self.repetitions % self.tt.dot_repetitions == 0):
+                sys.stdout.write('.')
+                sys.stdout.flush()
+            # repeatedly check data against self.inbuf if required
+            debug("receive comparison success (bytes=%d)" % len(data))
+            self.inbuf = self.inbuf[len(data):]
+            debug("receive leftover bytes (bytes=%d)" % len(self.inbuf))
+            self.repetitions -= 1
+            debug("receive remaining repetitions (reps=%d)" % self.repetitions)
+        if self.repetitions == 0 and len(self.inbuf) == 0:
             debug("successful verification")
-        return len(data) - len(self.inbuf)
+        # calculate the actual length of data remaining, including reps
+        debug("receive remaining bytes (bytes=%d)"
+              % (self.repetitions*len(data) - len(self.inbuf)))
+        return self.repetitions*len(data) - len(self.inbuf)
 
 
 class Source(Peer):
@@ -169,13 +192,19 @@ class Source(Peer):
     CONNECTING_THROUGH_PROXY = 2
     CONNECTED = 5
 
-    def __init__(self, tt, server, buf, proxy=None):
+    def __init__(self, tt, server, buf, proxy=None, repetitions=1):
         super(Source, self).__init__(Peer.SOURCE, tt)
         self.state = self.NOT_CONNECTED
         self.data = buf
         self.outbuf = ''
         self.inbuf = ''
         self.proxy = proxy
+        self.repetitions = repetitions
+        # sanity checks
+        if len(self.data) == 0:
+            self.repetitions = 0
+        if self.repetitions == 0:
+            self.data = {}
         self.connect(server)
 
     def connect(self, endpoint):
@@ -200,9 +229,14 @@ class Source(Peer):
                     debug("proxy handshake successful (fd=%d)" % self.fd())
                     self.state = self.CONNECTED
                     self.inbuf = ''
-                    self.outbuf = self.data
                     debug("successfully connected (fd=%d)" % self.fd())
-                    return 1    # Keep us around for writing.
+                    # if we have no reps or no data, skip sending actual data
+                    if self.want_to_write():
+                        return 1    # Keep us around for writing.
+                    else:
+                        # shortcut write when we don't ever expect any data
+                        debug("no connection required - no data")
+                        return 0
                 else:
                     debug("proxy handshake failed (0x%x)! (fd=%d)" %
                           (ord(self.inbuf[1]), self.fd()))
@@ -210,10 +244,11 @@ class Source(Peer):
                     return -1
             assert(8 - len(self.inbuf) > 0)
             return 8 - len(self.inbuf)
-        return 1                # Keep us around for writing.
+        return self.want_to_write()     # Keep us around for writing if needed
 
     def want_to_write(self):
-        return self.state == self.CONNECTING or len(self.outbuf) > 0
+        return (self.state == self.CONNECTING or len(self.outbuf) > 0
+                or (self.repetitions > 0 and len(self.data) > 0))
 
     def on_writable(self):
         """Invoked when the socket becomes writable.
@@ -224,11 +259,21 @@ class Source(Peer):
         if self.state == self.CONNECTING:
             if self.proxy is None:
                 self.state = self.CONNECTED
-                self.outbuf = self.data
                 debug("successfully connected (fd=%d)" % self.fd())
             else:
                 self.state = self.CONNECTING_THROUGH_PROXY
                 self.outbuf = socks_cmd(self.dest)
+                # we write socks_cmd() to the proxy, then read the response
+                # if we get the correct response, we're CONNECTED
+        if self.state == self.CONNECTED:
+            # repeat self.data into self.outbuf if required
+            if (len(self.outbuf) < len(self.data) and self.repetitions > 0):
+                self.outbuf += self.data
+                self.repetitions -= 1
+                debug("adding more data to send (bytes=%d)" % len(self.data))
+                debug("now have data to send (bytes=%d)" % len(self.outbuf))
+                debug("send repetitions remaining (reps=%d)"
+                      % self.repetitions)
         try:
             n = self.s.send(self.outbuf)
         except socket.error as e:
@@ -236,10 +281,19 @@ class Source(Peer):
                 debug("connection refused (fd=%d)" % self.fd())
                 return -1
             raise
+        # sometimes, this debug statement prints 0
+        # it should print length of the data sent
+        # but the code works regardless of this error
+        debug("successfully sent (bytes=%d)" % n)
         self.outbuf = self.outbuf[n:]
         if self.state == self.CONNECTING_THROUGH_PROXY:
             return 1            # Keep us around.
-        return len(self.outbuf)  # When 0, we're being removed.
+        debug("bytes remaining on outbuf (bytes=%d)" % len(self.outbuf))
+        # calculate the actual length of data remaining, including reps
+        # When 0, we're being removed.
+        debug("bytes remaining overall (bytes=%d)"
+              % (self.repetitions*len(self.data) + len(self.outbuf)))
+        return self.repetitions*len(self.data) + len(self.outbuf)
 
 
 class TrafficTester():
@@ -252,12 +306,24 @@ class TrafficTester():
     Return True if all tests succeed, else False.
     """
 
-    def __init__(self, endpoint, data={}, timeout=3):
+    def __init__(self,
+                 endpoint,
+                 data={},
+                 timeout=3,
+                 repetitions=1,
+                 dot_repetitions=0):
         self.listener = Listener(self, endpoint)
         self.pending_close = []
         self.timeout = timeout
         self.tests = TestSuite()
         self.data = data
+        self.repetitions = repetitions
+        # sanity checks
+        if len(self.data) == 0:
+            self.repetitions = 0
+        if self.repetitions == 0:
+            self.data = {}
+        self.dot_repetitions = dot_repetitions
         debug("listener fd=%d" % self.listener.fd())
         self.peers = {}         # fd:Peer
 
@@ -318,9 +384,16 @@ class TrafficTester():
                         self.tests.failure()
                         self.remove(p)
 
+        for fd in self.peers:
+            peer = self.peers[fd]
+            debug("peer fd=%d never pending close, never read or wrote" % fd)
+            self.pending_close.append(peer.s)
         self.listener.s.close()
         for s in self.pending_close:
             s.close()
+        if not debug_flag:
+            sys.stdout.write('\n')
+            sys.stdout.flush()
         return self.tests.all_done() and self.tests.failure_count() == 0