Commit 2072c30a authored by Christophe Massiot's avatar Christophe Massiot

aggregartp: Add support for null weights

parent f83d9aea
...@@ -2,6 +2,7 @@ Changes between 2.2 and 2.3: ...@@ -2,6 +2,7 @@ Changes between 2.2 and 2.3:
---------------------------- ----------------------------
* Fix potential segfault in aggregartp * Fix potential segfault in aggregartp
* Do not do retx in reordertp with multicast inputs * Do not do retx in reordertp with multicast inputs
* Add support for null weights in aggregartp and reordertp
Changes between 2.1 and 2.2: Changes between 2.1 and 2.2:
---------------------------- ----------------------------
......
...@@ -86,6 +86,7 @@ static void usage(void) ...@@ -86,6 +86,7 @@ static void usage(void)
{ {
msg_Raw( NULL, "Usage: aggregartp [-i <RT priority>] [-l <syslogtag>] [-t <ttl>] [-w] [-o <SSRC IP>] [-U] [-x <retx buffer>] [-X <retx URL>] [-m <payload size>] [-R <RTP header>] @<src host> <dest host 1>[,<weight 1>] ... [<dest host N>,<weight N>]" ); msg_Raw( NULL, "Usage: aggregartp [-i <RT priority>] [-l <syslogtag>] [-t <ttl>] [-w] [-o <SSRC IP>] [-U] [-x <retx buffer>] [-X <retx URL>] [-m <payload size>] [-R <RTP header>] @<src host> <dest host 1>[,<weight 1>] ... [<dest host N>,<weight N>]" );
msg_Raw( NULL, " host format: [<connect addr>[:<connect port>]][@[<bind addr][:<bind port>]]" ); msg_Raw( NULL, " host format: [<connect addr>[:<connect port>]][@[<bind addr][:<bind port>]]" );
msg_Raw( NULL, " weight: integer, higher value means more capacity, or 0 for all packets" );
msg_Raw( NULL, " -w: overwrite RTP timestamps" ); msg_Raw( NULL, " -w: overwrite RTP timestamps" );
msg_Raw( NULL, " -o: overwrite RTP SSRC" ); msg_Raw( NULL, " -o: overwrite RTP SSRC" );
msg_Raw( NULL, " -U: prepend RTP header" ); msg_Raw( NULL, " -U: prepend RTP header" );
...@@ -139,6 +140,17 @@ static void SendBlock( int i_fd, struct sockaddr *p_sout, ...@@ -139,6 +140,17 @@ static void SendBlock( int i_fd, struct sockaddr *p_sout,
} }
} }
/*****************************************************************************
* SendBlock0: send a block to all outputs with weight 0
*****************************************************************************/
static void SendBlock0( block_t *p_block )
{
int i;
for ( i = 0; i < i_nb_outputs; i++ )
if ( !p_outputs[i].i_weight )
SendBlock( p_outputs[i].i_fd, NULL, 0, p_block);
}
/***************************************************************************** /*****************************************************************************
* RetxQueue: store a packet in the retx queue * RetxQueue: store a packet in the retx queue
*****************************************************************************/ *****************************************************************************/
...@@ -476,13 +488,18 @@ int main( int i_argc, char **pp_argv ) ...@@ -476,13 +488,18 @@ int main( int i_argc, char **pp_argv )
} }
/* Output block */ /* Output block */
output_t *p_output = NextOutput(); SendBlock0( p_input_block );
SendBlock( p_output->i_fd, NULL, 0, p_input_block );
if ( i_max_weight )
{
output_t *p_output = NextOutput();
SendBlock( p_output->i_fd, NULL, 0, p_input_block );
p_output->i_weighted_size += (i_size + p_output->i_remainder) p_output->i_weighted_size += (i_size + p_output->i_remainder)
/ p_output->i_weight; / p_output->i_weight;
p_output->i_remainder = (i_size + p_output->i_remainder) p_output->i_remainder = (i_size + p_output->i_remainder)
% p_output->i_weight; % p_output->i_weight;
}
RetxQueue( p_input_block, i_current_date ); RetxQueue( p_input_block, i_current_date );
p_input_block = NULL; p_input_block = NULL;
......
...@@ -83,6 +83,7 @@ static int i_output_fd; ...@@ -83,6 +83,7 @@ static int i_output_fd;
static input_t *p_inputs = NULL; static input_t *p_inputs = NULL;
static int i_nb_inputs = 0; static int i_nb_inputs = 0;
static int b_udp = 0; static int b_udp = 0;
static int b_redundance = 0;
static block_t *p_first = NULL; static block_t *p_first = NULL;
static block_t **pp_retx = NULL; static block_t **pp_retx = NULL;
...@@ -114,9 +115,10 @@ static int i_last_retx_input = 0; ...@@ -114,9 +115,10 @@ static int i_last_retx_input = 0;
static void usage(void) static void usage(void)
{ {
msg_Raw( NULL, "Usage: reordertp [-i <RT priority>] [-l <syslogtag>] [-t <ttl>] [-b <buffer length>] [-U] [-g <max gap>] [-j <max jitter>] [-r <# of clock ref>] [-n <max retx burst>] [-x <reorder/retx delay>] [-X <retx URL>] [-m <payload size>] [-R <RTP header>] <src host 1> ... [<src host N>] <dest host>" ); msg_Raw( NULL, "Usage: reordertp [-i <RT priority>] [-l <syslogtag>] [-t <ttl>] [-b <buffer length>] [-U] [-D] [-g <max gap>] [-j <max jitter>] [-r <# of clock ref>] [-n <max retx burst>] [-x <reorder/retx delay>] [-X <retx URL>] [-m <payload size>] [-R <RTP header>] <src host 1> ... [<src host N>] <dest host>" );
msg_Raw( NULL, " host format: [<connect addr>[:<connect port>]][@[<bind addr][:<bind port>]]" ); msg_Raw( NULL, " host format: [<connect addr>[:<connect port>]][@[<bind addr][:<bind port>]]" );
msg_Raw( NULL, " -U: strip RTP header" ); msg_Raw( NULL, " -U: strip RTP header" );
msg_Raw( NULL, " -D: input has redundant packets" );
msg_Raw( NULL, " -b: buffer length in ms [default 400]" ); msg_Raw( NULL, " -b: buffer length in ms [default 400]" );
msg_Raw( NULL, " -g: max gap between two clock references in ms [default 300]" ); msg_Raw( NULL, " -g: max gap between two clock references in ms [default 300]" );
msg_Raw( NULL, " -j: max jitter in ms [default 150]" ); msg_Raw( NULL, " -j: max jitter in ms [default 150]" );
...@@ -282,7 +284,9 @@ static void RetxCheck( uint64_t i_current_date ) ...@@ -282,7 +284,9 @@ static void RetxCheck( uint64_t i_current_date )
if ( i_current_seqnum == i_prev_seqnum ) if ( i_current_seqnum == i_prev_seqnum )
{ {
msg_Dbg( NULL, "duplicate RTP packet %hu", i_current_seqnum ); if ( !b_redundance )
msg_Dbg( NULL, "duplicate RTP packet %hu",
i_current_seqnum );
RetxDereference( p_current ); RetxDereference( p_current );
free( p_current ); free( p_current );
continue; continue;
...@@ -459,7 +463,7 @@ int main( int i_argc, char **pp_argv ) ...@@ -459,7 +463,7 @@ int main( int i_argc, char **pp_argv )
pfd[i_nb_inputs - 1].fd = i_fd; \ pfd[i_nb_inputs - 1].fd = i_fd; \
pfd[i_nb_inputs - 1].events = POLLIN | POLLERR | POLLRDHUP | POLLHUP; pfd[i_nb_inputs - 1].events = POLLIN | POLLERR | POLLRDHUP | POLLHUP;
while ( (c = getopt( i_argc, pp_argv, "i:l:t:b:g:j:r:n:x:X:Um:R:h" )) != -1 ) while ( (c = getopt( i_argc, pp_argv, "i:l:t:b:g:j:r:n:x:X:UDm:R:h" )) != -1 )
{ {
switch ( c ) switch ( c )
{ {
...@@ -514,6 +518,10 @@ int main( int i_argc, char **pp_argv ) ...@@ -514,6 +518,10 @@ int main( int i_argc, char **pp_argv )
b_udp = 1; b_udp = 1;
break; break;
case 'D':
b_redundance = 1;
break;
case 'm': case 'm':
i_asked_payload_size = strtol( optarg, NULL, 0 ); i_asked_payload_size = strtol( optarg, NULL, 0 );
break; break;
......
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment