sscall.c (10503B)
1 /* See LICENSE file for copyright and license details */ 2 3 #include <stdio.h> 4 #include <stdlib.h> 5 #include <string.h> 6 #include <err.h> 7 #include <signal.h> 8 #include <unistd.h> 9 #include <fcntl.h> 10 #include <errno.h> 11 #include <sys/types.h> 12 #include <sys/socket.h> 13 #include <netinet/in.h> 14 #include <arpa/inet.h> 15 #include <netdb.h> 16 #include <sys/time.h> 17 #include <time.h> 18 19 #include <ao/ao.h> 20 #include <pthread.h> 21 22 #include "list.h" 23 24 /* Input/Output PCM buffer size */ 25 #define PCM_BUF_SIZE (8192) 26 /* Sleep at least 50ms between each sendto() call */ 27 #define UDELAY_SEND (50 * 1000) 28 29 /* Command line option, bits per sample */ 30 static int fbits; 31 /* Command line option, samples per second (in a single channel) */ 32 static int frate; 33 /* Command line option, number of channels */ 34 static int fchan; 35 /* Command line option, device driver ID */ 36 static int fdevid; 37 /* Command line option, verbosity flag */ 38 static int fverbose; 39 40 /* Libao handle */ 41 static ao_device *device; 42 /* Output PCM thread */ 43 static pthread_t output_pcm_thread; 44 /* Input PCM thread */ 45 static pthread_t input_pcm_thread; 46 47 /* Shared buf between do_output_pcm() 48 * and output_pcm thread */ 49 struct pcm_buf { 50 /* PCM buffer */ 51 void *buf; 52 /* PCM buffer size */ 53 size_t len; 54 struct list_head list; 55 } pcm_buf; 56 57 /* Private structure for the 58 * input_pcm thread */ 59 struct inp_pcm_priv { 60 /* Input file descriptor */ 61 int fd; 62 /* Client socket */ 63 int sockfd; 64 /* Client address info */ 65 struct addrinfo *servinfo; 66 } inp_pcm_priv; 67 68 /* Lock that protects pcm_buf */ 69 static pthread_mutex_t pcm_buf_lock; 70 /* Condition variable on which ao_play() blocks */ 71 static pthread_cond_t tx_pcm_cond; 72 73 /* State of the output_pcm thread */ 74 struct output_pcm_state { 75 int quit; 76 } output_pcm_state; 77 78 /* State of the input_pcm thread */ 79 struct input_pcm_state { 80 int quit; 81 } input_pcm_state; 82 83 /* Lock that protects output_pcm_state */ 84 static pthread_mutex_t output_pcm_state_lock; 85 /* Lock that protects input_pcm_state */ 86 static pthread_mutex_t input_pcm_state_lock; 87 88 /* Set to 1 when SIGINT is received */ 89 static volatile int handle_sigint; 90 91 /* Play back audio from the client */ 92 static void * 93 output_pcm(void *data) 94 { 95 struct pcm_buf *pctx; 96 struct list_head *iter, *q; 97 struct output_pcm_state *state = data; 98 struct timespec ts; 99 struct timeval tp; 100 int rc; 101 102 do { 103 pthread_mutex_lock(&pcm_buf_lock); 104 gettimeofday(&tp, NULL); 105 /* Convert from timeval to timespec */ 106 ts.tv_sec = tp.tv_sec; 107 ts.tv_nsec = tp.tv_usec * 1000; 108 /* Default to a 3 second wait internal */ 109 ts.tv_sec += 3; 110 111 if (list_empty(&pcm_buf.list)) { 112 /* Wait in the worst case 3 seconds to give some 113 * grace to perform cleanup if necessary */ 114 rc = pthread_cond_timedwait(&tx_pcm_cond, 115 &pcm_buf_lock, 116 &ts); 117 if (rc == ETIMEDOUT) 118 if (fverbose) 119 printf("Output thread is starving...\n"); 120 } 121 122 pthread_mutex_lock(&output_pcm_state_lock); 123 if (state->quit) { 124 pthread_mutex_unlock(&output_pcm_state_lock); 125 pthread_mutex_unlock(&pcm_buf_lock); 126 break; 127 } 128 pthread_mutex_unlock(&output_pcm_state_lock); 129 130 /* Dequeue and play buffers via libao */ 131 list_for_each_safe(iter, q, &pcm_buf.list) { 132 pctx = list_entry(iter, struct pcm_buf, 133 list); 134 ao_play(device, pctx->buf, pctx->len); 135 free(pctx->buf); 136 list_del(&pctx->list); 137 free(pctx); 138 } 139 pthread_mutex_unlock(&pcm_buf_lock); 140 } while (1); 141 142 pthread_exit(NULL); 143 144 return NULL; 145 } 146 147 /* Prepare for output PCM, enqueue buffer 148 * and signal output_pcm thread */ 149 static void 150 do_output_pcm(const void *buf, size_t len) 151 { 152 struct pcm_buf *pctx; 153 154 pctx = malloc(sizeof(*pctx)); 155 if (!pctx) 156 err(1, "malloc"); 157 memset(pctx, 0, sizeof(*pctx)); 158 159 pctx->buf = malloc(len); 160 if (!pctx->buf) 161 err(1, "malloc"); 162 163 pctx->len = len; 164 memcpy(pctx->buf, buf, len); 165 166 pthread_mutex_lock(&pcm_buf_lock); 167 INIT_LIST_HEAD(&pctx->list); 168 list_add_tail(&pctx->list, &pcm_buf.list); 169 pthread_cond_signal(&tx_pcm_cond); 170 pthread_mutex_unlock(&pcm_buf_lock); 171 } 172 173 /* Input PCM thread, outbound path */ 174 static void * 175 input_pcm(void *data) 176 { 177 char buf[PCM_BUF_SIZE]; 178 ssize_t bytes; 179 struct input_pcm_state *state = data; 180 181 do { 182 pthread_mutex_lock(&input_pcm_state_lock); 183 if (state->quit) { 184 pthread_mutex_unlock(&input_pcm_state_lock); 185 break; 186 } 187 pthread_mutex_unlock(&input_pcm_state_lock); 188 189 bytes = read(inp_pcm_priv.fd, buf, sizeof(buf)); 190 if (bytes > 0) { 191 bytes = sendto(inp_pcm_priv.sockfd, buf, 192 bytes, 0, 193 inp_pcm_priv.servinfo->ai_addr, 194 inp_pcm_priv.servinfo->ai_addrlen); 195 if (bytes < 0) 196 warn("sendto"); 197 usleep(UDELAY_SEND); 198 } 199 } while (1); 200 201 pthread_exit(NULL); 202 203 return NULL; 204 } 205 206 static void 207 usage(const char *s) 208 { 209 fprintf(stderr, 210 "usage: %s [OPTIONS] <remote-addr> <remote-port> <local-port>\n", s); 211 fprintf(stderr, " -b\tBits per sample\n"); 212 fprintf(stderr, " -r\tSamples per second (in a single channel)\n"); 213 fprintf(stderr, " -c\tNumber of channels\n"); 214 fprintf(stderr, " -d\tOverride default driver ID\n"); 215 fprintf(stderr, " -v\tEnable verbose output\n"); 216 fprintf(stderr, " -h\tThis help screen\n"); 217 } 218 219 static void 220 sig_handler(int signum) 221 { 222 switch (signum) { 223 case SIGINT: 224 handle_sigint = 1; 225 break; 226 default: 227 break; 228 } 229 } 230 231 void 232 set_nonblocking(int fd) 233 { 234 int opts; 235 236 opts = fcntl(fd, F_GETFL); 237 if (opts < 0) 238 err(1, "fcntl"); 239 opts = (opts | O_NONBLOCK); 240 if (fcntl(fd, F_SETFL, opts) < 0) 241 err(1, "fcntl"); 242 } 243 244 static void 245 init_libao(int rate, int bits, int chans, 246 int *devid) 247 { 248 ao_sample_format format; 249 int default_driver; 250 251 ao_initialize(); 252 253 default_driver = ao_default_driver_id(); 254 255 memset(&format, 0, sizeof(format)); 256 format.bits = bits; 257 format.channels = chans; 258 format.rate = rate; 259 format.byte_format = AO_FMT_LITTLE; 260 261 if (!*devid) 262 *devid = default_driver; 263 264 device = ao_open_live(*devid, &format, NULL); 265 if (!device) 266 errx(1, "Error opening output device: %d\n", 267 fdevid); 268 } 269 270 int 271 main(int argc, char *argv[]) 272 { 273 int recfd = STDIN_FILENO; 274 ssize_t bytes; 275 char buf[PCM_BUF_SIZE]; 276 int cli_sockfd, srv_sockfd; 277 struct addrinfo cli_hints, *cli_servinfo, *p0, *p1; 278 struct addrinfo srv_hints, *srv_servinfo; 279 int rv; 280 int ret; 281 socklen_t addr_len; 282 struct sockaddr_storage their_addr; 283 char *prog; 284 int c; 285 char host[NI_MAXHOST]; 286 int optval; 287 288 prog = *argv; 289 while ((c = getopt(argc, argv, "hb:c:r:d:v")) != -1) { 290 switch (c) { 291 case 'h': 292 usage(prog); 293 exit(0); 294 break; 295 case 'b': 296 fbits = strtol(optarg, NULL, 10); 297 break; 298 case 'c': 299 fchan = strtol(optarg, NULL, 10); 300 break; 301 case 'r': 302 frate = strtol(optarg, NULL, 10); 303 break; 304 case 'd': 305 fdevid = strtol(optarg, NULL, 10); 306 break; 307 case 'v': 308 fverbose = 1; 309 break; 310 case '?': 311 default: 312 exit(1); 313 } 314 } 315 argc -= optind; 316 argv += optind; 317 318 if (argc != 3) { 319 usage(prog); 320 exit(1); 321 } 322 323 if (!fbits) 324 fbits = 16; 325 326 if (!fchan) 327 fchan = 1; 328 329 if (!frate) 330 frate = 8000; 331 332 init_libao(frate, fbits, fchan, &fdevid); 333 334 if (fverbose) { 335 printf("Bits per sample: %d\n", fbits); 336 printf("Number of channels: %d\n", fchan); 337 printf("Sample rate: %d\n", frate); 338 printf("Default driver ID: %d\n", fdevid); 339 fflush(stdout); 340 } 341 342 memset(&cli_hints, 0, sizeof(cli_hints)); 343 cli_hints.ai_family = AF_INET; 344 cli_hints.ai_socktype = SOCK_DGRAM; 345 346 rv = getaddrinfo(argv[0], argv[1], &cli_hints, &cli_servinfo); 347 if (rv) 348 errx(1, "getaddrinfo: %s", gai_strerror(rv)); 349 350 for (p0 = cli_servinfo; p0; p0 = p0->ai_next) { 351 cli_sockfd = socket(p0->ai_family, p0->ai_socktype, 352 p0->ai_protocol); 353 if (cli_sockfd < 0) 354 continue; 355 break; 356 } 357 358 if (!p0) 359 errx(1, "failed to bind socket"); 360 361 memset(&srv_hints, 0, sizeof(srv_hints)); 362 srv_hints.ai_family = AF_INET; 363 srv_hints.ai_socktype = SOCK_DGRAM; 364 srv_hints.ai_flags = AI_PASSIVE; 365 366 rv = getaddrinfo(NULL, argv[2], &srv_hints, &srv_servinfo); 367 if (rv) 368 errx(1, "getaddrinfo: %s", gai_strerror(rv)); 369 370 for(p1 = srv_servinfo; p1; p1 = p1->ai_next) { 371 srv_sockfd = socket(p1->ai_family, p1->ai_socktype, 372 p1->ai_protocol); 373 if (srv_sockfd < 0) 374 continue; 375 optval = 1; 376 ret = setsockopt(srv_sockfd, SOL_SOCKET, 377 SO_REUSEADDR, &optval, sizeof(optval)); 378 if (ret < 0) { 379 close(srv_sockfd); 380 warn("setsockopt"); 381 continue; 382 } 383 if (bind(srv_sockfd, p1->ai_addr, p1->ai_addrlen) < 0) { 384 close(srv_sockfd); 385 warn("bind"); 386 continue; 387 } 388 break; 389 } 390 391 if (!p1) 392 errx(1, "failed to bind socket"); 393 394 INIT_LIST_HEAD(&pcm_buf.list); 395 396 pthread_mutex_init(&pcm_buf_lock, NULL); 397 pthread_cond_init(&tx_pcm_cond, NULL); 398 399 pthread_mutex_init(&output_pcm_state_lock, NULL); 400 pthread_mutex_init(&input_pcm_state_lock, NULL); 401 402 ret = pthread_create(&output_pcm_thread, NULL, 403 output_pcm, &output_pcm_state); 404 if (ret < 0) 405 err(1, "pthread_create"); 406 407 inp_pcm_priv.fd = recfd; 408 inp_pcm_priv.sockfd = cli_sockfd; 409 inp_pcm_priv.servinfo = p0; 410 411 set_nonblocking(inp_pcm_priv.fd); 412 413 ret = pthread_create(&input_pcm_thread, NULL, 414 input_pcm, &input_pcm_state); 415 if (ret < 0) 416 err(1, "pthread_create"); 417 418 if (signal(SIGINT, sig_handler) == SIG_ERR) 419 err(1, "signal"); 420 421 /* Receive audio data from other end and prepare 422 * for playback */ 423 do { 424 /* Handle SIGINT gracefully */ 425 if (handle_sigint) { 426 if (fverbose) 427 printf("Interrupted, exiting...\n"); 428 break; 429 } 430 431 addr_len = sizeof(their_addr); 432 bytes = recvfrom(srv_sockfd, buf, 433 sizeof(buf), MSG_DONTWAIT, 434 (struct sockaddr *)&their_addr, 435 &addr_len); 436 if (bytes > 0) { 437 if (fverbose) { 438 ret = getnameinfo((struct sockaddr *)&their_addr, 439 addr_len, host, 440 sizeof(host), NULL, 0, 0); 441 if (ret < 0) { 442 warn("getnameinfo"); 443 snprintf(host, sizeof(host), "unknown"); 444 } 445 printf("Received %zd bytes from %s\n", 446 bytes, host); 447 } 448 do_output_pcm(buf, bytes); 449 } 450 } while (1); 451 452 /* Prepare input thread to be killed */ 453 pthread_mutex_lock(&input_pcm_state_lock); 454 input_pcm_state.quit = 1; 455 pthread_mutex_unlock(&input_pcm_state_lock); 456 457 /* Wait for it */ 458 pthread_join(input_pcm_thread, NULL); 459 460 /* Prepare output thread to be killed */ 461 pthread_mutex_lock(&output_pcm_state_lock); 462 output_pcm_state.quit = 1; 463 pthread_mutex_unlock(&output_pcm_state_lock); 464 465 /* Wake up the output thread if it is 466 * sleeping */ 467 pthread_mutex_lock(&pcm_buf_lock); 468 pthread_cond_signal(&tx_pcm_cond); 469 pthread_mutex_unlock(&pcm_buf_lock); 470 471 /* Wait for it */ 472 pthread_join(output_pcm_thread, NULL); 473 474 ao_close(device); 475 ao_shutdown(); 476 477 freeaddrinfo(cli_servinfo); 478 freeaddrinfo(srv_servinfo); 479 480 return EXIT_SUCCESS; 481 }