code.delx.au - pulseaudio/blob - src/modules/xen/module-xenpv-sink.c

   1 /***
   2   This file is part of PulseAudio.
   3
   4   Copyright 2004-2006 Lennart Poettering
   5   Copyright 2011 George Boutsioukis for Xen
   6
   7   PulseAudio is free software; you can redistribute it and/or modify
   8   it under the terms of the GNU Lesser General Public License as published
   9   by the Free Software Foundation; either version 2.1 of the License,
  10   or (at your option) any later version.
  11
  12   PulseAudio is distributed in the hope that it will be useful, but
  13   WITHOUT ANY WARRANTY; without even the implied warranty of
  14   MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
  15   General Public License for more details.
  16
  17   You should have received a copy of the GNU Lesser General Public License
  18   along with PulseAudio; if not, write to the Free Software
  19   Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307
  20   USA.
  21 ***/
  22
  23 #ifdef HAVE_CONFIG_H
  24 #include <config.h>
  25 #endif
  26
  27 #include <stdlib.h>
  28 #include <sys/stat.h>
  29 #include <stdio.h>
  30 #include <errno.h>
  31 #include <string.h>
  32 #include <fcntl.h>
  33 #include <unistd.h>
  34 #include <limits.h>
  35 #include <sys/ioctl.h>
  36 #include <poll.h>
  37 #include <time.h>
  38
  39 #include <pulse/xmalloc.h>
  40
  41 #include <pulsecore/core-error.h>
  42 #include <pulsecore/sink.h>
  43 #include <pulsecore/module.h>
  44 #include <pulsecore/core-util.h>
  45 #include <pulsecore/modargs.h>
  46 #include <pulsecore/log.h>
  47 #include <pulsecore/thread.h>
  48 #include <pulsecore/thread-mq.h>
  49 #include <pulsecore/rtpoll.h>
  50
  51 #include <sys/select.h>
  52 #include <sys/mman.h>
  53 #include <xenctrl.h>
  54 #include <xs.h>
  55
  56 #include "module-xenpv-sink-symdef.h"
  57 #include "gntalloc.h"
  58 #include "gntdev.h"
  59
  60 PA_MODULE_AUTHOR("Giorgos Boutsioukis");
  61 PA_MODULE_DESCRIPTION("Xen PV audio sink");
  62 PA_MODULE_VERSION(PACKAGE_VERSION);
  63 PA_MODULE_LOAD_ONCE(FALSE);
  64 PA_MODULE_USAGE(
  65         "sink_name=<name for the sink> "
  66         "sink_properties=<properties for the sink> "
  67         "format=<sample format> "
  68         "rate=<sample rate>"
  69         "channels=<number of channels> "
  70         "channel_map=<channel map>");
  71
  72 #define DEFAULT_SINK_NAME "xenpv_output"
  73 #define DEFAULT_FILE_NAME "xenpv_output"
  74
  75 #define STATE_UNDEFINED 9999
  76
  77 int device_id = -1;
  78 enum xenbus_state {
  79         XenbusStateUnknown      = 0,
  80         XenbusStateInitialising = 1,
  81         XenbusStateInitWait     = 2,
  82         XenbusStateInitialised  = 3,
  83         XenbusStateConnected    = 4,
  84         XenbusStateClosing      = 5,
  85         XenbusStateClosed       = 6,
  86         XenbusStateReconfiguring = 7,
  87         XenbusStateReconfigured  = 8
  88 };
  89
  90 static const char* xenbus_names[] = {
  91     "XenbusStateUnknown",
  92     "XenbusStateInitialising",
  93     "XenbusStateInitWait",
  94     "XenbusStateInitialised",
  95     "XenbusStateConnected",
  96     "XenbusStateClosing",
  97     "XenbusStateClosed",
  98     "XenbusStateReconfiguring",
  99     "XenbusStateReconfigured"
 100 };
 101
 102 struct userdata {
 103     pa_core *core;
 104     pa_module *module;
 105     pa_sink *sink;
 106
 107     pa_thread *thread;
 108     pa_thread_mq thread_mq;
 109     pa_rtpoll *rtpoll;
 110
 111     pa_memchunk memchunk;
 112
 113     pa_rtpoll_item *rtpoll_item;
 114
 115     int write_type;
 116 };
 117
 118 pa_sample_spec ss;
 119 pa_channel_map map;
 120
 121 /* just to test non- frame-aligned size */
 122 #define BUFSIZE 2047
 123
 124 struct ring {
 125     uint32_t cons_indx, prod_indx;
 126     uint32_t usable_buffer_space; /* kept here for convenience */
 127     uint8_t buffer[BUFSIZE];
 128 } *ioring;
 129
 130 static const char* const valid_modargs[] = {
 131     "sink_name",
 132     "sink_properties",
 133     "file",
 134     "format",
 135     "rate",
 136     "channels",
 137     "channel_map",
 138     NULL
 139 };
 140
 141 /* Xen globals*/
 142 xc_interface* xch;
 143 xc_evtchn* xce;
 144 evtchn_port_or_error_t xen_evtchn_port;
 145 static struct xs_handle *xsh;
 146 struct ioctl_gntalloc_alloc_gref gref;
 147
 148 static int register_backend_state_watch(void);
 149 static int wait_for_backend_state_change(void);
 150 static int alloc_gref(struct ioctl_gntalloc_alloc_gref *gref, void **addr);
 151 static int ring_write(struct ring *r, void *src, int length);
 152 static int publish_spec(pa_sample_spec *ss);
 153 static int read_backend_default_spec(pa_sample_spec *ss);
 154 static int publish_param(const char *paramname, const char *value);
 155 static int publish_param_int(const char *paramname, const int value);
 156 static char* read_param(const char *paramname);
 157
 158 static int set_state(int state) {
 159     static int current_state = 0;
 160     pa_log_debug("State transition %s->%s\n",
 161             xenbus_names[current_state], xenbus_names[state]);
 162
 163     publish_param_int("state", state);
 164     current_state = state;
 165     return state;
 166 }
 167 #define NEGOTIATION_ERROR 2
 168 #define NEGOTIATION_OK 1
 169
 170 /* negotiation callbacks */
 171 static int state_unknown_cb() {
 172     pa_log_debug("Xen audio sink: Backend state was XenbusStateUnknown\n");
 173     set_state(XenbusStateInitialising);
 174
 175     return 0;
 176 }
 177
 178 static int state_initialising_cb() {
 179     pa_log_debug("Xen audio sink: Backend state was XenbusStateInitialising\n");
 180     set_state(XenbusStateInitialised);
 181     return 0;
 182 }
 183
 184 static int state_initwait_cb() {
 185     pa_log_debug("Xen audio sink: Backend state was XenbusStateInitWait\n");
 186     return 0;
 187 }
 188
 189 static int state_initialised_cb() {
 190     pa_log_debug("Xen audio sink: Backend state was XenbusStateInitialised\n");
 191     /*Remind the backend we are ready*/
 192     set_state(XenbusStateInitialised);
 193     return 0;
 194 }
 195
 196 static int state_connected_cb() {
 197     /* The backend accepted our parameters, sweet! */
 198     set_state(XenbusStateConnected);
 199     pa_log_debug("Xen audio sink: Backend state was XenbusStateConnected\n");
 200     return NEGOTIATION_OK;
 201 }
 202
 203 static int state_closing_cb() {
 204     pa_log_debug("Xen audio sink: Backend state was XenbusStateClosing\n");
 205     return 0;
 206 }
 207
 208 static int state_closed_cb() {
 209     pa_log_debug("Xen audio sink: Backend state was XenbusStateClosed\n");
 210     return 0;
 211 }
 212
 213 static int state_reconfiguring_cb() {
 214     /* The backend rejected our sample spec */
 215     pa_log_debug("Xen audio sink: Backend state was XenbusStateReconfiguring\n");
 216     /* fall back to the backend's default parameters*/
 217     read_backend_default_spec(&ss);
 218     /* backend should accept these now */
 219     publish_spec(&ss);
 220     set_state(XenbusStateInitialised);
 221     return 0;
 222 }
 223
 224 static int state_reconfigured_cb() {
 225     pa_log_debug("Xen audio sink: Backend state was XenbusStateReconfigured\n");
 226     return 0;
 227 }
 228
 229 int (*state_callbacks[9])(void) = {
 230     state_unknown_cb,
 231     state_initialising_cb,
 232     state_initwait_cb,
 233     state_initialised_cb,
 234     state_connected_cb,
 235     state_closing_cb,
 236     state_closed_cb,
 237     state_reconfiguring_cb,
 238     state_reconfigured_cb
 239 };
 240
 241 static void xen_cleanup() {
 242     char keybuf[64];
 243     /*XXX hardcoded*/
 244     munmap((void*)gref.index, 4096);
 245
 246     set_state(XenbusStateClosing);
 247     /* send one last event to unblock the backend */
 248     xc_evtchn_notify(xce, xen_evtchn_port);
 249     /* close xen interfaces */
 250     xc_evtchn_close(xce);
 251     xc_interface_close(xch);
 252
 253     /* delete xenstore keys */
 254     publish_param_int("state", XenbusStateClosed);
 255     snprintf(keybuf, sizeof(keybuf), "device/audio/%d", device_id);
 256     xs_rm(xsh, 0, keybuf);
 257     xs_daemon_close(xsh);
 258 }
 259
 260 static int sink_process_msg(pa_msgobject *o, int code, void *data, int64_t offset, pa_memchunk *chunk) {
 261     struct userdata *u = PA_SINK(o)->userdata;
 262
 263     switch (code) {
 264
 265         case PA_SINK_MESSAGE_GET_LATENCY: {
 266             size_t n = 0;
 267
 268             n += u->memchunk.length;
 269
 270             *((pa_usec_t*) data) = pa_bytes_to_usec(n, &u->sink->sample_spec);
 271             return 0;
 272         }
 273     }
 274
 275     return pa_sink_process_msg(o, code, data, offset, chunk);
 276 }
 277
 278 static int process_render(struct userdata *u) {
 279     pa_assert(u);
 280
 281
 282     if (u->memchunk.length <= 0)
 283         pa_sink_render(u->sink, ioring->usable_buffer_space, &u->memchunk);
 284
 285
 286     pa_assert(u->memchunk.length > 0);
 287
 288     xc_evtchn_notify(xce, xen_evtchn_port);
 289     for (;;) {
 290         ssize_t l;
 291         void *p;
 292
 293         p = pa_memblock_acquire(u->memchunk.memblock);
 294             /* xen: write data to ring buffer & notify backend */
 295         l = ring_write(ioring, (uint8_t*)p + u->memchunk.index, u->memchunk.length);
 296
 297         pa_memblock_release(u->memchunk.memblock);
 298
 299         pa_assert(l != 0);
 300
 301         if (l < 0) {
 302             if (errno == EINTR)
 303                 continue;
 304             else if (errno == EAGAIN)
 305                 return 0;
 306             else {
 307                 pa_log("Failed to write data to FIFO: %s", pa_cstrerror(errno));
 308                 return -1;
 309             }
 310
 311         } else {
 312
 313             u->memchunk.index += (size_t) l;
 314             u->memchunk.length -= (size_t) l;
 315
 316             if (u->memchunk.length <= 0) {
 317                 pa_memblock_unref(u->memchunk.memblock);
 318                 pa_memchunk_reset(&u->memchunk);
 319             }
 320         }
 321
 322         return 0;
 323     }
 324 }
 325
 326 static void thread_func(void *userdata) {
 327     struct userdata *u = userdata;
 328
 329     pa_assert(u);
 330
 331     pa_log_debug("Thread starting up");
 332
 333     pa_thread_mq_install(&u->thread_mq);
 334
 335     for(;;){
 336         struct pollfd *pollfd;
 337         int ret;
 338
 339         pollfd = pa_rtpoll_item_get_pollfd(u->rtpoll_item, NULL);
 340
 341         /* Render some data and write it to the fifo */
 342         if (PA_SINK_IS_OPENED(u->sink->thread_info.state)) {
 343
 344             if (u->sink->thread_info.rewind_requested)
 345                 pa_sink_process_rewind(u->sink, 0);
 346
 347             if (pollfd->revents) {
 348                 if (process_render(u) < 0)
 349                     goto fail;
 350
 351                 pollfd->revents = 0;
 352             }
 353         }
 354
 355         /* Hmm, nothing to do. Let's sleep */
 356
 357         pollfd->events = (short) (u->sink->thread_info.state == PA_SINK_RUNNING ? POLLOUT : 0);
 358
 359         if ((ret = pa_rtpoll_run(u->rtpoll, TRUE)) < 0)
 360             goto fail;
 361
 362         if (ret == 0)
 363             goto finish;
 364
 365         pollfd = pa_rtpoll_item_get_pollfd(u->rtpoll_item, NULL);
 366
 367         if (pollfd->revents & ~POLLOUT) {
 368             pa_log("FIFO shutdown.");
 369             goto fail;
 370         }
 371     }
 372
 373 fail:
 374     /* If this was no regular exit from the loop we have to continue
 375      * processing messages until we received PA_MESSAGE_SHUTDOWN */
 376     pa_asyncmsgq_post(u->thread_mq.outq, PA_MSGOBJECT(u->core), PA_CORE_MESSAGE_UNLOAD_MODULE, u->module, 0, NULL, NULL);
 377     pa_asyncmsgq_wait_for(u->thread_mq.inq, PA_MESSAGE_SHUTDOWN);
 378     pa_log_debug("Shutting down Xen...");
 379     xen_cleanup();
 380 finish:
 381     pa_log_debug("Thread shutting down");
 382 }
 383
 384 int pa__init(pa_module*m) {
 385
 386     struct userdata *u;
 387     pa_modargs *ma;
 388     pa_sink_new_data data;
 389     int backend_state;
 390     int ret;
 391     char strbuf[100];
 392
 393     pa_assert(m);
 394
 395     if (!(ma = pa_modargs_new(m->argument, valid_modargs))) {
 396         pa_log("Failed to parse module arguments.");
 397         goto fail;
 398     }
 399
 400     ss = m->core->default_sample_spec;
 401     map = m->core->default_channel_map;
 402
 403     /* user arguments override these */
 404     if (pa_modargs_get_sample_spec_and_channel_map(ma, &ss, &map, PA_CHANNEL_MAP_DEFAULT) < 0) {
 405         pa_log("Invalid sample format specification or channel map");
 406         return 1;
 407     }
 408
 409     /* Xen Basic init */
 410     xsh = xs_domain_open();
 411     if (xsh==NULL) {
 412         pa_log("xs_domain_open failed");
 413         goto fail;
 414     }
 415     set_state(XenbusStateUnknown);
 416
 417     xch = xc_interface_open(NULL, NULL, 0);
 418     if (xch==0) {
 419         pa_log("xc_interface_open failed");
 420         goto fail;
 421     }
 422
 423     xce = xc_evtchn_open(NULL, 0);
 424     if (xce==0) {
 425         pa_log("xc_evtchn_open failed");
 426         goto fail;
 427     }
 428
 429     /* use only dom0 as the backend for now */
 430     xen_evtchn_port = xc_evtchn_bind_unbound_port(xce, 0);
 431     if (xen_evtchn_port == 0) {
 432         pa_log("xc_evtchn_bind_unbound_port failed");
 433     }
 434
 435     /* get grant reference & map locally */
 436     if (alloc_gref(&gref, (void**)&ioring)) {
 437        pa_log("alloc_gref failed");
 438     };
 439     device_id = 0; /* hardcoded for now */
 440
 441     if (register_backend_state_watch()) {
 442         pa_log("Xen sink: register xenstore watch failed");
 443     };
 444
 445     publish_param_int("event-channel", xen_evtchn_port);
 446     publish_param_int("ring-ref", gref.gref_ids[0]);
 447
 448     /* let's ask for something absurd and deal with rejection */
 449     ss.rate = 192000;
 450
 451     publish_spec(&ss);
 452
 453     ret=0;
 454     while (!ret) {
 455         backend_state = wait_for_backend_state_change();
 456         if (backend_state == STATE_UNDEFINED) {
 457             pa_log("Xen Backend is taking long to respond, still waiting...");
 458             continue;
 459         } else if (backend_state == -1) {
 460             pa_log("Error while waiting for backend: %s", strerror(errno));
 461             break;
 462             goto fail;
 463         }
 464         ret = state_callbacks[backend_state]();
 465     }
 466     if (ret!=NEGOTIATION_OK) {
 467         pa_log("Negotiation with Xen backend failed!");
 468         return 1;
 469     }
 470
 471     pa_sample_spec_snprint(strbuf, 100, &ss);
 472     pa_log_debug("Negotiation ended, the result was: %s", strbuf);
 473
 474     /* End of Phase 2, begin playback cycle */
 475
 476     u = pa_xnew0(struct userdata, 1);
 477     u->core = m->core;
 478     u->module = m;
 479     m->userdata = u;
 480     pa_memchunk_reset(&u->memchunk);
 481     u->rtpoll = pa_rtpoll_new();
 482     pa_thread_mq_init(&u->thread_mq, m->core->mainloop, u->rtpoll);
 483     u->write_type = 0;
 484
 485     /* init ring buffer */
 486     ioring->prod_indx = ioring->cons_indx = 0;
 487     ioring->usable_buffer_space = BUFSIZE - BUFSIZE % pa_frame_size(&ss);
 488
 489     pa_sink_new_data_init(&data);
 490     data.driver = __FILE__;
 491     data.module = m;
 492     pa_sink_new_data_set_name(&data, pa_modargs_get_value(ma, "sink_name", DEFAULT_SINK_NAME));
 493     pa_proplist_sets(data.proplist, PA_PROP_DEVICE_STRING, "xensink");
 494     pa_proplist_setf(data.proplist, PA_PROP_DEVICE_DESCRIPTION, "Xen PV audio sink");
 495     pa_sink_new_data_set_sample_spec(&data, &ss);
 496     pa_sink_new_data_set_channel_map(&data, &map);
 497
 498     if (pa_modargs_get_proplist(ma, "sink_properties", data.proplist, PA_UPDATE_REPLACE) < 0) {
 499         pa_log("Invalid properties");
 500         pa_sink_new_data_done(&data);
 501         goto fail;
 502     }
 503
 504     u->sink = pa_sink_new(m->core, &data, PA_SINK_LATENCY);
 505     pa_sink_new_data_done(&data);
 506
 507     if (!u->sink) {
 508         pa_log("Failed to create sink.");
 509         goto fail;
 510     }
 511
 512     u->sink->parent.process_msg = sink_process_msg;
 513     u->sink->userdata = u;
 514
 515     pa_sink_set_asyncmsgq(u->sink, u->thread_mq.inq);
 516     pa_sink_set_rtpoll(u->sink, u->rtpoll);
 517     pa_sink_set_max_request(u->sink, ioring->usable_buffer_space);
 518     pa_sink_set_fixed_latency(u->sink, pa_bytes_to_usec(ioring->usable_buffer_space, &u->sink->sample_spec));
 519
 520     u->rtpoll_item = pa_rtpoll_item_new(u->rtpoll, PA_RTPOLL_NEVER, 1);
 521
 522     if (!(u->thread = pa_thread_new("xenpv-sink", thread_func, u))) {
 523         pa_log("Failed to create thread.");
 524         goto fail;
 525     }
 526
 527     pa_sink_put(u->sink);
 528
 529     pa_modargs_free(ma);
 530
 531     return 0;
 532
 533 fail:
 534     if (ma)
 535         pa_modargs_free(ma);
 536
 537     pa__done(m);
 538
 539     return -1;
 540 }
 541
 542 int pa__get_n_used(pa_module *m) {
 543     struct userdata *u;
 544
 545     pa_assert(m);
 546     pa_assert_se(u = m->userdata);
 547
 548     return pa_sink_linked_by(u->sink);
 549 }
 550
 551 void pa__done(pa_module*m) {
 552     struct userdata *u;
 553
 554     pa_assert(m);
 555
 556     if (!(u = m->userdata))
 557         return;
 558
 559     if (u->sink)
 560         pa_sink_unlink(u->sink);
 561
 562     if (u->thread) {
 563         pa_asyncmsgq_send(u->thread_mq.inq, NULL, PA_MESSAGE_SHUTDOWN, NULL, 0, NULL);
 564         pa_thread_free(u->thread);
 565     }
 566
 567     pa_thread_mq_done(&u->thread_mq);
 568
 569     if (u->sink)
 570         pa_sink_unref(u->sink);
 571
 572     if (u->memchunk.memblock)
 573        pa_memblock_unref(u->memchunk.memblock);
 574
 575     if (u->rtpoll_item)
 576         pa_rtpoll_item_free(u->rtpoll_item);
 577
 578     if (u->rtpoll)
 579         pa_rtpoll_free(u->rtpoll);
 580
 581     pa_xfree(u);
 582
 583     xen_cleanup();
 584
 585 }
 586
 587
 588 static int alloc_gref(struct ioctl_gntalloc_alloc_gref *gref_, void **addr) {
 589     int alloc_fd, dev_fd, rv;
 590
 591     alloc_fd = open("/dev/xen/gntalloc", O_RDWR);
 592     if (alloc_fd<=0) {
 593         perror("Could not open /dev/xen/gntalloc! Have you loaded the xen_gntalloc module?");
 594         return 1;
 595     }
 596
 597     dev_fd = open("/dev/xen/gntdev", O_RDWR);
 598     if (dev_fd<=0) {
 599         perror("Could not open /dev/xen/gntdev! Have you loaded the xen_gntdev module?");
 600         return 1;
 601     }
 602
 603     /* use dom0 */
 604     gref_->domid = 0;
 605     gref_->flags = GNTALLOC_FLAG_WRITABLE;
 606     gref_->count = 1;
 607
 608     rv = ioctl(alloc_fd, IOCTL_GNTALLOC_ALLOC_GREF, gref_);
 609     if (rv) {
 610         pa_log_debug("Xen audio sink: src-add error: %s (rv=%d)\n", strerror(errno), rv);
 611         return rv;
 612     }
 613
 614     /*addr=NULL(default),length, prot,             flags,    fd,         offset*/
 615     *addr = mmap(0, 4096, PROT_READ|PROT_WRITE, MAP_SHARED, alloc_fd, gref_->index);
 616     if (*addr == MAP_FAILED) {
 617         *addr = 0;
 618         pa_log_debug("Xen audio sink: mmap'ing shared page failed\n");
 619         return rv;
 620     }
 621
 622     pa_log_debug("Xen audio sink: Got grant #%d. Mapped locally at %Ld=%p\n",
 623             gref_->gref_ids[0], (long long)gref_->index, *addr);
 624
 625     /* skip this for now
 626        struct ioctl_gntalloc_unmap_notify uarg = {
 627        .index = gref->index + offsetof(struct shr_page, notifies[0]),
 628        .action = UNMAP_NOTIFY_CLEAR_BYTE
 629        };
 630
 631        rv = ioctl(a_fd, IOCTL_GNTALLOC_SET_UNMAP_NOTIFY, &uarg);
 632        if (rv)
 633        pa_log_debug("gntalloc unmap notify error: %s (rv=%d)\n", strerror(errno), rv);
 634        */
 635
 636     close(alloc_fd);
 637     close(dev_fd);
 638
 639     return rv;
 640 }
 641
 642 #define RING_FREE_BYTES ((r->usable_buffer_space - (r->prod_indx-r->cons_indx) -1) % r->usable_buffer_space)
 643 static int ring_write(struct ring *r, void *src, int length) {
 644     int full = 0;
 645     for (;;) {
 646         /* free space may be split over the end of the buffer */
 647         int first_chunk_size = (r->usable_buffer_space-r->prod_indx);
 648         int second_chunk_size = (r->cons_indx>=r->prod_indx)? (r->cons_indx) : 0;
 649         int l, fl, sl;
 650
 651         /* full? */
 652         if (RING_FREE_BYTES==0) {
 653             /*XXX hardcoded*/
 654             if (full>=100) {
 655                 errno = EINTR;
 656                 return -1;
 657             }
 658             /*XXX hardcoded */
 659             usleep(1000);
 660             /* should return in 100ms max; definitely not midstream */
 661             full++;
 662             continue;
 663         }
 664
 665         /* calculate lengths in case of a split buffer */
 666         l = PA_MIN((int)RING_FREE_BYTES, length);
 667         fl = PA_MIN(l, first_chunk_size);
 668         sl = PA_MIN(l-fl, second_chunk_size);
 669
 670         memcpy(r->buffer+r->prod_indx, src, fl);
 671         if (sl)
 672             memcpy(r->buffer, ((char*)src)+fl, sl);
 673         r->prod_indx = (r->prod_indx+fl+sl) % r->usable_buffer_space;
 674
 675         return sl+fl;
 676     }
 677 }
 678
 679 static int publish_param(const char *paramname, const char *value) {
 680     char keybuf[128], valbuf[32];
 681
 682     snprintf(keybuf, sizeof keybuf, "device/audio/%d/%s", device_id, paramname);
 683     snprintf(valbuf, sizeof valbuf, "%s", value);
 684     return xs_write(xsh, 0, keybuf, valbuf, strlen(valbuf));
 685 }
 686
 687 static int publish_param_int(const char *paramname, const int value) {
 688     char keybuf[128], valbuf[32];
 689     snprintf(keybuf, sizeof keybuf, "device/audio/%d/%s", device_id, paramname);
 690     snprintf(valbuf, sizeof valbuf, "%d", value);
 691     return xs_write(xsh, 0, keybuf, valbuf, strlen(valbuf));
 692 }
 693
 694 static char* read_param(const char *paramname) {
 695     char keybuf[128];
 696     unsigned int len;
 697     int my_domid;
 698
 699     my_domid = atoi(xs_read(xsh, 0, "domid", &len));
 700     snprintf(keybuf, sizeof(keybuf), "/local/domain/0/backend/audio/%d/%d/%s", my_domid, device_id, paramname);
 701     /* remember to free lvalue! */
 702     return xs_read(xsh, 0, keybuf, &len);
 703 }
 704
 705
 706 static int publish_spec(pa_sample_spec *sample_spec) {
 707     /* Publish spec and set state to XenbusStateInitWait*/
 708     int ret;
 709
 710     ret = publish_param("format", pa_sample_format_to_string(sample_spec->format));
 711     ret += publish_param_int("rate", sample_spec->rate);
 712     ret += publish_param_int("channels", sample_spec->channels);
 713
 714     return ret;
 715 }
 716
 717
 718 static int read_backend_default_spec(pa_sample_spec *sample_spec) {
 719     /* Read spec from backend */
 720     char *out;
 721
 722     out = read_param("default-format");
 723     sample_spec->format = pa_parse_sample_format(out);
 724     free(out);
 725
 726     out = read_param("default-rate");
 727     sample_spec->rate = atoi(out);
 728     free(out);
 729
 730     out = read_param("default-channels");
 731     sample_spec->channels = atoi(out);
 732     free(out);
 733
 734     return 0;
 735 }
 736
 737 static int register_backend_state_watch() {
 738     char keybuf[128];
 739     int my_domid;
 740     unsigned int len;
 741
 742     my_domid = atoi(xs_read(xsh, 0, "domid", &len));
 743     snprintf(keybuf, sizeof(keybuf), "/local/domain/0/backend/audio/%d/%d/state", my_domid, device_id);
 744     if (!xs_watch(xsh, keybuf, "xenpvaudiofrontendsinktoken")) {
 745         perror("xs_watch failed");
 746         return -EINVAL;
 747     }
 748     return 0;
 749 }
 750
 751 static int wait_for_backend_state_change() {
 752     char keybuf[128];
 753     int my_domid;
 754     unsigned int len;
 755
 756     int backend_state;
 757     int seconds;
 758     char *buf, **vec;
 759     int ret;
 760
 761     int xs_fd;
 762     struct timeval tv;
 763         fd_set watch_fdset;
 764     int start, now;
 765
 766     backend_state = STATE_UNDEFINED;
 767     xs_fd = xs_fileno(xsh);
 768     start = now = time(NULL);
 769
 770     my_domid = atoi(xs_read(xsh, 0, "domid", &len));
 771     snprintf(keybuf, sizeof(keybuf), "/local/domain/0/backend/audio/%d/%d/state", my_domid, device_id);
 772
 773     /*XXX: hardcoded */
 774     seconds = 10;
 775         do {
 776                 tv.tv_usec = 0;
 777                 tv.tv_sec = (start + seconds) - now;
 778                 FD_ZERO(&watch_fdset);
 779                 FD_SET(xs_fd, &watch_fdset);
 780         ret=select(xs_fd + 1, &watch_fdset, NULL, NULL, &tv);
 781
 782         if (ret==-1)
 783             /* error */
 784             return -1;
 785         else if (ret) {
 786
 787                         /* Read the watch to drain the buffer */
 788                         vec = xs_read_watch(xsh, &len);
 789
 790             buf = xs_read(xsh, XBT_NULL, vec[0], &len);
 791             if (buf == 0) {
 792                 /* usually means that the backend isn't there yet */
 793                 continue;
 794             };
 795             backend_state = atoi(buf);
 796
 797             free(buf);
 798             free(vec);
 799                 }
 800         /* else: timeout */
 801         } while (backend_state == STATE_UNDEFINED && \
 802             (now = time(NULL)) < start + seconds);
 803
 804     return backend_state;
 805 }