code.vuplus.com Git - vuplus_xbmc/blob - lib/cmyth/libcmyth/ringbuf.c

   1 /*
   2  *  Copyright (C) 2004-2012, Eric Lund
   3  *  http://www.mvpmc.org/
   4  *
   5  *  This library is free software; you can redistribute it and/or
   6  *  modify it under the terms of the GNU Lesser General Public
   7  *  License as published by the Free Software Foundation; either
   8  *  version 2.1 of the License, or (at your option) any later version.
   9  *
  10  *  This library is distributed in the hope that it will be useful,
  11  *  but WITHOUT ANY WARRANTY; without even the implied warranty of
  12  *  MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
  13  *  Lesser General Public License for more details.
  14  *
  15  *  You should have received a copy of the GNU Lesser General Public
  16  *  License along with this library; if not, write to the Free Software
  17  *  Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA  02111-1307  USA
  18  */
  19
  20 /*
  21  * ringbuf.c -    functions to handle operations on MythTV ringbuffers.  A
  22  *                MythTV Ringbuffer is the part of the backend that handles
  23  *                recording of live-tv for streaming to a MythTV frontend.
  24  *                This allows the watcher to do things like pause, rewind
  25  *                and so forth on live-tv.
  26  */
  27 #include <stdlib.h>
  28 #include <stdio.h>
  29 #include <errno.h>
  30 #include <string.h>
  31 #include <sys/types.h>
  32 #include <cmyth_local.h>
  33
  34 /*
  35  * cmyth_ringbuf_destroy(cmyth_ringbuf_t rb)
  36  *
  37  * Scope: PRIVATE (static)
  38  *
  39  * Description
  40  *
  41  * Clean up and free a ring buffer structure.  This should only be done
  42  * by the ref_release() code.  Everyone else should call
  43  * ref_release() because ring buffer structures are reference
  44  * counted.
  45  *
  46  * Return Value:
  47  *
  48  * None.
  49  */
  50 static void
  51 cmyth_ringbuf_destroy(cmyth_ringbuf_t rb)
  52 {
  53         cmyth_dbg(CMYTH_DBG_DEBUG, "%s\n", __FUNCTION__);
  54         if (!rb) {
  55                 return;
  56         }
  57
  58         if (rb->ringbuf_url) {
  59                 ref_release(rb->ringbuf_url);
  60         }
  61         if (rb->ringbuf_hostname) {
  62                 ref_release(rb->ringbuf_hostname);
  63         }
  64 }
  65
  66 /*
  67  * cmyth_ringbuf_create(void)
  68  *
  69  * Scope: PUBLIC
  70  *
  71  * Description
  72  *
  73  * Allocate and initialize a ring buffer structure.
  74  *
  75  * Return Value:
  76  *
  77  * Success: A non-NULL cmyth_ringbuf_t (this type is a pointer)
  78  *
  79  * Failure: A NULL cmyth_ringbuf_t
  80  */
  81 cmyth_ringbuf_t
  82 cmyth_ringbuf_create(void)
  83 {
  84         cmyth_ringbuf_t ret = ref_alloc(sizeof(*ret));
  85
  86         cmyth_dbg(CMYTH_DBG_DEBUG, "%s\n", __FUNCTION__);
  87         if (!ret) {
  88                 return NULL;
  89         }
  90
  91         ret->conn_data = NULL;
  92         ret->ringbuf_url = NULL;
  93         ret->ringbuf_size = 0;
  94         ret->ringbuf_fill = 0;
  95         ret->file_pos = 0;
  96         ret->file_id = 0;
  97         ret->ringbuf_hostname = NULL;
  98         ret->ringbuf_port = 0;
  99         ref_set_destroy(ret, (ref_destroy_t)cmyth_ringbuf_destroy);
 100         return ret;
 101 }
 102
 103 /*
 104  * cmyth_ringbuf_setup(cmyth_recorder_t old_rec)
 105  *
 106  * Scope: PUBLIC
 107  *
 108  * Description
 109  *
 110  * Set up the ring buffer inside a recorder for use in playing live
 111  * tv.  The recorder is supplied.  This will be duplicated and
 112  * released, so the caller can re-use the same variable to hold the
 113  * return.  The new copy of the recorder will have a ringbuffer set up
 114  * within it.
 115  *
 116  * Return Value:
 117  *
 118  * Success: A pointer to a new recorder structure with a ringbuffer
 119  *
 120  * Faiure: NULL
 121  */
 122 cmyth_recorder_t
 123 cmyth_ringbuf_setup(cmyth_recorder_t rec)
 124 {
 125         static const char service[]="rbuf://";
 126         cmyth_recorder_t new_rec = NULL;
 127         char *host = NULL;
 128         char *port = NULL;
 129         char *path = NULL;
 130         char tmp;
 131
 132         int err, count;
 133         int r;
 134         int64_t size, fill;
 135         char msg[256];
 136         char url[1024];
 137         char buf[32];
 138         cmyth_conn_t control;
 139
 140         if (!rec) {
 141                 cmyth_dbg(CMYTH_DBG_ERROR, "%s: no recorder connection\n",
 142                           __FUNCTION__);
 143                 return NULL;
 144         }
 145
 146         control = rec->rec_conn;
 147
 148         pthread_mutex_lock(&mutex);
 149
 150         snprintf(msg, sizeof(msg),
 151                  "QUERY_RECORDER %u[]:[]SETUP_RING_BUFFER[]:[]0",
 152                  rec->rec_id);
 153
 154         if ((err=cmyth_send_message(control, msg)) < 0) {
 155                 cmyth_dbg(CMYTH_DBG_ERROR,
 156                           "%s: cmyth_send_message() failed (%d)\n",
 157                           __FUNCTION__, err);
 158                 goto out;
 159         }
 160
 161         count = cmyth_rcv_length(control);
 162
 163         if (control->conn_version >= 16) {
 164                 r = cmyth_rcv_string(control, &err, buf, sizeof(buf)-1, count);
 165                 count -= r;
 166         }
 167         r = cmyth_rcv_string(control, &err, url, sizeof(url)-1, count);
 168         count -= r;
 169
 170         if ((r=cmyth_rcv_int64(control, &err, &size, count)) < 0) {
 171                 cmyth_dbg(CMYTH_DBG_ERROR,
 172                           "%s: cmyth_rcv_length() failed (%d)\n",
 173                           __FUNCTION__, r);
 174                 goto out;
 175         }
 176         count -= r;
 177
 178         if ((r=cmyth_rcv_int64(control, &err, &fill, count)) < 0) {
 179                 cmyth_dbg(CMYTH_DBG_ERROR,
 180                           "%s: cmyth_rcv_length() failed (%d)\n",
 181                           __FUNCTION__, r);
 182                 goto out;
 183         }
 184
 185         cmyth_dbg(CMYTH_DBG_DEBUG, "%s: url is: '%s'\n",
 186                   __FUNCTION__, url);
 187         path = url;
 188         if (strncmp(url, service, sizeof(service) - 1) == 0) {
 189                 /*
 190                  * The URL starts with rbuf://.  The rest looks like
 191                  * <host>:<port>/<filename>.
 192                  */
 193                 host = url + strlen(service);
 194                 port = strchr(host, ':');
 195                 if (!port) {
 196                         /*
 197                          * This does not seem to be a proper URL, so just
 198                          * assume it is a filename, and get out.
 199                          */
 200                         cmyth_dbg(CMYTH_DBG_DEBUG,
 201                                   "%s: 1 port %s, host = %s\n",
 202                                   __FUNCTION__, port, host);
 203                         goto out;
 204                 }
 205                 port = port + 1;
 206                 path = strchr(port, '/');
 207                 if (!path) {
 208                         /*
 209                          * This does not seem to be a proper URL, so just
 210                          * assume it is a filename, and get out.
 211                          */
 212                         cmyth_dbg(CMYTH_DBG_DEBUG, "%s: no path\n",
 213                                   __FUNCTION__);
 214                         goto out;
 215                 }
 216         }
 217
 218         new_rec = cmyth_recorder_dup(rec);
 219         if (new_rec == NULL) {
 220                 cmyth_dbg(CMYTH_DBG_DEBUG, "%s: cannot create recorder\n",
 221                           __FUNCTION__);
 222                 goto out;
 223         }
 224         ref_release(rec);
 225         new_rec->rec_ring = cmyth_ringbuf_create();
 226
 227         tmp = *(port - 1);
 228         *(port - 1) = '\0';
 229         new_rec->rec_ring->ringbuf_hostname = ref_strdup(host);
 230         *(port - 1) = tmp;
 231         tmp = *(path);
 232         *(path) = '\0';
 233         new_rec->rec_ring->ringbuf_port = atoi(port);
 234         *(path) = tmp;
 235         new_rec->rec_ring->ringbuf_url = ref_strdup(url);
 236         new_rec->rec_ring->ringbuf_size = size;
 237         new_rec->rec_ring->ringbuf_fill = fill;
 238
 239     out:
 240         pthread_mutex_unlock(&mutex);
 241
 242         return new_rec;
 243 }
 244
 245 char *
 246 cmyth_ringbuf_pathname(cmyth_recorder_t rec)
 247 {
 248         return ref_hold(rec->rec_ring->ringbuf_url);
 249 }
 250
 251 /*
 252  * cmyth_ringbuf_get_block(cmyth_recorder_t rec, char *buf, unsigned long len)
 253  * Scope: PUBLIC
 254  * Description
 255  * Read incoming file data off the network into a buffer of length len.
 256  *
 257  * Return Value:
 258  * Sucess: number of bytes read into buf
 259  * Failure: -1
 260  */
 261 int
 262 cmyth_ringbuf_get_block(cmyth_recorder_t rec, char *buf, unsigned long len)
 263 {
 264         struct timeval tv;
 265         fd_set fds;
 266
 267         if (rec == NULL)
 268                 return -EINVAL;
 269
 270         tv.tv_sec = 10;
 271         tv.tv_usec = 0;
 272         FD_ZERO(&fds);
 273         FD_SET(rec->rec_ring->conn_data->conn_fd, &fds);
 274         if (select((int)rec->rec_ring->conn_data->conn_fd+1,
 275                    NULL, &fds, NULL, &tv) == 0) {
 276                 rec->rec_ring->conn_data->conn_hang = 1;
 277                 return 0;
 278         } else {
 279                 rec->rec_ring->conn_data->conn_hang = 0;
 280         }
 281         return recv(rec->rec_ring->conn_data->conn_fd, buf, len, 0);
 282 }
 283
 284 int
 285 cmyth_ringbuf_select(cmyth_recorder_t rec, struct timeval *timeout)
 286 {
 287         fd_set fds;
 288         int ret;
 289         cmyth_socket_t fd;
 290         if (rec == NULL)
 291                 return -EINVAL;
 292
 293         fd = rec->rec_ring->conn_data->conn_fd;
 294
 295         FD_ZERO(&fds);
 296         FD_SET(fd, &fds);
 297
 298         ret = select((int)fd+1, &fds, NULL, NULL, timeout);
 299
 300         if (ret == 0)
 301                 rec->rec_ring->conn_data->conn_hang = 1;
 302         else
 303                 rec->rec_ring->conn_data->conn_hang = 0;
 304
 305         return ret;
 306 }
 307
 308 /*
 309  * cmyth_ringbuf_request_block(cmyth_ringbuf_t file, unsigned long len)
 310  *
 311  * Scope: PUBLIC
 312  *
 313  * Description
 314  *
 315  * Request a file data block of a certain size, and return when the
 316  * block has been transfered.
 317  *
 318  * Return Value:
 319  *
 320  * Sucess: number of bytes transfered
 321  *
 322  * Failure: an int containing -errno
 323  */
 324 int
 325 cmyth_ringbuf_request_block(cmyth_recorder_t rec, unsigned long len)
 326 {
 327         int err, count;
 328         int r;
 329         long c, ret;
 330         char msg[256];
 331
 332         if (!rec) {
 333                 cmyth_dbg(CMYTH_DBG_ERROR, "%s: no connection\n",
 334                           __FUNCTION__);
 335                 return -EINVAL;
 336         }
 337
 338         pthread_mutex_lock(&mutex);
 339
 340         if(len > (unsigned int)rec->rec_conn->conn_tcp_rcvbuf)
 341                 len = (unsigned int)rec->rec_conn->conn_tcp_rcvbuf;
 342
 343         snprintf(msg, sizeof(msg),
 344                  "QUERY_RECORDER %u[]:[]REQUEST_BLOCK_RINGBUF[]:[]%ld",
 345                  rec->rec_id, len);
 346
 347         if ((err = cmyth_send_message(rec->rec_conn, msg)) < 0) {
 348                 cmyth_dbg(CMYTH_DBG_ERROR,
 349                           "%s: cmyth_send_message() failed (%d)\n",
 350                           __FUNCTION__, err);
 351                 ret = err;
 352                 goto out;
 353         }
 354
 355         count = cmyth_rcv_length(rec->rec_conn);
 356         if ((r=cmyth_rcv_long(rec->rec_conn, &err, &c, count)) < 0) {
 357                 cmyth_dbg(CMYTH_DBG_ERROR,
 358                           "%s: cmyth_rcv_length() failed (%d)\n",
 359                           __FUNCTION__, r);
 360                 ret = err;
 361                 goto out;
 362         }
 363
 364         rec->rec_ring->file_pos += c;
 365         ret = c;
 366
 367     out:
 368         pthread_mutex_unlock(&mutex);
 369
 370         return ret;
 371 }
 372
 373 /*
 374  * cmyth_ringbuf_read (cmyth_recorder_t rec, char *buf, unsigned long len)
 375  *
 376  * Scope: PUBLIC
 377  *
 378  * Description
 379  *
 380  * Request and read a block of data from backend
 381  *
 382  * Return Value:
 383  *
 384  * Sucess: number of bytes transfered
 385  *
 386  * Failure: an int containing -errno
 387  */
 388 int cmyth_ringbuf_read(cmyth_recorder_t rec, char *buf, unsigned long len)
 389 {
 390         int err, count;
 391         int ret, req, nfds;
 392         char *end, *cur;
 393         char msg[256];
 394         struct timeval tv;
 395         fd_set fds;
 396
 397         if (!rec)
 398         {
 399                 cmyth_dbg (CMYTH_DBG_ERROR, "%s: no connection\n",
 400                            __FUNCTION__);
 401                 return -EINVAL;
 402         }
 403
 404         pthread_mutex_lock (&mutex);
 405
 406         snprintf(msg, sizeof(msg),
 407                  "QUERY_RECORDER %u[]:[]REQUEST_BLOCK_RINGBUF[]:[]%ld",
 408                  rec->rec_id, len);
 409
 410         if ( (err = cmyth_send_message (rec->rec_conn, msg) ) < 0)
 411         {
 412                 cmyth_dbg (CMYTH_DBG_ERROR,
 413                            "%s: cmyth_send_message() failed (%d)\n",
 414                            __FUNCTION__, err);
 415                 ret = err;
 416                 goto out;
 417         }
 418
 419         nfds = 0;
 420         req = 1;
 421         cur = buf;
 422         end = buf+len;
 423
 424         while (cur < end || req)
 425         {
 426                 tv.tv_sec = 20;
 427                 tv.tv_usec = 0;
 428                 FD_ZERO (&fds);
 429                 if(req) {
 430                         if((int)rec->rec_conn->conn_fd > nfds)
 431                                 nfds = (int)rec->rec_conn->conn_fd;
 432                         FD_SET (rec->rec_conn->conn_fd, &fds);
 433                 }
 434                 if((int)rec->rec_ring->conn_data->conn_fd > nfds)
 435                         nfds = (int)rec->rec_ring->conn_data->conn_fd;
 436                 FD_SET (rec->rec_ring->conn_data->conn_fd, &fds);
 437
 438                 if ((ret = select (nfds+1, &fds, NULL, NULL,&tv)) < 0)
 439                 {
 440                         cmyth_dbg (CMYTH_DBG_ERROR,
 441                                    "%s: select(() failed (%d)\n",
 442                                    __FUNCTION__, ret);
 443                         goto out;
 444                 }
 445
 446                 if (ret == 0)
 447                 {
 448                         rec->rec_ring->conn_data->conn_hang = 1;
 449                         rec->rec_conn->conn_hang = 1;
 450                         ret = -ETIMEDOUT;
 451                         goto out;
 452                 }
 453
 454                 /* check control connection */
 455                 if (FD_ISSET(rec->rec_conn->conn_fd, &fds) )
 456                 {
 457
 458                         if ((count = cmyth_rcv_length (rec->rec_conn)) < 0)
 459                         {
 460                                 cmyth_dbg (CMYTH_DBG_ERROR,
 461                                            "%s: cmyth_rcv_length() failed (%d)\n",
 462                                            __FUNCTION__, count);
 463                                 ret = count;
 464                                 goto out;
 465                         }
 466
 467                         if ((ret = cmyth_rcv_ulong (rec->rec_conn, &err, &len, count))< 0)
 468                         {
 469                                 cmyth_dbg (CMYTH_DBG_ERROR,
 470                                            "%s: cmyth_rcv_long() failed (%d)\n",
 471                                            __FUNCTION__, ret);
 472                                 ret = err;
 473                                 goto out;
 474                         }
 475
 476                         rec->rec_ring->file_pos += len;
 477                         req = 0;
 478                         end = buf+len;
 479                 }
 480
 481                 /* check data connection */
 482                 if (FD_ISSET(rec->rec_ring->conn_data->conn_fd, &fds))
 483                 {
 484
 485                         if ((ret = recv (rec->rec_ring->conn_data->conn_fd, cur, end-cur, 0)) < 0)
 486                         {
 487                                 cmyth_dbg (CMYTH_DBG_ERROR,
 488                                            "%s: recv() failed (%d)\n",
 489                                            __FUNCTION__, ret);
 490                                 goto out;
 491                         }
 492                         cur += ret;
 493                 }
 494         }
 495
 496         ret = end - buf;
 497 out:
 498         pthread_mutex_unlock (&mutex);
 499         return ret;
 500 }
 501
 502 /*
 503  * cmyth_ringbuf_seek(
 504  *                    cmyth_ringbuf_t file, long long offset, int whence)
 505  *
 506  * Scope: PUBLIC
 507  *
 508  * Description
 509  *
 510  * Seek to a new position in the file based on the value of whence:
 511  *      SEEK_SET
 512  *              The offset is set to offset bytes.
 513  *      SEEK_CUR
 514  *              The offset is set to the current position plus offset bytes.
 515  *      SEEK_END
 516  *              The offset is set to the size of the file minus offset bytes.
 517  *
 518  * Return Value:
 519  *
 520  * Sucess: 0
 521  *
 522  * Failure: an int containing -errno
 523  */
 524 long long
 525 cmyth_ringbuf_seek(cmyth_recorder_t rec,
 526                    long long offset, int whence)
 527 {
 528         char msg[128];
 529         int err;
 530         int count;
 531         int64_t c;
 532         long r;
 533         long long ret;
 534         cmyth_ringbuf_t ring;
 535
 536         if (rec == NULL)
 537                 return -EINVAL;
 538
 539         ring = rec->rec_ring;
 540
 541         if ((offset == 0) && (whence == SEEK_CUR))
 542                 return ring->file_pos;
 543
 544         pthread_mutex_lock(&mutex);
 545
 546         snprintf(msg, sizeof(msg),
 547                  "QUERY_RECORDER %u[]:[]SEEK_RINGBUF[]:[]%d[]:[]%d[]:[]%d[]:[]%d[]:[]%d",
 548                  rec->rec_id,
 549                  (int32_t)(offset >> 32),
 550                  (int32_t)(offset & 0xffffffff),
 551                  whence,
 552                  (int32_t)(ring->file_pos >> 32),
 553                  (int32_t)(ring->file_pos & 0xffffffff));
 554
 555         if ((err = cmyth_send_message(rec->rec_conn, msg)) < 0) {
 556                 cmyth_dbg(CMYTH_DBG_ERROR,
 557                           "%s: cmyth_send_message() failed (%d)\n",
 558                           __FUNCTION__, err);
 559                 ret = err;
 560                 goto out;
 561         }
 562
 563         count = cmyth_rcv_length(rec->rec_conn);
 564         if ((r=cmyth_rcv_int64(rec->rec_conn, &err, &c, count)) < 0) {
 565                 cmyth_dbg(CMYTH_DBG_ERROR,
 566                           "%s: cmyth_rcv_length() failed (%d)\n",
 567                           __FUNCTION__, r);
 568                 ret = err;
 569                 goto out;
 570         }
 571
 572         switch (whence) {
 573         case SEEK_SET:
 574                 ring->file_pos = offset;
 575                 break;
 576         case SEEK_CUR:
 577                 ring->file_pos += offset;
 578                 break;
 579         case SEEK_END:
 580                 ring->file_pos = ring->file_length - offset;
 581                 break;
 582         }
 583
 584         ret = ring->file_pos;
 585
 586     out:
 587         pthread_mutex_unlock(&mutex);
 588
 589         return ret;
 590 }