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     if (u->memchunk.length <= 0)
 282         pa_sink_render(u->sink, ioring->usable_buffer_space, &u->memchunk);
 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         if (PA_UNLIKELY(u->sink->thread_info.rewind_requested))
 340             pa_sink_process_rewind(u->sink, 0);
 341
 342         /* Render some data and write it to the fifo */
 343         if (PA_SINK_IS_OPENED(u->sink->thread_info.state)) {
 344             if (pollfd->revents) {
 345                 if (process_render(u) < 0)
 346                     goto fail;
 347
 348                 pollfd->revents = 0;
 349             }
 350         }
 351
 352         /* Hmm, nothing to do. Let's sleep */
 353
 354         pollfd->events = (short) (u->sink->thread_info.state == PA_SINK_RUNNING ? POLLOUT : 0);
 355
 356         if ((ret = pa_rtpoll_run(u->rtpoll, true)) < 0)
 357             goto fail;
 358
 359         if (ret == 0)
 360             goto finish;
 361
 362         pollfd = pa_rtpoll_item_get_pollfd(u->rtpoll_item, NULL);
 363
 364         if (pollfd->revents & ~POLLOUT) {
 365             pa_log("FIFO shutdown.");
 366             goto fail;
 367         }
 368     }
 369
 370 fail:
 371     /* If this was no regular exit from the loop we have to continue
 372      * processing messages until we received PA_MESSAGE_SHUTDOWN */
 373     pa_asyncmsgq_post(u->thread_mq.outq, PA_MSGOBJECT(u->core), PA_CORE_MESSAGE_UNLOAD_MODULE, u->module, 0, NULL, NULL);
 374     pa_asyncmsgq_wait_for(u->thread_mq.inq, PA_MESSAGE_SHUTDOWN);
 375     pa_log_debug("Shutting down Xen...");
 376     xen_cleanup();
 377 finish:
 378     pa_log_debug("Thread shutting down");
 379 }
 380
 381 int pa__init(pa_module*m) {
 382
 383     struct userdata *u;
 384     pa_modargs *ma;
 385     pa_sink_new_data data;
 386     int backend_state;
 387     int ret;
 388     char strbuf[100];
 389
 390     pa_assert(m);
 391
 392     if (!(ma = pa_modargs_new(m->argument, valid_modargs))) {
 393         pa_log("Failed to parse module arguments.");
 394         goto fail;
 395     }
 396
 397     ss = m->core->default_sample_spec;
 398     map = m->core->default_channel_map;
 399
 400     /* user arguments override these */
 401     if (pa_modargs_get_sample_spec_and_channel_map(ma, &ss, &map, PA_CHANNEL_MAP_DEFAULT) < 0) {
 402         pa_log("Invalid sample format specification or channel map");
 403         return 1;
 404     }
 405
 406     /* Xen Basic init */
 407     xsh = xs_domain_open();
 408     if (xsh==NULL) {
 409         pa_log("xs_domain_open failed");
 410         goto fail;
 411     }
 412     set_state(XenbusStateUnknown);
 413
 414     xch = xc_interface_open(NULL, NULL, 0);
 415     if (xch==0) {
 416         pa_log("xc_interface_open failed");
 417         goto fail;
 418     }
 419
 420     xce = xc_evtchn_open(NULL, 0);
 421     if (xce==0) {
 422         pa_log("xc_evtchn_open failed");
 423         goto fail;
 424     }
 425
 426     /* use only dom0 as the backend for now */
 427     xen_evtchn_port = xc_evtchn_bind_unbound_port(xce, 0);
 428     if (xen_evtchn_port == 0) {
 429         pa_log("xc_evtchn_bind_unbound_port failed");
 430     }
 431
 432     /* get grant reference & map locally */
 433     if (alloc_gref(&gref, (void**)&ioring)) {
 434        pa_log("alloc_gref failed");
 435     };
 436     device_id = 0; /* hardcoded for now */
 437
 438     if (register_backend_state_watch()) {
 439         pa_log("Xen sink: register xenstore watch failed");
 440     };
 441
 442     publish_param_int("event-channel", xen_evtchn_port);
 443     publish_param_int("ring-ref", gref.gref_ids[0]);
 444
 445     /* let's ask for something absurd and deal with rejection */
 446     ss.rate = 192000;
 447
 448     publish_spec(&ss);
 449
 450     ret=0;
 451     while (!ret) {
 452         backend_state = wait_for_backend_state_change();
 453         if (backend_state == STATE_UNDEFINED) {
 454             pa_log("Xen Backend is taking long to respond, still waiting...");
 455             continue;
 456         } else if (backend_state == -1) {
 457             pa_log("Error while waiting for backend: %s", strerror(errno));
 458             break;
 459             goto fail;
 460         }
 461         ret = state_callbacks[backend_state]();
 462     }
 463     if (ret!=NEGOTIATION_OK) {
 464         pa_log("Negotiation with Xen backend failed!");
 465         return 1;
 466     }
 467
 468     pa_sample_spec_snprint(strbuf, 100, &ss);
 469     pa_log_debug("Negotiation ended, the result was: %s", strbuf);
 470
 471     /* End of Phase 2, begin playback cycle */
 472
 473     u = pa_xnew0(struct userdata, 1);
 474     u->core = m->core;
 475     u->module = m;
 476     m->userdata = u;
 477     pa_memchunk_reset(&u->memchunk);
 478     u->rtpoll = pa_rtpoll_new();
 479     pa_thread_mq_init(&u->thread_mq, m->core->mainloop, u->rtpoll);
 480     u->write_type = 0;
 481
 482     /* init ring buffer */
 483     ioring->prod_indx = ioring->cons_indx = 0;
 484     ioring->usable_buffer_space = BUFSIZE - BUFSIZE % pa_frame_size(&ss);
 485
 486     pa_sink_new_data_init(&data);
 487     data.driver = __FILE__;
 488     data.module = m;
 489     pa_sink_new_data_set_name(&data, pa_modargs_get_value(ma, "sink_name", DEFAULT_SINK_NAME));
 490     pa_proplist_sets(data.proplist, PA_PROP_DEVICE_STRING, "xensink");
 491     pa_proplist_setf(data.proplist, PA_PROP_DEVICE_DESCRIPTION, "Xen PV audio sink");
 492     pa_sink_new_data_set_sample_spec(&data, &ss);
 493     pa_sink_new_data_set_channel_map(&data, &map);
 494
 495     if (pa_modargs_get_proplist(ma, "sink_properties", data.proplist, PA_UPDATE_REPLACE) < 0) {
 496         pa_log("Invalid properties");
 497         pa_sink_new_data_done(&data);
 498         goto fail;
 499     }
 500
 501     u->sink = pa_sink_new(m->core, &data, PA_SINK_LATENCY);
 502     pa_sink_new_data_done(&data);
 503
 504     if (!u->sink) {
 505         pa_log("Failed to create sink.");
 506         goto fail;
 507     }
 508
 509     u->sink->parent.process_msg = sink_process_msg;
 510     u->sink->userdata = u;
 511
 512     pa_sink_set_asyncmsgq(u->sink, u->thread_mq.inq);
 513     pa_sink_set_rtpoll(u->sink, u->rtpoll);
 514     pa_sink_set_max_request(u->sink, ioring->usable_buffer_space);
 515     pa_sink_set_fixed_latency(u->sink, pa_bytes_to_usec(ioring->usable_buffer_space, &u->sink->sample_spec));
 516
 517     u->rtpoll_item = pa_rtpoll_item_new(u->rtpoll, PA_RTPOLL_NEVER, 1);
 518
 519     if (!(u->thread = pa_thread_new("xenpv-sink", thread_func, u))) {
 520         pa_log("Failed to create thread.");
 521         goto fail;
 522     }
 523
 524     pa_sink_put(u->sink);
 525
 526     pa_modargs_free(ma);
 527
 528     return 0;
 529
 530 fail:
 531     if (ma)
 532         pa_modargs_free(ma);
 533
 534     pa__done(m);
 535
 536     return -1;
 537 }
 538
 539 int pa__get_n_used(pa_module *m) {
 540     struct userdata *u;
 541
 542     pa_assert(m);
 543     pa_assert_se(u = m->userdata);
 544
 545     return pa_sink_linked_by(u->sink);
 546 }
 547
 548 void pa__done(pa_module*m) {
 549     struct userdata *u;
 550
 551     pa_assert(m);
 552
 553     if (!(u = m->userdata))
 554         return;
 555
 556     if (u->sink)
 557         pa_sink_unlink(u->sink);
 558
 559     if (u->thread) {
 560         pa_asyncmsgq_send(u->thread_mq.inq, NULL, PA_MESSAGE_SHUTDOWN, NULL, 0, NULL);
 561         pa_thread_free(u->thread);
 562     }
 563
 564     pa_thread_mq_done(&u->thread_mq);
 565
 566     if (u->sink)
 567         pa_sink_unref(u->sink);
 568
 569     if (u->memchunk.memblock)
 570        pa_memblock_unref(u->memchunk.memblock);
 571
 572     if (u->rtpoll_item)
 573         pa_rtpoll_item_free(u->rtpoll_item);
 574
 575     if (u->rtpoll)
 576         pa_rtpoll_free(u->rtpoll);
 577
 578     pa_xfree(u);
 579
 580     xen_cleanup();
 581
 582 }
 583
 584 static int alloc_gref(struct ioctl_gntalloc_alloc_gref *gref_, void **addr) {
 585     int alloc_fd, dev_fd, rv;
 586
 587     alloc_fd = open("/dev/xen/gntalloc", O_RDWR);
 588     if (alloc_fd<=0) {
 589         perror("Could not open /dev/xen/gntalloc! Have you loaded the xen_gntalloc module?");
 590         return 1;
 591     }
 592
 593     dev_fd = open("/dev/xen/gntdev", O_RDWR);
 594     if (dev_fd<=0) {
 595         perror("Could not open /dev/xen/gntdev! Have you loaded the xen_gntdev module?");
 596         return 1;
 597     }
 598
 599     /* use dom0 */
 600     gref_->domid = 0;
 601     gref_->flags = GNTALLOC_FLAG_WRITABLE;
 602     gref_->count = 1;
 603
 604     rv = ioctl(alloc_fd, IOCTL_GNTALLOC_ALLOC_GREF, gref_);
 605     if (rv) {
 606         pa_log_debug("Xen audio sink: src-add error: %s (rv=%d)\n", strerror(errno), rv);
 607         return rv;
 608     }
 609
 610     /*addr=NULL(default),length, prot,             flags,    fd,         offset*/
 611     *addr = mmap(0, 4096, PROT_READ|PROT_WRITE, MAP_SHARED, alloc_fd, gref_->index);
 612     if (*addr == MAP_FAILED) {
 613         *addr = 0;
 614         pa_log_debug("Xen audio sink: mmap'ing shared page failed\n");
 615         return rv;
 616     }
 617
 618     pa_log_debug("Xen audio sink: Got grant #%d. Mapped locally at %Ld=%p\n",
 619             gref_->gref_ids[0], (long long)gref_->index, *addr);
 620
 621     /* skip this for now
 622        struct ioctl_gntalloc_unmap_notify uarg = {
 623        .index = gref->index + offsetof(struct shr_page, notifies[0]),
 624        .action = UNMAP_NOTIFY_CLEAR_BYTE
 625        };
 626
 627        rv = ioctl(a_fd, IOCTL_GNTALLOC_SET_UNMAP_NOTIFY, &uarg);
 628        if (rv)
 629        pa_log_debug("gntalloc unmap notify error: %s (rv=%d)\n", strerror(errno), rv);
 630        */
 631
 632     close(alloc_fd);
 633     close(dev_fd);
 634
 635     return rv;
 636 }
 637
 638 #define RING_FREE_BYTES ((r->usable_buffer_space - (r->prod_indx-r->cons_indx) -1) % r->usable_buffer_space)
 639 static int ring_write(struct ring *r, void *src, int length) {
 640     int full = 0;
 641     for (;;) {
 642         /* free space may be split over the end of the buffer */
 643         int first_chunk_size = (r->usable_buffer_space-r->prod_indx);
 644         int second_chunk_size = (r->cons_indx>=r->prod_indx)? (r->cons_indx) : 0;
 645         int l, fl, sl;
 646
 647         /* full? */
 648         if (RING_FREE_BYTES==0) {
 649             /*XXX hardcoded*/
 650             if (full>=100) {
 651                 errno = EINTR;
 652                 return -1;
 653             }
 654             /*XXX hardcoded */
 655             usleep(1000);
 656             /* should return in 100ms max; definitely not midstream */
 657             full++;
 658             continue;
 659         }
 660
 661         /* calculate lengths in case of a split buffer */
 662         l = PA_MIN((int)RING_FREE_BYTES, length);
 663         fl = PA_MIN(l, first_chunk_size);
 664         sl = PA_MIN(l-fl, second_chunk_size);
 665
 666         memcpy(r->buffer+r->prod_indx, src, fl);
 667         if (sl)
 668             memcpy(r->buffer, ((char*)src)+fl, sl);
 669         r->prod_indx = (r->prod_indx+fl+sl) % r->usable_buffer_space;
 670
 671         return sl+fl;
 672     }
 673 }
 674
 675 static int publish_param(const char *paramname, const char *value) {
 676     char keybuf[128], valbuf[32];
 677
 678     snprintf(keybuf, sizeof keybuf, "device/audio/%d/%s", device_id, paramname);
 679     snprintf(valbuf, sizeof valbuf, "%s", value);
 680     return xs_write(xsh, 0, keybuf, valbuf, strlen(valbuf));
 681 }
 682
 683 static int publish_param_int(const char *paramname, const int value) {
 684     char keybuf[128], valbuf[32];
 685     snprintf(keybuf, sizeof keybuf, "device/audio/%d/%s", device_id, paramname);
 686     snprintf(valbuf, sizeof valbuf, "%d", value);
 687     return xs_write(xsh, 0, keybuf, valbuf, strlen(valbuf));
 688 }
 689
 690 static char* read_param(const char *paramname) {
 691     char keybuf[128];
 692     unsigned int len;
 693     int my_domid;
 694
 695     my_domid = atoi(xs_read(xsh, 0, "domid", &len));
 696     snprintf(keybuf, sizeof(keybuf), "/local/domain/0/backend/audio/%d/%d/%s", my_domid, device_id, paramname);
 697     /* remember to free lvalue! */
 698     return xs_read(xsh, 0, keybuf, &len);
 699 }
 700
 701 static int publish_spec(pa_sample_spec *sample_spec) {
 702     /* Publish spec and set state to XenbusStateInitWait*/
 703     int ret;
 704
 705     ret = publish_param("format", pa_sample_format_to_string(sample_spec->format));
 706     ret += publish_param_int("rate", sample_spec->rate);
 707     ret += publish_param_int("channels", sample_spec->channels);
 708
 709     return ret;
 710 }
 711
 712 static int read_backend_default_spec(pa_sample_spec *sample_spec) {
 713     /* Read spec from backend */
 714     char *out;
 715
 716     out = read_param("default-format");
 717     sample_spec->format = pa_parse_sample_format(out);
 718     free(out);
 719
 720     out = read_param("default-rate");
 721     sample_spec->rate = atoi(out);
 722     free(out);
 723
 724     out = read_param("default-channels");
 725     sample_spec->channels = atoi(out);
 726     free(out);
 727
 728     return 0;
 729 }
 730
 731 static int register_backend_state_watch() {
 732     char keybuf[128];
 733     int my_domid;
 734     unsigned int len;
 735
 736     my_domid = atoi(xs_read(xsh, 0, "domid", &len));
 737     snprintf(keybuf, sizeof(keybuf), "/local/domain/0/backend/audio/%d/%d/state", my_domid, device_id);
 738     if (!xs_watch(xsh, keybuf, "xenpvaudiofrontendsinktoken")) {
 739         perror("xs_watch failed");
 740         return -EINVAL;
 741     }
 742     return 0;
 743 }
 744
 745 static int wait_for_backend_state_change() {
 746     char keybuf[128];
 747     int my_domid;
 748     unsigned int len;
 749
 750     int backend_state;
 751     int seconds;
 752     char *buf, **vec;
 753     int ret;
 754
 755     int xs_fd;
 756     struct timeval tv;
 757     fd_set watch_fdset;
 758     int start, now;
 759
 760     backend_state = STATE_UNDEFINED;
 761     xs_fd = xs_fileno(xsh);
 762     start = now = time(NULL);
 763
 764     my_domid = atoi(xs_read(xsh, 0, "domid", &len));
 765     snprintf(keybuf, sizeof(keybuf), "/local/domain/0/backend/audio/%d/%d/state", my_domid, device_id);
 766
 767     /*XXX: hardcoded */
 768     seconds = 10;
 769     do {
 770         tv.tv_usec = 0;
 771         tv.tv_sec = (start + seconds) - now;
 772         FD_ZERO(&watch_fdset);
 773         FD_SET(xs_fd, &watch_fdset);
 774         ret=select(xs_fd + 1, &watch_fdset, NULL, NULL, &tv);
 775
 776         if (ret==-1)
 777             /* error */
 778             return -1;
 779         else if (ret) {
 780
 781             /* Read the watch to drain the buffer */
 782             vec = xs_read_watch(xsh, &len);
 783
 784             buf = xs_read(xsh, XBT_NULL, vec[0], &len);
 785             if (buf == 0) {
 786                 /* usually means that the backend isn't there yet */
 787                 continue;
 788             };
 789             backend_state = atoi(buf);
 790
 791             free(buf);
 792             free(vec);
 793         }
 794     /* else: timeout */
 795     } while (backend_state == STATE_UNDEFINED && \
 796             (now = time(NULL)) < start + seconds);
 797
 798     return backend_state;
 799 }