added bandwidth usage reporting for bandwidth used by peers

src/apps/relay/ns_ioalib_engine_impl.c

@@ -3562,6 +3562,13 @@ void turn_report_allocation_delete(void *a)
 						snprintf(key, sizeof(key), "turn/user/%s/allocation/%018llu/total_traffic", (char*)ss->username, (unsigned long long)ss->id);
 					}
 					send_message_to_redis(e->rch, "publish", key, "rcvp=%lu, rcvb=%lu, sentp=%lu, sentb=%lu", (unsigned long)(ss->t_received_packets), (unsigned long)(ss->t_received_bytes), (unsigned long)(ss->t_sent_packets), (unsigned long)(ss->t_sent_bytes));
+					if (ss->realm_options.name[0]) {
+						snprintf(key, sizeof(key), "turn/realm/%s/user/%s/allocation/%018llu/total_traffic/peer", ss->realm_options.name, (char*)ss->username, (unsigned long long)(ss->id));
+					}
+					else {
+						snprintf(key, sizeof(key), "turn/user/%s/allocation/%018llu/total_traffic/peer", (char*)ss->username, (unsigned long long)(ss->id));
+					}
+					send_message_to_redis(e->rch, "publish", key, "rcvp=%lu, rcvb=%lu, sentp=%lu, sentb=%lu", (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));
 				}
 #endif
 			}
@@ -3576,9 +3583,10 @@ void turn_report_session_usage(void *session, int force_invalid)
 		turn_turnserver *server = (turn_turnserver*)ss->server;
 		if(server && (ss->received_packets || ss->sent_packets || force_invalid)) {
 			ioa_engine_handle e = turn_server_get_engine(server);
-			if(((ss->received_packets+ss->sent_packets)&2047)==0 || force_invalid) {
+			if(((ss->received_packets+ss->sent_packets+ss->peer_received_packets+ss->peer_sent_packets)&4095)==0 || force_invalid) {
 				if(e && e->verbose) {
 					TURN_LOG_FUNC(TURN_LOG_LEVEL_INFO,"session %018llu: usage: realm=<%s>, username=<%s>, rp=%lu, rb=%lu, sp=%lu, sb=%lu\n", (unsigned long long)(ss->id), (char*)ss->realm_options.name, (char*)ss->username, (unsigned long)(ss->received_packets), (unsigned long)(ss->received_bytes),(unsigned long)(ss->sent_packets),(unsigned long)(ss->sent_bytes));
+					TURN_LOG_FUNC(TURN_LOG_LEVEL_INFO, "session %018llu: peer usage: realm=<%s>, username=<%s>, rp=%lu, rb=%lu, sp=%lu, sb=%lu\n", (unsigned long long)(ss->id), (char*)ss->realm_options.name, (char*)ss->username, (unsigned long)(ss->peer_received_packets), (unsigned long)(ss->peer_received_bytes), (unsigned long)(ss->peer_sent_packets), (unsigned long)(ss->peer_sent_bytes));
 				}
 #if !defined(TURN_NO_HIREDIS)
 				{
@@ -3589,12 +3597,23 @@ void turn_report_session_usage(void *session, int force_invalid)
 						snprintf(key,sizeof(key),"turn/user/%s/allocation/%018llu/traffic",(char*)ss->username, (unsigned long long)(ss->id));
 					}
 					send_message_to_redis(e->rch, "publish", key, "rcvp=%lu, rcvb=%lu, sentp=%lu, sentb=%lu",(unsigned long)(ss->received_packets), (unsigned long)(ss->received_bytes),(unsigned long)(ss->sent_packets),(unsigned long)(ss->sent_bytes));
+					if (ss->realm_options.name[0]) {
+						snprintf(key, sizeof(key), "turn/realm/%s/user/%s/allocation/%018llu/traffic/peer", ss->realm_options.name, (char*)ss->username, (unsigned long long)(ss->id));
+					}
+					else {
+						snprintf(key, sizeof(key), "turn/user/%s/allocation/%018llu/traffic/peer", (char*)ss->username, (unsigned long long)(ss->id));
+					}
+					send_message_to_redis(e->rch, "publish", key, "rcvp=%lu, rcvb=%lu, sentp=%lu, sentb=%lu", (unsigned long)(ss->peer_received_packets), (unsigned long)(ss->peer_received_bytes), (unsigned long)(ss->peer_sent_packets), (unsigned long)(ss->peer_sent_bytes));
 				}
 #endif
 				ss->t_received_packets += ss->received_packets;
 				ss->t_received_bytes += ss->received_bytes;
 				ss->t_sent_packets += ss->sent_packets;
 				ss->t_sent_bytes += ss->sent_bytes;
+				ss->t_peer_received_packets += ss->peer_received_packets;
+				ss->t_peer_received_bytes += ss->peer_received_bytes;
+				ss->t_peer_sent_packets += ss->peer_sent_packets;
+				ss->t_peer_sent_bytes += ss->peer_sent_bytes;
 
 				{
 					turn_time_t ct = get_turn_server_time(server);
@@ -3603,6 +3622,9 @@ void turn_report_session_usage(void *session, int force_invalid)
 						ss->received_rate = (uint32_t)(ss->t_received_bytes / ct);
 						ss->sent_rate = (uint32_t)(ss->t_sent_bytes / ct);
 						ss->total_rate = ss->received_rate + ss->sent_rate;
+						ss->peer_received_rate = (uint32_t)(ss->t_peer_received_bytes / ct);
+						ss->peer_sent_rate = (uint32_t)(ss->t_peer_sent_bytes / ct);
+						ss->peer_total_rate = ss->peer_received_rate + ss->peer_sent_rate;
 					}
 				}
 
@@ -3612,6 +3634,10 @@ void turn_report_session_usage(void *session, int force_invalid)
 				ss->received_bytes=0;
 				ss->sent_packets=0;
 				ss->sent_bytes=0;
+				ss->peer_received_packets = 0;
+				ss->peer_received_bytes = 0;
+				ss->peer_sent_packets = 0;
+				ss->peer_sent_bytes = 0;
 			}
 		}
 	}

src/server/ns_turn_server.c

@@ -504,10 +504,33 @@ int turn_session_info_copy_from(struct turn_session_info* tsi, ts_ur_super_sessi
 			else
 				tsi->sent_bytes = ss->sent_bytes;
 
+			if (ss->t_peer_received_packets > ss->peer_received_packets)
+				tsi->peer_received_packets = ss->t_peer_received_packets;
+			else
+				tsi->peer_received_packets = ss->peer_received_packets;
+
+			if (ss->t_peer_sent_packets > ss->peer_sent_packets)
+				tsi->peer_sent_packets = ss->t_peer_sent_packets;
+			else
+				tsi->peer_sent_packets = ss->peer_sent_packets;
+
+			if (ss->t_peer_received_bytes > ss->peer_received_bytes)
+				tsi->peer_received_bytes = ss->t_peer_received_bytes;
+			else
+				tsi->peer_received_bytes = ss->peer_received_bytes;
+
+			if (ss->t_peer_sent_bytes > ss->peer_sent_bytes)
+				tsi->peer_sent_bytes = ss->t_peer_sent_bytes;
+			else
+				tsi->peer_sent_bytes = ss->peer_sent_bytes;
+
 			{
 				tsi->received_rate = ss->received_rate;
 				tsi->sent_rate = ss->sent_rate;
 				tsi->total_rate = tsi->received_rate + tsi->sent_rate;
+				tsi->peer_received_rate = ss->peer_received_rate;
+				tsi->peer_sent_rate = ss->peer_sent_rate;
+				tsi->peer_total_rate = tsi->peer_received_rate + tsi->peer_sent_rate;
 			}
 
 			tsi->is_mobile = ss->is_mobile;
@@ -1904,12 +1927,20 @@ static void tcp_peer_input_handler(ioa_socket_handle s, int event_type, ioa_net_
 
 	uint32_t bytes = (uint32_t)ioa_network_buffer_get_size(nbh);
 
+	if (ss) {
+		++(ss->peer_received_packets);
+		ss->peer_received_bytes += bytes;
+	}
+
 	int ret = send_data_from_ioa_socket_nbh(tc->client_s, NULL, nbh, TTL_IGNORE, TOS_IGNORE, NULL);
 	if (ret < 0) {
 		set_ioa_socket_tobeclosed(s);
 	} else if(ss) {
 		++(ss->sent_packets);
 		ss->sent_bytes += bytes;
+	}
+
+	if (ss) {
 		turn_report_session_usage(ss, 0);
 	}
 }
@@ -1938,17 +1969,23 @@ static void tcp_client_input_handler_rfc6062data(ioa_socket_handle s, int event_
 	ioa_network_buffer_handle nbh = in_buffer->nbh;
 	in_buffer->nbh = NULL;
 
+	uint32_t bytes = (uint32_t)ioa_network_buffer_get_size(nbh);
 	if(ss) {
-		uint32_t bytes = (uint32_t)ioa_network_buffer_get_size(nbh);
 		++(ss->received_packets);
 		ss->received_bytes += bytes;
 	}
 
-	int ret = send_data_from_ioa_socket_nbh(tc->peer_s, NULL, nbh, TTL_IGNORE, TOS_IGNORE, NULL);
+	int skip = 0;
+	int ret = send_data_from_ioa_socket_nbh(tc->peer_s, NULL, nbh, TTL_IGNORE, TOS_IGNORE, &skip);
 	if (ret < 0) {
 		set_ioa_socket_tobeclosed(s);
 	}
 
+	if (!skip) {
+		++(ss->peer_sent_packets);
+		ss->peer_sent_bytes += bytes;
+	}
+
 	turn_report_session_usage(ss, 0);
 }
 
@@ -2998,7 +3035,13 @@ static int handle_turn_send(turn_turnserver *server, ts_ur_super_session *ss,
 					ioa_network_buffer_set_size(nbh,len);
 				}
 				ioa_network_buffer_header_init(nbh);
-				send_data_from_ioa_socket_nbh(get_relay_socket_ss(ss,peer_addr.ss.sa_family), &peer_addr, nbh, in_buffer->recv_ttl-1, in_buffer->recv_tos, NULL);
+				int skip = 0;
+				send_data_from_ioa_socket_nbh(get_relay_socket_ss(ss,peer_addr.ss.sa_family), &peer_addr, nbh, in_buffer->recv_ttl-1, in_buffer->recv_tos, &skip);
+				if (!skip) {
+					++(ss->peer_sent_packets);
+					ss->peer_sent_bytes += len;
+					turn_report_session_usage(ss, 0);
+				}
 				in_buffer->nbh = NULL;
 			}
 
@@ -4087,7 +4130,15 @@ static int write_to_peerchannel(ts_ur_super_session* ss, uint16_t chnum, ioa_net
 
 			ioa_network_buffer_header_init(nbh);
 
-			rc = send_data_from_ioa_socket_nbh(get_relay_socket_ss(ss, chn->peer_addr.ss.sa_family), &(chn->peer_addr), nbh, in_buffer->recv_ttl-1, in_buffer->recv_tos, NULL);
+			int skip = 0;
+			rc = send_data_from_ioa_socket_nbh(get_relay_socket_ss(ss, chn->peer_addr.ss.sa_family), &(chn->peer_addr), nbh, in_buffer->recv_ttl-1, in_buffer->recv_tos, &skip);
+
+			if (!skip) {
+				++(ss->peer_sent_packets);
+				ss->peer_sent_bytes += (uint32_t)ioa_network_buffer_get_size(in_buffer->nbh);
+				turn_report_session_usage(ss, 0);
+			}
+
 			in_buffer->nbh = NULL;
 		}
 	}
@@ -4718,6 +4769,9 @@ static void peer_input_handler(ioa_socket_handle s, int event_type,
 			(int)(ioa_network_buffer_get_capacity_udp() - offset));
 
 	if (ilen >= 0) {
+		++(ss->peer_received_packets);
+		ss->peer_received_bytes += ilen;
+		turn_report_session_usage(ss, 0);
 
 		allocation* a = get_allocation_ss(ss);
 		if (is_allocation_valid(a)) {

src/server/ns_turn_session.h

@@ -101,6 +101,17 @@ struct _ts_ur_super_session {
   uint64_t received_rate;
   size_t sent_rate;
   size_t total_rate;
+  uint32_t peer_received_packets;
+  uint32_t peer_sent_packets;
+  uint32_t peer_received_bytes;
+  uint32_t peer_sent_bytes;
+  uint32_t t_peer_received_packets;
+  uint32_t t_peer_sent_packets;
+  uint32_t t_peer_received_bytes;
+  uint32_t t_peer_sent_bytes;
+  uint64_t peer_received_rate;
+  size_t peer_sent_rate;
+  size_t peer_total_rate;
   /* Mobile */
   int is_mobile;
   mobile_id_t mobile_id;
@@ -143,6 +154,13 @@ struct turn_session_info {
 	uint32_t received_rate;
 	uint32_t sent_rate;
 	uint32_t total_rate;
+	uint64_t peer_received_packets;
+	uint64_t peer_sent_packets;
+	uint64_t peer_received_bytes;
+	uint64_t peer_sent_bytes;
+	uint32_t peer_received_rate;
+	uint32_t peer_sent_rate;
+	uint32_t peer_total_rate;
 /* Mobile */
 	int is_mobile;
 /* Peers */