plugins/elements/gstelements_private.c

   1 /* GStreamer
   2  * Copyright (C) 2011 David Schleef <ds@schleef.org>
   3  * Copyright (C) 2011 Tim-Philipp Müller <tim.muller@collabora.co.uk>
   4  * Copyright (C) 2014 Tim-Philipp Müller <tim@centricular.com>
   5  * Copyright (C) 2014 Vincent Penquerc'h <vincent@collabora.co.uk>
   6  *
   7  * gstelements_private.c: Shared code for core elements
   8  *
   9  * This library is free software; you can redistribute it and/or
  10  * modify it under the terms of the GNU Library General Public
  11  * License as published by the Free Software Foundation; either
  12  * version 2 of the License, or (at your option) any later version.
  13  *
  14  * This library is distributed in the hope that it will be useful,
  15  * but WITHOUT ANY WARRANTY; without even the implied warranty of
  16  * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
  17  * Library General Public License for more details.
  18  *
  19  * You should have received a copy of the GNU Library General Public
  20  * License along with this library; if not, write to the
  21  * Free Software Foundation, Inc., 51 Franklin St, Fifth Floor,
  22  * Boston, MA 02110-1301, USA.
  23  */
  24
  25 #ifdef HAVE_CONFIG_H
  26 # include "config.h"
  27 #endif
  28 #include <stdio.h>
  29 #ifdef HAVE_UNISTD_H
  30 #include <unistd.h>
  31 #endif
  32 #ifdef HAVE_SYS_UIO_H
  33 #include <sys/uio.h>
  34 #endif
  35 #include <errno.h>
  36 #include <string.h>
  37 #include <string.h>
  38 #include "gst/gst.h"
  39 #include "gstelements_private.h"
  40
  41 #ifdef G_OS_WIN32
  42 #  define WIN32_LEAN_AND_MEAN   /* prevents from including too many things */
  43 #  include <windows.h>
  44 #  undef WIN32_LEAN_AND_MEAN
  45 #  ifndef EWOULDBLOCK
  46 #  define EWOULDBLOCK EAGAIN
  47 #  endif
  48 #endif /* G_OS_WIN32 */
  49
  50 #define BUFFER_FLAG_SHIFT 4
  51
  52 G_STATIC_ASSERT ((1 << BUFFER_FLAG_SHIFT) == GST_MINI_OBJECT_FLAG_LAST);
  53
  54 /* Returns a newly allocated string describing the flags on this buffer */
  55 gchar *
  56 gst_buffer_get_flags_string (GstBuffer * buffer)
  57 {
  58   static const char flag_strings[] =
  59       "\000\000\000\000live\000decode-only\000discont\000resync\000corrupted\000"
  60       "marker\000header\000gap\000droppable\000delta-unit\000tag-memory\000"
  61       "FIXME";
  62   static const guint8 flag_idx[] = { 0, 1, 2, 3, 4, 9, 21, 29, 36, 46, 53,
  63     60, 64, 74, 85, 96
  64   };
  65   int i, max_bytes;
  66   char *flag_str, *end;
  67
  68   /* max size is all flag strings plus a space or terminator after each one */
  69   max_bytes = sizeof (flag_strings);
  70   flag_str = g_malloc (max_bytes);
  71
  72   end = flag_str;
  73   end[0] = '\0';
  74   for (i = BUFFER_FLAG_SHIFT; i < G_N_ELEMENTS (flag_idx); i++) {
  75     if (GST_MINI_OBJECT_CAST (buffer)->flags & (1 << i)) {
  76       strcpy (end, flag_strings + flag_idx[i]);
  77       end += strlen (end);
  78       end[0] = ' ';
  79       end[1] = '\0';
  80       end++;
  81     }
  82   }
  83
  84   return flag_str;
  85 }
  86
  87 /* Returns a newly-allocated string describing the metas on this buffer, or NULL */
  88 gchar *
  89 gst_buffer_get_meta_string (GstBuffer * buffer)
  90 {
  91   gpointer state = NULL;
  92   GstMeta *meta;
  93   GString *s = NULL;
  94
  95   while ((meta = gst_buffer_iterate_meta (buffer, &state))) {
  96     const gchar *desc = g_type_name (meta->info->type);
  97
  98     if (s == NULL)
  99       s = g_string_new (NULL);
 100     else
 101       g_string_append (s, ", ");
 102
 103     g_string_append (s, desc);
 104   }
 105
 106   return (s != NULL) ? g_string_free (s, FALSE) : NULL;
 107 }
 108
 109 /* Define our own iovec structure here, so that we can use it unconditionally
 110  * in the code below and use almost the same code path for systems where
 111  * writev() is supported and those were it's not supported */
 112 #ifndef HAVE_SYS_UIO_H
 113 struct iovec
 114 {
 115   gpointer iov_base;
 116   gsize iov_len;
 117 };
 118 #endif
 119
 120 /* completely arbitrary thresholds */
 121 #define FDSINK_MAX_ALLOCA_SIZE (64 * 1024)      /* 64k */
 122 #define FDSINK_MAX_MALLOC_SIZE ( 8 * 1024 * 1024)       /*  8M */
 123
 124 /* UIO_MAXIOV is documented in writev(2), but <sys/uio.h> only
 125  * declares it on osx/ios if defined(KERNEL) */
 126 #ifndef UIO_MAXIOV
 127 #define UIO_MAXIOV 512
 128 #endif
 129
 130 static gssize
 131 gst_writev (gint fd, const struct iovec *iov, gint iovcnt, gsize total_bytes)
 132 {
 133   gssize written;
 134
 135 #ifdef HAVE_SYS_UIO_H
 136   if (iovcnt <= UIO_MAXIOV) {
 137     do {
 138       written = writev (fd, iov, iovcnt);
 139     } while (written < 0 && errno == EINTR);
 140   } else
 141 #endif
 142   {
 143     gint i;
 144
 145     /* We merge the memories here because technically write()/writev() is
 146      * supposed to be atomic, which it's not if we do multiple separate
 147      * write() calls. It's very doubtful anyone cares though in our use
 148      * cases, and it's not clear how that can be reconciled with the
 149      * possibility of short writes, so in any case we might want to
 150      * simplify this later or just remove it. */
 151     if (total_bytes <= FDSINK_MAX_MALLOC_SIZE) {
 152       gchar *mem, *p;
 153
 154       if (total_bytes <= FDSINK_MAX_ALLOCA_SIZE)
 155         mem = g_alloca (total_bytes);
 156       else
 157         mem = g_malloc (total_bytes);
 158
 159       p = mem;
 160       for (i = 0; i < iovcnt; ++i) {
 161         memcpy (p, iov[i].iov_base, iov[i].iov_len);
 162         p += iov[i].iov_len;
 163       }
 164
 165       do {
 166         written = write (fd, mem, total_bytes);
 167       } while (written < 0 && errno == EINTR);
 168
 169       if (total_bytes > FDSINK_MAX_ALLOCA_SIZE)
 170         g_free (mem);
 171     } else {
 172       gssize ret;
 173
 174       written = 0;
 175       for (i = 0; i < iovcnt; ++i) {
 176         do {
 177           ret = write (fd, iov[i].iov_base, iov[i].iov_len);
 178         } while (ret < 0 && errno == EINTR);
 179         if (ret > 0)
 180           written += ret;
 181         if (ret != iov[i].iov_len)
 182           break;
 183       }
 184     }
 185   }
 186
 187   return written;
 188 }
 189
 190 static gsize
 191 fill_vectors (struct iovec *vecs, GstMapInfo * maps, guint n, GstBuffer * buf)
 192 {
 193   GstMemory *mem;
 194   gsize size = 0;
 195   guint i;
 196
 197   g_assert (gst_buffer_n_memory (buf) == n);
 198
 199   for (i = 0; i < n; ++i) {
 200     mem = gst_buffer_peek_memory (buf, i);
 201     if (gst_memory_map (mem, &maps[i], GST_MAP_READ)) {
 202       vecs[i].iov_base = maps[i].data;
 203       vecs[i].iov_len = maps[i].size;
 204     } else {
 205       GST_WARNING ("Failed to map memory %p for reading", mem);
 206       vecs[i].iov_base = (void *) "";
 207       vecs[i].iov_len = 0;
 208     }
 209     size += vecs[i].iov_len;
 210   }
 211
 212   return size;
 213 }
 214
 215 GstFlowReturn
 216 gst_writev_buffers (GstObject * sink, gint fd, GstPoll * fdset,
 217     GstBuffer ** buffers, guint num_buffers, guint8 * mem_nums,
 218     guint total_mem_num, guint64 * bytes_written, guint64 skip)
 219 {
 220   struct iovec *vecs;
 221   GstMapInfo *map_infos;
 222   GstFlowReturn flow_ret;
 223   gsize size = 0;
 224   guint i, j;
 225
 226   GST_LOG_OBJECT (sink, "%u buffers, %u memories", num_buffers, total_mem_num);
 227
 228   vecs = g_newa (struct iovec, total_mem_num);
 229   map_infos = g_newa (GstMapInfo, total_mem_num);
 230
 231   /* populate output vectors */
 232   for (i = 0, j = 0; i < num_buffers; ++i) {
 233     size += fill_vectors (&vecs[j], &map_infos[j], mem_nums[i], buffers[i]);
 234     j += mem_nums[i];
 235   }
 236
 237   /* now write it all out! */
 238   {
 239     gssize ret, left;
 240     guint n_vecs = total_mem_num;
 241
 242     left = size;
 243
 244     if (skip) {
 245       ret = skip;
 246       errno = 0;
 247       goto skip_first;
 248     }
 249
 250     do {
 251 #ifndef HAVE_WIN32
 252       if (fdset != NULL) {
 253         do {
 254           GST_DEBUG_OBJECT (sink, "going into select, have %" G_GSSIZE_FORMAT
 255               " bytes to write", left);
 256           ret = gst_poll_wait (fdset, GST_CLOCK_TIME_NONE);
 257         } while (ret == -1 && (errno == EINTR || errno == EAGAIN));
 258
 259         if (ret == -1) {
 260           if (errno == EBUSY)
 261             goto stopped;
 262           else
 263             goto select_error;
 264         }
 265       }
 266 #endif
 267
 268       ret = gst_writev (fd, vecs, n_vecs, left);
 269
 270       if (ret > 0) {
 271         if (bytes_written)
 272           *bytes_written += ret;
 273       }
 274
 275     skip_first:
 276
 277       if (ret == left)
 278         break;
 279
 280       if (ret < 0 && (errno == EAGAIN || errno == EWOULDBLOCK)) {
 281         /* do nothing, try again */
 282       } else if (ret < 0) {
 283         goto write_error;
 284       } else if (ret < left) {
 285         /* skip vectors that have been written in full */
 286         while (ret >= vecs[0].iov_len) {
 287           ret -= vecs[0].iov_len;
 288           left -= vecs[0].iov_len;
 289           ++vecs;
 290           --n_vecs;
 291         }
 292         g_assert (n_vecs > 0);
 293         /* skip partially written vector data */
 294         if (ret > 0) {
 295           vecs[0].iov_len -= ret;
 296           vecs[0].iov_base = ((guint8 *) vecs[0].iov_base) + ret;
 297           left -= ret;
 298         }
 299       }
 300 #ifdef HAVE_WIN32
 301       /* do short sleep on windows where we don't use gst_poll(),
 302        * to avoid excessive busy looping */
 303       if (fdset != NULL)
 304         g_usleep (1000);
 305 #endif
 306
 307     }
 308     while (left > 0);
 309   }
 310
 311   flow_ret = GST_FLOW_OK;
 312
 313 out:
 314
 315   for (i = 0; i < total_mem_num; ++i)
 316     gst_memory_unmap (map_infos[i].memory, &map_infos[i]);
 317
 318   return flow_ret;
 319
 320 /* ERRORS */
 321 #ifndef HAVE_WIN32
 322 select_error:
 323   {
 324     GST_ELEMENT_ERROR (sink, RESOURCE, READ, (NULL),
 325         ("select on file descriptor: %s", g_strerror (errno)));
 326     GST_DEBUG_OBJECT (sink, "Error during select: %s", g_strerror (errno));
 327     flow_ret = GST_FLOW_ERROR;
 328     goto out;
 329   }
 330 stopped:
 331   {
 332     GST_DEBUG_OBJECT (sink, "Select stopped");
 333     flow_ret = GST_FLOW_FLUSHING;
 334     goto out;
 335   }
 336 #endif
 337 write_error:
 338   {
 339     switch (errno) {
 340       case ENOSPC:
 341         GST_ELEMENT_ERROR (sink, RESOURCE, NO_SPACE_LEFT, (NULL), (NULL));
 342         break;
 343       default:{
 344         GST_ELEMENT_ERROR (sink, RESOURCE, WRITE, (NULL),
 345             ("Error while writing to file descriptor %d: %s",
 346                 fd, g_strerror (errno)));
 347       }
 348     }
 349     flow_ret = GST_FLOW_ERROR;
 350     goto out;
 351   }
 352 }