?  search:
summary | log | graphiclog1 | graphiclog2 | commit | commitdiff | tree | refs | edit | fork
blame | history | raw | HEAD
usbmodeswitch: Updated to v.1.2.6 from shibby's branch.
[tomato.git] / release / src / router / udpxy / udpxy.c
blob8b38edaf60e0befc41d024d3c0d0e09e180f9f56
1 /* @(#) udpxy server: main module
2 *
3 * Copyright 2008-2011 Pavel V. Cherenkov (pcherenkov@gmail.com)
4 *
5 * This file is part of udpxy.
6 *
7 * udpxy is free software: you can redistribute it and/or modify
8 * it under the terms of the GNU General Public License as published by
9 * the Free Software Foundation, either version 3 of the License, or
10 * (at your option) any later version.
11 *
12 * udpxy is distributed in the hope that it will be useful,
13 * but WITHOUT ANY WARRANTY; without even the implied warranty of
14 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
15 * GNU General Public License for more details.
16 *
17 * You should have received a copy of the GNU General Public License
18 * along with udpxy. If not, see <http://www.gnu.org/licenses/>.
19 */
20
21 #include "osdef.h" /* os-specific definitions */
22
23 #include <sys/types.h>
24 #include <sys/socket.h>
25 #include <arpa/inet.h>
26 #include <net/if.h>
27 #include <sys/wait.h>
28 #include <sys/stat.h>
29
30 #include <netinet/in.h>
31 #include <sys/time.h>
32
33 /* #include <sys/select.h> */
34
35 #include <unistd.h>
36 #include <stdio.h>
37 #include <string.h>
38 #include <stdlib.h>
39 #include <signal.h>
40 #include <errno.h>
41 #include <assert.h>
42 #include <fcntl.h>
43 #include <syslog.h>
44 #include <time.h>
45 #include <sys/uio.h>
46
47 #include "mtrace.h"
48 #include "rparse.h"
49 #include "util.h"
50 #include "prbuf.h"
51 #include "ifaddr.h"
52
53 #include "udpxy.h"
54 #include "ctx.h"
55 #include "mkpg.h"
56 #include "rtp.h"
57 #include "uopt.h"
58 #include "dpkt.h"
59 #include "netop.h"
60
61 /* external globals */
62
63 extern const char CMD_UDP[];
64 extern const char CMD_STATUS[];
65 extern const char CMD_RESTART[];
66 extern const char CMD_RTP[];
67
68 extern const size_t CMD_UDP_LEN;
69 extern const size_t CMD_STATUS_LEN;
70 extern const size_t CMD_RESTART_LEN;
71 extern const size_t CMD_RTP_LEN;
72
73 extern const char IPv4_ALL[];
74
75 extern const char UDPXY_COPYRIGHT_NOTICE[];
76 extern const char UDPXY_CONTACT[];
77 extern const char COMPILE_MODE[];
78 extern const char VERSION[];
79 extern const int BUILDNUM;
80 extern const char BUILD_TYPE[];
81
82 extern FILE* g_flog;
83 extern volatile sig_atomic_t g_quit;
84
85 typedef struct tmfd {
86 int fd;
87 time_t atime;
88 } tmfd_t;
89
90
91 /* globals
92 */
93
94 struct udpxy_opt g_uopt;
95
96 /* misc
97 */
98
99 static volatile sig_atomic_t g_childexit = 0;
100
101 static const int PID_RESET = 1;
102
103 /*********************************************************/
104
105 /* handler for signals to perform a graceful exit
106 */
107 static void
108 handle_quitsigs(int signo)
109 {
110 g_quit = (sig_atomic_t)1;
111 (void) &signo;
112
113 TRACE( (void)tmfprintf( g_flog, "*** Caught SIGNAL %d ***\n", signo ) );
114 return;
115 }
116
117
118 /* return 1 if the application must gracefully quit
119 */
120 static sig_atomic_t
121 must_quit() { return g_quit; }
122
123
124 /* handle SIGCHLD
125 */
126 static void
127 handle_sigchld(int signo)
128 {
129 (void) &signo;
130 g_childexit = (sig_atomic_t)1;
131
132 TRACE( (void)tmfprintf( g_flog, "*** Caught SIGCHLD (%d) ***\n", signo ) );
133 return;
134 }
135
136 /*
137 static int get_childexit() { return g_childexit; }
138 */
139
140 /* clear SIGCHLD flag and adjust context if needed
141 */
142 static void
143 wait_children( struct server_ctx* ctx, int options )
144 {
145 int status, n = 0;
146 pid_t pid;
147
148 assert( ctx );
149 if (0 == g_childexit) {
150 TRACE( (void)tmfputs ("No children exited since last check\n",
151 g_flog) );
152 return;
153 }
154
155 g_childexit = 0;
156
157 TRACE( (void)tmfputs ("Waiting on exited children\n", g_flog) );
158 while( 0 < (pid = waitpid( -1, &status, options )) ) {
159 TRACE( (void)tmfprintf( g_flog, "Client [%d] has exited.\n", pid) );
160 delete_client( ctx, pid );
161 ++n;
162 }
163
164 if( (-1 == pid) && ( ECHILD != errno ) ) {
165 mperror(g_flog, errno, "%s: waitpid", __func__);
166 }
167
168 if (n > 0) {
169 TRACE( (void)tmfprintf (g_flog, "Cleaned up %d children, "
170 "%ld still running\n", n, (long)(ctx->clmax - ctx->clfree)) );
171 }
172
173 return;
174 }
175
176
177 /* wait for all children to quit
178 */
179 static void
180 wait_all( struct server_ctx* ctx ) { wait_children( ctx, 0 ); }
181
182
183 /* wait for the children who already terminated
184 */
185 static void
186 wait_terminated( struct server_ctx* ctx ) { wait_children( ctx, WNOHANG ); }
187
188
189 /* print out array of accepted socket fd's */
190 static void
191 print_fds (FILE* fp, const char* msg, tmfd_t* asock, size_t len)
192 {
193 size_t i;
194
195 (void) fprintf (fp, "%s [%ld]: ", msg, (long)len);
196 for (i = 0; i < len; ++i) {
197 (void) fprintf (fp, "%s%d", (i ? "," : ""), asock[i].fd);
198 }
199 fputc ('\n', fp);
200
201 return;
202 }
203
204
205 /* read HTTP request from sockfd, parse it into command
206 * and its parameters (for instance, command='udp' and
207 * parameters being '192.168.0.1:5002')
208 */
209 static int
210 read_command( int sockfd, char* cmd, size_t clen,
211 char* param, size_t plen )
212 {
213 #define DBUF_SZ 2048 /* max size for raw data with HTTP request */
214 #define RBUF_SZ 512 /* max size for url-derived request */
215 char httpbuf[ DBUF_SZ ] = "\0", request[ RBUF_SZ ] = "\0";
216 ssize_t hlen;
217 size_t rlen;
218 int rc = 0;
219
220 assert( (sockfd > 0) && cmd && clen && param && plen );
221 cmd[0] = '\0'; /* forget previous command */
222
223 TRACE( (void)tmfprintf( g_flog, "Reading command from socket [%d]\n",
224 sockfd ) );
225 hlen = recv( sockfd, httpbuf, sizeof(httpbuf), 0 );
226 if( 0>hlen ) {
227 rc = errno;
228 mperror(g_flog, rc, "%s - recv (%d)", __func__, rc);
229 return rc;
230 }
231 if (0 == hlen) {
232 (void) tmfprintf (g_flog, "%s: client closed socket [%d]\n",
233 __func__, sockfd);
234 return 1;
235 }
236
237 /* DEBUG - re-enable if needed */
238 TRACE( (void)tmfprintf( g_flog, "HTTP buffer [%ld bytes] received\n%s", (long)hlen, httpbuf ) );
239 /* TRACE( (void) save_buffer( httpbuf, hlen, "/tmp/httpbuf.dat" ) ); */
240
241 rlen = sizeof(request);
242 rc = get_request( httpbuf, (size_t)hlen, request, &rlen );
243 if (rc) return rc;
244
245 TRACE( (void)tmfprintf( g_flog, "Request=[%s], length=[%lu]\n",
246 request, (u_long)rlen ) );
247
248 rc = parse_param( request, rlen, cmd, clen, param, plen );
249 if( 0 == rc ) {
250 TRACE( (void)tmfprintf( g_flog, "Command [%s] with params [%s]"
251 " read from socket=[%d]\n", cmd, param, sockfd) );
252 }
253
254 return rc;
255 }
256
257
258 /* terminate the client process
259 */
260 static int
261 terminate( pid_t pid )
262 {
263 TRACE( (void)tmfprintf( g_flog, "Forcing client process [%d] to QUIT\n",
264 pid) );
265
266 if( pid <= 0 ) return 0;
267
268 if( 0 != kill( pid, SIGQUIT ) ) {
269 if( ESRCH != errno ) {
270 mperror(g_flog, errno, "%s - kill", __func__);
271 return ERR_INTERNAL;
272 }
273 /* ESRCH could mean client has quit already;
274 * if so we should wait for it */
275
276 TRACE( (void)tmfprintf( g_flog, "Process [%d] is not running.\n",
277 pid ) );
278 }
279
280 return 0;
281 }
282
283
284 /* terminate all clients
285 */
286 static void
287 terminate_all_clients( struct server_ctx* ctx )
288 {
289 size_t i;
290 pid_t pid;
291
292 for( i = 0; i < ctx->clmax; ++i ) {
293 pid = ctx->cl[i].pid;
294 if( pid > 0 ) (void) terminate( pid );
295 }
296
297 return;
298 }
299
300
301 /* send HTTP response to socket
302 */
303 static int
304 send_http_response( int sockfd, int code, const char* reason, ... )
305 {
306 static char msg[ 3072 ];
307 ssize_t nsent;
308 a_socklen_t msglen;
309 static const char CONTENT_TYPE[] = "Content-Type:application/octet-stream";
310
311 assert( (sockfd > 0) && code && reason );
312
313 msg[0] = '\0';
314
315 if ((200 == code) && g_uopt.h200_ftr[0]) {
316 msglen = snprintf( msg, sizeof(msg) - 1, "HTTP/1.1 %d %s\n%s\n%s\n\n",
317 code, reason, CONTENT_TYPE, g_uopt.h200_ftr);
318 }
319 else {
320 msglen = snprintf( msg, sizeof(msg) - 1, "HTTP/1.1 %d %s\n%s\n\n",
321 code, reason, CONTENT_TYPE );
322 }
323 if( msglen <= 0 ) return ERR_INTERNAL;
324
325 nsent = send( sockfd, msg, msglen, 0 );
326 if( -1 == nsent ) {
327 mperror(g_flog, errno, "%s - send", __func__);
328 return ERR_INTERNAL;
329 }
330
331 TRACE( (void)tmfprintf( g_flog, "Sent HTTP response code=[%d], "
332 "reason=[%s] to socket=[%d]\n%s\n",
333 code, reason, sockfd, msg) );
334 return 0;
335 }
336
337
338 /* renew multicast subscription if g_uopt.mcast_refresh seconds
339 * have passed since the last renewal
340 */
341 static void
342 check_mcast_refresh( int msockfd, time_t* last_tm,
343 const struct in_addr* mifaddr )
344 {
345 time_t now = 0;
346
347 if( NULL != g_uopt.srcfile ) /* reading from file */
348 return;
349
350 assert( (msockfd > 0) && last_tm && mifaddr );
351 now = time(NULL);
352
353 if( difftime( now, *last_tm ) >= (double)g_uopt.mcast_refresh ) {
354 (void) renew_multicast( msockfd, mifaddr );
355 *last_tm = now;
356 }
357
358 return;
359 }
360
361
362 /* analyze return value of I/O routines (read_data/write_data)
363 * and the pause-time marker to determine if we are in a
364 * PAUSE state
365 */
366 static int
367 pause_detect( int ntrans, time_t* p_pause )
368 {
369 time_t now = 0;
370 const double MAX_PAUSE_SEC = 5.0;
371
372 assert( p_pause );
373
374 /* timeshift: detect PAUSE by would-block error */
375 if (IO_BLK == ntrans) {
376 now = time(NULL);
377
378 if (*p_pause) {
379 if( difftime(now, *p_pause) > MAX_PAUSE_SEC ) {
380 TRACE( (void)tmfprintf( g_flog,
381 "PAUSE timed out after [%.0f] seconds\n",
382 MAX_PAUSE_SEC ) );
383 return -1;
384 }
385 }
386 else {
387 *p_pause = now;
388 TRACE( (void)tmfprintf( g_flog, "PAUSE started\n" ) );
389 }
390 }
391 else {
392 if (*p_pause) {
393 *p_pause = 0;
394 TRACE( (void)tmfprintf( g_flog, "PAUSE ended\n" ) );
395 }
396 }
397
398 return 0;
399 }
400
401
402 /* calculate values for:
403 * 1. number of messages to fit into data buffer
404 * 2. recommended (minimal) size of socket buffer
405 * (to read into the data buffer)
406 */
407 static int
408 calc_buf_settings( ssize_t* bufmsgs, size_t* sock_buflen )
409 {
410 ssize_t nmsgs = -1, max_buf_used = -1, env_snd_buflen = -1;
411 size_t buflen = 0;
412
413 /* how many messages should we process? */
414 nmsgs = (g_uopt.rbuf_msgs > 0) ? g_uopt.rbuf_msgs :
415 (int)g_uopt.rbuf_len / ETHERNET_MTU;
416
417 /* how many bytes could be written at once
418 * to the send socket */
419 max_buf_used = (g_uopt.rbuf_msgs > 0)
420 ? (ssize_t)(nmsgs * ETHERNET_MTU) : g_uopt.rbuf_len;
421 if (max_buf_used > g_uopt.rbuf_len) {
422 max_buf_used = g_uopt.rbuf_len;
423 }
424
425 assert( max_buf_used >= 0 );
426
427 env_snd_buflen = get_sizeval( "UDPXY_SOCKBUF_LEN", 0);
428 buflen = (env_snd_buflen > 0) ? (size_t)env_snd_buflen : (size_t)max_buf_used;
429
430 if (buflen < (size_t) MIN_SOCKBUF_LEN) {
431 buflen = (size_t) MIN_SOCKBUF_LEN;
432 }
433
434 /* cannot go below the size of effective usage */
435 if( buflen < (size_t)max_buf_used ) {
436 buflen = (size_t)max_buf_used;
437 }
438
439 if (bufmsgs) *bufmsgs = nmsgs;
440 if (sock_buflen) *sock_buflen = buflen;
441
442 TRACE( (void)tmfprintf( g_flog,
443 "min socket buffer = [%ld], "
444 "max space to use = [%ld], "
445 "Rmsgs = [%ld]\n",
446 (long)buflen, (long)max_buf_used, (long)nmsgs ) );
447
448 return 0;
449 }
450
451
452 /* make send-socket (dsockfd) buffer size no less
453 * than that of read-socket (ssockfd)
454 */
455 static int
456 sync_dsockbuf_len( int ssockfd, int dsockfd )
457 {
458 size_t curr_sendbuf_len = 0, curr_rcvbuf_len = 0;
459 int rc = 0;
460
461 if ( 0 != g_uopt.nosync_dbuf ) {
462 TRACE( (void)tmfprintf( g_flog,
463 "Must not adjust buffer size "
464 "for send socket [%d]\n", dsockfd) );
465 return 0;
466 }
467
468 assert( ssockfd && dsockfd );
469
470 rc = get_sendbuf( dsockfd, &curr_sendbuf_len );
471 if (0 != rc) return rc;
472
473 rc = get_rcvbuf( ssockfd, &curr_rcvbuf_len );
474 if (0 != rc) return rc;
475
476 if ( curr_rcvbuf_len > curr_sendbuf_len ) {
477 rc = set_sendbuf( dsockfd, curr_rcvbuf_len );
478 if (0 != rc) return rc;
479 }
480
481 return rc;
482 }
483
484
485 /* relay traffic from source to destination socket
486 *
487 */
488 static int
489 relay_traffic( int ssockfd, int dsockfd, struct server_ctx* ctx,
490 int dfilefd, const struct in_addr* mifaddr )
491 {
492 volatile sig_atomic_t quit = 0;
493
494 int rc = 0;
495 ssize_t nmsgs = -1;
496 ssize_t nrcv = 0, nsent = 0, nwr = 0,
497 lrcv = 0, lsent = 0;
498 char* data = NULL;
499 size_t data_len = g_uopt.rbuf_len;
500 struct rdata_opt ropt;
501 time_t pause_time = 0, rfr_tm = time(NULL);
502 sigset_t ubset;
503
504 const int ALLOW_PAUSES = get_flagval( "UDPXY_ALLOW_PAUSES", 0 );
505
506 /* permissible variation in data-packet size */
507 static const ssize_t t_delta = 0x20;
508
509 struct dstream_ctx ds;
510
511 static const int SET_PID = 1;
512 struct tps_data tps;
513
514 assert( ctx && mifaddr );
515
516 (void) sigemptyset (&ubset);
517 sigaddset (&ubset, SIGINT);
518 sigaddset (&ubset, SIGQUIT);
519 sigaddset (&ubset, SIGTERM);
520
521 /* restore the ability to receive *quit* signals */
522 rc = sigprocmask (SIG_UNBLOCK, &ubset, NULL);
523 if (0 != rc) {
524 mperror (g_flog, errno, "%s: sigprocmask", __func__);
525 return -1;
526 }
527
528 /* NOPs to eliminate warnings in lean version */
529 lrcv = t_delta - t_delta + lsent;
530
531 check_fragments( NULL, 0, 0, 0, 0, g_flog );
532
533 /* INIT
534 */
535
536 rc = calc_buf_settings( &nmsgs, NULL );
537 if (0 != rc) return -1;
538
539 TRACE( (void)tmfprintf( g_flog, "Data buffer will hold up to "
540 "[%d] messages\n", nmsgs ) );
541
542 rc = init_dstream_ctx( &ds, ctx->cmd, g_uopt.srcfile, nmsgs );
543 if( 0 != rc ) return -1;
544
545 (void) set_nice( g_uopt.nice_incr, g_flog );
546
547 do {
548 if( NULL == g_uopt.srcfile ) {
549 rc = set_timeouts( ssockfd, dsockfd,
550 ctx->rcv_tmout, 0,
551 ctx->snd_tmout, 0 );
552 if( 0 != rc ) break;
553 }
554
555 if( dsockfd > 0 ) {
556 rc = sync_dsockbuf_len( ssockfd, dsockfd );
557 if( 0 != rc ) break;
558
559 rc = send_http_response( dsockfd, 200, "OK" );
560 if( 0 != rc ) break;
561
562 /* timeshift: to detect PAUSE make destination
563 * socket non-blocking, otherwise make it blocking
564 * (since it might have been set unblocking earlier)
565 */
566 rc = set_nblock( dsockfd, (ALLOW_PAUSES ? 1 : 0) );
567 if( 0 != rc ) break;
568 }
569
570 data = malloc(data_len);
571 if( NULL == data ) {
572 mperror( g_flog, errno, "%s: malloc", __func__ );
573 break;
574 }
575
576 if( g_uopt.cl_tpstat )
577 tpstat_init( &tps, SET_PID );
578 } while(0);
579
580 TRACE( (void)tmfprintf( g_flog, "Relaying traffic from socket[%d] "
581 "to socket[%d], buffer size=[%d], Rmsgs=[%d], pauses=[%d]\n",
582 ssockfd, dsockfd, data_len, g_uopt.rbuf_msgs, ALLOW_PAUSES) );
583
584 /* RELAY LOOP
585 */
586 ropt.max_frgs = g_uopt.rbuf_msgs;
587 ropt.buf_tmout = g_uopt.dhold_tmout;
588
589 pause_time = 0;
590
591 while( (0 == rc) && !(quit = must_quit()) ) {
592 if( g_uopt.mcast_refresh > 0 ) {
593 check_mcast_refresh( ssockfd, &rfr_tm, mifaddr );
594 }
595
596 nrcv = read_data( &ds, ssockfd, data, data_len, &ropt );
597 if( -1 == nrcv ) break;
598
599 TRACE( check_fragments( "received new", data_len,
600 lrcv, nrcv, t_delta, g_flog ) );
601 lrcv = nrcv;
602
603 if( dsockfd && (nrcv > 0) ) {
604 nsent = write_data( &ds, data, nrcv, dsockfd );
605 if( -1 == nsent ) break;
606
607 if ( nsent < 0 ) {
608 if ( !ALLOW_PAUSES ) break;
609 if ( 0 != pause_detect( nsent, &pause_time ) ) break;
610 }
611
612 TRACE( check_fragments("sent", nrcv,
613 lsent, nsent, t_delta, g_flog) );
614 lsent = nsent;
615 }
616
617 if( (dfilefd > 0) && (nrcv > 0) ) {
618 nwr = write_data( &ds, data, nrcv, dfilefd );
619 if( -1 == nwr )
620 break;
621 TRACE( check_fragments( "wrote to file",
622 nrcv, lsent, nwr, t_delta, g_flog ) );
623 lsent = nwr;
624 }
625
626 if( ds.flags & F_SCATTERED ) reset_pkt_registry( &ds );
627
628 if( uf_TRUE == g_uopt.cl_tpstat )
629 tpstat_update( ctx, &tps, nsent );
630
631 } /* end of RELAY LOOP */
632
633 /* CLEANUP
634 */
635 TRACE( (void)tmfprintf( g_flog, "Exited relay loop: received=[%ld], "
636 "sent=[%ld], quit=[%ld]\n", (long)nrcv, (long)nsent, (long)quit ) );
637
638 free_dstream_ctx( &ds );
639 if( NULL != data ) free( data );
640
641 if( 0 != (quit = must_quit()) ) {
642 TRACE( (void)tmfprintf( g_flog, "Child process=[%d] must quit\n",
643 getpid()) );
644 }
645
646 return rc;
647 }
648
649
650 /* process command to relay udp traffic
651 *
652 */
653 static int
654 udp_relay( int sockfd, const char* param, size_t plen,
655 const struct in_addr* mifaddr,
656 struct server_ctx* ctx )
657 {
658 char mcast_addr[ IPADDR_STR_SIZE ];
659 struct sockaddr_in addr;
660
661 uint16_t port;
662 pid_t new_pid;
663 int rc = 0, flags;
664 int msockfd = -1, sfilefd = -1,
665 dfilefd = -1, srcfd = -1;
666 char dfile_name[ MAXPATHLEN ];
667 size_t rcvbuf_len = 0;
668
669 assert( (sockfd > 0) && param && plen && ctx );
670
671 TRACE( (void)tmfprintf( g_flog, "udp_relay : new_socket=[%d] param=[%s]\n",
672 sockfd, param) );
673 do {
674 rc = parse_udprelay( param, plen, mcast_addr, IPADDR_STR_SIZE, &port );
675 if( 0 != rc ) {
676 (void) tmfprintf( g_flog, "Error [%d] parsing parameters [%s]\n",
677 rc, param );
678 break;
679 }
680
681 if( 1 != inet_aton(mcast_addr, &addr.sin_addr) ) {
682 (void) tmfprintf( g_flog, "Invalid address: [%s]\n", mcast_addr );
683 rc = ERR_INTERNAL;
684 break;
685 }
686
687 addr.sin_family = AF_INET;
688 addr.sin_port = htons( (short)port );
689
690 } while(0);
691
692 if( 0 != rc ) {
693 (void) send_http_response( sockfd, 500, "Service error" );
694 return rc;
695 }
696
697 /* start the (new) process to relay traffic */
698
699 if( 0 != (new_pid = fork()) ) {
700 rc = add_client( ctx, new_pid, mcast_addr, port, sockfd );
701 return rc; /* parent returns */
702 }
703
704 /* child process:
705 */
706 TRACE( (void)tmfprintf( g_flog, "Client process=[%d] started "
707 "for socket=[%d]\n", getpid(), sockfd) );
708
709 (void) get_pidstr( PID_RESET, "c" );
710
711 (void)close( ctx->lsockfd );
712
713 /* close the reading end of the comm. pipe */
714 (void)close( ctx->cpipe[0] );
715 ctx->cpipe[0] = -1;
716
717 do {
718 /* make write end of pipe non-blocking (we don't want to
719 * block on pipe write while relaying traffic)
720 */
721 if( -1 == (flags = fcntl( ctx->cpipe[1], F_GETFL )) ||
722 -1 == fcntl( ctx->cpipe[1], F_SETFL, flags | O_NONBLOCK ) ) {
723 mperror( g_flog, errno, "%s: fcntl", __func__ );
724 rc = -1;
725 break;
726 }
727
728 if( NULL != g_uopt.dstfile ) {
729 (void) snprintf( dfile_name, MAXPATHLEN - 1,
730 "%s.%d", g_uopt.dstfile, getpid() );
731 dfilefd = creat( dfile_name, S_IRUSR | S_IWUSR | S_IRGRP );
732 if( -1 == dfilefd ) {
733 mperror( g_flog, errno, "%s: g_uopt.dstfile open", __func__ );
734 rc = -1;
735 break;
736 }
737
738 TRACE( (void)tmfprintf( g_flog,
739 "Dest file [%s] opened as fd=[%d]\n",
740 dfile_name, dfilefd ) );
741 }
742 else dfilefd = -1;
743
744 if( NULL != g_uopt.srcfile ) {
745 sfilefd = open( g_uopt.srcfile, O_RDONLY | O_NOCTTY );
746 if( -1 == sfilefd ) {
747 mperror( g_flog, errno, "%s: g_uopt.srcfile open", __func__ );
748 rc = -1;
749 }
750 else {
751 TRACE( (void) tmfprintf( g_flog, "Source file [%s] opened\n",
752 g_uopt.srcfile ) );
753 srcfd = sfilefd;
754 }
755 }
756 else {
757 rc = calc_buf_settings( NULL, &rcvbuf_len );
758 if (0 == rc ) {
759 rc = setup_mcast_listener( &addr, mifaddr, &msockfd,
760 (g_uopt.nosync_sbuf ? 0 : rcvbuf_len) );
761 srcfd = msockfd;
762 }
763 }
764 if( 0 != rc ) break;
765
766 rc = relay_traffic( srcfd, sockfd, ctx, dfilefd, mifaddr );
767 if( 0 != rc ) break;
768
769 } while(0);
770
771 if( msockfd > 0 ) {
772 close_mcast_listener( msockfd, mifaddr );
773 }
774 if( sfilefd > 0 ) {
775 (void) close( sfilefd );
776 TRACE( (void) tmfprintf( g_flog, "Source file [%s] closed\n",
777 g_uopt.srcfile ) );
778 }
779 if( dfilefd > 0 ) {
780 (void) close( dfilefd );
781 TRACE( (void) tmfprintf( g_flog, "Dest file [%s] closed\n",
782 dfile_name ) );
783 }
784
785 if( 0 != rc ) {
786 (void) send_http_response( sockfd, 500, "Service error" );
787 }
788
789 (void) close( sockfd );
790 free_server_ctx( ctx );
791
792 closelog();
793
794 TRACE( (void)tmfprintf( g_flog, "Child process=[%d] exits with rc=[%d]\n",
795 getpid(), rc) );
796
797 if( g_flog && (stderr != g_flog) ) {
798 (void) fclose(g_flog);
799 }
800
801 free_uopt( &g_uopt );
802
803 rc = ( 0 != rc ) ? ERR_INTERNAL : rc;
804 exit(rc); /* child exits */
805
806 return rc;
807 }
808
809
810 /* send server status as HTTP response to the given socket
811 */
812 static int
813 report_status( int sockfd, const struct server_ctx* ctx, int options )
814 {
815 char *buf = NULL;
816 int rc = 0;
817 ssize_t n, nsent;
818 size_t nlen = 0, bufsz, i;
819
820 static size_t BYTES_HDR = 2048;
821 static size_t BYTES_PER_CLI = 512;
822
823 assert( (sockfd > 0) && ctx );
824
825 for (bufsz=BYTES_HDR, i=0; i < ctx->clmax; ++i) {
826 bufsz+=BYTES_PER_CLI;
827 }
828 buf = malloc(bufsz);
829 if( !buf ) {
830 mperror(g_flog, ENOMEM, "malloc for %ld bytes for HTTP buffer "
831 "failed in %s", (long)bufsz, __func__ );
832 return ERR_INTERNAL;
833 }
834
835 (void) memset( buf, 0, sizeof(bufsz) );
836
837 nlen = bufsz;
838 rc = mk_status_page( ctx, buf, &nlen, options | MSO_HTTP_HEADER );
839
840 for( n = nsent = 0; (0 == rc) && (nsent < (ssize_t)nlen); ) {
841 errno = 0;
842 n = send( sockfd, buf, (int)nlen, 0 );
843
844 if( (-1 == n) && (EINTR != errno) ) {
845 mperror(g_flog, errno, "%s: send", __func__);
846 rc = ERR_INTERNAL;
847 break;
848 }
849
850 nsent += n;
851 }
852
853 if( 0 != rc ) {
854 TRACE( (void)tmfprintf( g_flog, "Error generating status report\n" ) );
855 }
856 else {
857 /* DEBUG only
858 TRACE( (void)tmfprintf( g_flog, "Saved status buffer to file\n" ) );
859 TRACE( (void)save_buffer(buf, nlen, "/tmp/status-udpxy.html") );
860 */
861 }
862
863 free(buf);
864 return rc;
865 }
866
867
868 /* process command within a request
869 */
870 static int
871 process_command( int new_sockfd, struct server_ctx* ctx,
872 const char* param, size_t plen )
873 {
874 int rc = 0;
875 const int STAT_OPTIONS = 0;
876 const int RESTART_OPTIONS = MSO_SKIP_CLIENTS | MSO_RESTART;
877
878 assert( (new_sockfd > 0) && ctx && param );
879
880 if( 0 == strncmp( ctx->cmd, CMD_UDP, sizeof(ctx->cmd) ) ||
881 0 == strncmp( ctx->cmd, CMD_RTP, sizeof(ctx->cmd) ) ) {
882 if( ctx->clfree ) {
883 rc = udp_relay( new_sockfd, param, plen,
884 &(ctx->mcast_inaddr), ctx );
885 }
886 else {
887 send_http_response( new_sockfd, 401, "Bad request" );
888 (void)tmfprintf( g_flog, "Client limit [%d] has been reached.\n",
889 ctx->clmax);
890 }
891 }
892 else if( 0 == strncmp( ctx->cmd, CMD_STATUS, sizeof(ctx->cmd) ) ) {
893 rc = report_status( new_sockfd, ctx, STAT_OPTIONS );
894 }
895 else if( 0 == strncmp( ctx->cmd, CMD_RESTART, sizeof(ctx->cmd) ) ) {
896 (void) report_status( new_sockfd, ctx, RESTART_OPTIONS );
897
898 terminate_all_clients( ctx );
899 wait_all( ctx );
900 }
901 else {
902 TRACE( (void)tmfprintf( g_flog, "Unrecognized command [%s]"
903 " - ignoring.\n", ctx->cmd) );
904 send_http_response( new_sockfd, 401, "Unrecognized request" );
905 }
906
907 return rc;
908 }
909
910
911 static void
912 accept_requests (int sockfd, tmfd_t* asock, size_t* alen)
913 {
914 int rc = 0, new_sockfd = -1, err = 0, peer_port = -1,
915 wmark = g_uopt.ss_rlwmark;
916 size_t nmax = *alen, naccepted = 0;
917 struct sockaddr_in cliaddr;
918 a_socklen_t addrlen = sizeof (cliaddr);
919 char peer_addr [128] = "#undef#";
920
921 while (naccepted < nmax) {
922 TRACE( (void)tmfputs ("Accepting new connection\n", g_flog) );
923
924 new_sockfd = accept (sockfd, (struct sockaddr*)&cliaddr, &addrlen );
925 if (-1 == new_sockfd) {
926 err = errno;
927 if ((EWOULDBLOCK == err) || (EAGAIN == err)) {
928 TRACE((void)tmfputs ("Nothing more to accept\n", g_flog));
929 rc = 0; break;
930 }
931 if ((ECONNABORTED == err) || (ECONNRESET == err) || (EPROTO == err)) {
932 TRACE( (void)tmfprintf (g_flog, "Connection aborted/reset "
933 "at accept point, errno=%d\n", err) );
934 rc = 0; continue;
935 }
936
937 mperror(g_flog, err, "%s: accept", __func__);
938 rc = err; break;
939 }
940
941 if (0 != set_nblock (new_sockfd, 1)) {
942 (void) close (new_sockfd); /* TODO: error-aware close */
943 continue;
944 }
945 /*
946 if (0 != set_timeouts(new_sockfd, new_sockfd, g_uopt.sr_tmout, 0,
947 g_uopt.sw_tmout, 0)) {
948 (void) close (new_sockfd);
949 continue;
950 }
951 */
952 if (wmark > 0) {
953 if (0 != setsockopt (new_sockfd, SOL_SOCKET, SO_RCVLOWAT,
954 (char*)&wmark, sizeof(wmark))) {
955 mperror (g_flog, errno, "%s: setsockopt SO_RCVLOWAT [%d]",
956 __func__, wmark);
957 (void) close (new_sockfd); /* TODO: error-aware close */
958 continue;
959 } else {
960 TRACE( (void)tmfprintf (g_flog, "Receive LOW WATERMARK [%d] appleid "
961 "to newly-accepted socket [%d]\n", wmark, new_sockfd) );
962 }
963 }
964
965 asock [naccepted].fd = new_sockfd;
966 asock [naccepted].atime = time (NULL);
967 ++naccepted;
968
969 (void) get_peerinfo (new_sockfd, peer_addr,
970 sizeof(peer_addr)-1, &peer_port);
971
972 TRACE( (void)tmfprintf( g_flog, "Accepted socket=[%d] from %s:%d "
973 "n=%ld/nmax=%ld\n", new_sockfd, peer_addr, peer_port,
974 (long)naccepted, (long)nmax) );
975 } /* while */
976
977 if (naccepted >= nmax) {
978 (void)tmfprintf (g_flog, "Accept limit max=[%d] reached, "
979 "%ld already accepted", (long)nmax, (long)naccepted);
980 }
981
982 *alen = naccepted;
983 TRACE( (void)tmfprintf (g_flog, "%s: Sockets accepted: [%ld]\n",
984 __func__, (long)naccepted));
985 return;
986 }
987
988
989 static void
990 shrink_asock (tmfd_t* asock, size_t* alen, size_t nserved)
991 {
992 size_t nmax = *alen;
993 long j = 0, v = 0;
994
995 (void) &print_fds;
996
997 /* uncomment to DEBUG
998 TRACE( print_fds (g_flog, "unshrunk accepted sockets", asock, *alen) );
999 */
1000
1001 (void) nserved;
1002
1003 for (; j < (long)nmax; ++j) {
1004 if (-1 == asock[j].fd) {
1005 ++v;
1006 continue;
1007 }
1008
1009 if (v > 0) {
1010 asock[j - v].fd = asock[j].fd;
1011 asock[j - v].atime = asock[j].atime;
1012 asock[j].fd = -1;
1013 }
1014 }
1015
1016 assert ((long)nserved == v);
1017 *alen = nmax - (size_t)v;
1018
1019 TRACE ( (void)tmfprintf (g_flog, "%s: %ld shrunk, was %ld now %ld\n",
1020 __func__, (long)v, (long)nmax, (long)*alen ) );
1021
1022 /* uncomment to DEBUG
1023 TRACE( print_fds (g_flog, "remaining accepted sockets", asock, *alen) );
1024 */
1025 }
1026
1027
1028 static void
1029 tmout_requests (tmfd_t* asock, size_t *alen)
1030 {
1031 size_t nmax = *alen, i = 0, nout = 0;
1032 time_t now = time (NULL);
1033
1034 TRACE( (void)tmfprintf (g_flog, "%s: BEGIN with %ld sockets\n",
1035 __func__, (long)*alen) );
1036
1037 for (; i < nmax; ++i) {
1038 assert ((asock[i].fd >= 0) && asock[i].atime);
1039 if ((asock[i].atime + g_uopt.ssel_tmout) < now) {
1040 TRACE( (void)tmfprintf (g_flog, "%s: timed out socket #%d [%d], "
1041 "atime/now/tmout=%ld/%ld/%ld\n", __func__, (i+1), asock[i].fd,
1042 (long)asock[i].atime, (long)now, g_uopt.ssel_tmout) );
1043
1044 (void) close (asock[i].fd);
1045 asock[i].fd = -1;
1046 ++nout;
1047 }
1048 }
1049 shrink_asock (asock, alen, nout); /* will adjust alen */
1050
1051 TRACE( (void)tmfprintf (g_flog, "%s: END with %ld sockets\n",
1052 __func__, (long)*alen) );
1053 return;
1054 }
1055
1056 static void
1057 process_requests (tmfd_t* asock, size_t *alen, fd_set* rset, struct server_ctx* srv)
1058 {
1059 size_t nmax = *alen, i = 0, nserved = 0;
1060 int rc = 0, served = 0;
1061 char param[ 128 ] = "\0";
1062 time_t now = time (NULL);
1063
1064 /* uncomment to DEBUG */
1065 TRACE( print_fds (g_flog, "pre-process sockets", asock, nmax) );
1066
1067 for (; i < nmax; ++i, served = 0) {
1068 assert (asock[i].fd >= 0);
1069 assert (asock[i].atime > 0);
1070
1071 do {
1072 /* not selected - yet try to time it out */
1073 if (!FD_ISSET(asock[i].fd, rset)) {
1074 if ((asock[i].atime + g_uopt.ssel_tmout) < now) {
1075 TRACE( (void)tmfprintf (g_flog,
1076 "%s: accepted socket [%ld] timed out\n",
1077 __func__, (long)asock[i].fd) );
1078 ++served; /* timed out - must close */
1079 }
1080 break;
1081 }
1082
1083 /* selected */
1084 TRACE( (void)tmfprintf (g_flog, "acting on accepted socket "
1085 "[%d] (%d/%d)\n", asock[i].fd, i+1, nmax) );
1086
1087 ++served; /* selected - must close regardless */
1088 rc = read_command( asock[i].fd, srv->cmd, sizeof(srv->cmd),
1089 param, sizeof(param) );
1090 if( 0 != rc ) break;
1091
1092 rc = process_command (asock[i].fd, srv, param, sizeof(param) );
1093 } while (0);
1094
1095 if (0 != rc) {
1096 TRACE( (void)tmfprintf (g_flog, "error [%d] processing "
1097 "client socket [%d]\n", rc, asock[i]));
1098 }
1099
1100 TRACE( (void)tmfprintf (g_flog, "%s: %s accepted "
1101 "socket [%d]\n", __func__, (served ? "closing" : "skipping"),
1102 asock[i].fd) );
1103
1104 if (served) {
1105 (void) close (asock[i].fd);
1106 asock[i].fd = -1;
1107 ++nserved;
1108 }
1109 } /* for */
1110
1111 TRACE( (void)tmfprintf (g_flog, "Processed [%ld/%ld] accepted sockets\n",
1112 (long)nserved, (long)nmax) );
1113 TRACE( print_fds (g_flog, "newly-accepted sockets", asock, nmax) );
1114
1115 if (nserved >= nmax) {
1116 *alen = 0;
1117 TRACE( (void)tmfputs ("All accepted sockets processed\n", g_flog) );
1118 }
1119 else {
1120 shrink_asock (asock, alen, nserved);
1121 }
1122
1123 return;
1124 }
1125
1126
1127 /* process client requests
1128 */
1129 static int
1130 server_loop( const char* ipaddr, int port,
1131 const char* mcast_addr )
1132 {
1133 int rc, maxfd, err, nrdy, i;
1134 struct in_addr mcast_inaddr;
1135 struct server_ctx srv;
1136 fd_set rset;
1137 struct timespec tmout, *ptmout = NULL;
1138 tmfd_t *asock = NULL;
1139 size_t n = 0, nasock = 0, max_nasock = LQ_BACKLOG;
1140 sigset_t oset, bset;
1141
1142 assert( (port > 0) && mcast_addr && ipaddr );
1143
1144 (void)tmfprintf( g_flog, "Server is starting up, max clients = [%u]\n",
1145 g_uopt.max_clients );
1146 asock = calloc (max_nasock, sizeof(*asock));
1147 if (!asock) {
1148 mperror (g_flog, ENOMEM, "%s: calloc", __func__);
1149 return ERR_INTERNAL;
1150 }
1151
1152 if( 1 != inet_aton(mcast_addr, &mcast_inaddr) ) {
1153 mperror(g_flog, errno, "%s: inet_aton", __func__);
1154 return ERR_INTERNAL;
1155 }
1156
1157 init_server_ctx( &srv, g_uopt.max_clients,
1158 (ipaddr[0] ? ipaddr : "0.0.0.0") , (uint16_t)port, mcast_addr );
1159
1160 srv.rcv_tmout = (u_short)g_uopt.rcv_tmout;
1161 srv.snd_tmout = RLY_SOCK_TIMEOUT;
1162 srv.mcast_inaddr = mcast_inaddr;
1163
1164 /* NB: server socket is non-blocking! */
1165 if( 0 != (rc = setup_listener( ipaddr, port, &srv.lsockfd,
1166 g_uopt.lq_backlog )) ) {
1167 return rc;
1168 }
1169
1170 sigemptyset (&bset);
1171 sigaddset (&bset, SIGINT);
1172 sigaddset (&bset, SIGQUIT);
1173 sigaddset (&bset, SIGCHLD);
1174 sigaddset (&bset, SIGTERM);
1175
1176 tmout.tv_sec = g_uopt.ssel_tmout;
1177 tmout.tv_nsec = 0;
1178
1179 if (ptmout) {
1180 TRACE( (void)tmfprintf (g_flog, "select() timeout set to "
1181 "[%ld] seconds\n", tmout.tv_sec) );
1182 }
1183
1184 (void) sigprocmask (SIG_BLOCK, &bset, &oset);
1185
1186 TRACE( (void)tmfprintf( g_flog, "Entering server loop\n") );
1187 while (1) {
1188 FD_ZERO( &rset );
1189 FD_SET( srv.lsockfd, &rset );
1190 FD_SET( srv.cpipe[0], &rset );
1191
1192 maxfd = (srv.lsockfd > srv.cpipe[0] ) ? srv.lsockfd : srv.cpipe[0];
1193 for (i = 0; (size_t)i < nasock; ++i) {
1194 assert (asock[i].fd >= 0);
1195 FD_SET (asock[i].fd, &rset);
1196 if (asock[i].fd > maxfd) maxfd = asock[i].fd;
1197 }
1198
1199 /* if there are accepted sockets - apply specified time-out
1200 */
1201 ptmout = ((nasock > 0) && (g_uopt.ssel_tmout > 0)) ? &tmout : NULL;
1202
1203 TRACE( (void)tmfprintf( g_flog, "Waiting for input from [%ld] fd's, "
1204 "%s timeout\n", (long)(2 + nasock), (ptmout ? "with" : "NO")));
1205
1206 nrdy = pselect (maxfd + 1, &rset, NULL, NULL, ptmout, &oset);
1207 err = errno;
1208
1209 if( must_quit() ) {
1210 TRACE( (void)tmfputs( "Must quit now\n", g_flog ) );
1211 rc = 0; break;
1212 }
1213 wait_terminated( &srv );
1214
1215 if( nrdy < 0 ) {
1216 if (EINTR == err) {
1217 TRACE( (void)tmfputs ("INTERRUPTED, yet "
1218 "will continue.\n", g_flog) );
1219 rc = 0; continue;
1220 }
1221
1222 mperror( g_flog, err, "%s: select", __func__ );
1223 break;
1224 }
1225
1226 TRACE( (void)tmfprintf (g_flog, "Got %ld requests\n", (long)nrdy) );
1227 if (0 == nrdy) { /* time-out */
1228 tmout_requests (asock, &nasock);
1229 rc = 0; continue;
1230 }
1231
1232 if( FD_ISSET(srv.cpipe[0], &rset) ) {
1233 (void) tpstat_read( &srv );
1234 if (--nrdy <= 0) continue;
1235 }
1236
1237 if ((0 < nasock) &&
1238 (0 < (nrdy - (FD_ISSET(srv.lsockfd, &rset) ? 1 : 0)))) {
1239 process_requests (asock, &nasock, &rset, &srv);
1240 /* n now contains # (yet) unprocessed accepted sockets */
1241 }
1242
1243 if (FD_ISSET(srv.lsockfd, &rset)) {
1244 if (nasock >= max_nasock) {
1245 (void) tmfprintf (g_flog, "Cannot accept sockets beyond "
1246 "the limit [%ld/%ld], skipping\n",
1247 (long)nasock, (long)max_nasock);
1248 }
1249 else {
1250 n = max_nasock - nasock; /* append asock */
1251 accept_requests (srv.lsockfd, &(asock[nasock]), &n);
1252 nasock += n;
1253 }
1254 }
1255
1256 } /* server loop */
1257
1258 TRACE( (void)tmfprintf( g_flog, "Exited server loop\n") );
1259
1260 for (i = 0; (size_t)i < nasock; ++i) {
1261 if (asock[i].fd > 0) (void) close (asock[i].fd);
1262 }
1263 free (asock);
1264
1265 /* receive additional (blocked signals) */
1266 (void) sigprocmask (SIG_SETMASK, &oset, NULL);
1267 wait_terminated( &srv );
1268 terminate_all_clients( &srv );
1269 wait_all( &srv );
1270
1271 if (0 != close( srv.lsockfd )) {
1272 mperror (g_flog, errno, "server socket close");
1273 }
1274
1275 free_server_ctx( &srv );
1276
1277 (void)tmfprintf( g_flog, "Server exits with rc=[%d]\n", rc );
1278 return rc;
1279 }
1280
1281
1282 static void
1283 usage( const char* app, FILE* fp )
1284 {
1285 (void) fprintf (fp, "%s %s (%s %d) %s [%s]\n", app, VERSION, BUILD_TYPE, BUILDNUM,
1286 COMPILE_MODE, get_sysinfo(NULL));
1287 (void) fprintf (fp, "usage: %s [-vTS] [-a listenaddr] -p port "
1288 "[-m mcast_ifc_addr] [-c clients] [-l logfile] "
1289 "[-B sizeK] [-n nice_incr]\n", app );
1290 (void) fprintf(fp,
1291 "\t-v : enable verbose output [default = disabled]\n"
1292 "\t-S : enable client statistics [default = disabled]\n"
1293 "\t-T : do NOT run as a daemon [default = daemon if root]\n"
1294 "\t-a : (IPv4) address/interface to listen on [default = %s]\n"
1295 "\t-p : port to listen on\n"
1296 "\t-m : (IPv4) address/interface of (multicast) "
1297 "source [default = %s]\n"
1298 "\t-c : max clients to serve [default = %d, max = %d]\n",
1299 IPv4_ALL, IPv4_ALL, DEFAULT_CLIENT_COUNT, MAX_CLIENT_COUNT);
1300 (void)fprintf(fp,
1301 "\t-l : log output to file [default = stderr]\n"
1302 "\t-B : buffer size (65536, 32Kb, 1Mb) for inbound (multicast) data [default = %ld bytes]\n"
1303 "\t-R : maximum messages to store in buffer (-1 = all) "
1304 "[default = %d]\n"
1305 "\t-H : maximum time (sec) to hold data in buffer "
1306 "(-1 = unlimited) [default = %d]\n"
1307 "\t-n : nice value increment [default = %d]\n"
1308 "\t-M : periodically renew multicast subscription (skip if 0 sec) [default = %d sec]\n",
1309 (long)DEFAULT_CACHE_LEN, g_uopt.rbuf_msgs, DHOLD_TIMEOUT, g_uopt.nice_incr,
1310 (int)g_uopt.mcast_refresh );
1311 (void) fprintf( fp, "Examples:\n"
1312 " %s -p 4022 \n"
1313 "\tlisten for HTTP requests on port 4022, all network interfaces\n"
1314 " %s -a lan0 -p 4022 -m lan1\n"
1315 "\tlisten for HTTP requests on interface lan0, port 4022;\n"
1316 "\tsubscribe to multicast groups on interface lan1\n",
1317 app, app);
1318 (void) fprintf( fp, "\n %s\n", UDPXY_COPYRIGHT_NOTICE );
1319 (void) fprintf( fp, " %s\n\n", UDPXY_CONTACT );
1320 return;
1321 }
1322
1323
1324 extern int udpxy_main( int argc, char* const argv[] );
1325
1326 int
1327 udpxy_main( int argc, char* const argv[] )
1328 {
1329 int rc = 0, ch = 0, port = -1,
1330 custom_log = 0, no_daemon = 0;
1331
1332 char ipaddr[IPADDR_STR_SIZE] = "\0",
1333 mcast_addr[IPADDR_STR_SIZE] = "\0";
1334
1335 char pidfile[ MAXPATHLEN ] = "\0";
1336 u_short MIN_MCAST_REFRESH = 0, MAX_MCAST_REFRESH = 0;
1337 char udpxy_finfo[ 80 ] = {0};
1338
1339 /* support for -r -w (file read/write) option is disabled by default;
1340 * those features are experimental and for dev debugging ONLY
1341 * */
1342 #ifdef UDPXY_FILEIO
1343 static const char UDPXY_OPTMASK[] = "TvSa:l:p:m:c:B:n:R:r:w:H:M:";
1344 #else
1345 static const char UDPXY_OPTMASK[] = "TvSa:l:p:m:c:B:n:R:H:M:";
1346 #endif
1347
1348 struct sigaction qact, iact, cact, oldact;
1349
1350 extern const char g_udpxy_app[];
1351
1352 (void) get_pidstr( PID_RESET, "S" );
1353
1354 rc = init_uopt( &g_uopt );
1355 while( (0 == rc) && (-1 != (ch = getopt(argc, argv, UDPXY_OPTMASK))) ) {
1356 switch( ch ) {
1357 case 'v': set_verbose( &g_uopt.is_verbose );
1358 break;
1359 case 'T': no_daemon = 1;
1360 break;
1361 case 'S': g_uopt.cl_tpstat = uf_TRUE;
1362 break;
1363 case 'a':
1364 rc = get_ipv4_address( optarg, ipaddr, sizeof(ipaddr) );
1365 if( 0 != rc ) {
1366 (void) fprintf( stderr, "Invalid address: [%s]\n",
1367 optarg );
1368 rc = ERR_PARAM;
1369 }
1370 break;
1371
1372 case 'p':
1373 port = atoi( optarg );
1374 if( port <= 0 ) {
1375 (void) fprintf( stderr, "Invalid port number: [%d]\n",
1376 port );
1377 rc = ERR_PARAM;
1378 }
1379 break;
1380
1381 case 'm':
1382 rc = get_ipv4_address( optarg, mcast_addr,
1383 sizeof(mcast_addr) );
1384 if( 0 != rc ) {
1385 (void) fprintf( stderr, "Invalid multicast address: "
1386 "[%s]\n", optarg );
1387 rc = ERR_PARAM;
1388 }
1389 break;
1390
1391 case 'c':
1392 g_uopt.max_clients = atoi( optarg );
1393 if( (g_uopt.max_clients < MIN_CLIENT_COUNT) ||
1394 (g_uopt.max_clients > MAX_CLIENT_COUNT) ) {
1395 (void) fprintf( stderr,
1396 "Client count should be between %d and %d\n",
1397 MIN_CLIENT_COUNT, MAX_CLIENT_COUNT );
1398 rc = ERR_PARAM;
1399 }
1400 break;
1401
1402 case 'l':
1403 g_flog = fopen( optarg, "a" );
1404 if( NULL == g_flog ) {
1405 rc = errno;
1406 (void) fprintf( stderr, "Error opening logfile "
1407 "[%s]: %s\n",
1408 optarg, strerror(rc) );
1409 rc = ERR_PARAM;
1410 break;
1411 }
1412
1413 Setlinebuf( g_flog );
1414 custom_log = 1;
1415 break;
1416
1417 case 'B':
1418 rc = a2size(optarg, &g_uopt.rbuf_len);
1419 if( 0 != rc ) {
1420 (void) fprintf( stderr, "Invalid buffer size: [%s]\n",
1421 optarg );
1422 exit( ERR_PARAM );
1423 }
1424 else if( (g_uopt.rbuf_len < MIN_MCACHE_LEN) ||
1425 (g_uopt.rbuf_len > MAX_MCACHE_LEN) ) {
1426 fprintf(stderr, "Buffer size "
1427 "must be within [%ld-%ld] bytes\n",
1428 (long)MIN_MCACHE_LEN, (long)MAX_MCACHE_LEN );
1429 rc = ERR_PARAM;
1430 }
1431 break;
1432
1433 case 'n':
1434 g_uopt.nice_incr = atoi( optarg );
1435 if( 0 == g_uopt.nice_incr ) {
1436 (void) fprintf( stderr,
1437 "Invalid nice-value increment: [%s]\n", optarg );
1438 rc = ERR_PARAM;
1439 break;
1440 }
1441 break;
1442
1443 case 'R':
1444 g_uopt.rbuf_msgs = atoi( optarg );
1445 if( (g_uopt.rbuf_msgs <= 0) && (-1 != g_uopt.rbuf_msgs) ) {
1446 (void) fprintf( stderr,
1447 "Invalid Rmsgs size: [%s]\n", optarg );
1448 rc = ERR_PARAM;
1449 break;
1450 }
1451 break;
1452
1453 case 'H':
1454 g_uopt.dhold_tmout = (time_t)atoi( optarg );
1455 if( (0 == g_uopt.dhold_tmout) ||
1456 ((g_uopt.dhold_tmout) < 0 && (-1 != g_uopt.dhold_tmout)) ) {
1457 (void) fprintf( stderr, "Invalid value for max time "
1458 "to hold buffered data: [%s]\n", optarg );
1459 rc = ERR_PARAM;
1460 break;
1461 }
1462 break;
1463
1464 #ifdef UDPXY_FILEIO
1465 case 'r':
1466 if( 0 != access(optarg, R_OK) ) {
1467 perror("source file - access");
1468 rc = ERR_PARAM;
1469 break;
1470 }
1471
1472 g_uopt.srcfile = strdup( optarg );
1473 break;
1474
1475 case 'w':
1476 g_uopt.dstfile = strdup( optarg );
1477 break;
1478 #endif /* UDPXY_FILEIO */
1479 case 'M':
1480 g_uopt.mcast_refresh = (u_short)atoi( optarg );
1481
1482 MIN_MCAST_REFRESH = 30;
1483 MAX_MCAST_REFRESH = 64000;
1484 if( g_uopt.mcast_refresh &&
1485 (g_uopt.mcast_refresh < MIN_MCAST_REFRESH ||
1486 g_uopt.mcast_refresh > MAX_MCAST_REFRESH )) {
1487 (void) fprintf( stderr,
1488 "Invalid multicast refresh period [%d] seconds, "
1489 "min=[%d] sec, max=[%d] sec\n",
1490 (int)g_uopt.mcast_refresh,
1491 (int)MIN_MCAST_REFRESH, (int)MAX_MCAST_REFRESH );
1492 rc = ERR_PARAM;
1493 break;
1494 }
1495 break;
1496
1497 case ':':
1498 (void) fprintf( stderr,
1499 "Option [-%c] requires an argument\n",
1500 optopt );
1501 rc = ERR_PARAM;
1502 break;
1503 case '?':
1504 (void) fprintf( stderr,
1505 "Unrecognized option: [-%c]\n", optopt );
1506 rc = ERR_PARAM;
1507 break;
1508
1509 default:
1510 usage( argv[0], stderr );
1511 rc = ERR_PARAM;
1512 break;
1513 }
1514 } /* while getopt */
1515
1516 if (rc) {
1517 free_uopt( &g_uopt );
1518 return rc;
1519 }
1520
1521 openlog( g_udpxy_app, LOG_CONS | LOG_PID, LOG_LOCAL0 );
1522
1523 do {
1524 if( (argc < 2) || (port <= 0) || (rc != 0) ) {
1525 usage( argv[0], stderr );
1526 rc = ERR_PARAM; break;
1527 }
1528
1529 if( '\0' == mcast_addr[0] ) {
1530 (void) strncpy( mcast_addr, IPv4_ALL, sizeof(mcast_addr) - 1 );
1531 }
1532
1533 if( !custom_log ) {
1534 /* in debug mode output goes to stderr, otherwise to /dev/null */
1535 g_flog = ((uf_TRUE == g_uopt.is_verbose)
1536 ? stderr
1537 : fopen( "/dev/null", "a" ));
1538 if( NULL == g_flog ) {
1539 perror("fopen");
1540 rc = ERR_INTERNAL; break;
1541 }
1542 }
1543
1544 if( 0 == geteuid() ) {
1545 if( !no_daemon ) {
1546 if( stderr == g_flog ) {
1547 (void) fprintf( stderr,
1548 "Logfile must be specified to run "
1549 "in verbose mode in background\n" );
1550 rc = ERR_PARAM; break;
1551 }
1552
1553 if( 0 != (rc = daemonize(0, g_flog)) ) {
1554 rc = ERR_INTERNAL; break;
1555 }
1556 }
1557
1558 rc = set_pidfile( g_udpxy_app, port, pidfile, sizeof(pidfile) );
1559 if( 0 != rc ) {
1560 mperror( g_flog, errno, "set_pidfile" );
1561 rc = ERR_INTERNAL; break;
1562 }
1563
1564 if( 0 != (rc = make_pidfile( pidfile, getpid(), g_flog )) )
1565 break;
1566 }
1567
1568 qact.sa_handler = handle_quitsigs;
1569 sigemptyset(&qact.sa_mask);
1570 qact.sa_flags = 0;
1571
1572 if( (sigaction(SIGTERM, &qact, &oldact) < 0) ||
1573 (sigaction(SIGQUIT, &qact, &oldact) < 0) ||
1574 (sigaction(SIGINT, &qact, &oldact) < 0)) {
1575 perror("sigaction-quit");
1576 rc = ERR_INTERNAL; break;
1577 }
1578
1579 iact.sa_handler = SIG_IGN;
1580 sigemptyset(&iact.sa_mask);
1581 iact.sa_flags = 0;
1582
1583 if( (sigaction(SIGPIPE, &iact, &oldact) < 0) ) {
1584 perror("sigaction-ignore");
1585 rc = ERR_INTERNAL; break;
1586 }
1587
1588 cact.sa_handler = handle_sigchld;
1589 sigemptyset(&cact.sa_mask);
1590 cact.sa_flags = 0;
1591
1592 if( sigaction(SIGCHLD, &cact, &oldact) < 0 ) {
1593 perror("sigaction-sigchld");
1594 rc = ERR_INTERNAL; break;
1595 }
1596
1597 (void) snprintf( udpxy_finfo, sizeof(udpxy_finfo),
1598 "%s %s (%s %d) %s [%s]", g_udpxy_app, VERSION,
1599 BUILD_TYPE, BUILDNUM,
1600 COMPILE_MODE, get_sysinfo(NULL) );
1601
1602 syslog( LOG_NOTICE, "%s is starting\n",udpxy_finfo );
1603 TRACE( printcmdln( g_flog, udpxy_finfo, argc, argv ) );
1604
1605 rc = server_loop( ipaddr, port, mcast_addr );
1606
1607 syslog( LOG_NOTICE, "%s is exiting with rc=[%d]\n",
1608 udpxy_finfo, rc);
1609 TRACE( tmfprintf( g_flog, "%s is exiting with rc=[%d]\n",
1610 g_udpxy_app, rc ) );
1611 TRACE( printcmdln( g_flog, udpxy_finfo, argc, argv ) );
1612 } while(0);
1613
1614 if( '\0' != pidfile[0] ) {
1615 if( -1 == unlink(pidfile) ) {
1616 mperror( g_flog, errno, "unlink [%s]", pidfile );
1617 }
1618 }
1619
1620 if( g_flog && (stderr != g_flog) ) {
1621 (void) fclose(g_flog);
1622 }
1623
1624 closelog();
1625 free_uopt( &g_uopt );
1626
1627 return rc;
1628 }
1629
1630 /* __EOF__ */
1631
Tomato firmware and modifications.