Merge pull request #91 from unnatisa/logrtt-feature

Add --log-rtt and --logrtt-entries-per-flow option to rr/crr tests.

Author: hi-rajat-kumar

check_all_options.c

@@ -86,6 +86,8 @@ void check_options_rr(struct options *opts, struct callbacks *cb)
               "Request size must be positive.");
         CHECK(cb, opts->response_size > 0,
               "Response size must be positive.");
+        CHECK(cb, (opts->log_rtt == NULL || opts->logrtt_entries_per_flow > 0),
+              "Log RTT entries per flow must be positive.");
 }
 
 void check_options_stream(struct options *opts, struct callbacks *cb)

define_all_flags.c

@@ -95,7 +95,11 @@ struct flags_parser *add_flags_rr(struct flags_parser *fp)
         DEFINE_FLAG_PRINTER(fp,              percentiles, percentiles_print);
         DEFINE_FLAG(fp, int,                 test_length,   10,                      'l', "Test length, >0 seconds, <0 transactions");
         DEFINE_FLAG(fp, int,                 buffer_size,   65536,                   'B', "Number of bytes that each read()/send() can transfer at once");
-        DEFINE_FLAG(fp, int,                 mark,          0,                       0, "SO_MARK value; use as 32-bit unsigned");
+        DEFINE_FLAG(fp, const char *,        log_rtt,       NULL,                     0,  "Logs per transaction rtt to txt/csv file");
+        DEFINE_FLAG_HAS_OPTIONAL_ARGUMENT(fp, log_rtt);
+        DEFINE_FLAG_PARSER(fp,               log_rtt, parse_log_rtt);
+        DEFINE_FLAG(fp, int,                 logrtt_entries_per_flow, 10000,          0,  "Number of RTT entries to log per flow. Logging stops when this limit is reached.");
+        DEFINE_FLAG(fp, int,                 mark,          0,                        0,  "SO_MARK value; use as 32-bit unsigned");
 
         /* Return the updated fp */
         return (fp);

flow.c

@@ -51,6 +51,8 @@ struct flow {
 	/* TCP RX zerocopy state. */
 	void *f_rx_zerocopy_buffer;
 	size_t f_rx_zerocopy_buffer_sz;
+
+        long rtt_log_count;
 };
 
 int flow_fd(const struct flow *f)
@@ -147,6 +149,16 @@ void flow_init_rx_zerocopy(struct flow *f, int buffer_size, struct callbacks *cb
         f->f_rx_zerocopy_buffer_sz = buffer_size;
 }
 
+long flow_rtt_log_count(const struct flow *f)
+{
+        return f->rtt_log_count;
+}
+
+void flow_increment_rtt_log_count(struct flow *f)
+{
+        f->rtt_log_count++;
+}
+
 struct flow *flow_create(const struct flow_create_args *args)
 {
         struct thread *t = args->thread;

flow.h

@@ -27,6 +27,13 @@ struct flow;  /* note: struct is defined opaquely within flow.c */
 struct neper_stat;
 struct thread;
 
+struct rtt_log {
+        double rtt;
+        int thread_id;
+        int flow_id;
+        struct timespec timestamp;
+};
+
 typedef void (*flow_handler)(struct flow *, uint32_t);
 
 /* Simple accessors. */
@@ -57,6 +64,8 @@ struct flow_create_args {
 };
 
 struct flow *flow_create(const struct flow_create_args *);
+long flow_rtt_log_count(const struct flow *f);
+void flow_increment_rtt_log_count(struct flow *f);
 void flow_delete(struct flow *);
 /* Adds duration to f's next event time. */
 void flow_update_next_event(struct flow *f, uint64_t duration);

lib.h

@@ -131,6 +131,10 @@ struct options {
 
         /* tcp_crr */
         bool nostats;
+
+        /* rr */
+        const char *log_rtt;
+        int logrtt_entries_per_flow;
 };
 
 #ifdef __cplusplus

parse.c

@@ -28,6 +28,14 @@ void parse_all_samples(const char *arg, void *out, struct callbacks *cb)
                 *(const char **)out = "samples.csv";
 }
 
+void parse_log_rtt(const char *arg, void *out, struct callbacks *cb)
+{
+        if (arg)
+                *(const char **)out = arg;
+        else
+                *(const char **)out = "rtt_samples.txt";
+}
+
 static long long parse_rate(const char *str, struct callbacks *cb)
 {
         const struct rate_conversion *conv;

parse.h

@@ -18,6 +18,7 @@
 #define THIRD_PARTY_NEPER_PARSE_H
 
 void parse_all_samples(const char *arg, void *out, struct callbacks *cb);
+void parse_log_rtt(const char *arg, void *out, struct callbacks *cb);
 void parse_max_pacing_rate(const char *arg, void *out, struct callbacks *cb);
 void parse_unit(const char *arg, void *out, struct callbacks *);
 void parse_duration(const char *arg, void *out, struct callbacks *cb);

rr.c

@@ -24,11 +24,14 @@
  * as sending a message, opening a socket connection, or gathering statistics.
  */
 
+#include <stdio.h>
+#include <string.h>
 #include "coef.h"
 #include "common.h"
 #include "countdown_cond.h"
 #include "flow.h"
 #include "histo.h"
+#include "logging.h"
 #include "percentiles.h"
 #include "print.h"
 #include "rr.h"
@@ -202,6 +205,15 @@ static void rr_state_init(struct thread *t, int fd,
         };
 
         flow_create(&args);
+
+        if (t->opts->client && t->opts->log_rtt && t->rtt_logs == NULL) {
+                t->rtt_log_capacity = (long)t->flow_limit *
+                                      t->opts->logrtt_entries_per_flow;
+                if (t->rtt_log_capacity > 0) {
+                        t->rtt_logs = calloc_or_die(
+                            t->rtt_log_capacity, sizeof(struct rtt_log), t->cb);
+                }
+        }
 }
 
 /*
@@ -327,6 +339,19 @@ static bool rr_do_compl(struct flow *f,
         struct neper_histo *histo = stat->histo(stat);
         neper_histo_event(histo, elapsed);
 
+        if (t->opts->client && t->rtt_logs) {
+                long count = flow_rtt_log_count(f);
+                if (count < t->opts->logrtt_entries_per_flow) {
+                        long offset = flow_id(f) * t->opts->logrtt_entries_per_flow;
+                        struct rtt_log *log = &t->rtt_logs[offset + count];
+                        log->rtt = elapsed;
+                        log->thread_id = t->index;
+                        log->flow_id = flow_id(f);
+                        log->timestamp = *now;
+                        flow_increment_rtt_log_count(f);
+                }
+        }
+
         if (t->data_pending) {
                 /* data vs time mode, last rr? */
                 if (!countdown_cond_commit(t->data_pending)) {
@@ -497,6 +522,54 @@ fn_add(struct neper_stat *stat, void *ptr)
         return 0;
 }
 
+static void rr_log_rtt(struct thread *tinfo, struct callbacks *cb,
+                       int num_transactions)
+{
+        const char *sep = " ";
+        const char *ext = strrchr(tinfo[0].opts->log_rtt, '.');
+        if (ext && !strcmp(ext, ".csv"))
+                sep = ",";
+
+        FILE *rtt_log_file = fopen(tinfo[0].opts->log_rtt, "w");
+        if (!rtt_log_file)
+                PLOG_FATAL(cb, "fopen %s", tinfo[0].opts->log_rtt);
+
+        fprintf(rtt_log_file, "%15s%s%10s%s%10s%s%12s\n", "timestamp", sep, "thread_id", sep, "flow_id", sep, "rtt");
+
+        long transactions_logged = 0;
+        const struct timespec *start_time = tinfo[0].time_start;
+
+        for (int i = 0; i < tinfo[0].opts->num_threads; i++) {
+                struct thread *t = &tinfo[i];
+                if (!t->rtt_logs)
+                        continue;
+                for (int j = 0; j < t->flow_count; j++) {
+                        struct flow *f = t->flows[j];
+                        if (!f)
+                                continue;
+                        transactions_logged += flow_rtt_log_count(f);
+                        long offset = flow_id(f) * t->opts->logrtt_entries_per_flow;
+                        for (long k = 0; k < flow_rtt_log_count(f); k++) {
+                                struct rtt_log *log = &t->rtt_logs[offset + k];
+                                double elapsed = seconds_between(start_time, &log->timestamp);
+                                fprintf(rtt_log_file, "%15.9f%s%10d%s%10d%s%12.9f\n",
+                                        elapsed, sep, log->thread_id, sep,
+                                        log->flow_id, sep, log->rtt);
+                        }
+                }
+                free(t->rtt_logs);
+                t->rtt_logs = NULL;
+        }
+        fclose(rtt_log_file);
+        PRINT(cb, "rtt_transactions_logged", "%ld", transactions_logged);
+
+        if (transactions_logged < num_transactions) {
+                LOG_INFO(cb,
+                         "rtt_transactions_logged (%ld) < num_transactions (%d)",
+                         transactions_logged, num_transactions);
+        }
+}
+
 int rr_report_stats(struct thread *tinfo)
 {
         const struct options *opts = tinfo[0].opts;
@@ -511,6 +584,9 @@ int rr_report_stats(struct thread *tinfo)
         int num_events = thread_stats_events(tinfo);
         PRINT(cb, "num_transactions", "%d", num_events);
 
+        if (opts->client && tinfo[0].opts->log_rtt)
+                rr_log_rtt(tinfo, cb, num_events);
+
         struct neper_histo *sum = neper_histo_new(tinfo, DEFAULT_K_BITS);
         for (i = 0; i < opts->num_threads; i++)
                 tinfo[i].stats->sumforeach(tinfo[i].stats, fn_add, sum);

thread.h

@@ -134,6 +134,8 @@ struct thread {
         int64_t rounding_ns; /* used to round to millisecond granularity */
         /* The duration between sends on a thread when using noburst. */
         int64_t gap_ns;
+        struct rtt_log *rtt_logs;
+        long rtt_log_capacity;
 };
 
 int thread_stats_events(const struct thread *);