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