src/pulsecore/ioline.c

   1 /***
   2   This file is part of PulseAudio.
   3
   4   Copyright 2004-2006 Lennart Poettering
   5
   6   PulseAudio is free software; you can redistribute it and/or modify
   7   it under the terms of the GNU Lesser General Public License as published
   8   by the Free Software Foundation; either version 2.1 of the License,
   9   or (at your option) any later version.
  10
  11   PulseAudio is distributed in the hope that it will be useful, but
  12   WITHOUT ANY WARRANTY; without even the implied warranty of
  13   MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
  14   General Public License for more details.
  15
  16   You should have received a copy of the GNU Lesser General Public License
  17   along with PulseAudio; if not, write to the Free Software
  18   Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307
  19   USA.
  20 ***/
  21
  22 #ifdef HAVE_CONFIG_H
  23 #include <config.h>
  24 #endif
  25
  26 #include <errno.h>
  27 #include <stdio.h>
  28 #include <stdlib.h>
  29 #include <string.h>
  30
  31 #include <pulse/xmalloc.h>
  32
  33 #include <pulsecore/socket.h>
  34 #include <pulsecore/core-error.h>
  35 #include <pulsecore/core-util.h>
  36 #include <pulsecore/log.h>
  37 #include <pulsecore/macro.h>
  38 #include <pulsecore/refcnt.h>
  39
  40 #include "ioline.h"
  41
  42 #define BUFFER_LIMIT (64*1024)
  43 #define READ_SIZE (1024)
  44
  45 struct pa_ioline {
  46     PA_REFCNT_DECLARE;
  47
  48     pa_iochannel *io;
  49     pa_defer_event *defer_event;
  50     pa_mainloop_api *mainloop;
  51
  52     char *wbuf;
  53     size_t wbuf_length, wbuf_index, wbuf_valid_length;
  54
  55     char *rbuf;
  56     size_t rbuf_length, rbuf_index, rbuf_valid_length;
  57
  58     pa_ioline_cb_t callback;
  59     void *userdata;
  60
  61     pa_ioline_drain_cb_t drain_callback;
  62     void *drain_userdata;
  63
  64     pa_bool_t dead:1;
  65     pa_bool_t defer_close:1;
  66 };
  67
  68 static void io_callback(pa_iochannel*io, void *userdata);
  69 static void defer_callback(pa_mainloop_api*m, pa_defer_event*e, void *userdata);
  70
  71 pa_ioline* pa_ioline_new(pa_iochannel *io) {
  72     pa_ioline *l;
  73     pa_assert(io);
  74
  75     l = pa_xnew(pa_ioline, 1);
  76     PA_REFCNT_INIT(l);
  77     l->io = io;
  78
  79     l->wbuf = NULL;
  80     l->wbuf_length = l->wbuf_index = l->wbuf_valid_length = 0;
  81
  82     l->rbuf = NULL;
  83     l->rbuf_length = l->rbuf_index = l->rbuf_valid_length = 0;
  84
  85     l->callback = NULL;
  86     l->userdata = NULL;
  87
  88     l->drain_callback = NULL;
  89     l->drain_userdata = NULL;
  90
  91     l->mainloop = pa_iochannel_get_mainloop_api(io);
  92
  93     l->defer_event = l->mainloop->defer_new(l->mainloop, defer_callback, l);
  94     l->mainloop->defer_enable(l->defer_event, 0);
  95
  96     l->dead = FALSE;
  97     l->defer_close = FALSE;
  98
  99     pa_iochannel_set_callback(io, io_callback, l);
 100
 101     return l;
 102 }
 103
 104 static void ioline_free(pa_ioline *l) {
 105     pa_assert(l);
 106
 107     if (l->io)
 108         pa_iochannel_free(l->io);
 109
 110     if (l->defer_event)
 111         l->mainloop->defer_free(l->defer_event);
 112
 113     pa_xfree(l->wbuf);
 114     pa_xfree(l->rbuf);
 115     pa_xfree(l);
 116 }
 117
 118 void pa_ioline_unref(pa_ioline *l) {
 119     pa_assert(l);
 120     pa_assert(PA_REFCNT_VALUE(l) >= 1);
 121
 122     if (PA_REFCNT_DEC(l) <= 0)
 123         ioline_free(l);
 124 }
 125
 126 pa_ioline* pa_ioline_ref(pa_ioline *l) {
 127     pa_assert(l);
 128     pa_assert(PA_REFCNT_VALUE(l) >= 1);
 129
 130     PA_REFCNT_INC(l);
 131     return l;
 132 }
 133
 134 void pa_ioline_close(pa_ioline *l) {
 135     pa_assert(l);
 136     pa_assert(PA_REFCNT_VALUE(l) >= 1);
 137
 138     l->dead = TRUE;
 139
 140     if (l->io) {
 141         pa_iochannel_free(l->io);
 142         l->io = NULL;
 143     }
 144
 145     if (l->defer_event) {
 146         l->mainloop->defer_free(l->defer_event);
 147         l->defer_event = NULL;
 148     }
 149
 150     if (l->callback)
 151         l->callback = NULL;
 152 }
 153
 154 void pa_ioline_puts(pa_ioline *l, const char *c) {
 155     size_t len;
 156
 157     pa_assert(l);
 158     pa_assert(PA_REFCNT_VALUE(l) >= 1);
 159     pa_assert(c);
 160
 161     if (l->dead)
 162         return;
 163
 164     len = strlen(c);
 165     if (len > BUFFER_LIMIT - l->wbuf_valid_length)
 166         len = BUFFER_LIMIT - l->wbuf_valid_length;
 167
 168     if (len) {
 169         pa_assert(l->wbuf_length >= l->wbuf_valid_length);
 170
 171         /* In case the allocated buffer is too small, enlarge it. */
 172         if (l->wbuf_valid_length + len > l->wbuf_length) {
 173             size_t n = l->wbuf_valid_length+len;
 174             char *new = pa_xnew(char, (unsigned) n);
 175
 176             if (l->wbuf) {
 177                 memcpy(new, l->wbuf+l->wbuf_index, l->wbuf_valid_length);
 178                 pa_xfree(l->wbuf);
 179             }
 180
 181             l->wbuf = new;
 182             l->wbuf_length = n;
 183             l->wbuf_index = 0;
 184         } else if (l->wbuf_index + l->wbuf_valid_length + len > l->wbuf_length) {
 185
 186             /* In case the allocated buffer fits, but the current index is too far from the start, move it to the front. */
 187             memmove(l->wbuf, l->wbuf+l->wbuf_index, l->wbuf_valid_length);
 188             l->wbuf_index = 0;
 189         }
 190
 191         pa_assert(l->wbuf_index + l->wbuf_valid_length + len <= l->wbuf_length);
 192
 193         /* Append the new string */
 194         memcpy(l->wbuf + l->wbuf_index + l->wbuf_valid_length, c, len);
 195         l->wbuf_valid_length += len;
 196
 197         l->mainloop->defer_enable(l->defer_event, 1);
 198     }
 199 }
 200
 201 void pa_ioline_set_callback(pa_ioline*l, pa_ioline_cb_t callback, void *userdata) {
 202     pa_assert(l);
 203     pa_assert(PA_REFCNT_VALUE(l) >= 1);
 204
 205     if (l->dead)
 206         return;
 207
 208     l->callback = callback;
 209     l->userdata = userdata;
 210 }
 211
 212 void pa_ioline_set_drain_callback(pa_ioline*l, pa_ioline_drain_cb_t callback, void *userdata) {
 213     pa_assert(l);
 214     pa_assert(PA_REFCNT_VALUE(l) >= 1);
 215
 216     if (l->dead)
 217         return;
 218
 219     l->drain_callback = callback;
 220     l->drain_userdata = userdata;
 221 }
 222
 223 static void failure(pa_ioline *l, pa_bool_t process_leftover) {
 224     pa_assert(l);
 225     pa_assert(PA_REFCNT_VALUE(l) >= 1);
 226     pa_assert(!l->dead);
 227
 228     if (process_leftover && l->rbuf_valid_length > 0) {
 229         /* Pass the last missing bit to the client */
 230
 231         if (l->callback) {
 232             char *p = pa_xstrndup(l->rbuf+l->rbuf_index, l->rbuf_valid_length);
 233             l->callback(l, p, l->userdata);
 234             pa_xfree(p);
 235         }
 236     }
 237
 238     if (l->callback) {
 239         l->callback(l, NULL, l->userdata);
 240         l->callback = NULL;
 241     }
 242
 243     pa_ioline_close(l);
 244 }
 245
 246 static void scan_for_lines(pa_ioline *l, size_t skip) {
 247     pa_assert(l);
 248     pa_assert(PA_REFCNT_VALUE(l) >= 1);
 249     pa_assert(skip < l->rbuf_valid_length);
 250
 251     while (!l->dead && l->rbuf_valid_length > skip) {
 252         char *e, *p;
 253         size_t m;
 254
 255         if (!(e = memchr(l->rbuf + l->rbuf_index + skip, '\n', l->rbuf_valid_length - skip)))
 256             break;
 257
 258         *e = 0;
 259
 260         p = l->rbuf + l->rbuf_index;
 261         m = strlen(p);
 262
 263         l->rbuf_index += m+1;
 264         l->rbuf_valid_length -= m+1;
 265
 266         /* A shortcut for the next time */
 267         if (l->rbuf_valid_length == 0)
 268             l->rbuf_index = 0;
 269
 270         if (l->callback)
 271             l->callback(l, pa_strip_nl(p), l->userdata);
 272
 273         skip = 0;
 274     }
 275
 276     /* If the buffer became too large and still no newline was found, drop it. */
 277     if (l->rbuf_valid_length >= BUFFER_LIMIT)
 278         l->rbuf_index = l->rbuf_valid_length = 0;
 279 }
 280
 281 static int do_write(pa_ioline *l);
 282
 283 static int do_read(pa_ioline *l) {
 284     pa_assert(l);
 285     pa_assert(PA_REFCNT_VALUE(l) >= 1);
 286
 287     while (l->io && !l->dead && pa_iochannel_is_readable(l->io)) {
 288         ssize_t r;
 289         size_t len;
 290
 291         len = l->rbuf_length - l->rbuf_index - l->rbuf_valid_length;
 292
 293         /* Check if we have to enlarge the read buffer */
 294         if (len < READ_SIZE) {
 295             size_t n = l->rbuf_valid_length+READ_SIZE;
 296
 297             if (n >= BUFFER_LIMIT)
 298                 n = BUFFER_LIMIT;
 299
 300             if (l->rbuf_length >= n) {
 301                 /* The current buffer is large enough, let's just move the data to the front */
 302                 if (l->rbuf_valid_length)
 303                     memmove(l->rbuf, l->rbuf+l->rbuf_index, l->rbuf_valid_length);
 304             } else {
 305                 /* Enlarge the buffer */
 306                 char *new = pa_xnew(char, (unsigned) n);
 307                 if (l->rbuf_valid_length)
 308                     memcpy(new, l->rbuf+l->rbuf_index, l->rbuf_valid_length);
 309                 pa_xfree(l->rbuf);
 310                 l->rbuf = new;
 311                 l->rbuf_length = n;
 312             }
 313
 314             l->rbuf_index = 0;
 315         }
 316
 317         len = l->rbuf_length - l->rbuf_index - l->rbuf_valid_length;
 318
 319         pa_assert(len >= READ_SIZE);
 320
 321         /* Read some data */
 322         if ((r = pa_iochannel_read(l->io, l->rbuf+l->rbuf_index+l->rbuf_valid_length, len)) <= 0) {
 323
 324             if (r < 0 && errno == EAGAIN)
 325                 return 0;
 326
 327             if (r < 0 && errno != ECONNRESET) {
 328                 pa_log("read(): %s", pa_cstrerror(errno));
 329                 failure(l, FALSE);
 330             } else
 331                 failure(l, TRUE);
 332
 333             return -1;
 334         }
 335
 336         l->rbuf_valid_length += (size_t) r;
 337
 338         /* Look if a line has been terminated in the newly read data */
 339         scan_for_lines(l, l->rbuf_valid_length - (size_t) r);
 340     }
 341
 342     return 0;
 343 }
 344
 345 /* Try to flush the buffer */
 346 static int do_write(pa_ioline *l) {
 347     ssize_t r;
 348
 349     pa_assert(l);
 350     pa_assert(PA_REFCNT_VALUE(l) >= 1);
 351
 352     while (l->io && !l->dead && pa_iochannel_is_writable(l->io) && l->wbuf_valid_length > 0) {
 353
 354         if ((r = pa_iochannel_write(l->io, l->wbuf+l->wbuf_index, l->wbuf_valid_length)) <= 0) {
 355
 356             if (r < 0 && errno == EAGAIN)
 357                 break;
 358
 359             if (r < 0 && errno != EPIPE)
 360                 pa_log("write(): %s", pa_cstrerror(errno));
 361
 362             failure(l, FALSE);
 363
 364             return -1;
 365         }
 366
 367         l->wbuf_index += (size_t) r;
 368         l->wbuf_valid_length -= (size_t) r;
 369
 370         /* A shortcut for the next time */
 371         if (l->wbuf_valid_length == 0)
 372             l->wbuf_index = 0;
 373     }
 374
 375     if (l->wbuf_valid_length <= 0 && l->drain_callback)
 376         l->drain_callback(l, l->drain_userdata);
 377
 378     return 0;
 379 }
 380
 381 /* Try to flush read/write data */
 382 static void do_work(pa_ioline *l) {
 383     pa_assert(l);
 384     pa_assert(PA_REFCNT_VALUE(l) >= 1);
 385
 386     pa_ioline_ref(l);
 387
 388     l->mainloop->defer_enable(l->defer_event, 0);
 389
 390     if (!l->dead)
 391         do_read(l);
 392
 393     if (!l->dead)
 394         do_write(l);
 395
 396     if (l->defer_close && !l->wbuf_valid_length)
 397         failure(l, TRUE);
 398
 399     pa_ioline_unref(l);
 400 }
 401
 402 static void io_callback(pa_iochannel*io, void *userdata) {
 403     pa_ioline *l = userdata;
 404
 405     pa_assert(io);
 406     pa_assert(l);
 407     pa_assert(PA_REFCNT_VALUE(l) >= 1);
 408
 409     do_work(l);
 410 }
 411
 412 static void defer_callback(pa_mainloop_api*m, pa_defer_event*e, void *userdata) {
 413     pa_ioline *l = userdata;
 414
 415     pa_assert(l);
 416     pa_assert(PA_REFCNT_VALUE(l) >= 1);
 417     pa_assert(l->mainloop == m);
 418     pa_assert(l->defer_event == e);
 419
 420     do_work(l);
 421 }
 422
 423 void pa_ioline_defer_close(pa_ioline *l) {
 424     pa_assert(l);
 425     pa_assert(PA_REFCNT_VALUE(l) >= 1);
 426
 427     l->defer_close = TRUE;
 428
 429     if (!l->wbuf_valid_length)
 430         l->mainloop->defer_enable(l->defer_event, 1);
 431 }
 432
 433 void pa_ioline_printf(pa_ioline *l, const char *format, ...) {
 434     char *t;
 435     va_list ap;
 436
 437     pa_assert(l);
 438     pa_assert(PA_REFCNT_VALUE(l) >= 1);
 439
 440     va_start(ap, format);
 441     t = pa_vsprintf_malloc(format, ap);
 442     va_end(ap);
 443
 444     pa_ioline_puts(l, t);
 445     pa_xfree(t);
 446 }
 447
 448 pa_iochannel* pa_ioline_detach_iochannel(pa_ioline *l) {
 449     pa_iochannel *r;
 450
 451     pa_assert(l);
 452
 453     if (!l->io)
 454         return NULL;
 455
 456     r = l->io;
 457     l->io = NULL;
 458
 459     pa_iochannel_set_callback(r, NULL, NULL);
 460
 461     return r;
 462 }
 463
 464 pa_bool_t pa_ioline_is_drained(pa_ioline *l) {
 465     pa_assert(l);
 466
 467     return l->wbuf_valid_length <= 0;
 468 }