update kernel read timer for multiple interfaces

daemon/call.c

@@ -42,7 +42,7 @@
 struct iterator_helper {
 	GSList			*del_timeout;
 	GSList			*del_scheduled;
-	struct stream_fd	*ports[0x10000];
+	GHashTable		*addr_sfd;
 };
 struct xmlrpc_helper {
 	enum xmlrpc_format fmt;
@@ -224,10 +224,9 @@ static void call_timer_iterator(void *key, void *val, void *ptr) {
 		if (css == CSS_ICE)
 			timestamp = &ps->media->ice_agent->last_activity;
 
-		if (hlp->ports[sfd->socket.local.port])
+		if (g_hash_table_contains(hlp->addr_sfd, &sfd->socket.local))
 			goto next;
-		hlp->ports[sfd->socket.local.port] = sfd;
-		obj_hold(sfd);
+		g_hash_table_insert(hlp->addr_sfd, &sfd->socket.local, obj_get(sfd));
 
 no_sfd:
 		if (good)
@@ -461,7 +460,7 @@ destroy:
 static void callmaster_timer(void *ptr) {
 	struct callmaster *m = ptr;
 	struct iterator_helper hlp;
-	GList *i;
+	GList *i, *l;
 	struct rtpengine_list_entry *ke;
 	struct packet_stream *ps, *sink;
 	struct stats tmpstats;
@@ -469,8 +468,10 @@ static void callmaster_timer(void *ptr) {
 	struct stream_fd *sfd;
 	struct rtp_stats *rs;
 	unsigned int pt;
+	endpoint_t ep;
 
 	ZERO(hlp);
+	hlp.addr_sfd = g_hash_table_new(g_endpoint_hash, g_endpoint_eq);
 
 	rwlock_lock_r(&m->hashlock);
 	g_hash_table_foreach(m->callhash, call_timer_iterator, &hlp);
@@ -488,7 +489,8 @@ static void callmaster_timer(void *ptr) {
 	while (i) {
 		ke = i->data;
 
-		sfd = hlp.ports[ke->target.local.port]; // XXX fix for multiple addresses
+		kernel2endpoint(&ep, &ke->target.local);
+		sfd = g_hash_table_lookup(hlp.addr_sfd, &ep);
 		if (!sfd)
 			goto next;
 
@@ -569,16 +571,18 @@ static void callmaster_timer(void *ptr) {
 			redis_update(ps->call, m->conf.redis);
 
 next:
-		hlp.ports[ke->target.local.port] = NULL;
+		g_hash_table_remove(hlp.addr_sfd, &ep);
 		g_slice_free1(sizeof(*ke), ke);
 		i = g_list_delete_link(i, i);
 		if (sfd)
 			obj_put(sfd);
 	}
 
-	for (j = 0; j < (sizeof(hlp.ports) / sizeof(*hlp.ports)); j++)
-		if (hlp.ports[j])
-			obj_put(hlp.ports[j]);
+	l = g_hash_table_get_values(hlp.addr_sfd);
+	for (i = l; i; i = i->next)
+		obj_put((struct stream_fd *) i->data);
+	g_list_free(l);
+	g_hash_table_destroy(hlp.addr_sfd);
 
 	kill_calls_timer(hlp.del_scheduled, NULL);
 	kill_calls_timer(hlp.del_timeout, m);

daemon/media_socket.c

@@ -623,17 +623,8 @@ static int __k_srtp_decrypt(struct rtpengine_srtp *s, struct packet_stream *stre
 	return __k_srtp_crypt(s, &stream->selected_sfd->crypto);
 }
 
-/* XXX unify this */
-INLINE void __re_address_translate(struct re_address *o, const sockaddr_t *address) {
-	o->family = address->family->af;
-	if (o->family == AF_INET)
-		o->u.ipv4 = address->u.ipv4.s_addr;
-	else
-		memcpy(o->u.ipv6, &address->u.ipv6, sizeof(o->u.ipv6));
-}
 INLINE void __re_address_translate_ep(struct re_address *o, const endpoint_t *ep) {
-	__re_address_translate(o, &ep->address);
-	o->port = ep->port;
+	ep->address.family->endpoint2kernel(o, ep);
 }
 
 static int __rtp_stats_pt_sort(const void *ap, const void *bp) {
@@ -653,7 +644,6 @@ void kernelize(struct packet_stream *stream) {
 	struct call *call = stream->call;
 	struct callmaster *cm = call->callmaster;
 	struct packet_stream *sink = NULL;
-	const struct local_intf *ifa;
 	const char *nk_warn_msg;
 
 	if (PS_ISSET(stream, KERNELIZED))
@@ -706,10 +696,8 @@ void kernelize(struct packet_stream *stream) {
 	reti.dtls = MEDIA_ISSET(stream->media, DTLS);
 	reti.stun = stream->media->ice_agent ? 1 : 0;
 
-	ifa = sink->selected_sfd->local_intf;
 	__re_address_translate_ep(&reti.dst_addr, &sink->endpoint);
-	__re_address_translate(&reti.src_addr, &ifa->spec->address.addr);
-	reti.src_addr.port = sink->selected_sfd->socket.local.port;
+	__re_address_translate_ep(&reti.src_addr, &sink->selected_sfd->socket.local);
 	reti.ssrc = sink->crypto.ssrc;
 
 	stream->handler->in->kernel(&reti.decrypt, stream);

daemon/socket.c

@@ -4,6 +4,7 @@
 #include <errno.h>
 #include "str.h"
 #include "media_socket.h"
+#include "xt_RTPENGINE.h"
 
 static int __ip4_addr_parse(sockaddr_t *dst, const char *src);
 static int __ip6_addr_parse(sockaddr_t *dst, const char *src);
@@ -29,6 +30,10 @@ static ssize_t __ip_sendmsg(socket_t *s, struct msghdr *mh, const endpoint_t *ep
 static ssize_t __ip_sendto(socket_t *s, const void *buf, size_t len, const endpoint_t *ep);
 static int __ip4_tos(socket_t *, unsigned int);
 static int __ip6_tos(socket_t *, unsigned int);
+static void __ip4_endpoint2kernel(struct re_address *, const endpoint_t *);
+static void __ip6_endpoint2kernel(struct re_address *, const endpoint_t *);
+static void __ip4_kernel2endpoint(endpoint_t *ep, const struct re_address *ra);
+static void __ip6_kernel2endpoint(endpoint_t *ep, const struct re_address *ra);
 
 
 
@@ -61,6 +66,8 @@ static struct socket_family __socket_families[__SF_LAST] = {
 		.sendmsg		= __ip_sendmsg,
 		.sendto			= __ip_sendto,
 		.tos			= __ip4_tos,
+		.endpoint2kernel	= __ip4_endpoint2kernel,
+		.kernel2endpoint	= __ip4_kernel2endpoint,
 	},
 	[SF_IP6] = {
 		.af			= AF_INET6,
@@ -83,6 +90,8 @@ static struct socket_family __socket_families[__SF_LAST] = {
 		.sendmsg		= __ip_sendmsg,
 		.sendto			= __ip_sendto,
 		.tos			= __ip6_tos,
+		.endpoint2kernel	= __ip6_endpoint2kernel,
+		.kernel2endpoint	= __ip6_kernel2endpoint,
 	},
 };
 
@@ -249,7 +258,32 @@ static int __ip6_tos(socket_t *s, unsigned int tos) {
 	setsockopt(s->fd, IPPROTO_IPV6, IPV6_TCLASS, &tos, sizeof(tos));
 	return 0;
 }
-
+static void __ip4_endpoint2kernel(struct re_address *ra, const endpoint_t *ep) {
+	ra->family = AF_INET;
+	ra->u.ipv4 = ep->address.u.ipv4.s_addr;
+	ra->port = ep->port;
+}
+static void __ip6_endpoint2kernel(struct re_address *ra, const endpoint_t *ep) {
+	ra->family = AF_INET6;
+	memcpy(ra->u.ipv6, &ep->address.u.ipv6, sizeof(ra->u.ipv6));
+	ra->port = ep->port;
+}
+void kernel2endpoint(endpoint_t *ep, const struct re_address *ra) {
+	if (ra->family == AF_INET)
+		ep->address.family = __get_socket_family_enum(SF_IP4);
+	else if (ra->family == AF_INET6)
+		ep->address.family = __get_socket_family_enum(SF_IP6);
+	else
+		abort();
+	ep->port = ra->port;
+	ep->address.family->kernel2endpoint(ep, ra);
+}
+static void __ip4_kernel2endpoint(endpoint_t *ep, const struct re_address *ra) {
+	ep->address.u.ipv4.s_addr = ra->u.ipv4;
+}
+static void __ip6_kernel2endpoint(endpoint_t *ep, const struct re_address *ra) {
+	memcpy(&ep->address.u.ipv6, ra->u.ipv6, sizeof(ep->address.u.ipv6));
+}
 
 
 

daemon/socket.h

@@ -21,6 +21,7 @@ struct socket_type;
 struct socket_family;
 struct endpoint;
 struct socket;
+struct re_address;
 
 typedef struct socket_address sockaddr_t;
 typedef struct endpoint endpoint_t;
@@ -62,6 +63,8 @@ struct socket_family {
 	ssize_t				(*sendmsg)(socket_t *, struct msghdr *, const endpoint_t *);
 	ssize_t				(*sendto)(socket_t *, const void *, size_t, const endpoint_t *);
 	int				(*tos)(socket_t *, unsigned int);
+	void				(*endpoint2kernel)(struct re_address *, const endpoint_t *);
+	void				(*kernel2endpoint)(endpoint_t *, const struct re_address *);
 };
 struct socket_address {
 	sockfamily_t			*family;
@@ -180,6 +183,7 @@ int sockaddr_parse_any(sockaddr_t *dst, const char *src);
 int sockaddr_parse_any_str(sockaddr_t *dst, const str *src);
 int sockaddr_parse_str(sockaddr_t *dst, sockfamily_t *fam, const str *src);
 int endpoint_parse_any(endpoint_t *, const char *);
+void kernel2endpoint(endpoint_t *ep, const struct re_address *ra);
 
 unsigned int sockaddr_hash(const sockaddr_t *);
 int sockaddr_eq(const sockaddr_t *, const sockaddr_t *); /* true/false */