Merge branch 'latency_stats' into 'master'

Add reporting of RX latency

Closes #258

See merge request openconnect/ocserv!145

.gitlab-ci.yml

@@ -65,7 +65,7 @@ Ubuntu18.04:
   script:
   - git submodule update --init
   - autoreconf -fvi
-  - ./configure --without-nuttcp-tests --without-docker-tests --enable-oidc-auth
+  - ./configure --without-nuttcp-tests --without-docker-tests --enable-oidc-auth --enable-latency-stats
   - make -j$JOBS
   - make check -j$JOBS
   tags:

configure.ac

@@ -540,6 +540,28 @@ AC_CHECK_FUNCS(sigaction)
 AC_CHECK_FUNCS(longjmp setjmp)
 AC_CHECK_FUNCS(sigaltstack)
 
+AC_CHECK_HEADERS([linux/net_tstamp.h], [net_tstamp_header=yes], [net_tstamp_header=no])
+AC_CHECK_TYPES([struct scm_timestamping],[scm_timestamping_defined=yes],[scm_timestamping_defined=no],
+[#include <time.h>
+#include <linux/errqueue.h>])
+
+AC_ARG_ENABLE([latency-stats],
+  [AS_HELP_STRING([--enable-latency-stats],
+                  [enable capturing latency stats (default is no)])],
+  [enable_latency_stats=$enableval],
+  [enable_latency_stats=no]
+)
+
+have_latency_stats=no
+
+if test "x$enable_latency_stats" = xyes && test "x$net_tstamp_header" = xyes && test "x$scm_timestamping_defined" = xyes; then
+  AC_DEFINE([CAPTURE_LATENCY_SUPPORT], [1], [Enable capturing latency stats])
+  AC_LIB_HAVE_LINKFLAGS([m])
+  have_latency_stats=yes
+fi
+
+AM_CONDITIONAL(ENABLE_LATENCY_SUPPORT,  test "x$have_latency_stats" = xyes)
+
 AC_ARG_WITH(pcl-lib,
   AS_HELP_STRING([--without-pcl-lib], [use the included PCL library]),
   test_for_pcl_lib=$withval,
@@ -672,6 +694,7 @@ Summary of build options:
   local PCL library:    ${with_local_pcl}
   local http-parser:    ${with_local_http_parser}
   seccomp trap:		${use_seccomp_trap}
+  capture latency stats ${have_latency_stats}
 ])
 
 if test "${warn_leak}" = "yes";then

doc/design.md

@@ -189,3 +189,20 @@ Cookies are valid for the value configured in `cookie-timeout` option, after
 a client disconnects due to timeout. Their purpose is to allow mobile clients to
 roam between networks without significant disruption in the VPN service.
 
+## When compile with --enable-latency-stats
+
+The ocserv server gathers statistical data about the latency incurred while processing received DTLS packets. Due to the volume of data being collected, processing is perfomed in batches. Batch size is a tradeoff of memory usage and statistical accuracy. All values are stored in microseconds (10^-6 seconds).
+
+* Latency samples are first batched by the ocserv-worker, which gathers LATENCY_SAMPLE_SIZE (1024) of latency data.
+
+* After LATENCY_SAMPLE_SIZE samples have been gathered, median, mean and STDEV (RMS of delta from median) are computed for that sample set. Totals are maintained for mean and STDEV as well as count of sample sets processed.
+
+* After more than LATENCY_WORKER_AGGREGATION_TIME (60) seconds have passed (or when the worker process ends) the totals for mean, STDEV and sample count are sent to the main process.
+
+* The ocserv-main process accumulates a running total of mean, STDEV and sample count from all the worker processes for the current sampling interval. Every LATENCY_AGGREGATION_TIME (60s) the current running total is stored as the latency data for the previous interval.
+
+* The main_server_st stats.current_latency_stats stores weighted mean of the latency the server experienced for the last LATENCY_AGGREGATION_TIME interval. Values are stored as total and sample count, permitting the consumer of the stats to better compute confidence interval and mean value for latency and variation of latency.
+
+* Latency information is emitted to the log and can also be queried via occtl. Mean latency for an interval can be computed as latency_median_total/latency_sample_count and mean STDEV can be computed as latency_rms_total/latency_sample_count.
+
+* Latency information can be used as a metric to measure how the ocserv is performing and to measure effective server load.

src/Makefile.am

@@ -63,6 +63,10 @@ CORE_SOURCES = $(CCAN_SOURCES) $(HTTP_PARSER_SOURCES) \
 	sup-config/radius.h tlslib.c tlslib.h tun.c tun.h valid-hostname.c \
 	vasprintf.c vasprintf.h vhost.h vpn.h 
 
+if ENABLE_LATENCY_SUPPORT
+ocserv_worker_SOURCES += worker-latency.c worker-latency.h
+endif
+
 if ENABLE_COMPRESSION
 CORE_SOURCES += lzs.c lzs.h
 endif
@@ -84,6 +88,10 @@ CORE_LDADD += $(LIBGNUTLS_LIBS) $(PAM_LIBS) $(LIBUTIL) \
 	$(LIBEV_LIBS) libipc.a $(NEEDED_LIBPROTOBUF_LIBS) \
 	$(CODE_COVERAGE_LDFLAGS) 
 	
+if ENABLE_LATENCY_SUPPORT
+ocserv_worker_LDADD += $(LIBM)
+endif
+
 if ENABLE_OIDC_AUTH
 CORE_LDADD += $(LIBCURL_LIBS) $(CJOSE_LIBS) $(JANSSON_LIBS)
 endif

src/ctl.proto

@@ -31,6 +31,10 @@ message status_rep
 	required uint64 auth_failures = 23;
 	required uint64 total_sessions_closed = 24;
 	required uint64 total_auth_failures = 25;
+
+	optional uint64 latency_median_total = 26;
+	optional uint64 latency_rms_total = 27;
+	optional uint64 latency_sample_count = 28;
 }
 
 message bool_msg

src/defs.h

@@ -74,6 +74,7 @@ typedef enum {
 	CMD_SESSION_INFO = 13,
 	CMD_BAN_IP = 16,
 	CMD_BAN_IP_REPLY = 17,
+	CMD_LATENCY_STATS_DELTA = 18,	
 
 	/* from worker to sec-mod */
 	CMD_SEC_AUTH_INIT = 120,

src/gettime.h

@@ -43,6 +43,19 @@ struct timeval tv;
 #endif
 }
 
+inline static void
+gettime_realtime (struct timespec *t)
+{
+#if defined(HAVE_CLOCK_GETTIME) && defined(CLOCK_REALTIME)
+  clock_gettime (CLOCK_REALTIME, t);
+#else
+struct timeval tv;
+  gettimeofday (&tv, NULL);
+  t->tv_sec = tv.tv_sec;
+  t->tv_nsec = tv.tv_usec * 1000;
+#endif
+}
+
 inline static
 unsigned int
 timespec_sub_ms (struct timespec *a, struct timespec *b)
@@ -51,4 +64,13 @@ timespec_sub_ms (struct timespec *a, struct timespec *b)
           (b->tv_sec * 1000 + b->tv_nsec / (1000 * 1000)));
 }
 
+
+inline static
+uint64_t
+timespec_sub_us (struct timespec *a, struct timespec *b)
+{
+  return (a->tv_sec * 1000000ULL  + a->tv_nsec / (1000) -
+          (b->tv_sec * 1000000ULL + b->tv_nsec / (1000)));
+}
+
 #endif

src/ipc.proto

@@ -212,6 +212,14 @@ message ban_ip_reply_msg
 	optional bytes sid = 2; /* sec-mod needs it */
 }
 
+/* WORKER_LATENCY_STATS_DELTA: sent from worker to main */
+message latency_stats_delta
+{
+	required uint64 median_delta = 1;
+	required uint64 rms_delta = 2;
+	required uint64 sample_count_delta = 3;
+}
+
 /* Messages to and from the security module */
 
 /*

src/main-ctl-unix.c

@@ -203,6 +203,14 @@ static void method_status(method_ctx *ctx, int cfd, uint8_t * msg,
 	rep.auth_failures = ctx->s->stats.auth_failures;
 	rep.total_auth_failures = ctx->s->stats.total_auth_failures;
 	rep.total_sessions_closed = ctx->s->stats.total_sessions_closed;
+#if defined(CAPTURE_LATENCY_SUPPORT)
+	rep.latency_median_total = ctx->s->stats.current_latency_stats.median_total;
+	rep.has_latency_median_total = true;
+	rep.latency_rms_total = ctx->s->stats.current_latency_stats.rms_total;
+	rep.has_latency_rms_total = true;
+	rep.latency_sample_count = ctx->s->stats.current_latency_stats.sample_count;
+	rep.has_latency_sample_count = true;
+#endif
 
 	ret = send_msg(ctx->pool, cfd, CTL_CMD_STATUS_REP, &rep,
 		       (pack_size_func) status_rep__get_packed_size,

src/main-worker-cmd.c

@@ -451,6 +451,32 @@ int handle_worker_commands(main_server_st * s, struct proc_st *proc)
 
 		break;
 
+#if defined(CAPTURE_LATENCY_SUPPORT)
+	case CMD_LATENCY_STATS_DELTA:{
+			LatencyStatsDelta * tmsg;
+ 
+			if (proc->status != PS_AUTH_COMPLETED) {
+				mslog(s, proc, LOG_ERR,
+					"received LATENCY STATS DELTA in unauthenticated state.");
+				ret = ERR_BAD_COMMAND;
+				goto cleanup;
+			}
+
+			tmsg = latency_stats_delta__unpack(&pa, raw_len, raw);
+			if (tmsg == NULL) {
+				mslog(s, proc, LOG_ERR, "error unpacking latency stats delta data");
+				ret = ERR_BAD_COMMAND;
+				goto cleanup;
+			}
+			
+			s->stats.delta_latency_stats.median_total += tmsg->median_delta;
+			s->stats.delta_latency_stats.rms_total += tmsg->rms_delta;
+			s->stats.delta_latency_stats.sample_count += tmsg->sample_count_delta;
+
+			latency_stats_delta__free_unpacked(tmsg, &pa);
+		}
+		break;
+#endif
 	default:
 		mslog(s, proc, LOG_ERR, "unknown CMD from worker: 0x%x", (unsigned)cmd);
 		ret = ERR_BAD_COMMAND;

src/main.c

@@ -100,6 +100,9 @@ ev_signal term_sig_watcher;
 ev_signal int_sig_watcher;
 ev_signal reload_sig_watcher;
 ev_child child_watcher;
+#if defined(CAPTURE_LATENCY_SUPPORT)
+ev_timer latency_watcher;
+#endif
 
 static int set_env_from_ws(main_server_st * ws);
 
@@ -527,6 +530,9 @@ void clear_lists(main_server_st *s)
 		ev_io_stop (loop, &sec_mod_watcher);
 		ev_child_stop (loop, &child_watcher);
 		ev_timer_stop(loop, &maintenance_watcher);
+#if defined(CAPTURE_LATENCY_SUPPORT)
+		ev_timer_stop(loop, &latency_watcher);		
+#endif
 		/* free memory and descriptors by the event loop */
 		ev_loop_destroy (loop);
 	}
@@ -1196,6 +1202,25 @@ static void maintenance_watcher_cb(EV_P_ ev_timer *w, int revents)
 	perform_maintenance(s);
 }
 
+#if defined(CAPTURE_LATENCY_SUPPORT)
+static void latency_watcher_cb(EV_P_ ev_timer *w, int revents)
+{
+	main_server_st *s = ev_userdata(loop);
+	s->stats.current_latency_stats = s->stats.delta_latency_stats;
+	s->stats.delta_latency_stats.median_total = 0;
+	s->stats.delta_latency_stats.rms_total = 0;
+	s->stats.delta_latency_stats.sample_count = 0;
+	mslog(
+		s, 
+		NULL, 
+		LOG_INFO, 
+		"Latency: Median Total %ld RMS Total %ld Sample Count %ld", 
+		s->stats.current_latency_stats.median_total, 
+		s->stats.current_latency_stats.rms_total, 
+		s->stats.current_latency_stats.sample_count);
+}
+#endif
+
 static void maintenance_sig_watcher_cb(struct ev_loop *loop, ev_signal *w, int revents)
 {
 	main_server_st *s = ev_userdata(loop);
@@ -1457,6 +1482,12 @@ int main(int argc, char** argv)
 	ev_timer_set(&maintenance_watcher, MAIN_MAINTENANCE_TIME, MAIN_MAINTENANCE_TIME);
 	ev_timer_start(loop, &maintenance_watcher);
 
+#if defined(CAPTURE_LATENCY_SUPPORT)
+	ev_init(&latency_watcher, latency_watcher_cb);
+	ev_timer_set(&latency_watcher, LATENCY_AGGREGATION_TIME, LATENCY_AGGREGATION_TIME);
+	ev_timer_start(loop, &latency_watcher);
+#endif
+
 	/* allow forcing maintenance with SIGUSR2 */
 	ev_init (&maintenance_sig_watcher, maintenance_sig_watcher_cb);
 	ev_signal_set (&maintenance_sig_watcher, SIGUSR2);

src/main.h

@@ -54,6 +54,10 @@ extern ev_timer maintainance_watcher;
 
 int cmd_parser (void *pool, int argc, char **argv, struct list_head *head, bool worker);
 
+#if defined(CAPTURE_LATENCY_SUPPORT)
+#define LATENCY_AGGREGATION_TIME (60)
+#endif
+
 struct listener_st {
 	ev_io io;
 	struct list_node list;
@@ -192,6 +196,14 @@ struct proc_hash_db_st {
 	unsigned total;
 };
 
+#if defined(CAPTURE_LATENCY_SUPPORT)
+struct latency_stats_st {
+	uint64_t median_total;
+	uint64_t rms_total;
+	uint64_t sample_count;
+};
+#endif
+
 struct main_stats_st {
 	uint64_t session_timeouts; /* sessions with timeout */
 	uint64_t session_idle_timeouts; /* sessions with idle timeout */
@@ -219,6 +231,11 @@ struct main_stats_st {
 	/* These are counted since start time */
 	uint64_t total_auth_failures; /* authentication failures since start_time */
 	uint64_t total_sessions_closed; /* sessions closed since start_time */
+
+#if defined(CAPTURE_LATENCY_SUPPORT)
+	struct latency_stats_st current_latency_stats;
+	struct latency_stats_st delta_latency_stats;
+#endif
 };
 
 typedef struct main_server_st {

src/occtl/occtl.c

@@ -240,6 +240,21 @@ double data;
 	}
 }
 
+void
+time2human(uint64_t microseconds, char* output, unsigned output_size)
+{
+	if (microseconds < 1000) {
+		snprintf(output, output_size, "<1ms");
+		return;
+	} else if (microseconds < 1000000) {
+		snprintf(output, output_size, "%ldms", microseconds / 1000);
+		return;
+	} else {
+		snprintf(output, output_size, "%lds", microseconds / 1000000);
+		return;
+	}
+}
+
 static int handle_help_cmd(CONN_TYPE * conn, const char *arg, cmd_params_st *params)
 {
 	print_commands(1);

src/occtl/occtl.h

@@ -44,6 +44,9 @@ void print_pair_value(FILE *out, cmd_params_st *params, const char *name1, const
 void
 bytes2human(unsigned long bytes, char* output, unsigned output_size, const char* suffix);
 
+void
+time2human(uint64_t microseconds, char* output, unsigned output_size);
+
 char* search_for_id(unsigned idx, const char* match, int match_size);
 char* search_for_user(unsigned idx, const char* match, int match_size);
 void entries_add(void *pool, const char* user, unsigned user_size, unsigned id);

src/occtl/unix.c

@@ -252,9 +252,31 @@ int handle_status_cmd(struct unix_ctx *ctx, const char *arg, cmd_params_st *para
 		print_single_value_int(stdout, params, "IPs in ban list", rep->banned_ips, 1);
 		if (params && params->debug) {
 			print_single_value_int(stdout, params, "Sec-mod client entries", rep->secmod_client_entries, 1);
+#if defined(CAPTURE_LATENCY_SUPPORT)
 			print_single_value_int(stdout, params, "TLS DB entries", rep->stored_tls_sessions, 1);
+#else
+			print_single_value_int(stdout, params, "TLS DB entries", rep->stored_tls_sessions, 0);
+#endif
 		}
 
+#if defined(CAPTURE_LATENCY_SUPPORT)
+		if (rep->has_latency_sample_count) {
+			unsigned int median_latency = (unsigned int)(rep->latency_sample_count ? rep->latency_median_total / rep->latency_sample_count : 0);
+			unsigned int stdev_latency = (unsigned int)(rep->latency_sample_count ? rep->latency_rms_total / rep->latency_sample_count : 0);
+
+			time2human(median_latency, buf, sizeof(buf));
+			print_single_value(stdout, params, "Median latency", buf, 1);
+			if (HAVE_JSON(params)) 
+				print_single_value_int(stdout, params, "raw_median_latency", median_latency, 1);
+
+			time2human(stdev_latency, buf, sizeof(buf));
+			print_single_value(stdout, params, "STDEV latency", buf, 1);
+			if (HAVE_JSON(params)) 
+				print_single_value_int(stdout, params, "raw_stdev_latency", stdev_latency, 1);
+
+		}
+#endif
+
 		print_separator(stdout, params);
 		if (NO_JSON(params))
 			printf("Current stats period:\n");

src/vpn.h

@@ -421,6 +421,11 @@ struct main_server_st;
 #define MAX_GROUPNAME_SIZE MAX_USERNAME_SIZE
 #define MAX_SESSION_DATA_SIZE (4*1024)
 
+#if defined(CAPTURE_LATENCY_SUPPORT)
+#define LATENCY_SAMPLE_SIZE 1024
+#define LATENCY_WORKER_AGGREGATION_TIME 60
+#endif
+
 #define DEFAULT_CONFIG_ENTRIES 96
 
 #include <tun.h>

src/worker-latency.c

@@ -0,0 +1,145 @@
+/*
+ * Copyright (C) 2013-2018 Nikos Mavrogiannopoulos
+ * Copyright (C) 2015, 2016 Red Hat, Inc.
+ *
+ * This file is part of ocserv.
+ *
+ * ocserv is free software: you can redistribute it and/or modify it
+ * under the terms of the GNU General Public License as published by
+ * the Free Software Foundation, either version 2 of the License, or
+ * (at your option) any later version.
+ *
+ * ocserv is distributed in the hope that it will be useful, but
+ * WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
+ * General Public License for more details.
+ *
+ * You should have received a copy of the GNU General Public License
+ * along with this program.  If not, see <http://www.gnu.org/licenses/>.
+ */
+
+#include <config.h>
+
+#include <gnutls/gnutls.h>
+#include <gnutls/dtls.h>
+#include <math.h>
+#include <linux/net_tstamp.h>
+#include <linux/errqueue.h>
+#include <worker.h>
+#include <worker-latency.h>
+
+
+ssize_t dtls_pull_latency(gnutls_transport_ptr_t ptr, void *data, size_t size)
+{
+	int err;
+	dtls_transport_ptr *p = ptr;
+	p->rx_time.tv_sec = 0;
+	p->rx_time.tv_nsec = 0;
+
+	if (p->msg) {
+		ssize_t need = p->msg->data.len;
+		if (need > size) {
+			need = size;
+		}
+		memcpy(data, p->msg->data.data, need);
+
+		udp_fd_msg__free_unpacked(p->msg, NULL);
+		p->msg = NULL;
+		return need;
+	}
+
+	char controlbuf[1024];
+	struct cmsghdr * cmsg;
+
+	struct iovec io = {
+		.iov_base = data,
+		.iov_len = size,
+	};
+	struct msghdr hdr = {
+		.msg_iov = &io,
+		.msg_iovlen = 1,
+		.msg_control = controlbuf,
+		.msg_controllen = sizeof(controlbuf)
+	};
+	err = recvmsg(p->fd, &hdr, 0);
+	if (err >= 0) {
+		for (cmsg = CMSG_FIRSTHDR(&hdr); cmsg != NULL; cmsg = CMSG_NXTHDR(&hdr, cmsg)) {
+			struct scm_timestamping *tss = NULL;
+			if (cmsg->cmsg_level != SOL_SOCKET || cmsg->cmsg_type != SCM_TIMESTAMPING) {
+				continue;
+			}
+			tss = (struct scm_timestamping *) CMSG_DATA(cmsg);
+			p->rx_time = tss->ts[0];
+		}
+	}
+	return err;
+}
+
+
+void send_latency_stats_delta_to_main(worker_st * ws, time_t now)
+{
+	LatencyStatsDelta msg = LATENCY_STATS_DELTA__INIT;
+	
+	if (ws->latency.sample_set_count == 0) {
+		return;
+	}
+
+	msg.median_delta = ws->latency.median_total;
+	msg.rms_delta = ws->latency.rms_total;
+	msg.sample_count_delta = ws->latency.sample_set_count;
+	
+	ws->latency.median_total = 0;
+	ws->latency.rms_total = 0;
+	ws->latency.sample_set_count = 0;
+
+	send_msg_to_main(ws, CMD_LATENCY_STATS_DELTA, &msg,
+			 (pack_size_func) latency_stats_delta__get_packed_size,
+			 (pack_func) latency_stats_delta__pack);
+
+	ws->latency.last_stats_msg = now;
+}
+
+static int greater_than(const void * a, const void * b)
+{
+    const unsigned long lhs = *(const unsigned long*)a;
+    const unsigned long rhs = *(const unsigned long*)b;
+    return rhs - lhs;
+}
+
+void capture_latency_sample(struct worker_st* ws, struct timespec *processing_start_time)
+{
+	struct timespec now;
+	gettime_realtime(&now);
+	unsigned long sample = (unsigned long)timespec_sub_us(&now, processing_start_time);
+	if (ws->latency.next_sample == LATENCY_SAMPLE_SIZE) {
+		unsigned long median;
+		uint64_t total = 0;
+		long double sum_of_squares = 0;
+		uint64_t mean = 0;
+		uint64_t rms = 0;
+		int i;
+
+		ws->latency.next_sample = 0;
+		qsort(ws->latency.samples, LATENCY_SAMPLE_SIZE, sizeof(ws->latency.samples[0]), greater_than);
+		median = ws->latency.samples[LATENCY_SAMPLE_SIZE - 1];
+
+		for (i = 0; i < LATENCY_SAMPLE_SIZE; i ++) {
+			total += ws->latency.samples[i];
+		}
+
+		mean = total / LATENCY_SAMPLE_SIZE;
+		for (i = 0; i < LATENCY_SAMPLE_SIZE; i ++) {
+			long double delta = (long double)ws->latency.samples[i];
+			delta -= mean;
+			sum_of_squares += delta * delta;
+		}
+
+		rms = (uint64_t)sqrt(sum_of_squares / LATENCY_SAMPLE_SIZE);
+
+		(ws->latency.median_total) += median;
+		(ws->latency.rms_total) += rms;
+		(ws->latency.sample_set_count) ++;
+    }
+    ws->latency.samples[(ws->latency.next_sample)++] = sample;
+
+}

src/worker-latency.h

@@ -0,0 +1,23 @@
+/*
+ * Copyright (C) 2013-2018 Nikos Mavrogiannopoulos
+ * Copyright (C) 2015, 2016 Red Hat, Inc.
+ *
+ * This file is part of ocserv.
+ *
+ * ocserv is free software: you can redistribute it and/or modify it
+ * under the terms of the GNU General Public License as published by
+ * the Free Software Foundation, either version 2 of the License, or
+ * (at your option) any later version.
+ *
+ * ocserv is distributed in the hope that it will be useful, but
+ * WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
+ * General Public License for more details.
+ *
+ * You should have received a copy of the GNU General Public License
+ * along with this program.  If not, see <http://www.gnu.org/licenses/>.
+ */
+
+ssize_t dtls_pull_latency(gnutls_transport_ptr_t ptr, void *data, size_t size);
+void send_latency_stats_delta_to_main(worker_st * ws, time_t now);
+void capture_latency_sample(struct worker_st* ws, struct timespec *processing_start_time);

src/worker-vpn.c

@@ -47,6 +47,7 @@
 #include <worker-bandwidth.h>
 #include <signal.h>
 #include <poll.h>
+#include <math.h>
 
 #if defined(__linux__) && !defined(IPV6_PATHMTU)
 # define IPV6_PATHMTU 61
@@ -56,9 +57,14 @@
 #include "ipc.pb-c.h"
 #include <worker.h>
 #include <tlslib.h>
-
 #include <http_parser.h>
 
+#if defined(CAPTURE_LATENCY_SUPPORT)
+#include <linux/net_tstamp.h>
+#include <linux/errqueue.h>
+#include <worker-latency.h>
+#endif
+
 #define MIN_MTU(ws) (((ws)->vinfo.ipv6!=NULL)?1280:800)
 
 #define PERIODIC_CHECK_TIME 30
@@ -330,6 +336,9 @@ static int setup_dtls_connection(struct worker_st *ws)
 {
 	int ret;
 	gnutls_session_t session;
+#if defined(CAPTURE_LATENCY_SUPPORT)
+	int ts_socket_opt = SOF_TIMESTAMPING_RX_SOFTWARE | SOF_TIMESTAMPING_SOFTWARE;
+#endif
 
 	/* DTLS cookie verified.
 	 * Initialize session.
@@ -360,7 +369,11 @@ static int setup_dtls_connection(struct worker_st *ws)
 	}
 
 	gnutls_transport_set_push_function(session, dtls_push);
+#if defined(CAPTURE_LATENCY_SUPPORT)
+	gnutls_transport_set_pull_function(session, dtls_pull_latency);
+#else
 	gnutls_transport_set_pull_function(session, dtls_pull);
+#endif
 	gnutls_transport_set_pull_timeout_function(session, dtls_pull_timeout);
 	gnutls_transport_set_ptr(session, &ws->dtls_tptr);
 
@@ -371,6 +384,12 @@ static int setup_dtls_connection(struct worker_st *ws)
 
 	ws->udp_state = UP_HANDSHAKE;
 
+#if defined(CAPTURE_LATENCY_SUPPORT)
+	ret = setsockopt(ws->dtls_tptr.fd, SOL_SOCKET, SO_TIMESTAMPING, &ts_socket_opt, sizeof(ts_socket_opt));
+	if (ret == -1)
+		oclog(ws, LOG_DEBUG, "setsockopt(UDP, SO_TIMESTAMPING), failed.");
+#endif
+
 	/* Setup the fd settings */
 	if (WSCONFIG(ws)->output_buffer > 0) {
 		int t = MIN(2048, ws->link_mtu * WSCONFIG(ws)->output_buffer);
@@ -1224,6 +1243,12 @@ int periodic_check(worker_st * ws, struct timespec *tnow, unsigned dpd)
 		send_stats_to_secmod(ws, now, 0);
 	}
 
+#if defined(CAPTURE_LATENCY_SUPPORT)
+	if (now - ws->latency.last_stats_msg >= LATENCY_WORKER_AGGREGATION_TIME) {
+		send_latency_stats_delta_to_main(ws, now);
+	}
+#endif
+
 	/* check DPD. Otherwise exit */
 	if (ws->udp_state == UP_ACTIVE &&
 	    now - ws->last_msg_udp > DPD_TRIES * dpd && dpd > 0) {
@@ -2432,6 +2457,14 @@ static int connect_handler(worker_st * ws)
 				terminate_reason = REASON_ERROR;
 				goto exit;
 			}
+
+#if defined(CAPTURE_LATENCY_SUPPORT)
+			if (ws->dtls_tptr.rx_time.tv_sec != 0) {
+				capture_latency_sample(ws, &ws->dtls_tptr.rx_time);
+				ws->dtls_tptr.rx_time.tv_sec = 0;
+				ws->dtls_tptr.rx_time.tv_nsec = 0;
+			}
+#endif
 		}
 
 		/* read commands from command fd */

src/worker.h

@@ -161,6 +161,9 @@ typedef struct dtls_transport_ptr {
 	int fd;
 	UdpFdMsg *msg; /* holds the data of the first client hello */
 	int consumed;
+#if defined(CAPTURE_LATENCY_SUPPORT)
+	struct timespec rx_time;
+#endif
 } dtls_transport_ptr;
 
 /* Given a base MTU, this macro provides the DTLS plaintext data we can send;
@@ -295,6 +298,18 @@ typedef struct worker_st {
 	unsigned default_route;
 	
 	void *main_pool; /* to be used only on deinitialization */
+
+#if defined(CAPTURE_LATENCY_SUPPORT)
+	/* latency stats */
+	struct {
+		uint64_t median_total;
+		uint64_t rms_total;
+		uint64_t sample_set_count;
+		size_t next_sample;
+		time_t last_stats_msg;
+		uint32_t samples[LATENCY_SAMPLE_SIZE];
+	} latency;
+#endif
 } worker_st;
 
 void vpn_server(struct worker_st* ws);