Merge remote-tracking branch 'karsten/morestats5'

doc/tor.1.txt

@@ -2061,6 +2061,9 @@ The following options are used for running a testing Tor network.
        TestingDescriptorMaxDownloadTries 80
        TestingMicrodescMaxDownloadTries 80
        TestingCertMaxDownloadTries 80
+       TestingEnableConnBwEvent 1
+       TestingEnableCellStatsEvent 1
+       TestingEnableTbEmptyEvent 1
 
 [[TestingV3AuthInitialVotingInterval]] **TestingV3AuthInitialVotingInterval** __N__ **minutes**|**hours**::
     Like V3AuthVotingInterval, but for initial voting interval before the first
@@ -2155,6 +2158,22 @@ The following options are used for running a testing Tor network.
     In order for this option to have any effect, **TestingTorNetwork**
     has to be set.
 
+[[TestingEnableConnBwEvent]] **TestingEnableConnBwEvent** **0**|**1**::
+    If this option is set, then Tor controllers may register for CONN_BW
+    events.  Changing this requires that **TestingTorNetwork** is set.
+    (Default: 0)
+
+[[TestingEnableCellStatsEvent]] **TestingEnableCellStatsEvent** **0**|**1**::
+    If this option is set, then Tor controllers may register for CELL_STATS
+    events.  Changing this requires that **TestingTorNetwork** is set.
+    (Default: 0)
+
+[[TestingEnableTbEmptyEvent]] **TestingEnableTbEmptyEvent** **0**|**1**::
+    If this option is set, then Tor controllers may register for TB_EMPTY
+    events.  Changing this requires that **TestingTorNetwork** is set.
+    (Default: 0)
+
+
 SIGNALS
 -------
 

src/or/circuitmux.c

@@ -1872,7 +1872,7 @@ circuitmux_append_destroy_cell(channel_t *chan,
   cell.command = CELL_DESTROY;
   cell.payload[0] = (uint8_t) reason;
 
-  cell_queue_append_packed_copy(&cmux->destroy_cell_queue, &cell,
+  cell_queue_append_packed_copy(NULL, &cmux->destroy_cell_queue, 0, &cell,
                                 chan->wide_circ_ids, 0);
 
   /* Destroy entering the queue, update counters */

src/or/command.c

@@ -53,6 +53,33 @@ static void command_process_created_cell(cell_t *cell, channel_t *chan);
 static void command_process_relay_cell(cell_t *cell, channel_t *chan);
 static void command_process_destroy_cell(cell_t *cell, channel_t *chan);
 
+/** Convert the cell <b>command</b> into a lower-case, human-readable
+ * string. */
+const char *
+cell_command_to_string(uint8_t command)
+{
+  switch (command) {
+    case CELL_PADDING: return "padding";
+    case CELL_CREATE: return "create";
+    case CELL_CREATED: return "created";
+    case CELL_RELAY: return "relay";
+    case CELL_DESTROY: return "destroy";
+    case CELL_CREATE_FAST: return "create_fast";
+    case CELL_CREATED_FAST: return "created_fast";
+    case CELL_VERSIONS: return "versions";
+    case CELL_NETINFO: return "netinfo";
+    case CELL_RELAY_EARLY: return "relay_early";
+    case CELL_CREATE2: return "create2";
+    case CELL_CREATED2: return "created2";
+    case CELL_VPADDING: return "vpadding";
+    case CELL_CERTS: return "certs";
+    case CELL_AUTH_CHALLENGE: return "auth_challenge";
+    case CELL_AUTHENTICATE: return "authenticate";
+    case CELL_AUTHORIZE: return "authorize";
+    default: return "unrecognized";
+  }
+}
+
 #ifdef KEEP_TIMING_STATS
 /** This is a wrapper function around the actual function that processes the
  * <b>cell</b> that just arrived on <b>conn</b>. Increment <b>*time</b>

src/or/command.h

@@ -19,6 +19,8 @@ void command_process_var_cell(channel_t *chan, var_cell_t *cell);
 void command_setup_channel(channel_t *chan);
 void command_setup_listener(channel_listener_t *chan_l);
 
+const char *cell_command_to_string(uint8_t command);
+
 extern uint64_t stats_n_padding_cells_processed;
 extern uint64_t stats_n_create_cells_processed;
 extern uint64_t stats_n_created_cells_processed;

src/or/config.c

@@ -220,6 +220,9 @@ static config_var_t option_vars_[] = {
   VPORT(DNSPort,                     LINELIST, NULL),
   V(DNSListenAddress,            LINELIST, NULL),
   V(DownloadExtraInfo,           BOOL,     "0"),
+  V(TestingEnableConnBwEvent,    BOOL,     "0"),
+  V(TestingEnableCellStatsEvent, BOOL,     "0"),
+  V(TestingEnableTbEmptyEvent,   BOOL,     "0"),
   V(EnforceDistinctSubnets,      BOOL,     "1"),
   V(EntryNodes,                  ROUTERSET,   NULL),
   V(EntryStatistics,             BOOL,     "0"),
@@ -503,6 +506,9 @@ static const config_var_t testing_tor_network_defaults[] = {
   V(TestingDescriptorMaxDownloadTries, UINT, "80"),
   V(TestingMicrodescMaxDownloadTries, UINT, "80"),
   V(TestingCertMaxDownloadTries, UINT, "80"),
+  V(TestingEnableConnBwEvent,    BOOL,     "1"),
+  V(TestingEnableCellStatsEvent, BOOL,     "1"),
+  V(TestingEnableTbEmptyEvent,   BOOL,     "1"),
   VAR("___UsingTestNetworkDefaults", BOOL, UsingTestNetworkDefaults_, "1"),
 
   { NULL, CONFIG_TYPE_OBSOLETE, 0, NULL }
@@ -3467,6 +3473,24 @@ options_validate(or_options_t *old_options, or_options_t *options,
     COMPLAIN("TestingCertMaxDownloadTries is insanely high.");
   }
 
+  if (options->TestingEnableConnBwEvent &&
+      !options->TestingTorNetwork && !options->UsingTestNetworkDefaults_) {
+    REJECT("TestingEnableConnBwEvent may only be changed in testing "
+           "Tor networks!");
+  }
+
+  if (options->TestingEnableCellStatsEvent &&
+      !options->TestingTorNetwork && !options->UsingTestNetworkDefaults_) {
+    REJECT("TestingEnableCellStatsEvent may only be changed in testing "
+           "Tor networks!");
+  }
+
+  if (options->TestingEnableTbEmptyEvent &&
+      !options->TestingTorNetwork && !options->UsingTestNetworkDefaults_) {
+    REJECT("TestingEnableTbEmptyEvent may only be changed in testing "
+           "Tor networks!");
+  }
+
   if (options->TestingTorNetwork) {
     log_warn(LD_CONFIG, "TestingTorNetwork is set. This will make your node "
                         "almost unusable in the public Tor network, and is "

src/or/connection.c

@@ -18,6 +18,7 @@
  * part of a subclass (channel_tls_t).
  */
 #define TOR_CHANNEL_INTERNAL_
+#define CONNECTION_PRIVATE
 #include "channel.h"
 #include "channeltls.h"
 #include "circuitbuild.h"
@@ -2589,6 +2590,35 @@ record_num_bytes_transferred(connection_t *conn,
 #endif
 
 #ifndef USE_BUFFEREVENTS
+/** Last time at which the global or relay buckets were emptied in msec
+ * since midnight. */
+static uint32_t global_relayed_read_emptied = 0,
+                global_relayed_write_emptied = 0,
+                global_read_emptied = 0,
+                global_write_emptied = 0;
+
+/** Helper: convert given <b>tvnow</b> time value to milliseconds since
+ * midnight. */
+static uint32_t
+msec_since_midnight(const struct timeval *tvnow)
+{
+  return (uint32_t)(((tvnow->tv_sec % 86400L) * 1000L) +
+         ((uint32_t)tvnow->tv_usec / (uint32_t)1000L));
+}
+
+/** Check if a bucket which had <b>tokens_before</b> tokens and which got
+ * <b>tokens_removed</b> tokens removed at timestamp <b>tvnow</b> has run
+ * out of tokens, and if so, note the milliseconds since midnight in
+ * <b>timestamp_var</b> for the next TB_EMPTY event. */
+void
+connection_buckets_note_empty_ts(uint32_t *timestamp_var,
+                                 int tokens_before, size_t tokens_removed,
+                                 const struct timeval *tvnow)
+{
+  if (tokens_before > 0 && (uint32_t)tokens_before <= tokens_removed)
+    *timestamp_var = msec_since_midnight(tvnow);
+}
+
 /** We just read <b>num_read</b> and wrote <b>num_written</b> bytes
  * onto <b>conn</b>. Decrement buckets appropriately. */
 static void
@@ -2611,6 +2641,30 @@ connection_buckets_decrement(connection_t *conn, time_t now,
   if (!connection_is_rate_limited(conn))
     return; /* local IPs are free */
 
+  /* If one or more of our token buckets ran dry just now, note the
+   * timestamp for TB_EMPTY events. */
+  if (get_options()->TestingEnableTbEmptyEvent) {
+    struct timeval tvnow;
+    tor_gettimeofday_cached(&tvnow);
+    if (connection_counts_as_relayed_traffic(conn, now)) {
+      connection_buckets_note_empty_ts(&global_relayed_read_emptied,
+                         global_relayed_read_bucket, num_read, &tvnow);
+      connection_buckets_note_empty_ts(&global_relayed_write_emptied,
+                         global_relayed_write_bucket, num_written, &tvnow);
+    }
+    connection_buckets_note_empty_ts(&global_read_emptied,
+                       global_read_bucket, num_read, &tvnow);
+    connection_buckets_note_empty_ts(&global_write_emptied,
+                       global_write_bucket, num_written, &tvnow);
+    if (connection_speaks_cells(conn) && conn->state == OR_CONN_STATE_OPEN) {
+      or_connection_t *or_conn = TO_OR_CONN(conn);
+      connection_buckets_note_empty_ts(&or_conn->read_emptied_time,
+                         or_conn->read_bucket, num_read, &tvnow);
+      connection_buckets_note_empty_ts(&or_conn->write_emptied_time,
+                         or_conn->write_bucket, num_written, &tvnow);
+    }
+  }
+
   if (connection_counts_as_relayed_traffic(conn, now)) {
     global_relayed_read_bucket -= (int)num_read;
     global_relayed_write_bucket -= (int)num_written;
@@ -2717,6 +2771,28 @@ connection_bucket_refill_helper(int *bucket, int rate, int burst,
   }
 }
 
+/** Helper: return the time in milliseconds since <b>last_empty_time</b>
+ * when a bucket ran empty that previously had <b>tokens_before</b> tokens
+ * now has <b>tokens_after</b> tokens after refilling at timestamp
+ * <b>tvnow</b>, capped at <b>milliseconds_elapsed</b> milliseconds since
+ * last refilling that bucket.  Return 0 if the bucket has not been empty
+ * since the last refill or has not been refilled. */
+uint32_t
+bucket_millis_empty(int tokens_before, uint32_t last_empty_time,
+                    int tokens_after, int milliseconds_elapsed,
+                    const struct timeval *tvnow)
+{
+  uint32_t result = 0, refilled;
+  if (tokens_before <= 0 && tokens_after > tokens_before) {
+    refilled = msec_since_midnight(tvnow);
+    result = (uint32_t)((refilled + 86400L * 1000L - last_empty_time) %
+             (86400L * 1000L));
+    if (result > (uint32_t)milliseconds_elapsed)
+      result = (uint32_t)milliseconds_elapsed;
+  }
+  return result;
+}
+
 /** Time has passed; increment buckets appropriately. */
 void
 connection_bucket_refill(int milliseconds_elapsed, time_t now)
@@ -2725,6 +2801,12 @@ connection_bucket_refill(int milliseconds_elapsed, time_t now)
   smartlist_t *conns = get_connection_array();
   int bandwidthrate, bandwidthburst, relayrate, relayburst;
 
+  int prev_global_read = global_read_bucket;
+  int prev_global_write = global_write_bucket;
+  int prev_relay_read = global_relayed_read_bucket;
+  int prev_relay_write = global_relayed_write_bucket;
+  struct timeval tvnow; /*< Only used if TB_EMPTY events are enabled. */
+
   bandwidthrate = (int)options->BandwidthRate;
   bandwidthburst = (int)options->BandwidthBurst;
 
@@ -2759,12 +2841,42 @@ connection_bucket_refill(int milliseconds_elapsed, time_t now)
                                   milliseconds_elapsed,
                                   "global_relayed_write_bucket");
 
+  /* If buckets were empty before and have now been refilled, tell any
+   * interested controllers. */
+  if (get_options()->TestingEnableTbEmptyEvent) {
+    uint32_t global_read_empty_time, global_write_empty_time,
+             relay_read_empty_time, relay_write_empty_time;
+    tor_gettimeofday_cached(&tvnow);
+    global_read_empty_time = bucket_millis_empty(prev_global_read,
+                             global_read_emptied, global_read_bucket,
+                             milliseconds_elapsed, &tvnow);
+    global_write_empty_time = bucket_millis_empty(prev_global_write,
+                              global_write_emptied, global_write_bucket,
+                              milliseconds_elapsed, &tvnow);
+    control_event_tb_empty("GLOBAL", global_read_empty_time,
+                           global_write_empty_time, milliseconds_elapsed);
+    relay_read_empty_time = bucket_millis_empty(prev_relay_read,
+                            global_relayed_read_emptied,
+                            global_relayed_read_bucket,
+                            milliseconds_elapsed, &tvnow);
+    relay_write_empty_time = bucket_millis_empty(prev_relay_write,
+                             global_relayed_write_emptied,
+                             global_relayed_write_bucket,
+                             milliseconds_elapsed, &tvnow);
+    control_event_tb_empty("RELAY", relay_read_empty_time,
+                           relay_write_empty_time, milliseconds_elapsed);
+  }
+
   /* refill the per-connection buckets */
   SMARTLIST_FOREACH_BEGIN(conns, connection_t *, conn) {
     if (connection_speaks_cells(conn)) {
       or_connection_t *or_conn = TO_OR_CONN(conn);
       int orbandwidthrate = or_conn->bandwidthrate;
       int orbandwidthburst = or_conn->bandwidthburst;
+
+      int prev_conn_read = or_conn->read_bucket;
+      int prev_conn_write = or_conn->write_bucket;
+
       if (connection_bucket_should_increase(or_conn->read_bucket, or_conn)) {
         connection_bucket_refill_helper(&or_conn->read_bucket,
                                         orbandwidthrate,
@@ -2779,6 +2891,27 @@ connection_bucket_refill(int milliseconds_elapsed, time_t now)
                                         milliseconds_elapsed,
                                         "or_conn->write_bucket");
       }
+
+      /* If buckets were empty before and have now been refilled, tell any
+       * interested controllers. */
+      if (get_options()->TestingEnableTbEmptyEvent) {
+        char *bucket;
+        uint32_t conn_read_empty_time, conn_write_empty_time;
+        tor_asprintf(&bucket, "ORCONN ID="U64_FORMAT,
+                     U64_PRINTF_ARG(or_conn->base_.global_identifier));
+        conn_read_empty_time = bucket_millis_empty(prev_conn_read,
+                               or_conn->read_emptied_time,
+                               or_conn->read_bucket,
+                               milliseconds_elapsed, &tvnow);
+        conn_write_empty_time = bucket_millis_empty(prev_conn_write,
+                                or_conn->write_emptied_time,
+                                or_conn->write_bucket,
+                                milliseconds_elapsed, &tvnow);
+        control_event_tb_empty(bucket, conn_read_empty_time,
+                               conn_write_empty_time,
+                               milliseconds_elapsed);
+        tor_free(bucket);
+      }
     }
 
     if (conn->read_blocked_on_bw == 1 /* marked to turn reading back on now */
@@ -3189,14 +3322,37 @@ connection_read_to_buf(connection_t *conn, ssize_t *max_to_read,
      /* change *max_to_read */
     *max_to_read = at_most - n_read;
 
-    /* Update edge_conn->n_read */
+    /* Update edge_conn->n_read and ocirc->n_read_circ_bw */
     if (conn->type == CONN_TYPE_AP) {
       edge_connection_t *edge_conn = TO_EDGE_CONN(conn);
+      circuit_t *circ = circuit_get_by_edge_conn(edge_conn);
+      origin_circuit_t *ocirc;
+
       /* Check for overflow: */
       if (PREDICT_LIKELY(UINT32_MAX - edge_conn->n_read > n_read))
         edge_conn->n_read += (int)n_read;
       else
         edge_conn->n_read = UINT32_MAX;
+
+      if (circ && CIRCUIT_IS_ORIGIN(circ)) {
+        ocirc = TO_ORIGIN_CIRCUIT(circ);
+        if (PREDICT_LIKELY(UINT32_MAX - ocirc->n_read_circ_bw > n_read))
+          ocirc->n_read_circ_bw += (int)n_read;
+        else
+          ocirc->n_read_circ_bw = UINT32_MAX;
+      }
+    }
+
+    /* If CONN_BW events are enabled, update conn->n_read_conn_bw for
+     * OR/DIR/EXIT connections, checking for overflow. */
+    if (get_options()->TestingEnableConnBwEvent &&
+       (conn->type == CONN_TYPE_OR ||
+        conn->type == CONN_TYPE_DIR ||
+        conn->type == CONN_TYPE_EXIT)) {
+      if (PREDICT_LIKELY(UINT32_MAX - conn->n_read_conn_bw > n_read))
+        conn->n_read_conn_bw += (int)n_read;
+      else
+        conn->n_read_conn_bw = UINT32_MAX;
     }
   }
 
@@ -3636,12 +3792,34 @@ connection_handle_write_impl(connection_t *conn, int force)
 
   if (n_written && conn->type == CONN_TYPE_AP) {
     edge_connection_t *edge_conn = TO_EDGE_CONN(conn);
+    circuit_t *circ = circuit_get_by_edge_conn(edge_conn);
+    origin_circuit_t *ocirc;
 
     /* Check for overflow: */
     if (PREDICT_LIKELY(UINT32_MAX - edge_conn->n_written > n_written))
       edge_conn->n_written += (int)n_written;
     else
       edge_conn->n_written = UINT32_MAX;
+
+    if (circ && CIRCUIT_IS_ORIGIN(circ)) {
+      ocirc = TO_ORIGIN_CIRCUIT(circ);
+      if (PREDICT_LIKELY(UINT32_MAX - ocirc->n_written_circ_bw > n_written))
+        ocirc->n_written_circ_bw += (int)n_written;
+      else
+        ocirc->n_written_circ_bw = UINT32_MAX;
+    }
+  }
+
+  /* If CONN_BW events are enabled, update conn->n_written_conn_bw for
+   * OR/DIR/EXIT connections, checking for overflow. */
+  if (n_written && get_options()->TestingEnableConnBwEvent &&
+     (conn->type == CONN_TYPE_OR ||
+      conn->type == CONN_TYPE_DIR ||
+      conn->type == CONN_TYPE_EXIT)) {
+    if (PREDICT_LIKELY(UINT32_MAX - conn->n_written_conn_bw > n_written))
+      conn->n_written_conn_bw += (int)n_written;
+    else
+      conn->n_written_conn_bw = UINT32_MAX;
   }
 
   connection_buckets_decrement(conn, approx_time(), n_read, n_written);

src/or/connection.h

@@ -216,6 +216,15 @@ void connection_enable_rate_limiting(connection_t *conn);
 
 #ifdef CONNECTION_PRIVATE
 STATIC void connection_free_(connection_t *conn);
+
+/* Used only by connection.c and test*.c */
+uint32_t bucket_millis_empty(int tokens_before, uint32_t last_empty_time,
+                             int tokens_after, int milliseconds_elapsed,
+                             const struct timeval *tvnow);
+void connection_buckets_note_empty_ts(uint32_t *timestamp_var,
+                                      int tokens_before,
+                                      size_t tokens_removed,
+                                      const struct timeval *tvnow);
 #endif
 
 #endif

src/or/control.c

@@ -19,6 +19,7 @@
 #include "circuitlist.h"
 #include "circuitstats.h"
 #include "circuituse.h"
+#include "command.h"
 #include "config.h"
 #include "confparse.h"
 #include "connection.h"
@@ -190,6 +191,20 @@ log_severity_to_event(int severity)
   }
 }
 
+/** Helper: clear bandwidth counters of all origin circuits. */
+static void
+clear_circ_bw_fields(void)
+{
+  circuit_t *circ;
+  origin_circuit_t *ocirc;
+  TOR_LIST_FOREACH(circ, circuit_get_global_list(), head) {
+    if (!CIRCUIT_IS_ORIGIN(circ))
+      continue;
+    ocirc = TO_ORIGIN_CIRCUIT(circ);
+    ocirc->n_written_circ_bw = ocirc->n_read_circ_bw = 0;
+  }
+}
+
 /** Set <b>global_event_mask*</b> to the bitwise OR of each live control
  * connection's event_mask field. */
 void
@@ -215,8 +230,8 @@ control_update_global_event_mask(void)
    * we want to hear...*/
   control_adjust_event_log_severity();
 
-  /* ...then, if we've started logging stream bw, clear the appropriate
+  /* ...then, if we've started logging stream or circ bw, clear the
-   * fields. */
+   * appropriate fields. */
   if (! (old_mask & EVENT_STREAM_BANDWIDTH_USED) &&
       (new_mask & EVENT_STREAM_BANDWIDTH_USED)) {
     SMARTLIST_FOREACH(conns, connection_t *, conn,
@@ -227,6 +242,10 @@ control_update_global_event_mask(void)
       }
     });
   }
+  if (! (old_mask & EVENT_CIRC_BANDWIDTH_USED) &&
+      (new_mask & EVENT_CIRC_BANDWIDTH_USED)) {
+    clear_circ_bw_fields();
+  }
 }
 
 /** Adjust the log severities that result in control_event_logmsg being called
@@ -916,6 +935,10 @@ static const struct control_event_t control_event_table[] = {
   { EVENT_BUILDTIMEOUT_SET, "BUILDTIMEOUT_SET" },
   { EVENT_SIGNAL, "SIGNAL" },
   { EVENT_CONF_CHANGED, "CONF_CHANGED"},
+  { EVENT_CONN_BW, "CONN_BW" },
+  { EVENT_CELL_STATS, "CELL_STATS" },
+  { EVENT_TB_EMPTY, "TB_EMPTY" },
+  { EVENT_CIRC_BANDWIDTH_USED, "CIRC_BW" },
   { EVENT_TRANSPORT_LAUNCHED, "TRANSPORT_LAUNCHED" },
   { 0, NULL },
 };
@@ -3834,17 +3857,17 @@ control_event_or_conn_status(or_connection_t *conn, or_conn_status_event_t tp,
   }
   ncircs += connection_or_get_num_circuits(conn);
   if (ncircs && (tp == OR_CONN_EVENT_FAILED || tp == OR_CONN_EVENT_CLOSED)) {
-    tor_snprintf(ncircs_buf, sizeof(ncircs_buf), "%sNCIRCS=%d",
+    tor_snprintf(ncircs_buf, sizeof(ncircs_buf), " NCIRCS=%d", ncircs);
-                 reason ? " " : "", ncircs);
   }
 
   orconn_target_get_name(name, sizeof(name), conn);
   send_control_event(EVENT_OR_CONN_STATUS, ALL_FORMATS,
-                              "650 ORCONN %s %s %s%s%s\r\n",
+                              "650 ORCONN %s %s%s%s%s ID="U64_FORMAT"\r\n",
                               name, status,
                               reason ? " REASON=" : "",
                               orconn_end_reason_to_control_string(reason),
-                              ncircs_buf);
+                              ncircs_buf,
+                              U64_PRINTF_ARG(conn->base_.global_identifier));
 
   return 0;
 }
@@ -3855,6 +3878,8 @@ control_event_or_conn_status(or_connection_t *conn, or_conn_status_event_t tp,
 int
 control_event_stream_bandwidth(edge_connection_t *edge_conn)
 {
+  circuit_t *circ;
+  origin_circuit_t *ocirc;
   if (EVENT_IS_INTERESTING(EVENT_STREAM_BANDWIDTH_USED)) {
     if (!edge_conn->n_read && !edge_conn->n_written)
       return 0;
@@ -3865,6 +3890,12 @@ control_event_stream_bandwidth(edge_connection_t *edge_conn)
                        (unsigned long)edge_conn->n_read,
                        (unsigned long)edge_conn->n_written);
 
+    circ = circuit_get_by_edge_conn(edge_conn);
+    if (circ && CIRCUIT_IS_ORIGIN(circ)) {
+      ocirc = TO_ORIGIN_CIRCUIT(circ);
+      ocirc->n_read_circ_bw += edge_conn->n_read;
+      ocirc->n_written_circ_bw += edge_conn->n_written;
+    }
     edge_conn->n_written = edge_conn->n_read = 0;
   }
 
@@ -3902,6 +3933,235 @@ control_event_stream_bandwidth_used(void)
   return 0;
 }
 
+/** A second or more has elapsed: tell any interested control connections
+ * how much bandwidth origin circuits have used. */
+int
+control_event_circ_bandwidth_used(void)
+{
+  circuit_t *circ;
+  origin_circuit_t *ocirc;
+  if (!EVENT_IS_INTERESTING(EVENT_CIRC_BANDWIDTH_USED))
+    return 0;
+
+  TOR_LIST_FOREACH(circ, circuit_get_global_list(), head) {
+    if (!CIRCUIT_IS_ORIGIN(circ))
+      continue;
+    ocirc = TO_ORIGIN_CIRCUIT(circ);
+    if (!ocirc->n_read_circ_bw && !ocirc->n_written_circ_bw)
+      continue;
+    send_control_event(EVENT_CIRC_BANDWIDTH_USED, ALL_FORMATS,
+                       "650 CIRC_BW ID=%d READ=%lu WRITTEN=%lu\r\n",
+                       ocirc->global_identifier,
+                       (unsigned long)ocirc->n_read_circ_bw,
+                       (unsigned long)ocirc->n_written_circ_bw);
+    ocirc->n_written_circ_bw = ocirc->n_read_circ_bw = 0;
+  }
+
+  return 0;
+}
+
+/** Print out CONN_BW event for a single OR/DIR/EXIT <b>conn</b> and reset
+  * bandwidth counters. */
+int
+control_event_conn_bandwidth(connection_t *conn)
+{
+  const char *conn_type_str;
+  if (!get_options()->TestingEnableConnBwEvent ||
+      !EVENT_IS_INTERESTING(EVENT_CONN_BW))
+    return 0;
+  if (!conn->n_read_conn_bw && !conn->n_written_conn_bw)
+    return 0;
+  switch (conn->type) {
+    case CONN_TYPE_OR:
+      conn_type_str = "OR";
+      break;
+    case CONN_TYPE_DIR:
+      conn_type_str = "DIR";
+      break;
+    case CONN_TYPE_EXIT:
+      conn_type_str = "EXIT";
+      break;
+    default:
+      return 0;
+  }
+  send_control_event(EVENT_CONN_BW, ALL_FORMATS,
+                     "650 CONN_BW ID="U64_FORMAT" TYPE=%s "
+                     "READ=%lu WRITTEN=%lu\r\n",
+                     U64_PRINTF_ARG(conn->global_identifier),
+                     conn_type_str,
+                     (unsigned long)conn->n_read_conn_bw,
+                     (unsigned long)conn->n_written_conn_bw);
+  conn->n_written_conn_bw = conn->n_read_conn_bw = 0;
+  return 0;
+}
+
+/** A second or more has elapsed: tell any interested control
+ * connections how much bandwidth connections have used. */
+int
+control_event_conn_bandwidth_used(void)
+{
+  if (get_options()->TestingEnableConnBwEvent &&
+      EVENT_IS_INTERESTING(EVENT_CONN_BW)) {
+    SMARTLIST_FOREACH(get_connection_array(), connection_t *, conn,
+                      control_event_conn_bandwidth(conn));
+  }
+  return 0;
+}
+
+/** Helper: iterate over cell statistics of <b>circ</b> and sum up added
+ * cells, removed cells, and waiting times by cell command and direction.
+ * Store results in <b>cell_stats</b>.  Free cell statistics of the
+ * circuit afterwards. */
+void
+sum_up_cell_stats_by_command(circuit_t *circ, cell_stats_t *cell_stats)
+{
+  memset(cell_stats, 0, sizeof(cell_stats_t));
+  SMARTLIST_FOREACH_BEGIN(circ->testing_cell_stats,
+                          testing_cell_stats_entry_t *, ent) {
+    tor_assert(ent->command <= CELL_COMMAND_MAX_);
+    if (!ent->removed && !ent->exitward) {
+      cell_stats->added_cells_appward[ent->command] += 1;
+    } else if (!ent->removed && ent->exitward) {
+      cell_stats->added_cells_exitward[ent->command] += 1;
+    } else if (!ent->exitward) {
+      cell_stats->removed_cells_appward[ent->command] += 1;
+      cell_stats->total_time_appward[ent->command] += ent->waiting_time * 10;
+    } else {
+      cell_stats->removed_cells_exitward[ent->command] += 1;
+      cell_stats->total_time_exitward[ent->command] += ent->waiting_time * 10;
+    }
+    tor_free(ent);
+  } SMARTLIST_FOREACH_END(ent);
+  smartlist_free(circ->testing_cell_stats);
+  circ->testing_cell_stats = NULL;
+}
+
+/** Helper: append a cell statistics string to <code>event_parts</code>,
+ * prefixed with <code>key</code>=.  Statistics consist of comma-separated
+ * key:value pairs with lower-case command strings as keys and cell
+ * numbers or total waiting times as values.  A key:value pair is included
+ * if the entry in <code>include_if_non_zero</code> is not zero, but with
+ * the (possibly zero) entry from <code>number_to_include</code>.  Both
+ * arrays are expected to have a length of CELL_COMMAND_MAX_ + 1.  If no
+ * entry in <code>include_if_non_zero</code> is positive, no string will
+ * be added to <code>event_parts</code>. */
+void
+append_cell_stats_by_command(smartlist_t *event_parts, const char *key,
+                             const uint64_t *include_if_non_zero,
+                             const uint64_t *number_to_include)
+{
+  smartlist_t *key_value_strings = smartlist_new();
+  int i;
+  for (i = 0; i <= CELL_COMMAND_MAX_; i++) {
+    if (include_if_non_zero[i] > 0) {
+      smartlist_add_asprintf(key_value_strings, "%s:"U64_FORMAT,
+                             cell_command_to_string(i),
+                             U64_PRINTF_ARG(number_to_include[i]));
+    }
+  }
+  if (smartlist_len(key_value_strings) > 0) {
+    char *joined = smartlist_join_strings(key_value_strings, ",", 0, NULL);
+    smartlist_add_asprintf(event_parts, "%s=%s", key, joined);
+    SMARTLIST_FOREACH(key_value_strings, char *, cp, tor_free(cp));
+    tor_free(joined);
+  }
+  smartlist_free(key_value_strings);
+}
+
+/** Helper: format <b>cell_stats</b> for <b>circ</b> for inclusion in a
+ * CELL_STATS event and write result string to <b>event_string</b>. */
+void
+format_cell_stats(char **event_string, circuit_t *circ,
+                  cell_stats_t *cell_stats)
+{
+  smartlist_t *event_parts = smartlist_new();
+  if (CIRCUIT_IS_ORIGIN(circ)) {
+    origin_circuit_t *ocirc = TO_ORIGIN_CIRCUIT(circ);
+    smartlist_add_asprintf(event_parts, "ID=%lu",
+                 (unsigned long)ocirc->global_identifier);
+  } else if (TO_OR_CIRCUIT(circ)->p_chan) {
+    or_circuit_t *or_circ = TO_OR_CIRCUIT(circ);
+    smartlist_add_asprintf(event_parts, "InboundQueue=%lu",
+                 (unsigned long)or_circ->p_circ_id);
+    smartlist_add_asprintf(event_parts, "InboundConn="U64_FORMAT,
+                 U64_PRINTF_ARG(or_circ->p_chan->global_identifier));
+    append_cell_stats_by_command(event_parts, "InboundAdded",
+                                 cell_stats->added_cells_appward,
+                                 cell_stats->added_cells_appward);
+    append_cell_stats_by_command(event_parts, "InboundRemoved",
+                                 cell_stats->removed_cells_appward,
+                                 cell_stats->removed_cells_appward);
+    append_cell_stats_by_command(event_parts, "InboundTime",
+                                 cell_stats->removed_cells_appward,
+                                 cell_stats->total_time_appward);
+  }
+  if (circ->n_chan) {
+    smartlist_add_asprintf(event_parts, "OutboundQueue=%lu",
+                     (unsigned long)circ->n_circ_id);
+    smartlist_add_asprintf(event_parts, "OutboundConn="U64_FORMAT,
+                 U64_PRINTF_ARG(circ->n_chan->global_identifier));
+    append_cell_stats_by_command(event_parts, "OutboundAdded",
+                                 cell_stats->added_cells_exitward,
+                                 cell_stats->added_cells_exitward);
+    append_cell_stats_by_command(event_parts, "OutboundRemoved",
+                                 cell_stats->removed_cells_exitward,
+                                 cell_stats->removed_cells_exitward);
+    append_cell_stats_by_command(event_parts, "OutboundTime",
+                                 cell_stats->removed_cells_exitward,
+                                 cell_stats->total_time_exitward);
+  }
+  *event_string = smartlist_join_strings(event_parts, " ", 0, NULL);
+  SMARTLIST_FOREACH(event_parts, char *, cp, tor_free(cp));
+  smartlist_free(event_parts);
+}
+
+/** A second or more has elapsed: tell any interested control connection
+ * how many cells have been processed for a given circuit. */
+int
+control_event_circuit_cell_stats(void)
+{
+  circuit_t *circ;
+  cell_stats_t *cell_stats;
+  char *event_string;
+  if (!get_options()->TestingEnableCellStatsEvent ||
+      !EVENT_IS_INTERESTING(EVENT_CELL_STATS))
+    return 0;
+  cell_stats = tor_malloc(sizeof(cell_stats_t));;
+  TOR_LIST_FOREACH(circ, circuit_get_global_list(), head) {
+    if (!circ->testing_cell_stats)
+      continue;
+    sum_up_cell_stats_by_command(circ, cell_stats);
+    format_cell_stats(&event_string, circ, cell_stats);
+    send_control_event(EVENT_CELL_STATS, ALL_FORMATS,
+                       "650 CELL_STATS %s\r\n", event_string);
+    tor_free(event_string);
+  }
+  tor_free(cell_stats);
+  return 0;
+}
+
+/** Tokens in <b>bucket</b> have been refilled: the read bucket was empty
+ * for <b>read_empty_time</b> millis, the write bucket was empty for
+ * <b>write_empty_time</b> millis, and buckets were last refilled
+ * <b>milliseconds_elapsed</b> millis ago.  Only emit TB_EMPTY event if
+ * either read or write bucket have been empty before. */
+int
+control_event_tb_empty(const char *bucket, uint32_t read_empty_time,
+                       uint32_t write_empty_time,
+                       int milliseconds_elapsed)
+{
+  if (get_options()->TestingEnableTbEmptyEvent &&
+      EVENT_IS_INTERESTING(EVENT_TB_EMPTY) &&
+      (read_empty_time > 0 || write_empty_time > 0)) {
+    send_control_event(EVENT_TB_EMPTY, ALL_FORMATS,
+                       "650 TB_EMPTY %s READ=%d WRITTEN=%d "
+                       "LAST=%d\r\n",
+                       bucket, read_empty_time, write_empty_time,
+                       milliseconds_elapsed);
+  }
+  return 0;
+}
+
 /** A second or more has elapsed: tell any interested control
  * connections how much bandwidth we used. */
 int

src/or/control.h

@@ -50,6 +50,13 @@ int control_event_or_conn_status(or_connection_t *conn,
 int control_event_bandwidth_used(uint32_t n_read, uint32_t n_written);
 int control_event_stream_bandwidth(edge_connection_t *edge_conn);
 int control_event_stream_bandwidth_used(void);
+int control_event_circ_bandwidth_used(void);
+int control_event_conn_bandwidth(connection_t *conn);
+int control_event_conn_bandwidth_used(void);
+int control_event_circuit_cell_stats(void);
+int control_event_tb_empty(const char *bucket, uint32_t read_empty_time,
+                           uint32_t write_empty_time,
+                           int milliseconds_elapsed);
 void control_event_logmsg(int severity, uint32_t domain, const char *msg);
 int control_event_descriptors_changed(smartlist_t *routers);
 int control_event_address_mapped(const char *from, const char *to,
@@ -128,6 +135,10 @@ void control_free_all(void);
 #define EVENT_BUILDTIMEOUT_SET        0x0017
 #define EVENT_SIGNAL                  0x0018
 #define EVENT_CONF_CHANGED            0x0019
+#define EVENT_CONN_BW                 0x001A
+#define EVENT_CELL_STATS              0x001B
+#define EVENT_TB_EMPTY                0x001C
+#define EVENT_CIRC_BANDWIDTH_USED     0x001D
 #define EVENT_TRANSPORT_LAUNCHED      0x0020
 #define EVENT_MAX_                    0x0020
 /* If EVENT_MAX_ ever hits 0x0040, we need to make the mask into a
@@ -151,6 +162,30 @@ send_control_event_string,(uint16_t event, event_format_t which,
 
 void control_testing_set_global_event_mask(uint64_t mask);
 #endif
+
+/** Helper structure: temporarily stores cell statistics for a circuit. */
+typedef struct cell_stats_t {
+  /** Number of cells added in app-ward direction by command. */
+  uint64_t added_cells_appward[CELL_COMMAND_MAX_ + 1];
+  /** Number of cells added in exit-ward direction by command. */
+  uint64_t added_cells_exitward[CELL_COMMAND_MAX_ + 1];
+  /** Number of cells removed in app-ward direction by command. */
+  uint64_t removed_cells_appward[CELL_COMMAND_MAX_ + 1];
+  /** Number of cells removed in exit-ward direction by command. */
+  uint64_t removed_cells_exitward[CELL_COMMAND_MAX_ + 1];
+  /** Total waiting time of cells in app-ward direction by command. */
+  uint64_t total_time_appward[CELL_COMMAND_MAX_ + 1];
+  /** Total waiting time of cells in exit-ward direction by command. */
+  uint64_t total_time_exitward[CELL_COMMAND_MAX_ + 1];
+} cell_stats_t;
+void sum_up_cell_stats_by_command(circuit_t *circ,
+                                  cell_stats_t *cell_stats);
+void append_cell_stats_by_command(smartlist_t *event_parts,
+                                  const char *key,
+                                  const uint64_t *include_if_non_zero,
+                                  const uint64_t *number_to_include);
+void format_cell_stats(char **event_string, circuit_t *circ,
+                       cell_stats_t *cell_stats);
 #endif
 
 #endif

src/or/main.c

@@ -352,6 +352,8 @@ connection_remove(connection_t *conn)
             (int)conn->s, conn_type_to_string(conn->type),
             smartlist_len(connection_array));
 
+  control_event_conn_bandwidth(conn);
+
   tor_assert(conn->conn_array_index >= 0);
   current_index = conn->conn_array_index;
   connection_unregister_events(conn); /* This is redundant, but cheap. */
@@ -1668,6 +1670,9 @@ second_elapsed_callback(periodic_timer_t *timer, void *arg)
 
   control_event_bandwidth_used((uint32_t)bytes_read,(uint32_t)bytes_written);
   control_event_stream_bandwidth_used();
+  control_event_conn_bandwidth_used();
+  control_event_circ_bandwidth_used();
+  control_event_circuit_cell_stats();
 
   if (server_mode(options) &&
       !net_is_disabled() &&

src/or/or.h

@@ -880,6 +880,7 @@ typedef enum {
 #define CELL_AUTH_CHALLENGE 130
 #define CELL_AUTHENTICATE 131
 #define CELL_AUTHORIZE 132
+#define CELL_COMMAND_MAX_ 132
 
 /** How long to test reachability before complaining to the user. */
 #define TIMEOUT_UNTIL_UNREACHABILITY_COMPLAINT (20*60)
@@ -1136,6 +1137,21 @@ typedef struct insertion_time_queue_t {
   struct insertion_time_elem_t *last; /**< Last element in queue. */
 } insertion_time_queue_t;
 
+/** Number of cells with the same command consecutively added to a circuit
+ * queue; used for cell statistics only if CELL_STATS events are enabled. */
+typedef struct insertion_command_elem_t {
+  struct insertion_command_elem_t *next; /**< Next element in queue. */
+  /** Which command did these consecutively added cells have? */
+  uint8_t command;
+  unsigned counter; /**< How many cells were inserted? */
+} insertion_command_elem_t;
+
+/** Queue of insertion commands. */
+typedef struct insertion_command_queue_t {
+  struct insertion_command_elem_t *first; /**< First element in queue. */
+  struct insertion_command_elem_t *last; /**< Last element in queue. */
+} insertion_command_queue_t;
+
 /** A queue of cells on a circuit, waiting to be added to the
  * or_connection_t's outbuf. */
 typedef struct cell_queue_t {
@@ -1143,6 +1159,8 @@ typedef struct cell_queue_t {
   TOR_SIMPLEQ_HEAD(cell_simpleq, packed_cell_t) head;
   int n; /**< The number of cells in the queue. */
   insertion_time_queue_t *insertion_times; /**< Insertion times of cells. */
+ /** Commands of inserted cells. */
+  insertion_command_queue_t *insertion_commands;
 } cell_queue_t;
 
 /** Beginning of a RELAY cell payload. */
@@ -1279,6 +1297,14 @@ typedef struct connection_t {
 
   /** Unique identifier for this connection on this Tor instance. */
   uint64_t global_identifier;
+
+  /** Bytes read since last call to control_event_conn_bandwidth_used().
+   * Only used if we're configured to emit CONN_BW events. */
+  uint32_t n_read_conn_bw;
+
+  /** Bytes written since last call to control_event_conn_bandwidth_used().
+   * Only used if we're configured to emit CONN_BW events. */
+  uint32_t n_written_conn_bw;
 } connection_t;
 
 /** Subtype of connection_t; used for a listener socket. */
@@ -1522,6 +1548,12 @@ typedef struct or_connection_t {
 
   struct or_connection_t *next_with_same_id; /**< Next connection with same
                                               * identity digest as this one. */
+  /** Last emptied read token bucket in msec since midnight; only used if
+   * TB_EMPTY events are enabled. */
+  uint32_t read_emptied_time;
+  /** Last emptied write token bucket in msec since midnight; only used if
+   * TB_EMPTY events are enabled. */
+  uint32_t write_emptied_time;
 } or_connection_t;
 
 /** Subtype of connection_t for an "edge connection" -- that is, an entry (ap)
@@ -2785,6 +2817,19 @@ typedef struct {
 
 struct create_cell_t;
 
+/** Entry in the cell stats list of a circuit; used only if CELL_STATS
+ * events are enabled. */
+typedef struct testing_cell_stats_entry_t {
+  uint8_t command; /**< cell command number. */
+  /** Waiting time in centiseconds if this event is for a removed cell,
+   * or 0 if this event is for adding a cell to the queue.  22 bits can
+   * store more than 11 hours, enough to assume that a circuit with this
+   * delay would long have been closed. */
+  unsigned int waiting_time:22;
+  unsigned int removed:1; /**< 0 for added to, 1 for removed from queue. */
+  unsigned int exitward:1; /**< 0 for app-ward, 1 for exit-ward. */
+} testing_cell_stats_entry_t;
+
 /**
  * A circuit is a path over the onion routing
  * network. Applications can connect to one end of the circuit, and can
@@ -2918,6 +2963,11 @@ typedef struct circuit_t {
    * cells to n_conn.  NULL if we have no cells pending, or if we're not
    * linked to an OR connection. */
   struct circuit_t *prev_active_on_n_chan;
+
+  /** Various statistics about cells being added to or removed from this
+   * circuit's queues; used only if CELL_STATS events are enabled and
+   * cleared after being sent to control port. */
+  smartlist_t *testing_cell_stats;
 } circuit_t;
 
 /** Largest number of relay_early cells that we can send on a given
@@ -2988,6 +3038,17 @@ typedef struct origin_circuit_t {
   /** Linked list of AP streams (or EXIT streams if hidden service)
    * associated with this circuit. */
   edge_connection_t *p_streams;
+
+  /** Bytes read from any attached stream since last call to
+   * control_event_circ_bandwidth_used().  Only used if we're configured
+   * to emit CIRC_BW events. */
+  uint32_t n_read_circ_bw;
+
+  /** Bytes written to any attached stream since last call to
+   * control_event_circ_bandwidth_used().  Only used if we're configured
+   * to emit CIRC_BW events. */
+  uint32_t n_written_circ_bw;
+
   /** Build state for this circuit. It includes the intended path
    * length, the chosen exit router, rendezvous information, etc.
    */
@@ -4069,6 +4130,15 @@ typedef struct {
    * regardless of uptime and bandwidth. */
   routerset_t *TestingDirAuthVoteGuard;
 
+  /** Enable CONN_BW events.  Only altered on testing networks. */
+  int TestingEnableConnBwEvent;
+
+  /** Enable CELL_STATS events.  Only altered on testing networks. */
+  int TestingEnableCellStatsEvent;
+
+  /** Enable TB_EMPTY events.  Only altered on testing networks. */
+  int TestingEnableTbEmptyEvent;
+
   /** If true, and we have GeoIP data, and we're a bridge, keep a per-country
    * count of how many client addresses have contacted us so that we can help
    * the bridge authority guess which countries have blocked access to us. */

src/or/relay.c

@@ -2050,6 +2050,10 @@ static mp_pool_t *cell_pool = NULL;
  * statistics. */
 static mp_pool_t *it_pool = NULL;
 
+/** Memory pool to allocate insertion_command_elem_t objects used for cell
+ * statistics if CELL_STATS events are enabled. */
+static mp_pool_t *ic_pool = NULL;
+
 /** Allocate structures to hold cells. */
 void
 init_cell_pool(void)
@@ -2058,8 +2062,8 @@ init_cell_pool(void)
   cell_pool = mp_pool_new(sizeof(packed_cell_t), 128*1024);
 }
 
-/** Free all storage used to hold cells (and insertion times if we measure
+/** Free all storage used to hold cells (and insertion times/commands if we
- * cell statistics). */
+ * measure cell statistics and/or if CELL_STATS events are enabled). */
 void
 free_cell_pool(void)
 {
@@ -2072,6 +2076,10 @@ free_cell_pool(void)
     mp_pool_destroy(it_pool);
     it_pool = NULL;
   }
+  if (ic_pool) {
+    mp_pool_destroy(ic_pool);
+    ic_pool = NULL;
+  }
 }
 
 /** Free excess storage in cell pool. */
@@ -2144,14 +2152,68 @@ cell_queue_append(cell_queue_t *queue, packed_cell_t *cell)
   ++queue->n;
 }
 
-/** Append a newly allocated copy of <b>cell</b> to the end of <b>queue</b> */
+/** Append command of type <b>command</b> in direction to <b>queue</b> for
+ * CELL_STATS event. */
+static void
+cell_command_queue_append(cell_queue_t *queue, uint8_t command)
+{
+  insertion_command_queue_t *ic_queue = queue->insertion_commands;
+  if (!ic_pool)
+    ic_pool = mp_pool_new(sizeof(insertion_command_elem_t), 1024);
+  if (!ic_queue) {
+    ic_queue = tor_malloc_zero(sizeof(insertion_command_queue_t));
+    queue->insertion_commands = ic_queue;
+  }
+  if (ic_queue->last && ic_queue->last->command == command) {
+    ic_queue->last->counter++;
+  } else {
+    insertion_command_elem_t *elem = mp_pool_get(ic_pool);
+    elem->next = NULL;
+    elem->command = command;
+    elem->counter = 1;
+    if (ic_queue->last) {
+      ic_queue->last->next = elem;
+      ic_queue->last = elem;
+    } else {
+      ic_queue->first = ic_queue->last = elem;
+    }
+  }
+}
+
+/** Retrieve oldest command from <b>queue</b> and write it to
+ * <b>command</b> for CELL_STATS event.  Return 0 for success, -1
+ * otherwise. */
+static int
+cell_command_queue_pop(uint8_t *command, cell_queue_t *queue)
+{
+  int res = -1;
+  insertion_command_queue_t *ic_queue = queue->insertion_commands;
+  if (ic_queue && ic_queue->first) {
+    insertion_command_elem_t *ic_elem = ic_queue->first;
+    ic_elem->counter--;
+    if (ic_elem->counter < 1) {
+      ic_queue->first = ic_elem->next;
+      if (ic_elem == ic_queue->last)
+        ic_queue->last = NULL;
+      mp_pool_release(ic_elem);
+    }
+    *command = ic_elem->command;
+    res = 0;
+  }
+  return res;
+}
+
+/** Append a newly allocated copy of <b>cell</b> to the end of the
+ * <b>exitward</b> (or app-ward) <b>queue</b> of <b>circ</b>. */
 void
-cell_queue_append_packed_copy(cell_queue_t *queue, const cell_t *cell,
+cell_queue_append_packed_copy(circuit_t *circ, cell_queue_t *queue,
+                              int exitward, const cell_t *cell,
                               int wide_circ_ids, int use_stats)
 {
   packed_cell_t *copy = packed_cell_copy(cell, wide_circ_ids);
   /* Remember the time when this cell was put in the queue. */
-  if (get_options()->CellStatistics && use_stats) {
+  if ((get_options()->CellStatistics ||
+      get_options()->TestingEnableCellStatsEvent) && use_stats) {
     struct timeval now;
     uint32_t added;
     insertion_time_queue_t *it_queue = queue->insertion_times;
@@ -2180,6 +2242,18 @@ cell_queue_append_packed_copy(cell_queue_t *queue, const cell_t *cell,
       }
     }
   }
+  /* Remember that we added a cell to the queue, and remember the cell
+   * command. */
+  if (get_options()->TestingEnableCellStatsEvent) {
+    testing_cell_stats_entry_t *ent =
+                      tor_malloc_zero(sizeof(testing_cell_stats_entry_t));
+    ent->command = cell->command;
+    ent->exitward = exitward;
+    if (!circ->testing_cell_stats)
+      circ->testing_cell_stats = smartlist_new();
+    smartlist_add(circ->testing_cell_stats, ent);
+    cell_command_queue_append(queue, cell->command);
+  }
   cell_queue_append(queue, copy);
 }
 
@@ -2421,7 +2495,8 @@ channel_flush_from_first_active_circuit(channel_t *chan, int max)
     cell = cell_queue_pop(queue);
 
     /* Calculate the exact time that this cell has spent in the queue. */
-    if (get_options()->CellStatistics && !CIRCUIT_IS_ORIGIN(circ)) {
+    if (get_options()->CellStatistics ||
+        get_options()->TestingEnableCellStatsEvent) {
       struct timeval tvnow;
       uint32_t flushed;
       uint32_t cell_waiting_time;
@@ -2435,7 +2510,6 @@ channel_flush_from_first_active_circuit(channel_t *chan, int max)
                              "recently enabled.");
       } else {
         insertion_time_elem_t *elem = it_queue->first;
-        or_circ = TO_OR_CIRCUIT(circ);
         cell_waiting_time =
             (uint32_t)((flushed * 10L + SECONDS_IN_A_DAY * 1000L -
                         elem->insertion_time * 10L) %
@@ -2448,9 +2522,31 @@ channel_flush_from_first_active_circuit(channel_t *chan, int max)
             it_queue->last = NULL;
           mp_pool_release(elem);
         }
+        if (get_options()->CellStatistics && !CIRCUIT_IS_ORIGIN(circ)) {
+          or_circ = TO_OR_CIRCUIT(circ);
           or_circ->total_cell_waiting_time += cell_waiting_time;
           or_circ->processed_cells++;
         }
+        if (get_options()->TestingEnableCellStatsEvent) {
+          uint8_t command;
+          if (cell_command_queue_pop(&command, queue) < 0) {
+            log_info(LD_GENERAL, "Cannot determine command of cell. "
+                                 "Looks like the CELL_STATS event was "
+                                 "recently enabled.");
+          } else {
+            testing_cell_stats_entry_t *ent =
+                        tor_malloc_zero(sizeof(testing_cell_stats_entry_t));
+            ent->command = command;
+            ent->waiting_time = (unsigned int)cell_waiting_time / 10;
+            ent->removed = 1;
+            if (circ->n_chan == chan)
+              ent->exitward = 1;
+            if (!circ->testing_cell_stats)
+              circ->testing_cell_stats = smartlist_new();
+            smartlist_add(circ->testing_cell_stats, ent);
+          }
+        }
+      }
     }
 
     /* If we just flushed our queue and this circuit is used for a
@@ -2524,10 +2620,12 @@ append_cell_to_circuit_queue(circuit_t *circ, channel_t *chan,
   uint32_t tgt_max_middle_cells, p_len, n_len, tmp, hard_max_middle_cells;
 #endif
 
+  int exitward;
   if (circ->marked_for_close)
     return;
 
-  if (direction == CELL_DIRECTION_OUT) {
+  exitward = (direction == CELL_DIRECTION_OUT);
+  if (exitward) {
     queue = &circ->n_chan_cells;
     streams_blocked = circ->streams_blocked_on_n_chan;
   } else {
@@ -2618,7 +2716,8 @@ append_cell_to_circuit_queue(circuit_t *circ, channel_t *chan,
   }
 #endif
 
-  cell_queue_append_packed_copy(queue, cell, chan->wide_circ_ids, 1);
+  cell_queue_append_packed_copy(circ, queue, exitward, cell,
+                                chan->wide_circ_ids, 1);
 
   if (PREDICT_UNLIKELY(cell_queues_check_size())) {
     /* We ran the OOM handler */

src/or/relay.h

@@ -54,7 +54,8 @@ void packed_cell_free(packed_cell_t *cell);
 void cell_queue_init(cell_queue_t *queue);
 void cell_queue_clear(cell_queue_t *queue);
 void cell_queue_append(cell_queue_t *queue, packed_cell_t *cell);
-void cell_queue_append_packed_copy(cell_queue_t *queue, const cell_t *cell,
+void cell_queue_append_packed_copy(circuit_t *circ, cell_queue_t *queue,
+                                   int exitward, const cell_t *cell,
                                    int wide_circ_ids, int use_stats);
 
 void append_cell_to_circuit_queue(circuit_t *circ, channel_t *chan,

src/test/Makefile.nmake

@@ -12,9 +12,9 @@ LIBS = ..\..\..\build-alpha\lib\libevent.lib \
  crypt32.lib gdi32.lib user32.lib
 
 TEST_OBJECTS = test.obj test_addr.obj test_containers.obj \
-	test_crypto.obj test_data.obj test_dir.obj test_microdesc.obj \
+	test_controller_events.ogj test_crypto.obj test_data.obj test_dir.obj \
-	test_pt.obj test_util.obj test_config.obj test_cell_formats.obj \
+	test_microdesc.obj test_pt.obj test_util.obj test_config.obj \
-	test_replay.obj test_introduce.obj tinytest.obj
+	test_cell_formats.obj test_replay.obj test_introduce.obj tinytest.obj
 
 tinytest.obj: ..\ext\tinytest.c
 	$(CC) $(CFLAGS) /D snprintf=_snprintf /c ..\ext\tinytest.c

src/test/include.am

@@ -23,6 +23,7 @@ src_test_test_SOURCES = \
 	src/test/test_circuitlist.c \
 	src/test/test_circuitmux.c \
 	src/test/test_containers.c \
+	src/test/test_controller_events.c \
 	src/test/test_crypto.c \
 	src/test/test_cell_queue.c \
 	src/test/test_data.c \

src/test/test.c

@@ -1620,6 +1620,7 @@ extern struct testcase_t cell_queue_tests[];
 extern struct testcase_t options_tests[];
 extern struct testcase_t socks_tests[];
 extern struct testcase_t extorport_tests[];
+extern struct testcase_t controller_event_tests[];
 
 static struct testgroup_t testgroups[] = {
   { "", test_array },
@@ -1641,6 +1642,7 @@ static struct testgroup_t testgroups[] = {
   { "circuitmux/", circuitmux_tests },
   { "options/", options_tests },
   { "extorport/", extorport_tests },
+  { "control/", controller_event_tests },
   END_OF_GROUPS
 };
 

src/test/test_cell_queue.c

@@ -56,9 +56,11 @@ test_cq_manip(void *arg)
           "once-ler lerkim, sed do barbaloot tempor gluppitus ut labore et "
           "truffula magna aliqua.",
           sizeof(cell.payload));
-  cell_queue_append_packed_copy(&cq, &cell, 1 /*wide*/, 0 /*stats*/);
+  cell_queue_append_packed_copy(NULL /*circ*/, &cq, 0 /*exitward*/, &cell,
+                                1 /*wide*/, 0 /*stats*/);
   cell.circ_id = 0x2013;
-  cell_queue_append_packed_copy(&cq, &cell, 0 /*wide*/, 0 /*stats*/);
+  cell_queue_append_packed_copy(NULL /*circ*/, &cq, 0 /*exitward*/, &cell,
+                                0 /*wide*/, 0 /*stats*/);
   tt_int_op(cq.n, ==, 2);
 
   pc_tmp = cell_queue_pop(&cq);

src/test/test_controller_events.c

@@ -0,0 +1,287 @@
+/* Copyright (c) 2013, The Tor Project, Inc. */
+/* See LICENSE for licensing information */
+
+#define CONNECTION_PRIVATE
+#define TOR_CHANNEL_INTERNAL_
+#define CONTROL_PRIVATE
+#include "or.h"
+#include "channel.h"
+#include "channeltls.h"
+#include "connection.h"
+#include "control.h"
+#include "test.h"
+
+static void
+help_test_bucket_note_empty(uint32_t expected_msec_since_midnight,
+                            int tokens_before, size_t tokens_removed,
+                            uint32_t msec_since_epoch)
+{
+  uint32_t timestamp_var = 0;
+  struct timeval tvnow;
+  tvnow.tv_sec = msec_since_epoch / 1000;
+  tvnow.tv_usec = (msec_since_epoch % 1000) * 1000;
+  connection_buckets_note_empty_ts(&timestamp_var, tokens_before,
+                                   tokens_removed, &tvnow);
+  tt_int_op(expected_msec_since_midnight, ==, timestamp_var);
+
+ done:
+  ;
+}
+
+static void
+test_cntev_bucket_note_empty(void *arg)
+{
+  (void)arg;
+
+  /* Two cases with nothing to note, because bucket was empty before;
+   * 86442200 == 1970-01-02 00:00:42.200000 */
+  help_test_bucket_note_empty(0, 0, 0, 86442200);
+  help_test_bucket_note_empty(0, -100, 100, 86442200);
+
+  /* Nothing to note, because bucket has not been emptied. */
+  help_test_bucket_note_empty(0, 101, 100, 86442200);
+
+  /* Bucket was emptied, note 42200 msec since midnight. */
+  help_test_bucket_note_empty(42200, 101, 101, 86442200);
+  help_test_bucket_note_empty(42200, 101, 102, 86442200);
+}
+
+static void
+test_cntev_bucket_millis_empty(void *arg)
+{
+  struct timeval tvnow;
+  (void)arg;
+
+  /* 1970-01-02 00:00:42.200000 */
+  tvnow.tv_sec = 86400 + 42;
+  tvnow.tv_usec = 200000;
+
+  /* Bucket has not been refilled. */
+  tt_int_op(0, ==, bucket_millis_empty(0, 42120, 0, 100, &tvnow));
+  tt_int_op(0, ==, bucket_millis_empty(-10, 42120, -10, 100, &tvnow));
+
+  /* Bucket was not empty. */
+  tt_int_op(0, ==, bucket_millis_empty(10, 42120, 20, 100, &tvnow));
+
+  /* Bucket has been emptied 80 msec ago and has just been refilled. */
+  tt_int_op(80, ==, bucket_millis_empty(-20, 42120, -10, 100, &tvnow));
+  tt_int_op(80, ==, bucket_millis_empty(-10, 42120, 0, 100, &tvnow));
+  tt_int_op(80, ==, bucket_millis_empty(0, 42120, 10, 100, &tvnow));
+
+  /* Bucket has been emptied 180 msec ago, last refill was 100 msec ago
+   * which was insufficient to make it positive, so cap msec at 100. */
+  tt_int_op(100, ==, bucket_millis_empty(0, 42020, 1, 100, &tvnow));
+
+  /* 1970-01-02 00:00:00:050000 */
+  tvnow.tv_sec = 86400;
+  tvnow.tv_usec = 50000;
+
+  /* Last emptied 30 msec before midnight, tvnow is 50 msec after
+   * midnight, that's 80 msec in total. */
+  tt_int_op(80, ==, bucket_millis_empty(0, 86400000 - 30, 1, 100, &tvnow));
+
+ done:
+  ;
+}
+
+static void
+add_testing_cell_stats_entry(circuit_t *circ, uint8_t command,
+                             unsigned int waiting_time,
+                             unsigned int removed, unsigned int exitward)
+{
+  testing_cell_stats_entry_t *ent = tor_malloc_zero(
+                                    sizeof(testing_cell_stats_entry_t));
+  ent->command = command;
+  ent->waiting_time = waiting_time;
+  ent->removed = removed;
+  ent->exitward = exitward;
+  if (!circ->testing_cell_stats)
+    circ->testing_cell_stats = smartlist_new();
+  smartlist_add(circ->testing_cell_stats, ent);
+}
+
+static void
+test_cntev_sum_up_cell_stats(void *arg)
+{
+  or_circuit_t *or_circ;
+  circuit_t *circ;
+  cell_stats_t *cell_stats;
+  (void)arg;
+
+  or_circ = tor_malloc_zero(sizeof(or_circuit_t));
+  or_circ->base_.magic = OR_CIRCUIT_MAGIC;
+  or_circ->base_.purpose = CIRCUIT_PURPOSE_OR;
+  circ = TO_CIRCUIT(or_circ);
+
+  /* A single RELAY cell was added to the appward queue. */
+  cell_stats = tor_malloc_zero(sizeof(cell_stats_t));
+  add_testing_cell_stats_entry(circ, CELL_RELAY, 0, 0, 0);
+  sum_up_cell_stats_by_command(circ, cell_stats);
+  tt_int_op(1, ==, cell_stats->added_cells_appward[CELL_RELAY]);
+
+  /* A single RELAY cell was added to the exitward queue. */
+  add_testing_cell_stats_entry(circ, CELL_RELAY, 0, 0, 1);
+  sum_up_cell_stats_by_command(circ, cell_stats);
+  tt_int_op(1, ==, cell_stats->added_cells_exitward[CELL_RELAY]);
+
+  /* A single RELAY cell was removed from the appward queue where it spent
+   * 20 msec. */
+  add_testing_cell_stats_entry(circ, CELL_RELAY, 2, 1, 0);
+  sum_up_cell_stats_by_command(circ, cell_stats);
+  tt_int_op(20, ==, cell_stats->total_time_appward[CELL_RELAY]);
+  tt_int_op(1, ==, cell_stats->removed_cells_appward[CELL_RELAY]);
+
+  /* A single RELAY cell was removed from the exitward queue where it
+   * spent 30 msec. */
+  add_testing_cell_stats_entry(circ, CELL_RELAY, 3, 1, 1);
+  sum_up_cell_stats_by_command(circ, cell_stats);
+  tt_int_op(30, ==, cell_stats->total_time_exitward[CELL_RELAY]);
+  tt_int_op(1, ==, cell_stats->removed_cells_exitward[CELL_RELAY]);
+
+ done:
+  ;
+}
+
+static void
+test_cntev_append_cell_stats(void *arg)
+{
+  smartlist_t *event_parts;
+  const char *key = "Z";
+  uint64_t include_if_non_zero[CELL_COMMAND_MAX_ + 1],
+           number_to_include[CELL_COMMAND_MAX_ + 1];
+  (void)arg;
+
+  event_parts = smartlist_new();
+  memset(include_if_non_zero, 0,
+         (CELL_COMMAND_MAX_ + 1) * sizeof(uint64_t));
+  memset(number_to_include, 0,
+         (CELL_COMMAND_MAX_ + 1) * sizeof(uint64_t));
+
+  /* All array entries empty. */
+  append_cell_stats_by_command(event_parts, key,
+                               include_if_non_zero,
+                               number_to_include);
+  tt_int_op(0, ==, smartlist_len(event_parts));
+
+  /* There's a RELAY cell to include, but the corresponding field in
+   * include_if_non_zero is still zero. */
+  number_to_include[CELL_RELAY] = 1;
+  append_cell_stats_by_command(event_parts, key,
+                               include_if_non_zero,
+                               number_to_include);
+  tt_int_op(0, ==, smartlist_len(event_parts));
+
+  /* Now include single RELAY cell. */
+  include_if_non_zero[CELL_RELAY] = 2;
+  append_cell_stats_by_command(event_parts, key,
+                               include_if_non_zero,
+                               number_to_include);
+  tt_str_op("Z=relay:1", ==, smartlist_pop_last(event_parts));
+
+  /* Add four CREATE cells. */
+  include_if_non_zero[CELL_CREATE] = 3;
+  number_to_include[CELL_CREATE] = 4;
+  append_cell_stats_by_command(event_parts, key,
+                               include_if_non_zero,
+                               number_to_include);
+  tt_str_op("Z=create:4,relay:1", ==, smartlist_pop_last(event_parts));
+
+ done:
+  ;
+}
+
+static void
+test_cntev_format_cell_stats(void *arg)
+{
+  char *event_string;
+  origin_circuit_t *ocirc;
+  or_circuit_t *or_circ;
+  cell_stats_t *cell_stats;
+  channel_tls_t *n_chan, *p_chan;
+  (void)arg;
+
+  n_chan = tor_malloc_zero(sizeof(channel_tls_t));
+  n_chan->base_.global_identifier = 1;
+
+  ocirc = tor_malloc_zero(sizeof(origin_circuit_t));
+  ocirc->base_.magic = ORIGIN_CIRCUIT_MAGIC;
+  ocirc->base_.purpose = CIRCUIT_PURPOSE_C_GENERAL;
+  ocirc->global_identifier = 2;
+  ocirc->base_.n_circ_id = 3;
+  ocirc->base_.n_chan = &(n_chan->base_);
+
+  /* Origin circuit was completely idle. */
+  cell_stats = tor_malloc_zero(sizeof(cell_stats_t));
+  format_cell_stats(&event_string, TO_CIRCUIT(ocirc), cell_stats);
+  tt_str_op("ID=2 OutboundQueue=3 OutboundConn=1", ==, event_string);
+
+  /* Origin circuit had 4 RELAY cells added to its exitward queue. */
+  cell_stats->added_cells_exitward[CELL_RELAY] = 4;
+  format_cell_stats(&event_string, TO_CIRCUIT(ocirc), cell_stats);
+  tt_str_op("ID=2 OutboundQueue=3 OutboundConn=1 OutboundAdded=relay:4",
+            ==, event_string);
+
+  /* Origin circuit also had 5 CREATE2 cells added to its exitward
+   * queue. */
+  cell_stats->added_cells_exitward[CELL_CREATE2] = 5;
+  format_cell_stats(&event_string, TO_CIRCUIT(ocirc), cell_stats);
+  tt_str_op("ID=2 OutboundQueue=3 OutboundConn=1 OutboundAdded=relay:4,"
+            "create2:5", ==, event_string);
+
+  /* Origin circuit also had 7 RELAY cells removed from its exitward queue
+   * which together spent 6 msec in the queue. */
+  cell_stats->total_time_exitward[CELL_RELAY] = 6;
+  cell_stats->removed_cells_exitward[CELL_RELAY] = 7;
+  format_cell_stats(&event_string, TO_CIRCUIT(ocirc), cell_stats);
+  tt_str_op("ID=2 OutboundQueue=3 OutboundConn=1 OutboundAdded=relay:4,"
+            "create2:5 OutboundRemoved=relay:7 OutboundTime=relay:6",
+            ==, event_string);
+
+  p_chan = tor_malloc_zero(sizeof(channel_tls_t));
+  p_chan->base_.global_identifier = 2;
+
+  or_circ = tor_malloc_zero(sizeof(or_circuit_t));
+  or_circ->base_.magic = OR_CIRCUIT_MAGIC;
+  or_circ->base_.purpose = CIRCUIT_PURPOSE_OR;
+  or_circ->p_circ_id = 8;
+  or_circ->p_chan = &(p_chan->base_);
+  or_circ->base_.n_circ_id = 9;
+  or_circ->base_.n_chan = &(n_chan->base_);
+
+  /* OR circuit was idle. */
+  cell_stats = tor_malloc_zero(sizeof(cell_stats_t));
+  format_cell_stats(&event_string, TO_CIRCUIT(or_circ), cell_stats);
+  tt_str_op("InboundQueue=8 InboundConn=2 OutboundQueue=9 OutboundConn=1",
+            ==, event_string);
+
+  /* OR circuit had 3 RELAY cells added to its appward queue. */
+  cell_stats->added_cells_appward[CELL_RELAY] = 3;
+  format_cell_stats(&event_string, TO_CIRCUIT(or_circ), cell_stats);
+  tt_str_op("InboundQueue=8 InboundConn=2 InboundAdded=relay:3 "
+            "OutboundQueue=9 OutboundConn=1", ==, event_string);
+
+  /* OR circuit had 7 RELAY cells removed from its appward queue which
+   * together spent 6 msec in the queue. */
+  cell_stats->total_time_appward[CELL_RELAY] = 6;
+  cell_stats->removed_cells_appward[CELL_RELAY] = 7;
+  format_cell_stats(&event_string, TO_CIRCUIT(or_circ), cell_stats);
+  tt_str_op("InboundQueue=8 InboundConn=2 InboundAdded=relay:3 "
+            "InboundRemoved=relay:7 InboundTime=relay:6 "
+            "OutboundQueue=9 OutboundConn=1", ==, event_string);
+
+ done:
+  ;
+}
+
+#define TEST(name, flags)                                               \
+  { #name, test_cntev_ ## name, flags, 0, NULL }
+
+struct testcase_t controller_event_tests[] = {
+  TEST(bucket_note_empty, 0),
+  TEST(bucket_millis_empty, 0),
+  TEST(sum_up_cell_stats, 0),
+  TEST(append_cell_stats, 0),
+  TEST(format_cell_stats, 0),
+  END_OF_TESTCASES
+};
+