relay-throughput-testing/src/throughput-logging.patch

diff --git a/src/app/config/config.c b/src/app/config/config.c
index deda2448b..059ab0d76 100644
--- a/src/app/config/config.c
+++ b/src/app/config/config.c
@@ -759,6 +759,7 @@ static const config_var_t option_vars_[] = {
   V(TestingDirAuthVoteHSDirIsStrict,  BOOL,     "0"),
   VAR_INVIS("___UsingTestNetworkDefaults", BOOL, UsingTestNetworkDefaults_,
             "0"),
+  V(ThroughputLogFile, FILENAME, NULL),
 
   END_OF_CONFIG_VARS
 };
@@ -3247,6 +3248,8 @@ warn_about_relative_paths(or_options_t *options)
   n += warn_if_option_path_is_relative("PidFile",options->PidFile);
   n += warn_if_option_path_is_relative("ClientOnionAuthDir",
                                         options->ClientOnionAuthDir);
+  n += warn_if_option_path_is_relative("ThroughputLogFile",
+                                        options->ThroughputLogFile);
 
   for (config_line_t *hs_line = options->RendConfigLines; hs_line;
        hs_line = hs_line->next) {
diff --git a/src/app/config/or_options_st.h b/src/app/config/or_options_st.h
index 32dcd9fb1..5ae69f68a 100644
--- a/src/app/config/or_options_st.h
+++ b/src/app/config/or_options_st.h
@@ -1109,6 +1109,8 @@ struct or_options_t {
    **/
   int DormantCanceledByStartup;
 
+  char *ThroughputLogFile;
+
   /**
    * Configuration objects for individual modules.
    *
diff --git a/src/core/include.am b/src/core/include.am
index 9b4b251c8..9c2796173 100644
--- a/src/core/include.am
+++ b/src/core/include.am
@@ -27,6 +27,7 @@ LIBTOR_APP_A_SOURCES = 				\
 	src/core/mainloop/mainloop_sys.c	\
 	src/core/mainloop/netstatus.c		\
 	src/core/mainloop/periodic.c		\
+	src/core/mainloop/throughput_logging.c	\
 	src/core/or/address_set.c		\
 	src/core/or/channel.c			\
 	src/core/or/channelpadding.c		\
@@ -233,6 +234,7 @@ noinst_HEADERS +=					\
 	src/core/mainloop/mainloop_sys.h		\
 	src/core/mainloop/netstatus.h			\
 	src/core/mainloop/periodic.h			\
+	src/core/mainloop/throughput_logging.h		\
 	src/core/or/addr_policy_st.h			\
 	src/core/or/address_set.h			\
 	src/core/or/cell_queue_st.h			\
diff --git a/src/core/mainloop/connection.c b/src/core/mainloop/connection.c
index 6094f33e4..39d5ac951 100644
--- a/src/core/mainloop/connection.c
+++ b/src/core/mainloop/connection.c
@@ -71,6 +71,7 @@
 #include "core/mainloop/connection.h"
 #include "core/mainloop/mainloop.h"
 #include "core/mainloop/netstatus.h"
+#include "core/mainloop/throughput_logging.h"
 #include "core/or/channel.h"
 #include "core/or/channeltls.h"
 #include "core/or/circuitbuild.h"
@@ -175,7 +176,8 @@ static int connection_finished_connecting(connection_t *conn);
 static int connection_reached_eof(connection_t *conn);
 static int connection_buf_read_from_socket(connection_t *conn,
                                            ssize_t *max_to_read,
-                                           int *socket_error);
+                                           int *socket_error,
+                                           monotime_coarse_t *now);
 static int connection_process_inbuf(connection_t *conn, int package_partial);
 static void client_check_address_changed(tor_socket_t sock);
 static void set_constrained_socket_buffers(tor_socket_t sock, int size);
@@ -3580,7 +3582,9 @@ connection_handle_read_impl(connection_t *conn)
 
   conn->timestamp_last_read_allowed = approx_time();
 
-  connection_bucket_refill_single(conn, monotime_coarse_get_stamp());
+  monotime_coarse_t now;
+  monotime_coarse_get(&now);
+  connection_bucket_refill_single(conn, monotime_coarse_to_stamp(&now));
 
   switch (conn->type) {
     case CONN_TYPE_OR_LISTENER:
@@ -3607,7 +3611,7 @@ connection_handle_read_impl(connection_t *conn)
   tor_assert(!conn->marked_for_close);
 
   before = buf_datalen(conn->inbuf);
-  if (connection_buf_read_from_socket(conn, &max_to_read, &socket_error) < 0) {
+  if (connection_buf_read_from_socket(conn, &max_to_read, &socket_error, &now) < 0) {
     /* There's a read error; kill the connection.*/
     if (conn->type == CONN_TYPE_OR) {
       connection_or_notify_error(TO_OR_CONN(conn),
@@ -3704,7 +3708,7 @@ connection_handle_read(connection_t *conn)
  */
 static int
 connection_buf_read_from_socket(connection_t *conn, ssize_t *max_to_read,
-                       int *socket_error)
+                       int *socket_error, monotime_coarse_t *now)
 {
   int result;
   ssize_t at_most = *max_to_read;
@@ -3802,6 +3806,7 @@ connection_buf_read_from_socket(connection_t *conn, ssize_t *max_to_read,
     tor_tls_get_n_raw_bytes(or_conn->tls, &n_read, &n_written);
     log_debug(LD_GENERAL, "After TLS read of %d: %ld read, %ld written",
               result, (long)n_read, (long)n_written);
+    log_recv_bytes(result, now);
   } else if (conn->linked) {
     if (conn->linked_conn) {
       result = buf_move_to_buf(conn->inbuf, conn->linked_conn->outbuf,
@@ -4009,7 +4014,9 @@ connection_handle_write_impl(connection_t *conn, int force)
 
   conn->timestamp_last_write_allowed = now;
 
-  connection_bucket_refill_single(conn, monotime_coarse_get_stamp());
+  monotime_coarse_t now_coarse;
+  monotime_coarse_get(&now_coarse);
+  connection_bucket_refill_single(conn, monotime_coarse_to_stamp(&now_coarse));
 
   /* Sometimes, "writable" means "connected". */
   if (connection_state_is_connecting(conn)) {
@@ -4148,6 +4155,7 @@ connection_handle_write_impl(connection_t *conn, int force)
      * the *_buf_tls functions, we should make them return ssize_t or size_t
      * or something. */
     result = (int)(initial_size-buf_datalen(conn->outbuf));
+    log_sent_bytes(result, &now_coarse);
   } else {
     CONN_LOG_PROTECT(conn,
                      result = buf_flush_to_socket(conn->outbuf, conn->s,
diff --git a/src/core/mainloop/mainloop.c b/src/core/mainloop/mainloop.c
index c47e44077..48b415db5 100644
--- a/src/core/mainloop/mainloop.c
+++ b/src/core/mainloop/mainloop.c
@@ -57,6 +57,7 @@
 #include "core/mainloop/mainloop.h"
 #include "core/mainloop/netstatus.h"
 #include "core/mainloop/periodic.h"
+#include "core/mainloop/throughput_logging.h"
 #include "core/or/channel.h"
 #include "core/or/channelpadding.h"
 #include "core/or/channeltls.h"
@@ -2339,6 +2340,14 @@ do_main_loop(void)
 
   periodic_events_connect_all();
 
+  bool logging_throughput = (get_options()->ThroughputLogFile != NULL &&
+                             strlen(get_options()->ThroughputLogFile) != 0);
+
+  if (logging_throughput) {
+    init_throughput_logging(1);
+    init_thread_throughput_logging(0);
+  }
+
   struct timeval one_second = { 1, 0 };
   initialize_periodic_events_event = tor_evtimer_new(
                   tor_libevent_get_base(),
@@ -2391,7 +2400,15 @@ do_main_loop(void)
   }
 #endif /* defined(ENABLE_RESTART_DEBUGGING) */
 
-  return run_main_loop_until_done();
+  int rv = run_main_loop_until_done();
+
+  if (logging_throughput) {
+    destroy_thread_throughput_logging();
+    write_throughput_log(get_options()->ThroughputLogFile);
+    destroy_throughput_logging();
+  }
+
+  return rv;
 }
 
 #ifndef _WIN32
diff --git a/src/core/mainloop/throughput_logging.c b/src/core/mainloop/throughput_logging.c
new file mode 100644
index 000000000..ee99de1e5
--- /dev/null
+++ b/src/core/mainloop/throughput_logging.c
@@ -0,0 +1,271 @@
+#include <stdbool.h>
+#include <stddef.h>
+#include <stdint.h>
+#include <stdio.h>
+#include <string.h>
+#include <sys/time.h>
+
+#include "core/mainloop/throughput_logging.h"
+
+#include "lib/lock/compat_mutex.h"
+#include "lib/smartlist_core/smartlist_core.h"
+#include "lib/thread/threads.h"
+#include "lib/malloc/malloc.h"
+#include "lib/log/util_bug.h"
+
+const unsigned int timestep_ms = 500;
+
+bool throughput_logging_enabled = false;
+monotime_coarse_t throughput_logging_coarse_start_time;
+// NOTE: we don't lock these variables, so make sure they are set
+// before any threads have started, and that they don't change
+// while threads are running
+
+double throughput_logging_wall_start_time;
+tor_mutex_t throughput_logging_lock;
+
+smartlist_t **sent_lists = NULL;
+smartlist_t **recv_lists = NULL;
+tor_mutex_t *bytes_list_mutexes = NULL;
+int relay_bytes_lists_len = -1;
+
+tor_threadlocal_t thread_sent_list;
+tor_threadlocal_t thread_recv_list;
+tor_threadlocal_t thread_logging_mutex;
+
+// only call if no threads are running
+void
+init_throughput_logging(int num_threads)
+{
+  tor_assert(!throughput_logging_enabled);
+
+  tor_mutex_init(&throughput_logging_lock);
+  tor_mutex_acquire(&throughput_logging_lock);
+
+  relay_bytes_lists_len = num_threads;
+
+  monotime_coarse_get(&throughput_logging_coarse_start_time);
+  struct timeval ts;
+  gettimeofday(&ts, NULL);
+  throughput_logging_wall_start_time = ts.tv_sec+(ts.tv_usec/1000000.0);
+
+  sent_lists = tor_malloc_zero(num_threads*sizeof(smartlist_t *));
+  recv_lists = tor_malloc_zero(num_threads*sizeof(smartlist_t *));
+  bytes_list_mutexes = tor_malloc_zero(num_threads*sizeof(tor_mutex_t));
+  for (int i=0; i<num_threads; i++) {
+    tor_mutex_init(&bytes_list_mutexes[i]);
+    tor_mutex_acquire(&bytes_list_mutexes[i]);
+    sent_lists[i] = smartlist_new();
+    recv_lists[i] = smartlist_new();
+    tor_mutex_release(&bytes_list_mutexes[i]);
+  }
+
+  tor_threadlocal_init(&thread_sent_list);
+  tor_threadlocal_init(&thread_recv_list);
+  tor_threadlocal_init(&thread_logging_mutex);
+
+  throughput_logging_enabled = true;
+  tor_mutex_release(&throughput_logging_lock);
+}
+
+// only call if no threads are running
+void
+destroy_throughput_logging(void)
+{
+  tor_assert(throughput_logging_enabled);
+
+  tor_mutex_acquire(&throughput_logging_lock);
+
+  for (int i=0; i<relay_bytes_lists_len; i++) {
+    tor_mutex_acquire(&bytes_list_mutexes[i]);
+
+    smartlist_free(sent_lists[i]);
+    smartlist_free(recv_lists[i]);
+    sent_lists[i] = NULL;
+    recv_lists[i] = NULL;
+
+    tor_mutex_release(&bytes_list_mutexes[i]);
+    tor_mutex_uninit(&bytes_list_mutexes[i]);
+  }
+
+  tor_free(bytes_list_mutexes);
+  tor_free(sent_lists);
+  tor_free(recv_lists);
+  relay_bytes_lists_len = -1;
+
+  tor_threadlocal_destroy(&thread_sent_list);
+  tor_threadlocal_destroy(&thread_recv_list);
+  tor_threadlocal_destroy(&thread_logging_mutex);
+
+  throughput_logging_enabled = false;
+
+  tor_mutex_release(&throughput_logging_lock);
+  tor_mutex_uninit(&throughput_logging_lock);
+}
+
+void
+init_thread_throughput_logging(int thread_id)
+{
+  tor_assert(throughput_logging_enabled);
+
+  tor_mutex_acquire(&throughput_logging_lock);
+
+  tor_assert(thread_id < relay_bytes_lists_len && thread_id >= 0);
+  tor_threadlocal_set(&thread_logging_mutex, &bytes_list_mutexes[thread_id]);
+  tor_mutex_acquire(&bytes_list_mutexes[thread_id]);
+  // we acquire this mutex for the lifetime of the thread, hope nobody
+  // tries to acquire it :)
+
+  tor_threadlocal_set(&thread_sent_list, sent_lists[thread_id]);
+  tor_threadlocal_set(&thread_recv_list, recv_lists[thread_id]);
+
+  tor_mutex_release(&throughput_logging_lock);
+}
+
+void
+destroy_thread_throughput_logging(void)
+{
+  tor_assert(throughput_logging_enabled);
+
+  tor_threadlocal_set(&thread_sent_list, NULL);
+  tor_threadlocal_set(&thread_recv_list, NULL);
+
+  tor_mutex_t *mutex = tor_threadlocal_get(&thread_logging_mutex);
+  if (mutex != NULL) {
+    tor_mutex_release(mutex);
+    tor_threadlocal_set(&thread_logging_mutex, NULL);
+  }
+}
+
+static void
+log_throughput(smartlist_t *list, uint32_t bytes, monotime_coarse_t *time)
+{
+  tor_assert(throughput_logging_enabled);
+
+  int64_t ms_since_start = monotime_coarse_diff_msec(&throughput_logging_coarse_start_time, time);
+  int list_index = ms_since_start/timestep_ms;
+
+  if (list_index >= smartlist_len(list)) {
+    // need to grow the list
+    int additional_elements = (60000-1)/timestep_ms + 1;
+    // want an extra 60 seconds, and we want the ceil
+    int new_size = (list_index+1)+additional_elements;
+    // want enough room to store the current value, plus an extra 60 seconds
+    smartlist_grow(list, new_size);
+  }
+
+  uint32_t old_bytes = (intptr_t)smartlist_get(list, list_index);
+  uint32_t new_bytes = old_bytes+bytes;
+  if (new_bytes < old_bytes) {
+    new_bytes = UINT32_MAX;
+  }
+  smartlist_set(list, list_index, (void *)(intptr_t)new_bytes);
+}
+
+void
+log_sent_bytes(uint32_t bytes, monotime_coarse_t *now)
+{
+  if (bytes > 0 && throughput_logging_enabled) {
+    smartlist_t *sent_list = tor_threadlocal_get(&thread_sent_list);
+    tor_assert(sent_list != NULL);
+    log_throughput(sent_list, bytes, now);
+  }
+}
+
+void
+log_recv_bytes(uint32_t bytes, monotime_coarse_t *now)
+{
+  if (bytes > 0 && throughput_logging_enabled) {
+    smartlist_t *recv_list = tor_threadlocal_get(&thread_recv_list);
+    tor_assert(recv_list != NULL);
+    log_throughput(recv_list, bytes, now);
+  }
+}
+
+// only run this function when the threads have finished
+void
+write_throughput_log(char *file_name)
+{
+  if (!throughput_logging_enabled) {
+    log_warn(LD_CONFIG, "Throughput logging was not set up, so didn't write to log file");
+    return;
+  }
+
+  tor_mutex_acquire(&throughput_logging_lock);
+
+  if (file_name == NULL || strlen(file_name) == 0) {
+    log_warn(LD_CONFIG, "Was not given a file name for the throughput log");
+    tor_mutex_release(&throughput_logging_lock);
+    return;
+  }
+
+  FILE *log_file = fopen(file_name, "w");
+
+  if (log_file == NULL) {
+    log_warn(LD_CONFIG, "Could not open throughput log file %s", file_name);
+    tor_mutex_release(&throughput_logging_lock);
+    return;
+  }
+
+  for (int i=0; i<relay_bytes_lists_len; i++) {
+    tor_mutex_acquire(&bytes_list_mutexes[i]);
+    // this will block if any threads have not finished
+  }
+
+  struct timeval ts;
+  gettimeofday(&ts, NULL);
+  double current_time = ts.tv_sec+(ts.tv_usec/1000000.0);
+
+  // write header
+  fprintf(log_file, "time          ");
+  for (int i=0; i<relay_bytes_lists_len; i++) {
+    for (int j=0; j<2; j++) {
+      fprintf(log_file, ", thrd %d %s", 0, (j==0)?"sent":"recv");
+    }
+  }
+  fprintf(log_file, "\n");
+
+  // write data
+  bool thread_had_data = true;
+  int time_index = 0;
+  while (thread_had_data) {
+    // write line
+    thread_had_data = false;
+    double logging_time = throughput_logging_wall_start_time+(time_index*timestep_ms/1000.0);
+    fprintf(log_file, "%.3f", logging_time);
+
+    for (int i=0; i<relay_bytes_lists_len; i++) {
+      // write column
+      smartlist_t *sent_list = sent_lists[i];
+      smartlist_t *recv_list = recv_lists[i];
+      uint32_t bytes_sent = 0;
+      uint32_t bytes_recv = 0;
+
+      if (time_index < smartlist_len(sent_list)) {
+        bytes_sent = (intptr_t)smartlist_get(sent_list, time_index);
+        if (logging_time <= current_time || bytes_sent != 0) {
+          thread_had_data = true;
+        }
+      }
+      if (time_index < smartlist_len(recv_list)) {
+        bytes_recv = (intptr_t)smartlist_get(recv_list, time_index);
+        if (logging_time <= current_time || bytes_recv != 0) {
+          thread_had_data = true;
+        }
+      }
+
+      fprintf(log_file, ", %11"PRIu32", %11"PRIu32, bytes_sent, bytes_recv);
+    }
+
+    time_index += 1;
+    fprintf(log_file, "\n");
+  }
+
+  for (int i=0; i<relay_bytes_lists_len; i++) {
+    tor_mutex_release(&bytes_list_mutexes[i]);
+  }
+
+  fclose(log_file);
+
+  tor_mutex_release(&throughput_logging_lock);
+}
diff --git a/src/core/mainloop/throughput_logging.h b/src/core/mainloop/throughput_logging.h
new file mode 100644
index 000000000..3c39cd435
--- /dev/null
+++ b/src/core/mainloop/throughput_logging.h
@@ -0,0 +1,24 @@
+#ifndef MAINLOOP_THROUGHPUT_LOG_H
+#define MAINLOOP_THROUGHPUT_LOG_H
+
+#include "lib/time/compat_time.h"
+
+// the main thread should run the following before any threads have been
+// created
+void init_throughput_logging(int num_threads);
+// the main thread should run the following after all threads have completed
+void destroy_throughput_logging(void);
+
+// each thread should run the following
+void init_thread_throughput_logging(int thread_id);
+void destroy_thread_throughput_logging(void);
+
+// each thread should log the sent and received bytes with the following
+void log_sent_bytes(uint32_t bytes, monotime_coarse_t *now);
+void log_recv_bytes(uint32_t bytes, monotime_coarse_t *now);
+
+// the file should be written to after all threads have finished but before
+// calling 'destroy_throughput_logging()'
+void write_throughput_log(char *file_name);
+
+#endif