Merge pull request #60 from paullouisageneau/dscp

IP Differenciated Services support

include/juice/juice.h

@@ -82,6 +82,7 @@ JUICE_EXPORT int juice_set_remote_description(juice_agent_t *agent, const char *
 JUICE_EXPORT int juice_add_remote_candidate(juice_agent_t *agent, const char *sdp);
 JUICE_EXPORT int juice_set_remote_gathering_done(juice_agent_t *agent);
 JUICE_EXPORT int juice_send(juice_agent_t *agent, const char *data, size_t size);
+JUICE_EXPORT int juice_send_diffserv(juice_agent_t *agent, const char *data, size_t size, int ds);
 JUICE_EXPORT juice_state_t juice_get_state(juice_agent_t *agent);
 JUICE_EXPORT int juice_get_selected_candidates(juice_agent_t *agent, char *local, size_t local_size,
                                                char *remote, size_t remote_size);

src/agent.c

@@ -72,8 +72,10 @@ juice_agent_t *agent_create(const juice_config_t *config) {
 	agent->state = JUICE_STATE_DISCONNECTED;
 	agent->mode = AGENT_MODE_UNKNOWN;
 	agent->sock = INVALID_SOCKET;
+	agent->send_ds = 0;
 
 	mutex_init(&agent->mutex, MUTEX_RECURSIVE);
+	mutex_init(&agent->send_mutex, 0);
 
 #ifdef NO_ATOMICS
 	agent->selected_entry = NULL;
@@ -100,7 +102,10 @@ void agent_do_destroy(juice_agent_t *agent) {
 	JLOG_DEBUG("Destroying agent");
 	if (agent->sock != INVALID_SOCKET)
 		closesocket(agent->sock);
+
 	mutex_destroy(&agent->mutex);
+	mutex_destroy(&agent->send_mutex);
+
 	free(agent);
 
 #ifdef _WIN32
@@ -299,8 +304,8 @@ int agent_set_remote_gathering_done(juice_agent_t *agent) {
 	return 0;
 }
 
-int agent_send(juice_agent_t *agent, const char *data, size_t size) {
-	// For performance reasons, this function is lock-free if the platform has atomics
+int agent_send(juice_agent_t *agent, const char *data, size_t size, int ds) {
+	// For performance reasons, do not lock the global mutex if the platform has atomics
 #ifdef NO_ATOMICS
 	mutex_lock(&agent->mutex);
 	agent_stun_entry_t *selected_entry = agent->selected_entry;
@@ -315,10 +320,22 @@ int agent_send(juice_agent_t *agent, const char *data, size_t size) {
 
 	if (!selected_entry) {
 		JLOG_ERROR("Send called before ICE is connected");
-		mutex_unlock(&agent->mutex);
 		return -1;
 	}
 
+	// Lock the send-specific mutex
+	mutex_lock(&agent->send_mutex);
+
+	if (agent->send_ds >= 0 && agent->send_ds != ds) {
+		JLOG_VERBOSE("Setting Differentiated Services field to 0x%X", ds);
+		if(udp_set_diffserv(agent->sock, ds) < 0)
+			agent->send_ds = -1; // disable for next time
+		else
+			agent->send_ds = ds;
+	}
+
+	JLOG_VERBOSE("Sending datagram, size=%d", size);
+
 	const addr_record_t *record = &selected_entry->record;
 #if defined(_WIN32) || defined(__APPLE__)
 	addr_record_t tmp = *record;
@@ -330,6 +347,8 @@ int agent_send(juice_agent_t *agent, const char *data, size_t size) {
 #endif
 	if (ret < 0 && sockerrno != SEAGAIN && sockerrno != SEWOULDBLOCK)
 		JLOG_WARN("Send failed, errno=%d", sockerrno);
+
+	mutex_unlock(&agent->send_mutex);
 	return ret;
 }
 

src/agent.h

@@ -106,24 +106,31 @@ struct juice_agent {
 	socket_t sock;
 	thread_t thread;
 	mutex_t mutex;
-	uint64_t ice_tiebreaker;
+
 	ice_description_t local;
 	ice_description_t remote;
+
 	ice_candidate_pair_t candidate_pairs[MAX_CANDIDATE_PAIRS_COUNT];
 	ice_candidate_pair_t *ordered_pairs[MAX_CANDIDATE_PAIRS_COUNT];
 	ice_candidate_pair_t *selected_pair;
 	int candidate_pairs_count;
+
 	agent_stun_entry_t entries[MAX_STUN_ENTRIES_COUNT];
 	int entries_count;
 #ifdef NO_ATOMICS
 	agent_stun_entry_t *volatile selected_entry;
 #else
 	_Atomic(agent_stun_entry_t *) selected_entry;
 #endif
+
+	uint64_t ice_tiebreaker;
 	timestamp_t fail_timestamp;
 	bool gathering_done;
 	bool thread_started;
 	bool thread_stopped;
+
+	mutex_t send_mutex;
+	int send_ds;
 };
 
 juice_agent_t *agent_create(const juice_config_t *config);
@@ -134,7 +141,7 @@ int agent_get_local_description(juice_agent_t *agent, char *buffer, size_t size)
 int agent_set_remote_description(juice_agent_t *agent, const char *sdp);
 int agent_add_remote_candidate(juice_agent_t *agent, const char *sdp);
 int agent_set_remote_gathering_done(juice_agent_t *agent);
-int agent_send(juice_agent_t *agent, const char *data, size_t size);
+int agent_send(juice_agent_t *agent, const char *data, size_t size, int ds);
 juice_state_t agent_get_state(juice_agent_t *agent);
 int agent_get_selected_candidate_pair(juice_agent_t *agent, ice_candidate_t *local,
                                       ice_candidate_t *remote);

src/juice.c

@@ -88,7 +88,17 @@ JUICE_EXPORT int juice_send(juice_agent_t *agent, const char *data, size_t size)
 	if (!agent || (!data && size))
 		return JUICE_ERR_INVALID;
 
-	if (agent_send(agent, data, size) < 0)
+	if (agent_send(agent, data, size, 0) < 0)
+		return JUICE_ERR_FAILED;
+
+	return JUICE_ERR_SUCCESS;
+}
+
+JUICE_EXPORT int juice_send_diffserv(juice_agent_t *agent, const char *data, size_t size, int ds) {
+	if (!agent || (!data && size))
+		return JUICE_ERR_INVALID;
+
+	if (agent_send(agent, data, size, ds) < 0)
 		return JUICE_ERR_FAILED;
 
 	return JUICE_ERR_SUCCESS;

src/udp.c

@@ -104,7 +104,7 @@ socket_t udp_create_socket(const udp_socket_config_t *config) {
 
 	ctl_t b = 1;
 	if (ioctlsocket(sock, FIONBIO, &b)) {
-		JLOG_ERROR("setting non-blocking mode on UDP socket failed, errno=%d", sockerrno);
+		JLOG_ERROR("Setting non-blocking mode on UDP socket failed, errno=%d", sockerrno);
 		goto error;
 	}
 
@@ -142,6 +142,52 @@ socket_t udp_create_socket(const udp_socket_config_t *config) {
 	return INVALID_SOCKET;
 }
 
+int udp_set_diffserv(socket_t sock, int ds) {
+#ifdef _WIN32
+	// IP_TOS has been intentionally broken on Windows in favor of a convoluted proprietary
+	// mechanism called qWave. Thank you Microsoft!
+	// TODO: Investigate if DSCP can be still set directly without administrator flow configuration.
+	JLOG_INFO("IP Differentiated Services are not supported on Windows");
+	return -1;
+#else
+	addr_record_t name;
+	name.len = sizeof(name.addr);
+	if (getsockname(sock, (struct sockaddr *)&name.addr, &name.len) < 0) {
+		JLOG_WARN("getsockname failed, errno=%d", sockerrno);
+		return -1;
+	}
+
+	switch (name.addr.ss_family) {
+	case AF_INET:
+#ifdef IP_TOS
+		if (setsockopt(sock, IPPROTO_IP, IP_TOS, &ds, sizeof(ds)) < 0) {
+			JLOG_WARN("Setting IP ToS failed, errno=%d", sockerrno);
+			return -1;
+		}
+		return 0;
+#else
+		JLOG_INFO("Setting IP ToS is not supported");
+		return -1;
+#endif
+
+	case AF_INET6:
+#ifdef IPV6_TCLASS
+		if (setsockopt(sock, IPPROTO_IPV6, IPV6_TCLASS, &ds, sizeof(ds)) < 0) {
+			JLOG_WARN("Setting IPv6 traffic class failed, errno=%d", sockerrno);
+			return -1;
+		}
+		return 0;
+#else
+		JLOG_INFO("Setting IPv6 traffic class is not supported");
+		return -1;
+#endif
+
+	default:
+		return -1;
+	}
+#endif
+}
+
 uint16_t udp_get_port(socket_t sock) {
 	addr_record_t record;
 	record.len = sizeof(record.addr);
@@ -395,7 +441,7 @@ int udp_get_addrs(socket_t sock, addr_record_t *records, size_t count) {
 				break;
 			}
 		}
-	    for (struct addrinfo *ai = ai_list; ai; ai = ai->ai_next) {
+		for (struct addrinfo *ai = ai_list; ai; ai = ai->ai_next) {
 			if (!addr_is_local(ai->ai_addr) &&
 			    !(has_temp_inet6 && !addr_is_temp_inet6(ai->ai_addr))) {
 				if (!has_duplicate_addr(ai->ai_addr, records, current - records)) {

src/udp.h

@@ -30,6 +30,7 @@ typedef struct udp_socket_config {
 } udp_socket_config_t;
 
 socket_t udp_create_socket(const udp_socket_config_t *config);
+int udp_set_diffserv(socket_t sock, int ds);
 uint16_t udp_get_port(socket_t sock);
 int udp_get_local_addr(socket_t sock, addr_record_t *record);
 int udp_get_addrs(socket_t sock, addr_record_t *records, size_t count);