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
  71 static PA_DEFINE_CHECK_TYPE(connection, connection_check_type, pa_msgobject_check_type);
  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     MESSAGE_REQUEST_DATA,      /* data requested from sink input from the main loop */
  96     MESSAGE_POST_DATA,         /* data from source output to main loop */
  97     MESSAGE_DROP_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_free(pa_object *o) {
 107     connection *c = CONNECTION(o);
 108     pa_assert(c);
 109
 110     if (c->playback.current_memblock)
 111         pa_memblock_unref(c->playback.current_memblock);
 112
 113     if (c->io)
 114         pa_iochannel_free(c->io);
 115     if (c->input_memblockq)
 116         pa_memblockq_free(c->input_memblockq);
 117     if (c->output_memblockq)
 118         pa_memblockq_free(c->output_memblockq);
 119
 120     pa_xfree(c);
 121 }
 122
 123 static void connection_drop(connection *c) {
 124     pa_assert(c);
 125
 126     pa_idxset_remove_by_data(c->protocol->connections, c, NULL);
 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)))
 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
 195     return 0;
 196 }
 197
 198 static int do_write(connection *c) {
 199     pa_memchunk chunk;
 200     ssize_t r;
 201     void *p;
 202
 203     pa_assert(c);
 204
 205     if (!c->source_output)
 206         return 0;
 207
 208     if (pa_memblockq_peek(c->output_memblockq, &chunk) < 0) {
 209 /*         pa_log("peek failed"); */
 210         return 0;
 211     }
 212
 213     pa_assert(chunk.memblock);
 214     pa_assert(chunk.length);
 215
 216     p = pa_memblock_acquire(chunk.memblock);
 217     r = pa_iochannel_write(c->io, (uint8_t*) p+chunk.index, chunk.length);
 218     pa_memblock_release(chunk.memblock);
 219
 220     pa_memblock_unref(chunk.memblock);
 221
 222     if (r < 0) {
 223
 224         if (errno == EINTR || errno == EAGAIN)
 225             return 0;
 226
 227         pa_log("write(): %s", pa_cstrerror(errno));
 228         return -1;
 229     }
 230
 231     pa_memblockq_drop(c->output_memblockq, &chunk, r);
 232
 233     return 0;
 234 }
 235
 236 static void do_work(connection *c) {
 237     pa_assert(c);
 238
 239     if (c->dead)
 240         return;
 241
 242     if (pa_iochannel_is_readable(c->io)) {
 243         if (do_read(c) < 0)
 244             goto fail;
 245     } else if (pa_iochannel_is_hungup(c->io))
 246         goto fail;
 247
 248     if (pa_iochannel_is_writable(c->io)) {
 249         if (do_write(c) < 0)
 250             goto fail;
 251     }
 252
 253     return;
 254
 255 fail:
 256
 257     if (c->sink_input) {
 258
 259         /* If there is a sink input, we first drain what we already have read before shutting down the connection */
 260         c->dead = 1;
 261
 262         pa_iochannel_free(c->io);
 263         c->io = NULL;
 264
 265         pa_asyncmsgq_post(c->sink_input->sink->asyncmsgq, PA_MSGOBJECT(c->sink_input), SINK_INPUT_MESSAGE_DISABLE_PREBUF, NULL, NULL, NULL);
 266     } else
 267         connection_drop(c);
 268 }
 269
 270 static int connection_process_msg(pa_msgobject *o, int code, void*userdata, pa_memchunk *chunk) {
 271     connection *c = CONNECTION(o);
 272
 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             pa_atomic_store(&c->playback.missing, pa_memblockq_missing(c->input_memblockq));
 313
 314 /*             pa_log("got data, %u", pa_memblockq_get_length(c->input_memblockq)); */
 315
 316             return 0;
 317         }
 318
 319         case SINK_INPUT_MESSAGE_DISABLE_PREBUF: {
 320             pa_memblockq_prebuf_disable(c->input_memblockq);
 321             return 0;
 322         }
 323
 324         case PA_SINK_INPUT_MESSAGE_GET_LATENCY: {
 325             pa_usec_t *r = userdata;
 326
 327             *r = pa_bytes_to_usec(pa_memblockq_get_length(c->input_memblockq), &c->sink_input->sample_spec);
 328
 329             /* Fall through, the default handler will add in the extra
 330              * latency added by the resampler */
 331         }
 332
 333         default:
 334             return pa_sink_input_process_msg(o, code, userdata, chunk);
 335     }
 336 }
 337
 338 /* Called from thread context */
 339 static int sink_input_peek_cb(pa_sink_input *i, pa_memchunk *chunk) {
 340     connection*c;
 341     int r;
 342
 343     pa_assert(i);
 344     c = i->userdata;
 345     pa_assert(c);
 346     pa_assert(chunk);
 347
 348     r = pa_memblockq_peek(c->input_memblockq, chunk);
 349
 350 /*     pa_log("peeked %u %i", r >= 0 ? chunk->length: 0, r); */
 351
 352     if (c->dead && r < 0)
 353         pa_asyncmsgq_post(c->protocol->core->asyncmsgq, PA_MSGOBJECT(c), MESSAGE_DROP_CONNECTION, c, NULL, NULL);
 354
 355     return r;
 356 }
 357
 358 /* Called from thread context */
 359 static void sink_input_drop_cb(pa_sink_input *i, const pa_memchunk *chunk, size_t length) {
 360     connection*c = i->userdata;
 361     size_t old, new;
 362
 363     pa_assert(i);
 364     pa_assert(c);
 365     pa_assert(length);
 366
 367     old = pa_memblockq_missing(c->input_memblockq);
 368     pa_memblockq_drop(c->input_memblockq, chunk, length);
 369     new = pa_memblockq_missing(c->input_memblockq);
 370
 371     pa_atomic_store(&c->playback.missing, new);
 372
 373     if (new > old)
 374         pa_asyncmsgq_post(c->protocol->core->asyncmsgq, PA_MSGOBJECT(c), MESSAGE_REQUEST_DATA, NULL, NULL, NULL);
 375 }
 376
 377 /* Called from main context */
 378 static void sink_input_kill_cb(pa_sink_input *i) {
 379     pa_assert(i);
 380     pa_assert(i->userdata);
 381
 382     connection_drop((connection *) i->userdata);
 383 }
 384
 385 /*** source_output callbacks ***/
 386
 387 static void source_output_push_cb(pa_source_output *o, const pa_memchunk *chunk) {
 388     connection *c;
 389
 390     pa_assert(o);
 391     c = o->userdata;
 392     pa_assert(c);
 393     pa_assert(chunk);
 394
 395     pa_asyncmsgq_post(c->protocol->core->asyncmsgq, PA_MSGOBJECT(c), MESSAGE_POST_DATA, NULL, chunk, NULL);
 396 }
 397
 398 static void source_output_kill_cb(pa_source_output *o) {
 399     connection*c;
 400
 401     pa_assert(o);
 402     c = o->userdata;
 403     pa_assert(c);
 404
 405     connection_drop(c);
 406 }
 407
 408 static pa_usec_t source_output_get_latency_cb(pa_source_output *o) {
 409     connection*c;
 410
 411     pa_assert(o);
 412     c = o->userdata;
 413     pa_assert(c);
 414
 415     return pa_bytes_to_usec(pa_memblockq_get_length(c->output_memblockq), &c->source_output->sample_spec);
 416 }
 417
 418 /*** client callbacks ***/
 419
 420 static void client_kill_cb(pa_client *client) {
 421     connection*c;
 422
 423     pa_assert(client);
 424     c = client->userdata;
 425     pa_assert(c);
 426
 427     connection_drop(c);
 428 }
 429
 430 /*** pa_iochannel callbacks ***/
 431
 432 static void io_callback(pa_iochannel*io, void *userdata) {
 433     connection *c = userdata;
 434
 435     pa_assert(io);
 436     pa_assert(c);
 437
 438     do_work(c);
 439 }
 440
 441 /*** socket_server callbacks ***/
 442
 443 static void on_connection(pa_socket_server*s, pa_iochannel *io, void *userdata) {
 444     pa_protocol_simple *p = userdata;
 445     connection *c = NULL;
 446     char cname[256];
 447
 448     pa_assert(s);
 449     pa_assert(io);
 450     pa_assert(p);
 451
 452     if (pa_idxset_size(p->connections)+1 > MAX_CONNECTIONS) {
 453         pa_log("Warning! Too many connections (%u), dropping incoming connection.", MAX_CONNECTIONS);
 454         pa_iochannel_free(io);
 455         return;
 456     }
 457
 458     c = pa_msgobject_new(connection, connection_check_type);
 459     c->parent.parent.free = connection_free;
 460     c->parent.process_msg = connection_process_msg;
 461     c->io = io;
 462     c->sink_input = NULL;
 463     c->source_output = NULL;
 464     c->input_memblockq = c->output_memblockq = NULL;
 465     c->protocol = p;
 466     c->playback.current_memblock = NULL;
 467     c->playback.memblock_index = 0;
 468     c->playback.fragment_size = 0;
 469     c->dead = 0;
 470     pa_atomic_store(&c->playback.missing, 0);
 471
 472     pa_iochannel_socket_peer_to_string(io, cname, sizeof(cname));
 473     pa_assert_se(c->client = pa_client_new(p->core, __FILE__, cname));
 474     c->client->owner = p->module;
 475     c->client->kill = client_kill_cb;
 476     c->client->userdata = c;
 477
 478     if (p->mode & PLAYBACK) {
 479         pa_sink_input_new_data data;
 480         size_t l;
 481
 482         pa_sink_input_new_data_init(&data);
 483         data.driver = __FILE__;
 484         data.name = c->client->name;
 485         pa_sink_input_new_data_set_sample_spec(&data, &p->sample_spec);
 486         data.module = p->module;
 487         data.client = c->client;
 488
 489         if (!(c->sink_input = pa_sink_input_new(p->core, &data, 0))) {
 490             pa_log("Failed to create sink input.");
 491             goto fail;
 492         }
 493
 494         c->sink_input->parent.process_msg = sink_input_process_msg;
 495         c->sink_input->peek = sink_input_peek_cb;
 496         c->sink_input->drop = sink_input_drop_cb;
 497         c->sink_input->kill = sink_input_kill_cb;
 498         c->sink_input->userdata = c;
 499
 500         l = (size_t) (pa_bytes_per_second(&p->sample_spec)*PLAYBACK_BUFFER_SECONDS);
 501         c->input_memblockq = pa_memblockq_new(
 502                 0,
 503                 l,
 504                 0,
 505                 pa_frame_size(&p->sample_spec),
 506                 (size_t) -1,
 507                 l/PLAYBACK_BUFFER_FRAGMENTS,
 508                 NULL);
 509         pa_assert(c->input_memblockq);
 510         pa_iochannel_socket_set_rcvbuf(io, l/PLAYBACK_BUFFER_FRAGMENTS*5);
 511         c->playback.fragment_size = l/PLAYBACK_BUFFER_FRAGMENTS;
 512
 513         pa_atomic_store(&c->playback.missing, pa_memblockq_missing(c->input_memblockq));
 514
 515         pa_sink_input_put(c->sink_input);
 516     }
 517
 518     if (p->mode & RECORD) {
 519         pa_source_output_new_data data;
 520         size_t l;
 521
 522         pa_source_output_new_data_init(&data);
 523         data.driver = __FILE__;
 524         data.name = c->client->name;
 525         pa_source_output_new_data_set_sample_spec(&data, &p->sample_spec);
 526         data.module = p->module;
 527         data.client = c->client;
 528
 529         if (!(c->source_output = pa_source_output_new(p->core, &data, 0))) {
 530             pa_log("Failed to create source output.");
 531             goto fail;
 532         }
 533         c->source_output->push = source_output_push_cb;
 534         c->source_output->kill = source_output_kill_cb;
 535         c->source_output->get_latency = source_output_get_latency_cb;
 536         c->source_output->userdata = c;
 537
 538         l = (size_t) (pa_bytes_per_second(&p->sample_spec)*RECORD_BUFFER_SECONDS);
 539         c->output_memblockq = pa_memblockq_new(
 540                 0,
 541                 l,
 542                 0,
 543                 pa_frame_size(&p->sample_spec),
 544                 1,
 545                 0,
 546                 NULL);
 547         pa_iochannel_socket_set_sndbuf(io, l/RECORD_BUFFER_FRAGMENTS*2);
 548
 549         pa_source_output_put(c->source_output);
 550     }
 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_drop(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_drop(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