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