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