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 #include <pulsecore/atomic.h>
  44 #include <pulsecore/thread-mq.h>
  45
  46 #include "protocol-simple.h"
  47
  48 /* Don't allow more than this many concurrent connections */
  49 #define MAX_CONNECTIONS 10
  50
  51 typedef struct connection {
  52     pa_msgobject parent;
  53     pa_protocol_simple *protocol;
  54     pa_iochannel *io;
  55     pa_sink_input *sink_input;
  56     pa_source_output *source_output;
  57     pa_client *client;
  58     pa_memblockq *input_memblockq, *output_memblockq;
  59
  60     int dead;
  61
  62     struct {
  63         pa_memblock *current_memblock;
  64         size_t memblock_index, fragment_size;
  65         pa_atomic_t missing;
  66     } playback;
  67 } connection;
  68
  69 PA_DECLARE_CLASS(connection);
  70 #define CONNECTION(o) (connection_cast(o))
  71 static PA_DEFINE_CHECK_TYPE(connection, pa_msgobject);
  72
  73 struct pa_protocol_simple {
  74     pa_module *module;
  75     pa_core *core;
  76     pa_socket_server*server;
  77     pa_idxset *connections;
  78
  79     enum {
  80         RECORD = 1,
  81         PLAYBACK = 2,
  82         DUPLEX = 3
  83     } mode;
  84
  85     pa_sample_spec sample_spec;
  86     char *source_name, *sink_name;
  87 };
  88
  89 enum {
  90     SINK_INPUT_MESSAGE_POST_DATA = PA_SINK_INPUT_MESSAGE_MAX, /* data from main loop to sink input */
  91     SINK_INPUT_MESSAGE_DISABLE_PREBUF /* disabled prebuf, get playback started. */
  92 };
  93
  94 enum {
  95     CONNECTION_MESSAGE_REQUEST_DATA,      /* data requested from sink input from the main loop */
  96     CONNECTION_MESSAGE_POST_DATA,         /* data from source output to main loop */
  97     CONNECTION_MESSAGE_UNLINK_CONNECTION    /* Please drop a aconnection now */
  98 };
  99
 100
 101 #define PLAYBACK_BUFFER_SECONDS (.5)
 102 #define PLAYBACK_BUFFER_FRAGMENTS (10)
 103 #define RECORD_BUFFER_SECONDS (5)
 104 #define RECORD_BUFFER_FRAGMENTS (100)
 105
 106 static void connection_unlink(connection *c) {
 107     pa_assert(c);
 108
 109     if (!c->protocol)
 110         return;
 111
 112     if (c->sink_input) {
 113         pa_sink_input_unlink(c->sink_input);
 114         pa_sink_input_unref(c->sink_input);
 115         c->sink_input = NULL;
 116     }
 117
 118     if (c->source_output) {
 119         pa_source_output_unlink(c->source_output);
 120         pa_source_output_unref(c->source_output);
 121         c->source_output = NULL;
 122     }
 123
 124     if (c->client) {
 125         pa_client_free(c->client);
 126         c->client = NULL;
 127     }
 128
 129     if (c->io) {
 130         pa_iochannel_free(c->io);
 131         c->io = NULL;
 132     }
 133
 134     pa_assert_se(pa_idxset_remove_by_data(c->protocol->connections, c, NULL) == c);
 135     c->protocol = NULL;
 136     connection_unref(c);
 137 }
 138
 139 static void connection_free(pa_object *o) {
 140     connection *c = CONNECTION(o);
 141     pa_assert(c);
 142
 143     connection_unref(c);
 144
 145     if (c->playback.current_memblock)
 146         pa_memblock_unref(c->playback.current_memblock);
 147
 148     if (c->input_memblockq)
 149         pa_memblockq_free(c->input_memblockq);
 150     if (c->output_memblockq)
 151         pa_memblockq_free(c->output_memblockq);
 152
 153     pa_xfree(c);
 154 }
 155
 156 static int do_read(connection *c) {
 157     pa_memchunk chunk;
 158     ssize_t r;
 159     size_t l;
 160     void *p;
 161
 162     connection_assert_ref(c);
 163
 164     if (!c->sink_input || (l = pa_atomic_load(&c->playback.missing)) <= 0)
 165         return 0;
 166
 167     if (l > c->playback.fragment_size)
 168         l = c->playback.fragment_size;
 169
 170     if (c->playback.current_memblock)
 171         if (pa_memblock_get_length(c->playback.current_memblock) - c->playback.memblock_index < l) {
 172             pa_memblock_unref(c->playback.current_memblock);
 173             c->playback.current_memblock = NULL;
 174             c->playback.memblock_index = 0;
 175         }
 176
 177     if (!c->playback.current_memblock) {
 178         pa_assert_se(c->playback.current_memblock = pa_memblock_new(c->protocol->core->mempool, l));
 179         c->playback.memblock_index = 0;
 180     }
 181
 182     p = pa_memblock_acquire(c->playback.current_memblock);
 183     r = pa_iochannel_read(c->io, (uint8_t*) p + c->playback.memblock_index, l);
 184     pa_memblock_release(c->playback.current_memblock);
 185
 186     if (r <= 0) {
 187
 188         if (r < 0 && (errno == EINTR || errno == EAGAIN))
 189             return 0;
 190
 191         pa_log_debug("read(): %s", r == 0 ? "EOF" : pa_cstrerror(errno));
 192         return -1;
 193     }
 194
 195     chunk.memblock = c->playback.current_memblock;
 196     chunk.index = c->playback.memblock_index;
 197     chunk.length = r;
 198
 199     c->playback.memblock_index += r;
 200
 201     pa_asyncmsgq_post(c->sink_input->sink->asyncmsgq, PA_MSGOBJECT(c->sink_input), SINK_INPUT_MESSAGE_POST_DATA, NULL, 0, &chunk, NULL);
 202     pa_atomic_sub(&c->playback.missing, r);
 203
 204     return 0;
 205 }
 206
 207 static int do_write(connection *c) {
 208     pa_memchunk chunk;
 209     ssize_t r;
 210     void *p;
 211
 212     connection_assert_ref(c);
 213
 214     if (!c->source_output)
 215         return 0;
 216
 217     if (pa_memblockq_peek(c->output_memblockq, &chunk) < 0) {
 218 /*         pa_log("peek failed"); */
 219         return 0;
 220     }
 221
 222     pa_assert(chunk.memblock);
 223     pa_assert(chunk.length);
 224
 225     p = pa_memblock_acquire(chunk.memblock);
 226     r = pa_iochannel_write(c->io, (uint8_t*) p+chunk.index, chunk.length);
 227     pa_memblock_release(chunk.memblock);
 228
 229     pa_memblock_unref(chunk.memblock);
 230
 231     if (r < 0) {
 232
 233         if (errno == EINTR || errno == EAGAIN)
 234             return 0;
 235
 236         pa_log("write(): %s", pa_cstrerror(errno));
 237         return -1;
 238     }
 239
 240     pa_memblockq_drop(c->output_memblockq, r);
 241
 242     return 0;
 243 }
 244
 245 static void do_work(connection *c) {
 246     connection_assert_ref(c);
 247
 248     if (c->dead)
 249         return;
 250
 251     if (pa_iochannel_is_readable(c->io)) {
 252         if (do_read(c) < 0)
 253             goto fail;
 254     } else if (pa_iochannel_is_hungup(c->io))
 255         goto fail;
 256
 257     if (pa_iochannel_is_writable(c->io)) {
 258         if (do_write(c) < 0)
 259             goto fail;
 260     }
 261
 262     return;
 263
 264 fail:
 265
 266     if (c->sink_input) {
 267
 268         /* If there is a sink input, we first drain what we already have read before shutting down the connection */
 269         c->dead = 1;
 270
 271         pa_iochannel_free(c->io);
 272         c->io = NULL;
 273
 274         pa_asyncmsgq_post(c->sink_input->sink->asyncmsgq, PA_MSGOBJECT(c->sink_input), SINK_INPUT_MESSAGE_DISABLE_PREBUF, NULL, 0, NULL, NULL);
 275     } else
 276         connection_unlink(c);
 277 }
 278
 279 static int connection_process_msg(pa_msgobject *o, int code, void*userdata, int64_t offset, pa_memchunk *chunk) {
 280     connection *c = CONNECTION(o);
 281     connection_assert_ref(c);
 282
 283     switch (code) {
 284         case CONNECTION_MESSAGE_REQUEST_DATA:
 285             do_work(c);
 286             break;
 287
 288         case CONNECTION_MESSAGE_POST_DATA:
 289 /*             pa_log("got data %u", chunk->length); */
 290             pa_memblockq_push_align(c->output_memblockq, chunk);
 291             do_work(c);
 292             break;
 293
 294         case CONNECTION_MESSAGE_UNLINK_CONNECTION:
 295             connection_unlink(c);
 296             break;
 297     }
 298
 299     return 0;
 300 }
 301
 302 /*** sink_input callbacks ***/
 303
 304 /* Called from thread context */
 305 static int sink_input_process_msg(pa_msgobject *o, int code, void *userdata, int64_t offset, pa_memchunk *chunk) {
 306     pa_sink_input *i = PA_SINK_INPUT(o);
 307     connection*c;
 308
 309     pa_sink_input_assert_ref(i);
 310     c = CONNECTION(i->userdata);
 311     connection_assert_ref(c);
 312
 313     switch (code) {
 314
 315         case SINK_INPUT_MESSAGE_POST_DATA: {
 316             pa_assert(chunk);
 317
 318             /* New data from the main loop */
 319             pa_memblockq_push_align(c->input_memblockq, chunk);
 320
 321 /*             pa_log("got data, %u", pa_memblockq_get_length(c->input_memblockq)); */
 322
 323             return 0;
 324         }
 325
 326         case SINK_INPUT_MESSAGE_DISABLE_PREBUF: {
 327             pa_memblockq_prebuf_disable(c->input_memblockq);
 328             return 0;
 329         }
 330
 331         case PA_SINK_INPUT_MESSAGE_GET_LATENCY: {
 332             pa_usec_t *r = userdata;
 333
 334             *r = pa_bytes_to_usec(pa_memblockq_get_length(c->input_memblockq), &c->sink_input->sample_spec);
 335
 336             /* Fall through, the default handler will add in the extra
 337              * latency added by the resampler */
 338         }
 339
 340         default:
 341             return pa_sink_input_process_msg(o, code, userdata, offset, chunk);
 342     }
 343 }
 344
 345 /* Called from thread context */
 346 static int sink_input_pop_cb(pa_sink_input *i, size_t length, pa_memchunk *chunk) {
 347     connection *c;
 348     int r;
 349
 350     pa_assert(i);
 351     c = CONNECTION(i->userdata);
 352     connection_assert_ref(c);
 353     pa_assert(chunk);
 354
 355     if ((r = pa_memblockq_peek(c->input_memblockq, chunk)) < 0) {
 356
 357         if (c->dead && pa_sink_input_safe_to_remove(i))
 358             pa_asyncmsgq_post(pa_thread_mq_get()->outq, PA_MSGOBJECT(c), CONNECTION_MESSAGE_UNLINK_CONNECTION, NULL, 0, NULL, NULL);
 359
 360     } else {
 361         size_t old, new;
 362
 363         old = pa_memblockq_missing(c->input_memblockq);
 364         pa_memblockq_drop(c->input_memblockq, chunk->length);
 365         new = pa_memblockq_missing(c->input_memblockq);
 366
 367         if (new > old) {
 368             if (pa_atomic_add(&c->playback.missing, new - old) <= 0)
 369                 pa_asyncmsgq_post(pa_thread_mq_get()->outq, PA_MSGOBJECT(c), CONNECTION_MESSAGE_REQUEST_DATA, NULL, 0, NULL, NULL);
 370         }
 371     }
 372
 373     return r;
 374 }
 375
 376 /* Called from main context */
 377 static void sink_input_kill_cb(pa_sink_input *i) {
 378     pa_sink_input_assert_ref(i);
 379
 380     connection_unlink(CONNECTION(i->userdata));
 381 }
 382
 383 /*** source_output callbacks ***/
 384
 385 /* Called from thread context */
 386 static void source_output_push_cb(pa_source_output *o, const pa_memchunk *chunk) {
 387     connection *c;
 388
 389     pa_assert(o);
 390     c = CONNECTION(o->userdata);
 391     pa_assert(c);
 392     pa_assert(chunk);
 393
 394     pa_asyncmsgq_post(pa_thread_mq_get()->outq, PA_MSGOBJECT(c), CONNECTION_MESSAGE_POST_DATA, NULL, 0, chunk, NULL);
 395 }
 396
 397 /* Called from main context */
 398 static void source_output_kill_cb(pa_source_output *o) {
 399     pa_source_output_assert_ref(o);
 400
 401     connection_unlink(CONNECTION(o->userdata));
 402 }
 403
 404 /* Called from main context */
 405 static pa_usec_t source_output_get_latency_cb(pa_source_output *o) {
 406     connection*c;
 407
 408     pa_assert(o);
 409     c = CONNECTION(o->userdata);
 410     pa_assert(c);
 411
 412     return pa_bytes_to_usec(pa_memblockq_get_length(c->output_memblockq), &c->source_output->sample_spec);
 413 }
 414
 415 /*** client callbacks ***/
 416
 417 static void client_kill_cb(pa_client *client) {
 418     connection*c;
 419
 420     pa_assert(client);
 421     c = CONNECTION(client->userdata);
 422     pa_assert(c);
 423
 424     connection_unlink(c);
 425 }
 426
 427 /*** pa_iochannel callbacks ***/
 428
 429 static void io_callback(pa_iochannel*io, void *userdata) {
 430     connection *c = CONNECTION(userdata);
 431
 432     connection_assert_ref(c);
 433     pa_assert(io);
 434
 435     do_work(c);
 436 }
 437
 438 /*** socket_server callbacks ***/
 439
 440 static void on_connection(pa_socket_server*s, pa_iochannel *io, void *userdata) {
 441     pa_protocol_simple *p = userdata;
 442     connection *c = NULL;
 443     char cname[256];
 444
 445     pa_assert(s);
 446     pa_assert(io);
 447     pa_assert(p);
 448
 449     if (pa_idxset_size(p->connections)+1 > MAX_CONNECTIONS) {
 450         pa_log("Warning! Too many connections (%u), dropping incoming connection.", MAX_CONNECTIONS);
 451         pa_iochannel_free(io);
 452         return;
 453     }
 454
 455     c = pa_msgobject_new(connection);
 456     c->parent.parent.free = connection_free;
 457     c->parent.process_msg = connection_process_msg;
 458     c->io = io;
 459     c->sink_input = NULL;
 460     c->source_output = NULL;
 461     c->input_memblockq = c->output_memblockq = NULL;
 462     c->protocol = p;
 463     c->playback.current_memblock = NULL;
 464     c->playback.memblock_index = 0;
 465     c->playback.fragment_size = 0;
 466     c->dead = 0;
 467     pa_atomic_store(&c->playback.missing, 0);
 468
 469     pa_iochannel_socket_peer_to_string(io, cname, sizeof(cname));
 470     pa_assert_se(c->client = pa_client_new(p->core, __FILE__, cname));
 471     c->client->module = p->module;
 472     c->client->kill = client_kill_cb;
 473     c->client->userdata = c;
 474
 475     if (p->mode & PLAYBACK) {
 476         pa_sink_input_new_data data;
 477         size_t l;
 478         pa_sink *sink;
 479
 480         if (!(sink = pa_namereg_get(c->protocol->core, c->protocol->sink_name, PA_NAMEREG_SINK, TRUE))) {
 481             pa_log("Failed to get sink.");
 482             goto fail;
 483         }
 484
 485         pa_sink_input_new_data_init(&data);
 486         data.driver = __FILE__;
 487         data.module = p->module;
 488         data.client = c->client;
 489         data.sink = sink;
 490         pa_proplist_update(data.proplist, PA_UPDATE_MERGE, c->client->proplist);
 491         pa_sink_input_new_data_set_sample_spec(&data, &p->sample_spec);
 492
 493         c->sink_input = pa_sink_input_new(p->core, &data, 0);
 494         pa_sink_input_new_data_done(&data);
 495
 496         if (!c->sink_input) {
 497             pa_log("Failed to create sink input.");
 498             goto fail;
 499         }
 500
 501         c->sink_input->parent.process_msg = sink_input_process_msg;
 502         c->sink_input->pop = sink_input_pop_cb;
 503         c->sink_input->kill = sink_input_kill_cb;
 504         c->sink_input->userdata = c;
 505
 506         l = (size_t) (pa_bytes_per_second(&p->sample_spec)*PLAYBACK_BUFFER_SECONDS);
 507         c->input_memblockq = pa_memblockq_new(
 508                 0,
 509                 l,
 510                 0,
 511                 pa_frame_size(&p->sample_spec),
 512                 (size_t) -1,
 513                 l/PLAYBACK_BUFFER_FRAGMENTS,
 514                 0,
 515                 NULL);
 516         pa_iochannel_socket_set_rcvbuf(io, l/PLAYBACK_BUFFER_FRAGMENTS*5);
 517         c->playback.fragment_size = l/PLAYBACK_BUFFER_FRAGMENTS;
 518
 519         pa_atomic_store(&c->playback.missing, pa_memblockq_missing(c->input_memblockq));
 520
 521         pa_sink_input_put(c->sink_input);
 522     }
 523
 524     if (p->mode & RECORD) {
 525         pa_source_output_new_data data;
 526         size_t l;
 527         pa_source *source;
 528
 529         if (!(source = pa_namereg_get(c->protocol->core, c->protocol->source_name, PA_NAMEREG_SOURCE, TRUE))) {
 530             pa_log("Failed to get source.");
 531             goto fail;
 532         }
 533
 534         pa_source_output_new_data_init(&data);
 535         data.driver = __FILE__;
 536         data.module = p->module;
 537         data.client = c->client;
 538         data.source = source;
 539         pa_proplist_update(data.proplist, PA_UPDATE_MERGE, c->client->proplist);
 540         pa_source_output_new_data_set_sample_spec(&data, &p->sample_spec);
 541
 542         c->source_output = pa_source_output_new(p->core, &data, 0);
 543         pa_source_output_new_data_done(&data);
 544
 545         if (!c->source_output) {
 546             pa_log("Failed to create source output.");
 547             goto fail;
 548         }
 549         c->source_output->push = source_output_push_cb;
 550         c->source_output->kill = source_output_kill_cb;
 551         c->source_output->get_latency = source_output_get_latency_cb;
 552         c->source_output->userdata = c;
 553
 554         l = (size_t) (pa_bytes_per_second(&p->sample_spec)*RECORD_BUFFER_SECONDS);
 555         c->output_memblockq = pa_memblockq_new(
 556                 0,
 557                 l,
 558                 0,
 559                 pa_frame_size(&p->sample_spec),
 560                 1,
 561                 0,
 562                 0,
 563                 NULL);
 564         pa_iochannel_socket_set_sndbuf(io, l/RECORD_BUFFER_FRAGMENTS*2);
 565
 566         pa_source_output_put(c->source_output);
 567     }
 568
 569     pa_iochannel_set_callback(c->io, io_callback, c);
 570     pa_idxset_put(p->connections, c, NULL);
 571
 572     return;
 573
 574 fail:
 575     if (c)
 576         connection_unlink(c);
 577 }
 578
 579 pa_protocol_simple* pa_protocol_simple_new(pa_core *core, pa_socket_server *server, pa_module *m, pa_modargs *ma) {
 580     pa_protocol_simple* p = NULL;
 581     pa_bool_t enable;
 582
 583     pa_assert(core);
 584     pa_assert(server);
 585     pa_assert(ma);
 586
 587     p = pa_xnew0(pa_protocol_simple, 1);
 588     p->module = m;
 589     p->core = core;
 590     p->server = server;
 591     p->connections = pa_idxset_new(NULL, NULL);
 592
 593     p->sample_spec = core->default_sample_spec;
 594     if (pa_modargs_get_sample_spec(ma, &p->sample_spec) < 0) {
 595         pa_log("Failed to parse sample type specification.");
 596         goto fail;
 597     }
 598
 599     p->source_name = pa_xstrdup(pa_modargs_get_value(ma, "source", NULL));
 600     p->sink_name = pa_xstrdup(pa_modargs_get_value(ma, "sink", NULL));
 601
 602     enable = FALSE;
 603     if (pa_modargs_get_value_boolean(ma, "record", &enable) < 0) {
 604         pa_log("record= expects a numeric argument.");
 605         goto fail;
 606     }
 607     p->mode = enable ? RECORD : 0;
 608
 609     enable = 1;
 610     if (pa_modargs_get_value_boolean(ma, "playback", &enable) < 0) {
 611         pa_log("playback= expects a numeric argument.");
 612         goto fail;
 613     }
 614     p->mode |= enable ? PLAYBACK : 0;
 615
 616     if ((p->mode & (RECORD|PLAYBACK)) == 0) {
 617         pa_log("neither playback nor recording enabled for protocol.");
 618         goto fail;
 619     }
 620
 621     pa_socket_server_set_callback(p->server, on_connection, p);
 622
 623     return p;
 624
 625 fail:
 626     if (p)
 627         pa_protocol_simple_free(p);
 628
 629     return NULL;
 630 }
 631
 632
 633 void pa_protocol_simple_free(pa_protocol_simple *p) {
 634     connection *c;
 635     pa_assert(p);
 636
 637     if (p->connections) {
 638         while((c = pa_idxset_first(p->connections, NULL)))
 639             connection_unlink(c);
 640
 641         pa_idxset_free(p->connections, NULL, NULL);
 642     }
 643
 644     if (p->server)
 645         pa_socket_server_unref(p->server);
 646
 647     pa_xfree(p);
 648 }