src/pulsecore/protocol-simple.c

   1 /* $Id$ */
   2
   3 /***
   4   This file is part of PulseAudio.
   5
   6   Copyright 2004-2006 Lennart Poettering
   7
   8   PulseAudio is free software; you can redistribute it and/or modify
   9   it under the terms of the GNU Lesser General Public License as published
  10   by the Free Software Foundation; either version 2 of the License,
  11   or (at your option) any later version.
  12
  13   PulseAudio is distributed in the hope that it will be useful, but
  14   WITHOUT ANY WARRANTY; without even the implied warranty of
  15   MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
  16   General Public License for more details.
  17
  18   You should have received a copy of the GNU Lesser General Public License
  19   along with PulseAudio; if not, write to the Free Software
  20   Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307
  21   USA.
  22 ***/
  23
  24 #ifdef HAVE_CONFIG_H
  25 #include <config.h>
  26 #endif
  27
  28 #include <stdlib.h>
  29 #include <limits.h>
  30 #include <stdio.h>
  31 #include <errno.h>
  32 #include <string.h>
  33
  34 #include <pulse/xmalloc.h>
  35
  36 #include <pulsecore/sink-input.h>
  37 #include <pulsecore/source-output.h>
  38 #include <pulsecore/client.h>
  39 #include <pulsecore/sample-util.h>
  40 #include <pulsecore/namereg.h>
  41 #include <pulsecore/log.h>
  42 #include <pulsecore/core-error.h>
  43
  44 #include "protocol-simple.h"
  45
  46 /* Don't allow more than this many concurrent connections */
  47 #define MAX_CONNECTIONS 10
  48
  49 struct connection {
  50     pa_protocol_simple *protocol;
  51     pa_iochannel *io;
  52     pa_sink_input *sink_input;
  53     pa_source_output *source_output;
  54     pa_client *client;
  55     pa_memblockq *input_memblockq, *output_memblockq;
  56
  57     int dead;
  58
  59     struct {
  60         pa_memblock *current_memblock;
  61         size_t memblock_index, fragment_size;
  62         pa_atomic_int missing;
  63     } playback;
  64 };
  65
  66 struct pa_protocol_simple {
  67     pa_module *module;
  68     pa_core *core;
  69     pa_socket_server*server;
  70     pa_idxset *connections;
  71
  72     pa_asyncmsgq *asyncmsgq;
  73
  74     enum {
  75         RECORD = 1,
  76         PLAYBACK = 2,
  77         DUPLEX = 3
  78     } mode;
  79
  80     pa_sample_spec sample_spec;
  81     char *source_name, *sink_name;
  82 };
  83
  84 enum {
  85     SINK_INPUT_MESSAGE_POST_DATA = PA_SINK_INPUT_MESSAGE_MAX, /* data from main loop to sink input */
  86 };
  87
  88 enum {
  89     MESSAGE_REQUEST_DATA,   /* data from source output to main loop */
  90     MESSAGE_POST_DATA       /* data from source output to main loop */
  91 };
  92
  93
  94 #define PLAYBACK_BUFFER_SECONDS (.5)
  95 #define PLAYBACK_BUFFER_FRAGMENTS (10)
  96 #define RECORD_BUFFER_SECONDS (5)
  97 #define RECORD_BUFFER_FRAGMENTS (100)
  98
  99 static void connection_free(struct connection *c) {
 100     pa_assert(c);
 101
 102     pa_idxset_remove_by_data(c->protocol->connections, c, NULL);
 103
 104     if (c->sink_input) {
 105         pa_sink_input_disconnect(c->sink_input);
 106         pa_sink_input_unref(c->sink_input);
 107     }
 108
 109     if (c->source_output) {
 110         pa_source_output_disconnect(c->source_output);
 111         pa_source_output_unref(c->source_output);
 112     }
 113
 114     if (c->playback.current_memblock)
 115         pa_memblock_unref(c->playback.current_memblock);
 116
 117     if (c->client)
 118         pa_client_free(c->client);
 119     if (c->io)
 120         pa_iochannel_free(c->io);
 121     if (c->input_memblockq)
 122         pa_memblockq_free(c->input_memblockq);
 123     if (c->output_memblockq)
 124         pa_memblockq_free(c->output_memblockq);
 125
 126     pa_xfree(c);
 127 }
 128
 129 static int do_read(struct connection *c) {
 130     pa_memchunk chunk;
 131     ssize_t r;
 132     size_t l;
 133     void *p;
 134
 135     pa_assert(c);
 136
 137     if (!c->sink_input || !(l = pa_atomic_load(&c->playback.missing)))
 138         return 0;
 139
 140     if (l > c->playback.fragment_size)
 141         l = c->playback.fragment_size;
 142
 143     if (c->playback.current_memblock)
 144         if (pa_memblock_get_length(c->playback.current_memblock) - c->playback.memblock_index < l) {
 145             pa_memblock_unref(c->playback.current_memblock);
 146             c->playback.current_memblock = NULL;
 147             c->playback.memblock_index = 0;
 148         }
 149
 150     if (!c->playback.current_memblock) {
 151         pa_assert_se(c->playback.current_memblock = pa_memblock_new(c->protocol->core->mempool, l));
 152         c->playback.memblock_index = 0;
 153     }
 154
 155     p = pa_memblock_acquire(c->playback.current_memblock);
 156     r = pa_iochannel_read(c->io, (uint8_t*) p + c->playback.memblock_index, l);
 157     pa_memblock_release(c->playback.current_memblock);
 158
 159     if (r <= 0) {
 160
 161         if (errno == EINTR || errno == EAGAIN)
 162             return 0;
 163
 164         pa_log_debug("read(): %s", r == 0 ? "EOF" : pa_cstrerror(errno));
 165         return -1;
 166     }
 167
 168     chunk.memblock = c->playback.current_memblock;
 169     chunk.index = c->playback.memblock_index;
 170     chunk.length = r;
 171
 172     c->playback.memblock_index += r;
 173
 174     pa_asyncmsgq_post(c->protocol->asyncmsgq, c, MESSAGE_POST_DATA, NULL, &chunk, NULL, NULL);
 175
 176     return 0;
 177 }
 178
 179 static int do_write(struct connection *c) {
 180     pa_memchunk chunk;
 181     ssize_t r;
 182     void *p;
 183
 184     p_assert(c);
 185
 186     if (!c->source_output)
 187         return 0;
 188
 189     if (pa_memblockq_peek(c->output_memblockq, &chunk) < 0)
 190         return 0;
 191
 192     pa_assert(chunk.memblock);
 193     pa_assert(chunk.length);
 194
 195     p = pa_memblock_acquire(chunk.memblock);
 196     r = pa_iochannel_write(c->io, (uint8_t*) p+chunk.index, chunk.length);
 197     pa_memblock_release(chunk.memblock);
 198
 199     pa_memblock_unref(chunk.memblock);
 200
 201     if (r < 0) {
 202
 203         if (errno == EINTR || errno == EAGAIN)
 204             return 0;
 205
 206         pa_log("write(): %s", pa_cstrerror(errno));
 207         return -1;
 208     }
 209
 210     pa_memblockq_drop(c->output_memblockq, &chunk, r);
 211
 212     return 0;
 213 }
 214
 215 static void do_work(struct connection *c) {
 216     pa_assert(c);
 217
 218     if (c->dead)
 219         return;
 220
 221     if (pa_iochannel_is_readable(c->io)) {
 222         if (do_read(c) < 0)
 223             goto fail;
 224     } else if (pa_iochannel_is_hungup(c->io))
 225         goto fail;
 226
 227     if (pa_iochannel_is_writable(c->io)) {
 228         if (do_write(c) < 0)
 229             goto fail;
 230     }
 231
 232     return;
 233
 234 fail:
 235
 236     if (c->sink_input) {
 237
 238         /* If there is a sink input, we first drain what we already have read before shutting down the connection */
 239         c->dead = 1;
 240
 241         pa_iochannel_free(c->io);
 242         c->io = NULL;
 243
 244         pa_memblockq_prebuf_disable(c->input_memblockq);
 245     } else
 246         connection_free(c);
 247 }
 248
 249 /*** sink_input callbacks ***/
 250
 251 /* Called from thread context */
 252 static int sink_input_process_msg(pa_sink_input *i, int code, void *userdata, const pa_memchunk *chunk) {
 253     struct connection*c;
 254
 255     pa_assert(i);
 256     c = i->userdata;
 257     pa_assert(c);
 258
 259     switch (code) {
 260
 261         case SINK_INPUT_MESSAGE_POST_DATA: {
 262             pa_assert(chunk);
 263
 264             /* New data from the main loop */
 265             pa_memblockq_push_align(c->input_memblockq, chunk);
 266             return 0;
 267         }
 268
 269         case PA_SINK_INPUT_MESSAGE_GET_LATENCY: {
 270             pa_usec_t *r = userdata;
 271
 272             *r = pa_bytes_to_usec(pa_memblockq_get_length(c->input_memblockq), &c->sink_input->sample_spec);
 273
 274             /* Fall through, the default handler will add in the extra
 275              * latency added by the resampler */
 276         }
 277
 278         default:
 279             return pa_sink_input_process_msg(i, code, userdata);
 280     }
 281 }
 282
 283 /* Called from thread context */
 284 static int sink_input_peek_cb(pa_sink_input *i, pa_memchunk *chunk) {
 285     struct connection*c;
 286
 287     pa_assert(i);
 288     c = i->userdata;
 289     pa_assert(c);
 290     pa_assert(chunk);
 291
 292     r = pa_memblockq_peek(c->input_memblockq, chunk);
 293
 294     if (c->dead && r < 0)
 295         connection_free(c);
 296
 297     return r;
 298 }
 299
 300 /* Called from thread context */
 301 static void sink_input_drop_cb(pa_sink_input *i, const pa_memchunk *chunk, size_t length) {
 302     struct connection*c = i->userdata;
 303     size_t old, new;
 304
 305     pa_assert(i);
 306     pa_assert(c);
 307     pa_assert(length);
 308
 309     old = pa_memblockq_missing(c->input_memblockq);
 310     pa_memblockq_drop(c->input_memblockq, chunk, length);
 311     new = pa_memblockq_missing(c->input_memblockq);
 312
 313     pa_atomic_store(&c->playback.missing, &new);
 314
 315     if (new > old)
 316         pa_asyncmsgq_post(c->protocol->asyncmsgq, c, MESSAGE_REQUEST_DATA, NULL, NULL, NULL, NULL);
 317 }
 318
 319 /* Called from main context */
 320 static void sink_input_kill_cb(pa_sink_input *i) {
 321     pa_assert(i);
 322     pa_assert(i->userdata);
 323
 324     connection_free((struct connection *) i->userdata);
 325 }
 326
 327 /*** source_output callbacks ***/
 328
 329 static void source_output_push_cb(pa_source_output *o, const pa_memchunk *chunk) {
 330     struct connection *c;
 331
 332     pa_assert(o);
 333     c = o->userdata;
 334     pa_assert(c);
 335     pa_assert(chunk);
 336
 337     pa_asyncmsgq_post(c->protocol->asyncmsgq, c, MESSAGE_REQUEST_DATA, NULL, chunk, NULL, NULL);
 338 }
 339
 340 static void source_output_kill_cb(pa_source_output *o) {
 341     struct connection*c;
 342
 343     pa_assert(o);
 344     c = o->userdata;
 345     pa_assert(c);
 346
 347     connection_free(c);
 348 }
 349
 350 static pa_usec_t source_output_get_latency_cb(pa_source_output *o) {
 351     struct connection*c;
 352
 353     pa_assert(o);
 354     c = o->userdata;
 355     pa_assert(c);
 356
 357     return pa_bytes_to_usec(pa_memblockq_get_length(c->output_memblockq), &c->source_output->sample_spec);
 358 }
 359
 360 /*** client callbacks ***/
 361
 362 static void client_kill_cb(pa_client *client) {
 363     struct connection*c;
 364
 365     pa_assert(client);
 366     c = client->userdata;
 367     pa_assert(c);
 368
 369     connection_free(client);
 370 }
 371
 372 /*** pa_iochannel callbacks ***/
 373
 374 static void io_callback(pa_iochannel*io, void *userdata) {
 375     struct connection *c = userdata;
 376
 377     pa_assert(io);
 378     pa_assert(c);
 379
 380     do_work(c);
 381 }
 382
 383 /*** socket_server callbacks ***/
 384
 385 static void on_connection(pa_socket_server*s, pa_iochannel *io, void *userdata) {
 386     pa_protocol_simple *p = userdata;
 387     struct connection *c = NULL;
 388     char cname[256];
 389
 390     pa_assert(s);
 391     pa_assert(io);
 392     pa_assert(p);
 393
 394     if (pa_idxset_size(p->connections)+1 > MAX_CONNECTIONS) {
 395         pa_log("Warning! Too many connections (%u), dropping incoming connection.", MAX_CONNECTIONS);
 396         pa_iochannel_free(io);
 397         return;
 398     }
 399
 400     c = pa_xnew(struct connection, 1);
 401     c->io = io;
 402     c->sink_input = NULL;
 403     c->source_output = NULL;
 404     c->input_memblockq = c->output_memblockq = NULL;
 405     c->protocol = p;
 406     c->playback.current_memblock = NULL;
 407     c->playback.memblock_index = 0;
 408     c->playback.fragment_size = 0;
 409     c->dead = 0;
 410     pa_atomic_store(&c->playback.missing, 0);
 411
 412     pa_iochannel_socket_peer_to_string(io, cname, sizeof(cname));
 413     pa_assert_se(c->client = pa_client_new(p->core, __FILE__, cname));
 414     c->client->owner = p->module;
 415     c->client->kill = client_kill_cb;
 416     c->client->userdata = c;
 417
 418
 419     if (p->mode & PLAYBACK) {
 420         pa_sink_input_new_data data;
 421         size_t l;
 422
 423         pa_sink_input_new_data_init(&data);
 424         data.driver = __FILE__;
 425         data.name = c->client->name;
 426         pa_sink_input_new_data_set_sample_spec(&data, &p->sample_spec);
 427         data.module = p->module;
 428         data.client = c->client;
 429
 430         if (!(c->sink_input = pa_sink_input_new(p->core, &data, 0))) {
 431             pa_log("Failed to create sink input.");
 432             goto fail;
 433         }
 434
 435         c->sink_input->peek = sink_input_peek_cb;
 436         c->sink_input->drop = sink_input_drop_cb;
 437         c->sink_input->kill = sink_input_kill_cb;
 438         c->sink_input->get_latency = sink_input_get_latency_cb;
 439         c->sink_input->userdata = c;
 440
 441         l = (size_t) (pa_bytes_per_second(&p->sample_spec)*PLAYBACK_BUFFER_SECONDS);
 442         c->input_memblockq = pa_memblockq_new(
 443                 0,
 444                 l,
 445                 0,
 446                 pa_frame_size(&p->sample_spec),
 447                 (size_t) -1,
 448                 l/PLAYBACK_BUFFER_FRAGMENTS,
 449                 NULL);
 450         pa_assert(c->input_memblockq);
 451         pa_iochannel_socket_set_rcvbuf(io, l/PLAYBACK_BUFFER_FRAGMENTS*5);
 452         c->playback.fragment_size = l/10;
 453
 454         pa_atomic_store(&c->playback.missing, pa_memblockq_missing(c->input_memblockq));
 455
 456         pa_sink_input_put(c->sink_input);
 457     }
 458
 459     if (p->mode & RECORD) {
 460         pa_source_output_new_data data;
 461         size_t l;
 462
 463         pa_source_output_new_data_init(&data);
 464         data.driver = __FILE__;
 465         data.name = c->client->name;
 466         pa_source_output_new_data_set_sample_spec(&data, &p->sample_spec);
 467         data.module = p->module;
 468         data.client = c->client;
 469
 470         if (!(c->source_output = pa_source_output_new(p->core, &data, 0))) {
 471             pa_log("Failed to create source output.");
 472             goto fail;
 473         }
 474         c->source_output->push = source_output_push_cb;
 475         c->source_output->kill = source_output_kill_cb;
 476         c->source_output->get_latency = source_output_get_latency_cb;
 477         c->source_output->userdata = c;
 478
 479         l = (size_t) (pa_bytes_per_second(&p->sample_spec)*RECORD_BUFFER_SECONDS);
 480         c->output_memblockq = pa_memblockq_new(
 481                 0,
 482                 l,
 483                 0,
 484                 pa_frame_size(&p->sample_spec),
 485                 1,
 486                 0,
 487                 NULL);
 488         pa_iochannel_socket_set_sndbuf(io, l/RECORD_BUFFER_FRAGMENTS*2);
 489
 490         pa_source_output_put(c->source_output);
 491     }
 492
 493
 494     pa_iochannel_set_callback(c->io, io_callback, c);
 495     pa_idxset_put(p->connections, c, NULL);
 496
 497     return;
 498
 499 fail:
 500     if (c)
 501         connection_free(c);
 502 }
 503
 504 static void asyncmsgq_cb(pa_mainloop_api*api, pa_io_event* e, int fd, pa_io_event_flags_t events, void *userdata) {
 505     pa_protocol_simple *p = userdata;
 506     int do_some_work = 0;
 507
 508     pa_assert(pa_asyncmsgq_get_fd(p->asyncmsgq) == fd);
 509     pa_assert(events == PA_IO_EVENT_INPUT);
 510
 511     pa_asyncmsgq_after_poll(p->asyncmsgq);
 512
 513     for (;;) {
 514         int code;
 515         void *object, *data;
 516
 517         /* Check whether there is a message for us to process */
 518         while (pa_asyncmsgq_get(p->asyncmsgq, &object, &code, &data) == 0) {
 519
 520             connection *c = object;
 521
 522             pa_assert(c);
 523
 524             switch (code) {
 525
 526                 case MESSAGE_REQUEST_DATA:
 527                     do_work(c);
 528                     break;
 529
 530                 case MESSAGE_POST_DATA:
 531                     pa_memblockq_push(c->output_memblockq, chunk);
 532                     do_work(c);
 533                     break;
 534             }
 535
 536             pa_asyncmsgq_done(p->asyncmsgq);
 537         }
 538
 539         if (pa_asyncmsgq_before_poll(p->asyncmsgq) == 0)
 540             break;
 541     }
 542 }
 543
 544 pa_protocol_simple* pa_protocol_simple_new(pa_core *core, pa_socket_server *server, pa_module *m, pa_modargs *ma) {
 545     pa_protocol_simple* p = NULL;
 546     int enable;
 547
 548     pa_assert(core);
 549     pa_assert(server);
 550     pa_assert(ma);
 551
 552     p = pa_xnew0(pa_protocol_simple, 1);
 553     p->module = m;
 554     p->core = core;
 555     p->server = server;
 556     p->connections = pa_idxset_new(NULL, NULL);
 557     pa_assert_se(p->asyncmsgq = pa_asyncmsgq_new(0));
 558
 559     p->sample_spec = core->default_sample_spec;
 560     if (pa_modargs_get_sample_spec(ma, &p->sample_spec) < 0) {
 561         pa_log("Failed to parse sample type specification.");
 562         goto fail;
 563     }
 564
 565     p->source_name = pa_xstrdup(pa_modargs_get_value(ma, "source", NULL));
 566     p->sink_name = pa_xstrdup(pa_modargs_get_value(ma, "sink", NULL));
 567
 568     enable = 0;
 569     if (pa_modargs_get_value_boolean(ma, "record", &enable) < 0) {
 570         pa_log("record= expects a numeric argument.");
 571         goto fail;
 572     }
 573     p->mode = enable ? RECORD : 0;
 574
 575     enable = 1;
 576     if (pa_modargs_get_value_boolean(ma, "playback", &enable) < 0) {
 577         pa_log("playback= expects a numeric argument.");
 578         goto fail;
 579     }
 580     p->mode |= enable ? PLAYBACK : 0;
 581
 582     if ((p->mode & (RECORD|PLAYBACK)) == 0) {
 583         pa_log("neither playback nor recording enabled for protocol.");
 584         goto fail;
 585     }
 586
 587     pa_socket_server_set_callback(p->server, on_connection, p);
 588
 589     pa_assert_se(pa_asyncmsgq_before_poll(p->asyncmsgq) == 0);
 590     pa_assert_se(p->asyncmsgq_event = core->mainloop->io_event_new(core->mainloop, pa_asyncmsgq_get_fd(p->asyncmsgq), PA_IO_EVENT_INPUT, p));
 591
 592     return p;
 593
 594 fail:
 595     if (p)
 596         pa_protocol_simple_free(p);
 597
 598     return NULL;
 599 }
 600
 601
 602 void pa_protocol_simple_free(pa_protocol_simple *p) {
 603     struct connection *c;
 604     pa_assert(p);
 605
 606     if (p->connections) {
 607         while((c = pa_idxset_first(p->connections, NULL)))
 608             connection_free(c);
 609
 610         pa_idxset_free(p->connections, NULL, NULL);
 611     }
 612
 613     if (p->server)
 614         pa_socket_server_unref(p->server);
 615
 616     if (p->asyncmsgq) {
 617         c->mainloop->io_event_free(c->asyncmsgq_event);
 618         pa_asyncmsgq_after_poll(c->asyncmsgq);
 619         pa_asyncmsgq_free(p->asyncmsgq);
 620     }
 621
 622     pa_xfree(p);
 623 }
 624