r11812@catbus: nickm | 2007-02-14 11:22:08 -0500

Apply stream_bw patch from Robert Hogan.


svn:r9585

ChangeLog

@@ -40,11 +40,15 @@ Changes in version 0.1.2.8-alpha - 2007-??-??
       order: the EntryNodes that were guards before; the rest of the
       EntryNodes; the nodes that were guards before.
 
-  o Minor features:
+  o Minor features (controller):
     - Warn the user when an application uses the obsolete binary v0
       control protocol.  We're planning to remove support for it during
       the next development series, so it's good to give people some
       advance warning.
+    - Add STREAM_BW events to reprot per-entry-stream bandwidth use. (Patch
+      from Robert Hogan.)
+
+  o Minor features:
     - Remove some never-implemented options.  Mark PathlenCoinWeight as
       obsolete.
     - Implement proposal 106: Stop requiring clients to have well-formed

doc/spec/control-spec.txt

@@ -194,7 +194,7 @@ $Id$
      EventCode = "CIRC" / "STREAM" / "ORCONN" / "BW" / "DEBUG" /
          "INFO" / "NOTICE" / "WARN" / "ERR" / "NEWDESC" / "ADDRMAP" /
          "AUTHDIR_NEWDESCS" / "DESCCHANGED" / "STATUS_GENERAL" /
-         "STATUS_CLIENT" / "STATUS_SERVER" / "GUARDS" / "NS"
+         "STATUS_CLIENT" / "STATUS_SERVER" / "GUARDS" / "NS" / "STREAM_BW"
 
   Any events *not* listed in the SETEVENTS line are turned off; thus, sending
   SETEVENTS with an empty body turns off all event reporting.
@@ -1271,6 +1271,17 @@ $Id$
 
   [First added in 0.1.2.3-alpha]
 
+4.1.13. Bandwidth used on a stream
+
+  The syntax is:
+     "650" SP "STREAM_BW" SP StreamID SP BytesRead SP BytesWritten
+     BytesRead = 1*DIGIT
+     BytesWritten = 1*DIGIT
+
+  The number of bytes read and written since the last read or write event
+  on a stream.
+
+
 5. Implementation notes
 
 5.1. Authentication

src/or/connection.c

@@ -1571,6 +1571,13 @@ connection_read_to_buf(connection_t *conn, int *max_to_read)
     *max_to_read = at_most - n_read;
   }
 
+  if (CONN_IS_EDGE(conn)) {
+    if (conn->type == CONN_TYPE_AP) {
+        edge_connection_t *edge_conn = TO_EDGE_CONN(conn);
+        edge_conn->n_read += n_read;
+    }
+  }
+
   if (connection_is_rate_limited(conn)) {
     /* For non-local IPs, remember if we flushed any bytes over the wire. */
     time_t now = time(NULL);
@@ -1767,6 +1774,13 @@ connection_handle_write(connection_t *conn, int force)
     n_written = (size_t) result;
   }
 
+  if (CONN_IS_EDGE(conn)) {
+    if (conn->type == CONN_TYPE_AP) {
+      edge_connection_t *edge_conn = TO_EDGE_CONN(conn);
+      edge_conn->n_written += n_written;
+    }
+  }
+
   if (connection_is_rate_limited(conn)) {
     /* For non-local IPs, remember if we flushed any bytes over the wire. */
     time_t now = time(NULL);

src/or/control.c

@@ -89,7 +89,8 @@ const char control_c_id[] =
 #define EVENT_STATUS_SERVER    0x0011
 #define EVENT_STATUS_GENERAL   0x0012
 #define EVENT_GUARD            0x0013
-#define _EVENT_MAX             0x0013
+#define EVENT_STREAM_BANDWIDTH_USED   0x0014
+#define _EVENT_MAX             0x0014
 /* If _EVENT_MAX ever hits 0x0020, we need to make the mask wider. */
 
 /** Array mapping from message type codes to human-readable message
@@ -1104,6 +1105,8 @@ handle_control_setevents(control_connection_t *conn, uint32_t len,
           event_code = EVENT_STATUS_SERVER;
         else if (!strcasecmp(ev, "GUARD"))
           event_code = EVENT_GUARD;
+        else if (!strcasecmp(ev, "STREAM_BW"))
+          event_code = EVENT_STREAM_BANDWIDTH_USED;
         else {
           connection_printf_to_buf(conn, "552 Unrecognized event \"%s\"\r\n",
                                    ev);
@@ -3410,6 +3413,44 @@ control_event_or_conn_status(or_connection_t *conn,or_conn_status_event_t tp,
   return 0;
 }
 
+/** A second or more has elapsed: tell any interested control
+ * connections how much bandwidth streams have used. */
+int
+control_event_stream_bandwidth_used()
+{
+  connection_t **carray;
+  edge_connection_t *conn;
+  int n, i;
+  uint32_t justread, justwritten;
+
+  if (EVENT_IS_INTERESTING1(EVENT_STREAM_BANDWIDTH_USED)) {
+
+    get_connection_array(&carray, &n);
+
+    for (i = 0; i < n; ++i) {
+        if (carray[i]->type != CONN_TYPE_AP)
+          continue;
+        conn = TO_EDGE_CONN(carray[i]);
+        if (conn->p_read == conn->n_read && conn->p_written == conn->n_written)
+          continue;
+
+        justread = conn->n_read - conn->p_read;
+        conn->p_read = conn->n_read;
+        justwritten = conn->n_written - conn->p_written;
+        conn->p_written = conn->n_written;
+
+        send_control1_event(EVENT_STREAM_BANDWIDTH_USED, ALL_NAMES,
+                            "650 STREAM_BW %lu %lu %lu\r\n",
+                            (unsigned long)conn->global_identifier,
+                            (unsigned long)justread,
+                            (unsigned long)justwritten);
+
+    }
+  }
+
+  return 0;
+}
+
 /** A second or more has elapsed: tell any interested control
  * connections how much bandwidth we used. */
 int

src/or/main.c

@@ -998,6 +998,7 @@ second_elapsed_callback(int fd, short event, void *args)
   if (accounting_is_enabled(options) && seconds_elapsed >= 0)
     accounting_add_bytes(bytes_read, bytes_written, seconds_elapsed);
   control_event_bandwidth_used((uint32_t)bytes_read,(uint32_t)bytes_written);
+  control_event_stream_bandwidth_used();
 
   if (seconds_elapsed > 0)
     connection_bucket_refill(seconds_elapsed);

src/or/or.h

@@ -838,6 +838,14 @@ typedef struct edge_connection_t {
   /* XXXX NM This can get re-used after 2**32 streams */
   uint32_t global_identifier;
 
+  /** Bytes read */
+  uint32_t n_read;
+  uint32_t p_read;
+
+  /** Bytes written */
+  uint32_t n_written;
+  uint32_t p_written;
+
   /** Exit only: a dirserv connection that is tunneled over this connection
    * using a socketpair. */
   struct dir_connection_t *bridge_for_conn;
@@ -2329,6 +2337,7 @@ int control_tls_error_to_reason(int e);
 int control_event_or_conn_status(or_connection_t *conn,
                                  or_conn_status_event_t e, int reason);
 int control_event_bandwidth_used(uint32_t n_read, uint32_t n_written);
+int control_event_stream_bandwidth_used(void);
 void control_event_logmsg(int severity, unsigned int domain, const char *msg);
 int control_event_descriptors_changed(smartlist_t *routers);
 int control_event_address_mapped(const char *from, const char *to,