Add cpu pinning for relay threads and udp listen sockets

To reduce lock contention, one map per relay server.

Admin server periodically tells the relay servers to cycle to fresh maps,
and the admin server outputs the old maps for external analysis.

Also, clang-format earlier changes

src/apps/common/apputils.c

@@ -468,6 +468,12 @@ int set_raw_socket_tos(evutil_socket_t fd, int family, int tos) {
   return 0;
 }
 
+// Signal change to add cpu pinning
+
+int set_raw_socket_incoming_cpu(evutil_socket_t fd, int cpu) {
+  return setsockopt(fd, SOL_SOCKET, SO_INCOMING_CPU, &cpu, sizeof(cpu));
+}
+
 int is_stream_socket(int st) {
   switch (st) {
   case TCP_SOCKET:
@@ -528,6 +534,40 @@ const char *socket_type_name(SOCKET_TYPE st) {
   return "UNKNOWN";
 }
 
+const char *duration_name(unsigned long duration) {
+  if (duration < 60) {
+    return "1min";
+  } else if (duration < 600) {
+    return "10mins";
+  } else if (duration < 86400) {
+    return "24hrs";
+  } else {
+    return "days";
+  }
+}
+
+const char *rate_name(unsigned long rate_kbps) {
+  if (rate_kbps < 1) {
+    return "1kbps";
+  } else if (rate_kbps < 50) {
+    return "50kbps";
+  } else if (rate_kbps < 2500) {
+    return "2500kbps";
+  } else {
+    return "10000kbps";
+  }
+}
+
+const char *addr_family_name(int addr_family) {
+  if (addr_family == AF_INET) {
+    return "ipv4";
+  } else if (addr_family == AF_INET6) {
+    return "ipv6";
+  } else {
+    return "other";
+  }
+}
+
 /////////////////// MTU /////////////////////////////////////////
 
 int set_socket_df(evutil_socket_t fd, int family, int value) {
@@ -1123,7 +1163,7 @@ char *find_config_file(const char *config_file, int print_file_name) {
           size_t celen = strlen(c_execdir);
           fnsz = sizeof(char) * (dirlen + cflen + celen + 10);
           fn = (char *)malloc(fnsz + 1);
-          strncpy(fn, c_execdir, fnsz);
+          strncpy(fn, c_execdir, celen);
           size_t fnlen = strlen(fn);
           if (fnlen < fnsz) {
             strncpy(fn + fnlen, "/", fnsz - fnlen);

src/apps/common/apputils.h

@@ -205,6 +205,8 @@ int handle_socket_error(void);
 
 int set_raw_socket_tos(evutil_socket_t fd, int family, int tos);
 int set_raw_socket_ttl(evutil_socket_t fd, int family, int ttl);
+// Signal change to add cpu pinning
+int set_raw_socket_incoming_cpu(evutil_socket_t fd, int cpu);
 int get_raw_socket_tos(evutil_socket_t fd, int family);
 int get_raw_socket_ttl(evutil_socket_t fd, int family);
 

src/apps/relay/dtls_listener.c

@@ -768,6 +768,9 @@ static int create_server_socket(dtls_listener_relay_server_type *server, int rep
     set_raw_socket_ttl_options(udp_listen_fd, server->addr.ss.sa_family);
     set_raw_socket_tos_options(udp_listen_fd, server->addr.ss.sa_family);
 
+    // Signal change to add cpu pinning
+    set_raw_socket_incoming_cpu(udp_listen_fd, server->ts->id);
+
     {
       const int max_binding_time = 60;
       int addr_bind_cycle = 0;

src/apps/relay/netengine.c

@@ -43,6 +43,10 @@ static unsigned int barrier_count = 0;
 static pthread_barrier_t barrier;
 #endif
 
+// Signal change to add rtt metrics
+
+static pthread_barrier_t rtt_barrier;
+
 ////////////// Auth Server ////////////////
 
 typedef unsigned char authserver_id;
@@ -686,6 +690,37 @@ err:
   return ret;
 }
 
+// Signal change to add rtt metrics
+int send_cycle_rtt_map_to_relay(turnserver_id id) {
+  int ret = 0;
+
+  struct message_to_relay sm;
+  memset(&sm, 0, sizeof(struct message_to_relay));
+  sm.t = RMT_CYCLE_RTT_MAP;
+
+  struct relay_server *rs = get_relay_server(id);
+  if (!rs) {
+    TURN_LOG_FUNC(TURN_LOG_LEVEL_ERROR, "%s: can't find relay for turn_server_id: %d\n", __FUNCTION__, (int)id);
+    ret = -1;
+    goto err;
+  }
+
+  sm.relay_server = rs;
+
+  {
+    struct evbuffer *output = bufferevent_get_output(rs->out_buf);
+    if (output) {
+      evbuffer_add(output, &sm, sizeof(struct message_to_relay));
+    } else {
+      TURN_LOG_FUNC(TURN_LOG_LEVEL_ERROR, "%s: Empty output buffer\n", __FUNCTION__);
+      ret = -1;
+    }
+  }
+
+err:
+  return ret;
+}
+
 static int handle_relay_message(relay_server_handle rs, struct message_to_relay *sm) {
   if (rs && sm) {
 
@@ -766,6 +801,18 @@ static int handle_relay_message(relay_server_handle rs, struct message_to_relay
       sm->m.sm.nd.nbh = NULL;
       break;
     }
+    // Signal change to add rtt metric
+    case RMT_CYCLE_RTT_MAP: {
+      rs->server.rtt_ms_mins = ur_map_create();
+      int br = 0;
+      do {
+        br = pthread_barrier_wait(&rtt_barrier);
+        if ((br < 0) && (br != PTHREAD_BARRIER_SERIAL_THREAD)) {
+          perror("rtt barrier wait (message)");
+        }
+      } while ((br < 0) && (br != PTHREAD_BARRIER_SERIAL_THREAD));
+      break;
+    }
     default: {
       perror("Weird buffer type\n");
     }
@@ -1872,3 +1919,61 @@ void setup_server(void) {
 void init_listener(void) { memset(&turn_params.listener, 0, sizeof(struct listener_server)); }
 
 ///////////////////////////////
+
+// Signal change to add rtt metrics
+size_t cycle_rtt_ms_maps(ur_map **rtt_ms_maps, size_t len) {
+  if (len != 1 + ((turnserver_id)-1)) {
+    TURN_LOG_FUNC(TURN_LOG_LEVEL_ERROR, "cycle_rtt_ms_maps, length is %ld, must be %ld\n", len,
+                  1L + ((turnserver_id)-1));
+    return 0;
+  }
+  size_t count = 0;
+  for (size_t i = 0; i < get_real_general_relay_servers_number(); i++) {
+    if (general_relay_servers[i] && general_relay_servers[i]->server.rtt_ms_mins) {
+      rtt_ms_maps[count] = general_relay_servers[i]->server.rtt_ms_mins;
+      ++count;
+    }
+  }
+  for (size_t i = 0; i < get_real_udp_relay_servers_number(); i++) {
+    if (udp_relay_servers[i] && udp_relay_servers[i]->server.rtt_ms_mins) {
+      rtt_ms_maps[count] = udp_relay_servers[i]->server.rtt_ms_mins;
+      ++count;
+    }
+  }
+
+  static size_t last_count = 0;
+  if (last_count != count) {
+    if (last_count) {
+      if (pthread_barrier_destroy(&rtt_barrier) != 0) {
+        perror("rtt barrier destroy");
+        return 0;
+      }
+    }
+    if (pthread_barrier_init(&rtt_barrier, NULL, count + 1) != 0) {
+      perror("rtt barrier init");
+      return 0;
+    }
+    last_count = count;
+  }
+
+  for (size_t i = 0; i < get_real_general_relay_servers_number(); i++) {
+    if (general_relay_servers[i] && general_relay_servers[i]->server.rtt_ms_mins) {
+      send_cycle_rtt_map_to_relay(i);
+    }
+  }
+  for (size_t i = 0; i < get_real_udp_relay_servers_number(); i++) {
+    if (udp_relay_servers[i] && udp_relay_servers[i]->server.rtt_ms_mins) {
+      send_cycle_rtt_map_to_relay(i + TURNSERVER_ID_BOUNDARY_BETWEEN_TCP_AND_UDP);
+    }
+  }
+
+  int br = 0;
+  do {
+    br = pthread_barrier_wait(&rtt_barrier);
+    if ((br < 0) && (br != PTHREAD_BARRIER_SERIAL_THREAD)) {
+      perror("rtt barrier wait");
+    }
+  } while ((br < 0) && (br != PTHREAD_BARRIER_SERIAL_THREAD));
+
+  return count;
+}

src/apps/relay/ns_ioalib_engine_impl.c

@@ -3574,14 +3574,15 @@ void turn_report_allocation_set(void *a, turn_time_t lifetime, int refresh) {
 #endif
         {
           if (!refresh)
-            prom_inc_allocation(get_ioa_socket_type(ss->client_socket));
+            prom_inc_allocation(get_ioa_socket_type(ss->client_socket),
+                                get_ioa_socket_address_family(ss->client_socket));
         }
       }
     }
   }
 }
 
-void turn_report_allocation_delete(void *a, SOCKET_TYPE socket_type) {
+void turn_report_allocation_delete(void *a, SOCKET_TYPE socket_type, int family) {
   if (a) {
     ts_ur_super_session *ss = (ts_ur_super_session *)(((allocation *)a)->owner);
     if (ss) {
@@ -3630,27 +3631,33 @@ void turn_report_allocation_delete(void *a, SOCKET_TYPE socket_type) {
         }
 #endif
         {
+          // Signal change to add metrics
           if (ss->realm_options.name[0]) {
 
             // Set prometheus traffic metrics
             prom_set_finished_traffic(ss->realm_options.name, (const char *)ss->username,
                                       (unsigned long)(ss->t_received_packets), (unsigned long)(ss->t_received_bytes),
-                                      (unsigned long)(ss->t_sent_packets), (unsigned long)(ss->t_sent_bytes), false);
+                                      (unsigned long)(ss->t_sent_packets), (unsigned long)(ss->t_sent_bytes),
+                                      (unsigned long)ss->t_before_ping_packets, false);
             prom_set_finished_traffic(
                 ss->realm_options.name, (const char *)ss->username, (unsigned long)(ss->t_peer_received_packets),
                 (unsigned long)(ss->t_peer_received_bytes), (unsigned long)(ss->t_peer_sent_packets),
-                (unsigned long)(ss->t_peer_sent_bytes), true);
+                (unsigned long)(ss->t_peer_sent_bytes), 0, true);
           } else {
             // Set prometheus traffic metrics
             prom_set_finished_traffic(NULL, (const char *)ss->username, (unsigned long)(ss->t_received_packets),
                                       (unsigned long)(ss->t_received_bytes), (unsigned long)(ss->t_sent_packets),
-                                      (unsigned long)(ss->t_sent_bytes), false);
+                                      (unsigned long)(ss->t_sent_bytes), (unsigned long)ss->t_before_ping_packets,
+                                      false);
             prom_set_finished_traffic(NULL, (const char *)ss->username, (unsigned long)(ss->t_peer_received_packets),
                                       (unsigned long)(ss->t_peer_received_bytes),
                                       (unsigned long)(ss->t_peer_sent_packets), (unsigned long)(ss->t_peer_sent_bytes),
-                                      true);
+                                      0, true);
           }
-          prom_dec_allocation(socket_type);
+          turn_time_t ct = get_turn_server_time(server) - ss->start_time;
+          const uint32_t byte_to_kilobit = 125;
+          uint64_t sent_rate_kbps = ss->sent_rate / byte_to_kilobit;
+          prom_dec_allocation(socket_type, family, (unsigned long)ct, (unsigned long)sent_rate_kbps);
         }
       }
     }

src/apps/relay/ns_ioalib_impl.h

@@ -288,6 +288,10 @@ void *allocate_super_memory_engine_func(ioa_engine_handle e, size_t size, const
 
 /////////////////////////////////////////////////
 
+// Signal change to add rtt metrics
+int send_cycle_rtt_map_to_relay(turnserver_id id);
+size_t cycle_rtt_ms_maps(ur_map **rtt_ms_maps, size_t len);
+
 #ifdef __cplusplus
 }
 #endif

src/apps/relay/prom_server.c

@@ -33,8 +33,16 @@ prom_counter_t *turn_total_traffic_peer_rcvb;
 prom_counter_t *turn_total_traffic_peer_sentp;
 prom_counter_t *turn_total_traffic_peer_sentb;
 
+prom_counter_t *turn_total_sessions;
+
 prom_gauge_t *turn_total_allocations;
 
+// Signal change to add metrics
+prom_counter_t *turn_rtt_client[8];
+prom_counter_t *turn_rtt_peer[8];
+prom_counter_t *turn_rtt_combined[8];
+prom_counter_t *turn_with_no_ping_rcvp;
+
 void start_prometheus_server(void) {
   if (turn_params.prometheus == 0) {
     TURN_LOG_FUNC(TURN_LOG_LEVEL_INFO, "prometheus collector disabled, not started\n");
@@ -98,10 +106,73 @@ void start_prometheus_server(void) {
   turn_total_traffic_peer_sentb = prom_collector_registry_must_register_metric(
       prom_counter_new("turn_total_traffic_peer_sentb", "Represents total finished sessions peer sent bytes", 0, NULL));
 
+  // Create total completed session counter metric
+  const char *total_sessions_labels[] = {"duration", "sent_rate"};
+  turn_total_sessions = prom_collector_registry_must_register_metric(
+      prom_counter_new("turn_total_sessions", "Represents total completed sessions", 2, total_sessions_labels));
+
   // Create total allocations number gauge metric
-  const char *typeLabel[] = {"type"};
+  const char *total_allocations_labels[] = {"type", "client_addr_family"};
   turn_total_allocations = prom_collector_registry_must_register_metric(
-      prom_gauge_new("turn_total_allocations", "Represents current allocations number", 1, typeLabel));
+      prom_gauge_new("turn_total_allocations", "Represents current allocations number", 2, total_allocations_labels));
+
+  // Signal change to add metrics
+  // Create round trip time pseudo-histogram metrics
+  // values must be kept in sync with observation function below
+
+  turn_rtt_client[0] = prom_collector_registry_must_register_metric(prom_counter_new(
+      "turn_rtt_client_le_25ms", "Represents measured round trip time of client with channel", 0, NULL));
+  turn_rtt_client[1] = prom_collector_registry_must_register_metric(prom_counter_new(
+      "turn_rtt_client_le_50ms", "Represents measured round trip time of client with channel", 0, NULL));
+  turn_rtt_client[2] = prom_collector_registry_must_register_metric(prom_counter_new(
+      "turn_rtt_client_le_100ms", "Represents measured round trip time of client with channel", 0, NULL));
+  turn_rtt_client[3] = prom_collector_registry_must_register_metric(prom_counter_new(
+      "turn_rtt_client_le_200ms", "Represents measured round trip time of client with channel", 0, NULL));
+  turn_rtt_client[4] = prom_collector_registry_must_register_metric(prom_counter_new(
+      "turn_rtt_client_le_400ms", "Represents measured round trip time of client with channel", 0, NULL));
+  turn_rtt_client[5] = prom_collector_registry_must_register_metric(prom_counter_new(
+      "turn_rtt_client_le_800ms", "Represents measured round trip time of client with channel", 0, NULL));
+  turn_rtt_client[6] = prom_collector_registry_must_register_metric(prom_counter_new(
+      "turn_rtt_client_le_1500ms", "Represents measured round trip time of client with channel", 0, NULL));
+  turn_rtt_client[7] = prom_collector_registry_must_register_metric(
+      prom_counter_new("turn_rtt_client_more", "Represents measured round trip time of client with channel", 0, NULL));
+
+  turn_rtt_peer[0] = prom_collector_registry_must_register_metric(
+      prom_counter_new("turn_rtt_peer_le_25ms", "Represents measured round trip time of peer with channel", 0, NULL));
+  turn_rtt_peer[1] = prom_collector_registry_must_register_metric(
+      prom_counter_new("turn_rtt_peer_le_50ms", "Represents measured round trip time of peer with channel", 0, NULL));
+  turn_rtt_peer[2] = prom_collector_registry_must_register_metric(
+      prom_counter_new("turn_rtt_peer_le_100ms", "Represents measured round trip time of peer with channel", 0, NULL));
+  turn_rtt_peer[3] = prom_collector_registry_must_register_metric(
+      prom_counter_new("turn_rtt_peer_le_200ms", "Represents measured round trip time of peer with channel", 0, NULL));
+  turn_rtt_peer[4] = prom_collector_registry_must_register_metric(
+      prom_counter_new("turn_rtt_peer_le_400ms", "Represents measured round trip time of peer with channel", 0, NULL));
+  turn_rtt_peer[5] = prom_collector_registry_must_register_metric(
+      prom_counter_new("turn_rtt_peer_le_800ms", "Represents measured round trip time of peer with channel", 0, NULL));
+  turn_rtt_peer[6] = prom_collector_registry_must_register_metric(
+      prom_counter_new("turn_rtt_peer_le_1500ms", "Represents measured round trip time of peer with channel", 0, NULL));
+  turn_rtt_peer[7] = prom_collector_registry_must_register_metric(
+      prom_counter_new("turn_rtt_peer_more", "Represents measured round trip time of peer with channel", 0, NULL));
+
+  turn_rtt_combined[0] = prom_collector_registry_must_register_metric(
+      prom_counter_new("turn_rtt_combined_le_25ms", "Represents combined round trip time of channel", 0, NULL));
+  turn_rtt_combined[1] = prom_collector_registry_must_register_metric(
+      prom_counter_new("turn_rtt_combined_le_50ms", "Represents combined round trip time of channel", 0, NULL));
+  turn_rtt_combined[2] = prom_collector_registry_must_register_metric(
+      prom_counter_new("turn_rtt_combined_le_100ms", "Represents combined round trip time of channel", 0, NULL));
+  turn_rtt_combined[3] = prom_collector_registry_must_register_metric(
+      prom_counter_new("turn_rtt_combined_le_200ms", "Represents combined round trip time of channel", 0, NULL));
+  turn_rtt_combined[4] = prom_collector_registry_must_register_metric(
+      prom_counter_new("turn_rtt_combined_le_400ms", "Represents combined round trip time of channel", 0, NULL));
+  turn_rtt_combined[5] = prom_collector_registry_must_register_metric(
+      prom_counter_new("turn_rtt_combined_le_800ms", "Represents combined round trip time of channel", 0, NULL));
+  turn_rtt_combined[6] = prom_collector_registry_must_register_metric(
+      prom_counter_new("turn_rtt_combined_le_1500ms", "Represents combined round trip time of channel", 0, NULL));
+  turn_rtt_combined[7] = prom_collector_registry_must_register_metric(
+      prom_counter_new("turn_rtt_combined_more", "Represents combined round trip time of channel", 0, NULL));
+
+  turn_with_no_ping_rcvp = prom_collector_registry_must_register_metric(prom_counter_new(
+      "turn_with_no_ping_rcvp", "Count of packets received for TURN where no ICE ping has been observed", 0, NULL));
 
   promhttp_set_active_collector_registry(NULL);
 
@@ -138,8 +209,9 @@ void start_prometheus_server(void) {
   return;
 }
 
+// Signal change to add metrics
 void prom_set_finished_traffic(const char *realm, const char *user, unsigned long rsvp, unsigned long rsvb,
-                               unsigned long sentp, unsigned long sentb, bool peer) {
+                               unsigned long sentp, unsigned long sentb, unsigned long without_pingp, bool peer) {
   if (turn_params.prometheus == 1) {
 
     const char *label[] = {realm, NULL};
@@ -168,20 +240,26 @@ void prom_set_finished_traffic(const char *realm, const char *user, unsigned lon
       prom_counter_add(turn_total_traffic_sentp, sentp, NULL);
       prom_counter_add(turn_total_traffic_sentb, sentb, NULL);
     }
+    // Signal change to add metrics
+    if (without_pingp) {
+      prom_counter_add(turn_with_no_ping_rcvp, without_pingp, NULL);
+    }
   }
 }
 
-void prom_inc_allocation(SOCKET_TYPE type) {
+void prom_inc_allocation(SOCKET_TYPE type, int addr_family) {
   if (turn_params.prometheus == 1) {
-    const char *label[] = {socket_type_name(type)};
-    prom_gauge_inc(turn_total_allocations, label);
+    const char *labels[] = {socket_type_name(type), addr_family_name(addr_family)};
+    prom_gauge_inc(turn_total_allocations, labels);
   }
 }
 
-void prom_dec_allocation(SOCKET_TYPE type) {
+void prom_dec_allocation(SOCKET_TYPE type, int addr_family, unsigned long duration, unsigned long sent_rate_kbps) {
   if (turn_params.prometheus == 1) {
-    const char *label[] = {socket_type_name(type)};
-    prom_gauge_dec(turn_total_allocations, label);
+    const char *labels[] = {socket_type_name(type), addr_family_name(addr_family)};
+    prom_gauge_dec(turn_total_allocations, labels);
+    const char *total_sessions_labels[] = {duration_name(duration), rate_name(sent_rate_kbps)};
+    prom_counter_add(turn_total_sessions, 1, total_sessions_labels);
   }
 }
 
@@ -219,6 +297,50 @@ int is_ipv6_enabled(void) {
   return ret;
 }
 
+// Signal change to add metrics
+void prom_observe_rtt(prom_counter_t *counter[8], int microseconds) {
+  if (microseconds <= 25000) {
+    prom_counter_add(counter[0], 1, NULL);
+  }
+  if (microseconds <= 50000) {
+    prom_counter_add(counter[1], 1, NULL);
+  }
+  if (microseconds <= 100000) {
+    prom_counter_add(counter[2], 1, NULL);
+  }
+  if (microseconds <= 200000) {
+    prom_counter_add(counter[3], 1, NULL);
+  }
+  if (microseconds <= 400000) {
+    prom_counter_add(counter[4], 1, NULL);
+  }
+  if (microseconds <= 800000) {
+    prom_counter_add(counter[5], 1, NULL);
+  }
+  if (microseconds <= 1500000) {
+    prom_counter_add(counter[6], 1, NULL);
+  }
+  prom_counter_add(counter[7], 1, NULL);
+}
+
+void prom_observe_rtt_client(int microseconds) {
+  if (turn_params.prometheus == 1) {
+    prom_observe_rtt(turn_rtt_client, microseconds);
+  }
+}
+
+void prom_observe_rtt_peer(int microseconds) {
+  if (turn_params.prometheus == 1) {
+    prom_observe_rtt(turn_rtt_peer, microseconds);
+  }
+}
+
+void prom_observe_rtt_combined(int microseconds) {
+  if (turn_params.prometheus == 1) {
+    prom_observe_rtt(turn_rtt_combined, microseconds);
+  }
+}
+
 #else
 
 void start_prometheus_server(void) {
@@ -227,13 +349,14 @@ void start_prometheus_server(void) {
 }
 
 void prom_set_finished_traffic(const char *realm, const char *user, unsigned long rsvp, unsigned long rsvb,
-                               unsigned long sentp, unsigned long sentb, bool peer) {
+                               unsigned long sentp, unsigned long sentb, unsigned long without_pingp, bool peer) {
   UNUSED_ARG(realm);
   UNUSED_ARG(user);
   UNUSED_ARG(rsvp);
   UNUSED_ARG(rsvb);
   UNUSED_ARG(sentp);
   UNUSED_ARG(sentb);
+  UNUSED_ARG(without_pingp);
   UNUSED_ARG(peer);
 }
 

src/apps/relay/prom_server.h

@@ -53,17 +53,26 @@ extern prom_counter_t *turn_total_traffic_peer_sentb;
 
 extern prom_gauge_t *turn_total_allocations_number;
 
+// Signal change to add metrics
+extern prom_counter_t *turn_rtt_client[8];
+extern prom_counter_t *turn_rtt_peer[8];
+extern prom_counter_t *turn_rtt_combined[8];
+extern prom_counter_t *turn_with_no_ping_rcvp;
+
+#define TURN_ALLOC_STR_MAX_SIZE (20)
+
 #ifdef __cplusplus
 extern "C" {
 #endif
 
 void start_prometheus_server(void);
 
+// Signal change to add metrics
 void prom_set_finished_traffic(const char *realm, const char *user, unsigned long rsvp, unsigned long rsvb,
-                               unsigned long sentp, unsigned long sentb, bool peer);
+                               unsigned long sentp, unsigned long sentb, unsigned long without_pingp, bool peer);
 
-void prom_inc_allocation(SOCKET_TYPE type);
-void prom_dec_allocation(SOCKET_TYPE type);
+void prom_inc_allocation(SOCKET_TYPE type, int addr_family);
+void prom_dec_allocation(SOCKET_TYPE type, int addr_family, unsigned long duration, unsigned long sent_rate_kbps);
 
 int is_ipv6_enabled(void);
 
@@ -71,12 +80,19 @@ void prom_inc_stun_binding_request(void);
 void prom_inc_stun_binding_response(void);
 void prom_inc_stun_binding_error(void);
 
+// Signal change to add metrics
+void prom_observe_rtt(prom_counter_t *counter[8], int microseconds);
+void prom_observe_rtt_client(int microseconds);
+void prom_observe_rtt_peer(int microseconds);
+void prom_observe_rtt_combined(int microseconds);
+
 #else
 
 void start_prometheus_server(void);
 
+// Signal change to add metrics
 void prom_set_finished_traffic(const char *realm, const char *user, unsigned long rsvp, unsigned long rsvb,
-                               unsigned long sentp, unsigned long sentb, bool peer);
+                               unsigned long sentp, unsigned long sentb, unsigned long without_pingp, bool peer);
 
 void prom_inc_allocation(SOCKET_TYPE type);
 void prom_dec_allocation(SOCKET_TYPE type);

src/apps/relay/turn_admin_server.c

@@ -81,6 +81,9 @@
 
 #include "tls_listener.h"
 
+// Signal change to add rtt metrics
+#include <fcntl.h>
+#include <unistd.h>
 ///////////////////////////////
 
 struct admin_server adminserver;
@@ -1358,6 +1361,11 @@ void setup_admin_thread(void) {
   }
 
   adminserver.sessions = ur_map_create();
+
+  // Signal change to add rtt metrics
+  // run once a day
+  adminserver.rtt_ev =
+      set_ioa_timer(adminserver.e, 86400, 0, admin_server_rtt_timer_handler, NULL, 1, "admin_server_rtt_timer");
 }
 
 void admin_server_receive_message(struct bufferevent *bev, void *ptr) {
@@ -3735,3 +3743,82 @@ void send_https_socket(ioa_socket_handle s) {
 }
 
 ///////////////////////////////
+// Signal change to add rtt metrics
+
+ur_map *rtt_maps[1 + ((turnserver_id)-1)] = {0};
+size_t rtt_maps_count = 0;
+size_t rtt_map_current = 0;
+FILE *rtt_file;
+
+int rtt_foreach(ur_map_key_type key, ur_map_value_type value) {
+  if (!value) {
+    return 0;
+  }
+  ur_map_value_type min = value;
+  for (size_t i = rtt_map_current + 1; i < rtt_maps_count; ++i) {
+    if (ur_map_get(rtt_maps[i], key, &value)) {
+      ur_map_put(rtt_maps[i], key, 0);
+      if (value && value < min) {
+        min = value;
+      }
+    }
+  }
+  // value is stored as measured rtt in ms + 1
+  value -= 1;
+  char saddr[INET6_ADDRSTRLEN] = "\0";
+  if (key & (1L << 63)) {
+    struct sockaddr_in6 addr = {0};
+    addr.sin6_family = AF_INET6;
+    size_t i = 6;
+    while (i--) {
+      addr.sin6_addr.s6_addr[i] = key & 0xFF;
+      key >>= 8;
+    }
+    inet_ntop(AF_INET6, &addr.sin6_addr, saddr, sizeof(saddr));
+    fprintf(rtt_file, "%s/48,%ld\n", saddr, min);
+  } else {
+    struct sockaddr_in addr = {0};
+    addr.sin_family = AF_INET;
+    key <<= 8;
+    addr.sin_addr.s_addr = htonl(key & 0xFFFFFFFF);
+    inet_ntop(AF_INET, &addr.sin_addr, saddr, sizeof(saddr));
+    fprintf(rtt_file, "%s/24,%ld\n", saddr, min);
+  }
+  return 0;
+}
+
+void admin_server_rtt_timer_handler(ioa_engine *engine, void *arg) {
+  UNUSED_ARG(engine);
+  UNUSED_ARG(arg);
+  int fd = open("/var/tmp/rtt_dump.tmp", O_WRONLY | O_CREAT | O_NOFOLLOW | O_TRUNC, S_IRUSR | S_IWUSR);
+  if (fd == -1) {
+    TURN_LOG_FUNC(TURN_LOG_LEVEL_ERROR, "error opening temporary file during rtt timer (%d)\n", errno);
+    return;
+  }
+
+  rtt_file = fdopen(fd, "w");
+  if (rtt_file == NULL) {
+    TURN_LOG_FUNC(TURN_LOG_LEVEL_ERROR, "fdopen error during rtt timer\n");
+    close(fd);
+    return;
+  }
+
+  fprintf(rtt_file, "network,rtt_ms\n");
+
+  rtt_maps_count = cycle_rtt_ms_maps(rtt_maps, sizeof(rtt_maps) / sizeof(rtt_maps[0]));
+
+  for (rtt_map_current = 0; rtt_map_current < rtt_maps_count; ++rtt_map_current) {
+    ur_map_foreach(rtt_maps[rtt_map_current], rtt_foreach);
+    ur_map_free(&rtt_maps[rtt_map_current]);
+  }
+
+  if (fflush(rtt_file) != 0) {
+    TURN_LOG_FUNC(TURN_LOG_LEVEL_ERROR, "fflush /var/tmp/rtt_dump.tmp failed, not relinking (%d)\n", errno);
+  } else {
+    if (rename("/var/tmp/rtt_dump.tmp", "/var/tmp/rtt_dump") == -1) {
+      TURN_LOG_FUNC(TURN_LOG_LEVEL_ERROR, "could not rename rtt dump into /var/tmp/rtt_dump (%d)\n", errno);
+    }
+  }
+  fclose(rtt_file);
+  rtt_file = NULL;
+}

src/apps/relay/turn_admin_server.h

@@ -73,6 +73,8 @@ struct admin_server {
   struct bufferevent *https_out_buf;
   ur_map *sessions;
   pthread_t thr;
+  // Signal change to add rtt metrics
+  ioa_timer_handle rtt_ev;
 };
 
 ///////////////////////////////////////////
@@ -113,6 +115,10 @@ void https_admin_server_receive_message(struct bufferevent *bev, void *ptr);
 int send_turn_session_info(struct turn_session_info *tsi);
 void send_https_socket(ioa_socket_handle s);
 
+// Signal change to add rtt metrics
+int rtt_foreach(ur_map_key_type, ur_map_value_type);
+void admin_server_rtt_timer_handler(ioa_engine *, void *);
+
 ////////////////////////////////////////////
 
 #ifdef __cplusplus

src/server/ns_turn_allocation.c

@@ -48,11 +48,11 @@ void init_allocation(void *owner, allocation *a, ur_map *tcp_connections) {
   }
 }
 
-void clear_allocation(allocation *a, SOCKET_TYPE socket_type) {
+void clear_allocation(allocation *a, SOCKET_TYPE socket_type, int family) {
   if (a) {
 
     if (a->is_valid)
-      turn_report_allocation_delete(a, socket_type);
+      turn_report_allocation_delete(a, socket_type, family);
 
     if (a->tcs.elems) {
       size_t i;

src/server/ns_turn_allocation.h

@@ -137,6 +137,13 @@ void ch_map_clean(ch_map *map);
 
 ////////////////////////////
 
+// Signal change to add rtt metrics
+typedef struct _turn_ice_ping_info {
+  stun_tid tid;
+  struct timespec ts;
+  int lastrttus;
+} turn_ice_ping_info;
+
 typedef struct _turn_permission_info {
   int allocated;
   lm_map chns;
@@ -146,6 +153,8 @@ typedef struct _turn_permission_info {
   void *owner; // a
   int verbose;
   unsigned long long session_id;
+  // Signal change to add rtt metrics
+  turn_ice_ping_info pings[2]; // 0 measuring round trip to peer, 1 measuring round trip to client
 } turn_permission_info;
 
 typedef struct _turn_permission_slot {
@@ -192,7 +201,7 @@ void turn_channel_delete(ch_info *chn);
 /////////// ALLOCATION ////////////
 
 void init_allocation(void *owner, allocation *a, ur_map *tcp_connections);
-void clear_allocation(allocation *a, SOCKET_TYPE socket_type);
+void clear_allocation(allocation *a, SOCKET_TYPE socket_type, int family);
 
 void turn_permission_clean(turn_permission_info *tinfo);
 

src/server/ns_turn_ioalib.h

@@ -216,7 +216,7 @@ typedef enum _STUN_PROMETHEUS_METRIC_TYPE STUN_PROMETHEUS_METRIC_TYPE;
 void stun_report_binding(void *session, STUN_PROMETHEUS_METRIC_TYPE type);
 
 void turn_report_allocation_set(void *a, turn_time_t lifetime, int refresh);
-void turn_report_allocation_delete(void *a, SOCKET_TYPE socket_type);
+void turn_report_allocation_delete(void *a, SOCKET_TYPE socket_type, int family);
 void turn_report_session_usage(void *session, int force_invalid);
 
 /*
@@ -272,6 +272,9 @@ int is_stream_socket(int st);
 int is_tcp_socket(int st);
 int is_sctp_socket(int st);
 const char *socket_type_name(SOCKET_TYPE st);
+const char *duration_name(unsigned long duration);
+const char *rate_name(unsigned long rate_kbps);
+const char *addr_family_name(int family);
 const char *get_ioa_socket_cipher(ioa_socket_handle s);
 const char *get_ioa_socket_ssl_method(ioa_socket_handle s);
 SOCKET_TYPE get_ioa_socket_type(ioa_socket_handle s);

src/server/ns_turn_server.c

@@ -27,10 +27,15 @@
  * OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF
  * SUCH DAMAGE.
  */
+// Signal change to add cpu pinning
+#define _GNU_SOURCE
+#include <sched.h>
 
 #include "ns_turn_server.h"
 
 #include "../apps/relay/ns_ioalib_impl.h"
+// Signal change to add rtt metrics
+#include "../apps/relay/prom_server.h"
 #include "ns_turn_allocation.h"
 #include "ns_turn_ioalib.h"
 #include "ns_turn_utils.h"
@@ -777,12 +782,12 @@ static ts_ur_super_session *create_new_ss(turn_turnserver *server) {
   return ss;
 }
 
-static void delete_ur_map_ss(void *p, SOCKET_TYPE socket_type) {
+static void delete_ur_map_ss(void *p, SOCKET_TYPE socket_type, int family) {
   if (p) {
     ts_ur_super_session *ss = (ts_ur_super_session *)p;
     delete_session_from_map(ss);
     IOA_CLOSE_SOCKET(ss->client_socket);
-    clear_allocation(get_allocation_ss(ss), socket_type);
+    clear_allocation(get_allocation_ss(ss), socket_type, family);
     IOA_EVENT_DEL(ss->to_be_allocated_timeout_ev);
     free(p);
   }
@@ -790,7 +795,7 @@ static void delete_ur_map_ss(void *p, SOCKET_TYPE socket_type) {
 
 /////////// clean all /////////////////////
 
-static int turn_server_remove_all_from_ur_map_ss(ts_ur_super_session *ss, SOCKET_TYPE socket_type) {
+static int turn_server_remove_all_from_ur_map_ss(ts_ur_super_session *ss, SOCKET_TYPE socket_type, int family) {
   if (!ss)
     return 0;
   else {
@@ -804,7 +809,7 @@ static int turn_server_remove_all_from_ur_map_ss(ts_ur_super_session *ss, SOCKET
     if (get_relay_socket_ss(ss, AF_INET6)) {
       clear_ioa_socket_session_if(get_relay_socket_ss(ss, AF_INET6), ss);
     }
-    delete_ur_map_ss(ss, socket_type);
+    delete_ur_map_ss(ss, socket_type, family);
     return ret;
   }
 }
@@ -2879,6 +2884,106 @@ static int handle_turn_binding(turn_turnserver *server, ts_ur_super_session *ss,
   return 0;
 }
 
+// Signal change to add metrics
+/////////////// inspect relayed packets, they might be ICE binds ///////////////
+
+static int inspect_binds(turn_turnserver *server, ioa_net_data *in_buffer, turn_permission_info *tinfo, int from_peer,
+                         int is_channel) {
+  if (!in_buffer || !tinfo || !(from_peer == 0 || from_peer == 1)) {
+    return 0;
+  }
+  size_t len = ioa_network_buffer_get_size(in_buffer->nbh);
+  uint8_t *buf = ioa_network_buffer_data(in_buffer->nbh);
+  if (stun_is_command_message_str(buf, len) && (stun_get_method_str(buf, len) == STUN_METHOD_BINDING)) {
+    if (stun_is_request_str(buf, len)) {
+      stun_tid tid;
+      stun_tid_from_message_str(buf, len, &tid);
+
+      // only process if this is the first received request
+      if (!stun_tid_equals(&tid, &tinfo->pings[from_peer].tid)) {
+        stun_tid_cpy(&tinfo->pings[from_peer].tid, &tid);
+        clock_gettime(CLOCK_MONOTONIC, &tinfo->pings[from_peer].ts);
+      }
+    } else if (stun_is_response_str(buf, len)) {
+      // invert from_peer, because we're processing replies
+      int from_client;
+      if (from_peer) {
+        from_client = 0;
+      } else {
+        from_client = 1;
+      }
+
+      if (tinfo->pings[from_client].ts.tv_sec == 0) {
+        return 0;
+      }
+
+      stun_tid tid;
+      stun_tid_from_message_str(buf, len, &tid);
+
+      if (stun_tid_equals(&tid, &tinfo->pings[from_client].tid)) {
+        struct timespec now;
+        clock_gettime(CLOCK_MONOTONIC, &now);
+
+        if (now.tv_sec >= tinfo->pings[from_client].ts.tv_sec &&
+            now.tv_sec <= tinfo->pings[from_client].ts.tv_sec + 60) {
+
+          int diffus = (now.tv_sec - tinfo->pings[from_client].ts.tv_sec) * 1000000;
+          diffus += (now.tv_nsec - tinfo->pings[from_client].ts.tv_nsec) / 1000;
+          if (diffus > 0) {
+            tinfo->pings[from_client].lastrttus = diffus;
+
+#if !defined(TURN_NO_PROMETHEUS)
+            if (is_channel) {
+              if (from_client) {
+                prom_observe_rtt_client(diffus);
+              } else {
+                prom_observe_rtt_peer(diffus);
+              }
+              if (tinfo->pings[from_peer].lastrttus > 0) {
+                prom_observe_rtt_combined(diffus + tinfo->pings[from_peer].lastrttus);
+              }
+            }
+#endif
+            ur_map_key_type key = 0;
+            // add one to value to differentiate from zero
+            ur_map_value_type diffms = diffus / 1000 + 1;
+
+            if (in_buffer->src_addr.ss.sa_family == AF_INET) {
+              key = ntohl(((struct sockaddr_in *)&in_buffer->src_addr)->sin_addr.s_addr);
+              key >>= 8; // keep only the top 24 bits
+            } else if (in_buffer->src_addr.ss.sa_family == AF_INET6) {
+              // use the high 6 bytes (48 bits)
+              for (int i = 0; i < 6; ++i) {
+                key <<= 8;
+                key |= ((struct sockaddr_in6 *)&in_buffer->src_addr)->sin6_addr.s6_addr[i];
+              }
+              key |= (1L << 63);
+            }
+
+            // explicitly copy map pointer in case of concurrent access
+            ur_map *map = server->rtt_ms_mins;
+
+            ur_map_lock(map);
+            ur_map_value_type value = 0;
+            ur_map_get(map, key, &value);
+            if (value == 0 || diffms < value) {
+              ur_map_put(map, key, diffms);
+            }
+            ur_map_unlock(map);
+          }
+        }
+        // don't process retransmited responses
+        tinfo->pings[from_client].ts.tv_sec = 0;
+      }
+    }
+  } else {
+    if (tinfo->pings[0].lastrttus == 0 && tinfo->pings[1].lastrttus == 0) {
+      return 1;
+    }
+  }
+  return 0;
+}
+
 static int handle_turn_send(turn_turnserver *server, ts_ur_super_session *ss, int *err_code, const uint8_t **reason,
                             uint16_t *unknown_attrs, uint16_t *ua_num, ioa_net_data *in_buffer) {
 
@@ -2961,6 +3066,11 @@ static int handle_turn_send(turn_turnserver *server, ts_ur_super_session *ss, in
           len = 0;
           ioa_network_buffer_set_size(nbh, len);
         }
+        // Signal change to add rtt metrics
+        if (inspect_binds(server, in_buffer, tinfo, 0, 0)) {
+          ++(ss->t_before_ping_packets);
+        }
+
         ioa_network_buffer_header_init(nbh);
         int skip = 0;
         send_data_from_ioa_socket_nbh(get_relay_socket_ss(ss, peer_addr.ss.sa_family), &peer_addr, nbh,
@@ -4019,6 +4129,13 @@ static int write_to_peerchannel(ts_ur_super_session *ss, uint16_t chnum, ioa_net
       ioa_network_buffer_add_offset_size(in_buffer->nbh, STUN_CHANNEL_HEADER_LENGTH, 0,
                                          ioa_network_buffer_get_size(in_buffer->nbh) - STUN_CHANNEL_HEADER_LENGTH);
 
+      // Signal change to add rtt metrics
+      turn_turnserver *server = (turn_turnserver *)ss->server;
+      turn_permission_info *tinfo = (turn_permission_info *)(chn->owner);
+      if (inspect_binds(server, in_buffer, tinfo, 0, 1)) {
+        ++(ss->t_before_ping_packets);
+      }
+
       ioa_network_buffer_header_init(nbh);
 
       int skip = 0;
@@ -4051,6 +4168,7 @@ int shutdown_client_connection(turn_turnserver *server, ts_ur_super_session *ss,
     return -1;
 
   SOCKET_TYPE socket_type = get_ioa_socket_type(ss->client_socket);
+  int family = get_ioa_socket_address_family(ss->client_socket);
 
   turn_report_session_usage(ss, 1);
   dec_quota(ss);
@@ -4114,7 +4232,7 @@ int shutdown_client_connection(turn_turnserver *server, ts_ur_super_session *ss,
     }
   }
 
-  turn_server_remove_all_from_ur_map_ss(ss, socket_type);
+  turn_server_remove_all_from_ur_map_ss(ss, socket_type, family);
 
   FUNCEND;
 
@@ -4219,7 +4337,7 @@ static void client_ss_allocation_timeout_handler(ioa_engine_handle e, void *arg)
   turn_turnserver *server = (turn_turnserver *)(ss->server);
 
   if (!server) {
-    clear_allocation(a, get_ioa_socket_type(ss->client_socket));
+    clear_allocation(a, get_ioa_socket_type(ss->client_socket), get_ioa_socket_address_family(ss->client_socket));
     return;
   }
 
@@ -4711,6 +4829,10 @@ static void peer_input_handler(ioa_socket_handle s, int event_type, ioa_net_data
       turn_permission_info *tinfo = allocation_get_permission(a, &(in_buffer->src_addr));
       if (tinfo) {
         chnum = get_turn_channel_number(tinfo, &(in_buffer->src_addr));
+        // Signal change to add rtt metrics
+        if (inspect_binds(server, in_buffer, tinfo, 1, chnum != 0)) {
+          ++(ss->t_before_ping_packets);
+        }
       } else if (!(server->server_relay)) {
         return;
       }
@@ -4899,6 +5021,17 @@ void init_turn_server(turn_turnserver *server, turnserver_id id, int verbose, io
   server->response_origin_only_with_rfc5780 = response_origin_only_with_rfc5780;
 
   server->respond_http_unsupported = respond_http_unsupported;
+
+  // Signal change to add rtt metrics
+  server->rtt_ms_mins = ur_map_create();
+
+  // Signal change to add cpu pinning
+  cpu_set_t cpuset;
+  CPU_ZERO(&cpuset);
+  CPU_SET(id, &cpuset);
+  if (sched_setaffinity(0, sizeof(cpuset), &cpuset) == -1) {
+    TURN_LOG_FUNC(TURN_LOG_LEVEL_ERROR, "turn server id=%d, could not set cpu affinity\n", (int) id);
+  }
 }
 
 ioa_engine_handle turn_server_get_engine(turn_turnserver *s) {

src/server/ns_turn_server.h

@@ -67,7 +67,15 @@ extern int TURN_MAX_ALLOCATE_TIMEOUT_STUN_ONLY;
 
 typedef uint8_t turnserver_id;
 
-enum _MESSAGE_TO_RELAY_TYPE { RMT_UNKNOWN = 0, RMT_SOCKET, RMT_CB_SOCKET, RMT_MOBILE_SOCKET, RMT_CANCEL_SESSION };
+// Signal change to add rtt metrics
+enum _MESSAGE_TO_RELAY_TYPE {
+  RMT_UNKNOWN = 0,
+  RMT_SOCKET,
+  RMT_CB_SOCKET,
+  RMT_MOBILE_SOCKET,
+  RMT_CANCEL_SESSION,
+  RMT_CYCLE_RTT_MAP
+};
 typedef enum _MESSAGE_TO_RELAY_TYPE MESSAGE_TO_RELAY_TYPE;
 
 ///////// ALLOCATION DEFAULT ADDRESS FAMILY TYPES /////////////////////
@@ -198,6 +206,10 @@ struct _turn_turnserver {
   /* Return an HTTP 400 response to HTTP connections made to ports not
      otherwise handling HTTP. */
   vintp respond_http_unsupported;
+
+  // Signal change to add rtt metrics
+  /* measured round trip minimums per network */
+  ur_map *rtt_ms_mins;
 };
 
 const char *get_version(turn_turnserver *server);

src/server/ns_turn_session.h

@@ -119,6 +119,8 @@ struct _ts_ur_super_session {
   char s_mobile_id[33];
   /* Bandwidth */
   band_limit_t bps;
+  // Signal change to add metrics
+  uint64_t t_before_ping_packets;
 };
 
 ////// Session info for statistics //////