gstreamer/plugins/elements/gstelements_private.c

/* GStreamer
 * Copyright (C) 2011 David Schleef <ds@schleef.org>
 * Copyright (C) 2011 Tim-Philipp Müller <tim.muller@collabora.co.uk>
 * Copyright (C) 2014 Tim-Philipp Müller <tim@centricular.com>
 * Copyright (C) 2014 Vincent Penquerc'h <vincent@collabora.co.uk>
 *
 * gstelements_private.c: Shared code for core elements
 *
 * This library is free software; you can redistribute it and/or
 * modify it under the terms of the GNU Library General Public
 * License as published by the Free Software Foundation; either
 * version 2 of the License, or (at your option) any later version.
 *
 * This library is distributed in the hope that it will be useful,
 * but WITHOUT ANY WARRANTY; without even the implied warranty of
 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
 * Library General Public License for more details.
 *
 * You should have received a copy of the GNU Library General Public
 * License along with this library; if not, write to the
 * Free Software Foundation, Inc., 51 Franklin St, Fifth Floor,
 * Boston, MA 02110-1301, USA.
 */

#ifdef HAVE_CONFIG_H
# include "config.h"
#endif
#include <stdio.h>
#ifdef HAVE_UNISTD_H
#include <unistd.h>
#endif
#ifdef HAVE_SYS_UIO_H
#include <sys/uio.h>
#endif
#include <errno.h>
#include <string.h>
#include <string.h>
#include "gst/gst.h"
#include "gstelements_private.h"

#ifdef G_OS_WIN32
#  define WIN32_LEAN_AND_MEAN   /* prevents from including too many things */
#  include <windows.h>
#  undef WIN32_LEAN_AND_MEAN
#  ifndef EWOULDBLOCK
#  define EWOULDBLOCK EAGAIN
#  endif
#endif /* G_OS_WIN32 */

#define BUFFER_FLAG_SHIFT 4

G_STATIC_ASSERT ((1 << BUFFER_FLAG_SHIFT) == GST_MINI_OBJECT_FLAG_LAST);

/* Returns a newly allocated string describing the flags on this buffer */
char *
gst_buffer_get_flags_string (GstBuffer * buffer)
{
  static const char flag_strings[] =
      "\000\000\000\000live\000decode-only\000discont\000resync\000corrupted\000"
      "marker\000header\000gap\000droppable\000delta-unit\000tag-memory\000"
      "FIXME";
  static const guint8 flag_idx[] = { 0, 1, 2, 3, 4, 9, 21, 29, 36, 46, 53,
    60, 64, 74, 85, 96
  };
  int i, max_bytes;
  char *flag_str, *end;

  /* max size is all flag strings plus a space or terminator after each one */
  max_bytes = sizeof (flag_strings);
  flag_str = g_malloc (max_bytes);

  end = flag_str;
  end[0] = '\0';
  for (i = BUFFER_FLAG_SHIFT; i < G_N_ELEMENTS (flag_idx); i++) {
    if (GST_MINI_OBJECT_CAST (buffer)->flags & (1 << i)) {
      strcpy (end, flag_strings + flag_idx[i]);
      end += strlen (end);
      end[0] = ' ';
      end[1] = '\0';
      end++;
    }
  }

  return flag_str;
}

/* Define our own iovec structure here, so that we can use it unconditionally
 * in the code below and use almost the same code path for systems where
 * writev() is supported and those were it's not supported */
#ifndef HAVE_SYS_UIO_H
struct iovec
{
  gpointer iov_base;
  gsize iov_len;
};
#endif

/* completely arbitrary thresholds */
#define FDSINK_MAX_ALLOCA_SIZE (64 * 1024)      /* 64k */
#define FDSINK_MAX_MALLOC_SIZE ( 8 * 1024 * 1024)       /*  8M */

static gssize
gst_writev (gint fd, const struct iovec *iov, gint iovcnt, gsize total_bytes)
{
  gssize written;

#ifdef HAVE_SYS_UIO_H
  if (iovcnt <= UIO_MAXIOV) {
    do {
      written = writev (fd, iov, iovcnt);
    } while (written < 0 && errno == EINTR);
  } else
#endif
  {
    gint i;

    /* We merge the memories here because technically write()/writev() is
     * supposed to be atomic, which it's not if we do multiple separate
     * write() calls. It's very doubtful anyone cares though in our use
     * cases, and it's not clear how that can be reconciled with the
     * possibility of short writes, so in any case we might want to
     * simplify this later or just remove it. */
    if (total_bytes <= FDSINK_MAX_MALLOC_SIZE) {
      gchar *mem, *p;

      if (total_bytes <= FDSINK_MAX_ALLOCA_SIZE)
        mem = g_alloca (total_bytes);
      else
        mem = g_malloc (total_bytes);

      p = mem;
      for (i = 0; i < iovcnt; ++i) {
        memcpy (p, iov[i].iov_base, iov[i].iov_len);
        p += iov[i].iov_len;
      }

      do {
        written = write (fd, mem, total_bytes);
      } while (written < 0 && errno == EINTR);

      if (total_bytes > FDSINK_MAX_ALLOCA_SIZE)
        g_free (mem);
    } else {
      gssize ret;

      written = 0;
      for (i = 0; i < iovcnt; ++i) {
        do {
          ret = write (fd, iov[i].iov_base, iov[i].iov_len);
        } while (ret < 0 && errno == EINTR);
        if (ret > 0)
          written += ret;
        if (ret != iov[i].iov_len)
          break;
      }
    }
  }

  return written;
}

static gsize
fill_vectors (struct iovec *vecs, GstMapInfo * maps, guint n, GstBuffer * buf)
{
  GstMemory *mem;
  gsize size = 0;
  guint i;

  g_assert (gst_buffer_n_memory (buf) == n);

  for (i = 0; i < n; ++i) {
    mem = gst_buffer_peek_memory (buf, i);
    if (gst_memory_map (mem, &maps[i], GST_MAP_READ)) {
      vecs[i].iov_base = maps[i].data;
      vecs[i].iov_len = maps[i].size;
    } else {
      GST_WARNING ("Failed to map memory %p for reading", mem);
      vecs[i].iov_base = (void *) "";
      vecs[i].iov_len = 0;
    }
    size += vecs[i].iov_len;
  }

  return size;
}

GstFlowReturn
gst_writev_buffers (GstObject * sink, gint fd, GstPoll * fdset,
    GstBuffer ** buffers, guint num_buffers, guint8 * mem_nums,
    guint total_mem_num, guint64 * total_written, guint64 * cur_pos)
{
  struct iovec *vecs;
  GstMapInfo *map_infos;
  GstFlowReturn flow_ret;
  gsize size = 0;
  guint i, j;

  GST_LOG_OBJECT (sink, "%u buffers, %u memories", num_buffers, total_mem_num);

  vecs = g_newa (struct iovec, total_mem_num);
  map_infos = g_newa (GstMapInfo, total_mem_num);

  /* populate output vectors */
  for (i = 0, j = 0; i < num_buffers; ++i) {
    size += fill_vectors (&vecs[j], &map_infos[j], mem_nums[i], buffers[i]);
    j += mem_nums[i];
  }

  /* now write it all out! */
  {
    gssize ret, left;
    guint n_vecs = total_mem_num;

    left = size;
    do {
#ifndef HAVE_WIN32
      if (fdset != NULL) {
        do {
          GST_DEBUG_OBJECT (sink, "going into select, have %" G_GSSIZE_FORMAT
              " bytes to write", left);
          ret = gst_poll_wait (fdset, GST_CLOCK_TIME_NONE);
        } while (ret == -1 && (errno == EINTR || errno == EAGAIN));

        if (ret == -1) {
          if (errno == EBUSY)
            goto stopped;
          else
            goto select_error;
        }
      }
#endif

      ret = gst_writev (fd, vecs, n_vecs, left);

      if (ret > 0) {
        if (total_written)
          *total_written += ret;
        if (cur_pos)
          *cur_pos += ret;
      }

      if (ret == left)
        break;

      if (ret < 0 && (errno == EAGAIN || errno == EWOULDBLOCK)) {
        /* do nothing, try again */
      } else if (ret < 0) {
        goto write_error;
      } else if (ret < left) {
        /* skip vectors that have been written in full */
        while (ret >= vecs[0].iov_len) {
          ret -= vecs[0].iov_len;
          left -= vecs[0].iov_len;
          ++vecs;
          --n_vecs;
        }
        g_assert (n_vecs > 0);
        /* skip partially written vector data */
        if (ret > 0) {
          vecs[0].iov_len -= ret;
          vecs[0].iov_base = ((guint8 *) vecs[0].iov_base) + ret;
          left -= ret;
        }
      }
#ifdef HAVE_WIN32
      /* do short sleep on windows where we don't use gst_poll(),
       * to avoid excessive busy looping */
      if (fdset != NULL)
        g_usleep (1000);
#endif

    }
    while (left > 0);
  }

  flow_ret = GST_FLOW_OK;

out:

  for (i = 0; i < total_mem_num; ++i)
    gst_memory_unmap (map_infos[i].memory, &map_infos[i]);

  return flow_ret;

/* ERRORS */
#ifndef HAVE_WIN32
select_error:
  {
    GST_ELEMENT_ERROR (sink, RESOURCE, READ, (NULL),
        ("select on file descriptor: %s", g_strerror (errno)));
    GST_DEBUG_OBJECT (sink, "Error during select: %s", g_strerror (errno));
    flow_ret = GST_FLOW_ERROR;
    goto out;
  }
stopped:
  {
    GST_DEBUG_OBJECT (sink, "Select stopped");
    flow_ret = GST_FLOW_FLUSHING;
    goto out;
  }
#endif
write_error:
  {
    switch (errno) {
      case ENOSPC:
        GST_ELEMENT_ERROR (sink, RESOURCE, NO_SPACE_LEFT, (NULL), (NULL));
        break;
      default:{
        GST_ELEMENT_ERROR (sink, RESOURCE, WRITE, (NULL),
            ("Error while writing to file descriptor %d: %s",
                fd, g_strerror (errno)));
      }
    }
    flow_ret = GST_FLOW_ERROR;
    goto out;
  }
}