code.delx.au - pulseaudio/blob - 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_peek_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     r = pa_memblockq_peek(c->input_memblockq, chunk);
 356
 357 /*     pa_log("peeked %u %i", r >= 0 ? chunk->length: 0, r); */
 358
 359     if (c->dead && r < 0)
 360         pa_asyncmsgq_post(pa_thread_mq_get()->outq, PA_MSGOBJECT(c), CONNECTION_MESSAGE_UNLINK_CONNECTION, NULL, 0, NULL, NULL);
 361
 362     return r;
 363 }
 364
 365 /* Called from thread context */
 366 static void sink_input_drop_cb(pa_sink_input *i, size_t length) {
 367     connection *c;
 368     size_t old, new;
 369
 370     pa_assert(i);
 371     c = CONNECTION(i->userdata);
 372     connection_assert_ref(c);
 373     pa_assert(length);
 374
 375     old = pa_memblockq_missing(c->input_memblockq);
 376     pa_memblockq_drop(c->input_memblockq, length);
 377     new = pa_memblockq_missing(c->input_memblockq);
 378
 379     if (new > old) {
 380         if (pa_atomic_add(&c->playback.missing, new - old) <= 0)
 381             pa_asyncmsgq_post(pa_thread_mq_get()->outq, PA_MSGOBJECT(c), CONNECTION_MESSAGE_REQUEST_DATA, NULL, 0, NULL, NULL);
 382     }
 383 }
 384
 385 /* Called from main context */
 386 static void sink_input_kill_cb(pa_sink_input *i) {
 387     pa_sink_input_assert_ref(i);
 388
 389     connection_unlink(CONNECTION(i->userdata));
 390 }
 391
 392 /*** source_output callbacks ***/
 393
 394 /* Called from thread context */
 395 static void source_output_push_cb(pa_source_output *o, const pa_memchunk *chunk) {
 396     connection *c;
 397
 398     pa_assert(o);
 399     c = CONNECTION(o->userdata);
 400     pa_assert(c);
 401     pa_assert(chunk);
 402
 403     pa_asyncmsgq_post(pa_thread_mq_get()->outq, PA_MSGOBJECT(c), CONNECTION_MESSAGE_POST_DATA, NULL, 0, chunk, NULL);
 404 }
 405
 406 /* Called from main context */
 407 static void source_output_kill_cb(pa_source_output *o) {
 408     pa_source_output_assert_ref(o);
 409
 410     connection_unlink(CONNECTION(o->userdata));
 411 }
 412
 413 /* Called from main context */
 414 static pa_usec_t source_output_get_latency_cb(pa_source_output *o) {
 415     connection*c;
 416
 417     pa_assert(o);
 418     c = CONNECTION(o->userdata);
 419     pa_assert(c);
 420
 421     return pa_bytes_to_usec(pa_memblockq_get_length(c->output_memblockq), &c->source_output->sample_spec);
 422 }
 423
 424 /*** client callbacks ***/
 425
 426 static void client_kill_cb(pa_client *client) {
 427     connection*c;
 428
 429     pa_assert(client);
 430     c = CONNECTION(client->userdata);
 431     pa_assert(c);
 432
 433     connection_unlink(c);
 434 }
 435
 436 /*** pa_iochannel callbacks ***/
 437
 438 static void io_callback(pa_iochannel*io, void *userdata) {
 439     connection *c = CONNECTION(userdata);
 440
 441     connection_assert_ref(c);
 442     pa_assert(io);
 443
 444     do_work(c);
 445 }
 446
 447 /*** socket_server callbacks ***/
 448
 449 static void on_connection(pa_socket_server*s, pa_iochannel *io, void *userdata) {
 450     pa_protocol_simple *p = userdata;
 451     connection *c = NULL;
 452     char cname[256];
 453
 454     pa_assert(s);
 455     pa_assert(io);
 456     pa_assert(p);
 457
 458     if (pa_idxset_size(p->connections)+1 > MAX_CONNECTIONS) {
 459         pa_log("Warning! Too many connections (%u), dropping incoming connection.", MAX_CONNECTIONS);
 460         pa_iochannel_free(io);
 461         return;
 462     }
 463
 464     c = pa_msgobject_new(connection);
 465     c->parent.parent.free = connection_free;
 466     c->parent.process_msg = connection_process_msg;
 467     c->io = io;
 468     c->sink_input = NULL;
 469     c->source_output = NULL;
 470     c->input_memblockq = c->output_memblockq = NULL;
 471     c->protocol = p;
 472     c->playback.current_memblock = NULL;
 473     c->playback.memblock_index = 0;
 474     c->playback.fragment_size = 0;
 475     c->dead = 0;
 476     pa_atomic_store(&c->playback.missing, 0);
 477
 478     pa_iochannel_socket_peer_to_string(io, cname, sizeof(cname));
 479     pa_assert_se(c->client = pa_client_new(p->core, __FILE__, cname));
 480     c->client->owner = p->module;
 481     c->client->kill = client_kill_cb;
 482     c->client->userdata = c;
 483
 484     if (p->mode & PLAYBACK) {
 485         pa_sink_input_new_data data;
 486         size_t l;
 487
 488         pa_sink_input_new_data_init(&data);
 489         data.driver = __FILE__;
 490         data.name = c->client->name;
 491         pa_sink_input_new_data_set_sample_spec(&data, &p->sample_spec);
 492         data.module = p->module;
 493         data.client = c->client;
 494
 495         if (!(c->sink_input = pa_sink_input_new(p->core, &data, 0))) {
 496             pa_log("Failed to create sink input.");
 497             goto fail;
 498         }
 499
 500         c->sink_input->parent.process_msg = sink_input_process_msg;
 501         c->sink_input->peek = sink_input_peek_cb;
 502         c->sink_input->drop = sink_input_drop_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                 NULL);
 515         pa_iochannel_socket_set_rcvbuf(io, l/PLAYBACK_BUFFER_FRAGMENTS*5);
 516         c->playback.fragment_size = l/PLAYBACK_BUFFER_FRAGMENTS;
 517
 518         pa_atomic_store(&c->playback.missing, pa_memblockq_missing(c->input_memblockq));
 519
 520         pa_sink_input_put(c->sink_input);
 521     }
 522
 523     if (p->mode & RECORD) {
 524         pa_source_output_new_data data;
 525         size_t l;
 526
 527         pa_source_output_new_data_init(&data);
 528         data.driver = __FILE__;
 529         data.name = c->client->name;
 530         pa_source_output_new_data_set_sample_spec(&data, &p->sample_spec);
 531         data.module = p->module;
 532         data.client = c->client;
 533
 534         if (!(c->source_output = pa_source_output_new(p->core, &data, 0))) {
 535             pa_log("Failed to create source output.");
 536             goto fail;
 537         }
 538         c->source_output->push = source_output_push_cb;
 539         c->source_output->kill = source_output_kill_cb;
 540         c->source_output->get_latency = source_output_get_latency_cb;
 541         c->source_output->userdata = c;
 542
 543         l = (size_t) (pa_bytes_per_second(&p->sample_spec)*RECORD_BUFFER_SECONDS);
 544         c->output_memblockq = pa_memblockq_new(
 545                 0,
 546                 l,
 547                 0,
 548                 pa_frame_size(&p->sample_spec),
 549                 1,
 550                 0,
 551                 NULL);
 552         pa_iochannel_socket_set_sndbuf(io, l/RECORD_BUFFER_FRAGMENTS*2);
 553
 554         pa_source_output_put(c->source_output);
 555     }
 556
 557     pa_iochannel_set_callback(c->io, io_callback, c);
 558     pa_idxset_put(p->connections, c, NULL);
 559
 560     return;
 561
 562 fail:
 563     if (c)
 564         connection_unlink(c);
 565 }
 566
 567 pa_protocol_simple* pa_protocol_simple_new(pa_core *core, pa_socket_server *server, pa_module *m, pa_modargs *ma) {
 568     pa_protocol_simple* p = NULL;
 569     pa_bool_t enable;
 570
 571     pa_assert(core);
 572     pa_assert(server);
 573     pa_assert(ma);
 574
 575     p = pa_xnew0(pa_protocol_simple, 1);
 576     p->module = m;
 577     p->core = core;
 578     p->server = server;
 579     p->connections = pa_idxset_new(NULL, NULL);
 580
 581     p->sample_spec = core->default_sample_spec;
 582     if (pa_modargs_get_sample_spec(ma, &p->sample_spec) < 0) {
 583         pa_log("Failed to parse sample type specification.");
 584         goto fail;
 585     }
 586
 587     p->source_name = pa_xstrdup(pa_modargs_get_value(ma, "source", NULL));
 588     p->sink_name = pa_xstrdup(pa_modargs_get_value(ma, "sink", NULL));
 589
 590     enable = FALSE;
 591     if (pa_modargs_get_value_boolean(ma, "record", &enable) < 0) {
 592         pa_log("record= expects a numeric argument.");
 593         goto fail;
 594     }
 595     p->mode = enable ? RECORD : 0;
 596
 597     enable = 1;
 598     if (pa_modargs_get_value_boolean(ma, "playback", &enable) < 0) {
 599         pa_log("playback= expects a numeric argument.");
 600         goto fail;
 601     }
 602     p->mode |= enable ? PLAYBACK : 0;
 603
 604     if ((p->mode & (RECORD|PLAYBACK)) == 0) {
 605         pa_log("neither playback nor recording enabled for protocol.");
 606         goto fail;
 607     }
 608
 609     pa_socket_server_set_callback(p->server, on_connection, p);
 610
 611     return p;
 612
 613 fail:
 614     if (p)
 615         pa_protocol_simple_free(p);
 616
 617     return NULL;
 618 }
 619
 620
 621 void pa_protocol_simple_free(pa_protocol_simple *p) {
 622     connection *c;
 623     pa_assert(p);
 624
 625     if (p->connections) {
 626         while((c = pa_idxset_first(p->connections, NULL)))
 627             connection_unlink(c);
 628
 629         pa_idxset_free(p->connections, NULL, NULL);
 630     }
 631
 632     if (p->server)
 633         pa_socket_server_unref(p->server);
 634
 635     pa_xfree(p);
 636 }