/* GStreamer * Copyright (C) 2011 David Schleef * Copyright (C) 2011 Tim-Philipp Müller * Copyright (C) 2014 Tim-Philipp Müller * Copyright (C) 2014 Vincent Penquerc'h * * 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 #ifdef HAVE_UNISTD_H #include #endif #ifdef HAVE_SYS_UIO_H #include #endif #include #include #include #include "gst/gst.h" #include "gstelements_private.h" #ifdef G_OS_WIN32 # define WIN32_LEAN_AND_MEAN /* prevents from including too many things */ # include # 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 */ gchar * 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; } /* Returns a newly-allocated string describing the metas on this buffer, or NULL */ gchar * gst_buffer_get_meta_string (GstBuffer * buffer) { gpointer state = NULL; GstMeta *meta; GString *s = NULL; while ((meta = gst_buffer_iterate_meta (buffer, &state))) { const gchar *desc = g_type_name (meta->info->type); if (s == NULL) s = g_string_new (NULL); else g_string_append (s, ", "); g_string_append (s, desc); } return (s != NULL) ? g_string_free (s, FALSE) : NULL; } /* 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 */ /* UIO_MAXIOV is documented in writev(2), but only * declares it on osx/ios if defined(KERNEL) */ #ifndef UIO_MAXIOV #define UIO_MAXIOV 512 #endif 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 * bytes_written, guint64 skip) { 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; if (skip) { ret = skip; errno = 0; goto skip_first; } 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 (bytes_written) *bytes_written += ret; } skip_first: 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; } }