TT#37394 support SO_INCOMING_CPU

Change-Id: I8017929519cc3b721272cc122bee4fdc36c7317a

daemon/call.c

@@ -72,6 +72,8 @@ GHashTable *rtpe_callhash;
 struct call_iterator_list rtpe_call_iterators[NUM_CALL_ITERATORS];
 static struct mqtt_timer *global_mqtt_timer;
 
+unsigned int call_socket_cpu_affinity = 0;
+
 /* ********** */
 
 static void __monologue_destroy(struct call_monologue *monologue, int recurse);
@@ -896,6 +898,11 @@ alloc:
 
 		while ((sock = g_queue_pop_head(&il->list))) {
 			set_tos(sock, media->call->tos);
+			if (media->call->cpu_affinity >= 0) {
+				if (socket_cpu_affinity(sock, media->call->cpu_affinity))
+					ilog(LOG_ERR | LOG_FLAG_LIMIT, "Failed to set socket CPU "
+							"affinity: %s", strerror(errno));
+			}
 			sfd = stream_fd_new(sock, media->call, il->local_intf);
 			g_queue_push_tail(&em_il->list, sfd); /* not referenced */
 		}
@@ -3114,6 +3121,10 @@ static struct call *call_create(const str *callid) {
 	c->created = rtpe_now;
 	c->dtls_cert = dtls_cert();
 	c->tos = rtpe_config.default_tos;
+	if (rtpe_config.cpu_affinity)
+		c->cpu_affinity = call_socket_cpu_affinity++ % rtpe_config.cpu_affinity;
+	else
+		c->cpu_affinity = -1;
 
 	for (int i = 0; i < NUM_CALL_ITERATORS; i++) {
 		mutex_init(&c->iterator[i].next_lock);

daemon/main.c

@@ -557,6 +557,9 @@ static void options(int *argc, char ***argv) {
 		{ "mqtt-publish-scope",0,0,G_OPTION_ARG_STRING,	&mqtt_publish_scope,	"Scope for published mosquitto messages","global|call|media"},
 #endif
 		{ "mos",0,0,		G_OPTION_ARG_STRING,	&mos,		"Type of MOS calculation","CQ|LQ"},
+#ifdef SO_INCOMING_CPU
+		{ "socket-cpu-affinity",0,0,G_OPTION_ARG_INT,	&rtpe_config.cpu_affinity,"CPU affinity for media sockets","INT"},
+#endif
 
 		{ NULL, }
 	};
@@ -1129,6 +1132,12 @@ no_kernel:
 	if (rtpe_config.num_threads < 1)
 		rtpe_config.num_threads = num_cpu_cores(4);
 
+	if (rtpe_config.cpu_affinity < 0) {
+		rtpe_config.cpu_affinity = num_cpu_cores(0);
+		if (rtpe_config.cpu_affinity <= 0)
+			die("Number of CPU cores is unknown, cannot auto-set socket CPU affinity");
+	}
+
 	if (websocket_init())
 		die("Failed to init websocket listener");
 

daemon/rtpengine.pod

@@ -985,6 +985,17 @@ RTT into account and therefore requires peers to correctly send RTCP. If set to
 B<LQ> (listening quality) RTT is ignored, allowing a MOS to be calculated in
 the absence of RTCP.
 
+=item B<--socket-cpu-affinity=>I<INT>
+
+Enables setting the socket CPU affinity via the B<SO_INCOMING_CPU> socket
+option if available. The default value is zero which disables this feature. If
+set to a positive number then the CPU affinity for all sockets belonging to the
+same call will be set to the same value. The number specifies the upper limit
+of the affinity to be set, and values will be used in a round-robin fashion
+(e.g. if set to B<8> then the values B<0> through B<7> will be used to set the
+affinity). If this option is set to a negative number, then the number of
+available CPU cores will be used.
+
 =back
 
 =head1 INTERFACES

include/call.h

@@ -502,6 +502,7 @@ struct call {
 	str			metadata;
 
 	struct call_iterator_entry iterator[NUM_CALL_ITERATORS];
+	int			cpu_affinity;
 
 	// ipv4/ipv6 media flags
 	unsigned int		is_ipv4_media_offer:1;

include/main.h

@@ -147,6 +147,7 @@ struct rtpengine_config {
 		MOS_CQ = 0,
 		MOS_LQ,
 	}			mos;
+	int			cpu_affinity;
 };
 
 

lib/socket.h

@@ -192,6 +192,14 @@ INLINE void ipv6only(int fd, int yn) {
 	// coverity[check_return : FALSE]
 	setsockopt(fd, IPPROTO_IPV6, IPV6_V6ONLY, &yn, sizeof(yn));
 }
+INLINE int socket_cpu_affinity(socket_t *s, int cpu) {
+#ifndef SO_INCOMING_CPU
+	errno = ENOTSUP;
+	return -1;
+#else
+	return setsockopt(s->fd, SOL_SOCKET, SO_INCOMING_CPU, &cpu, sizeof(cpu));
+#endif
+}