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 *
gst_buffer_get_flags_string(GstBuffer * buffer)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 "sync-after\000non-droppable\000FIXME";
62 static const guint8 flag_idx[] = { 0, 1, 2, 3, 4, 9, 21, 29, 36, 46, 53,
63 60, 64, 74, 85, 96, 107, 121,
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 *
gst_buffer_get_meta_string(GstBuffer * buffer)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 /* Adapted from GLib (gio/gioprivate.h)
125 *
126 * POSIX defines IOV_MAX/UIO_MAXIOV as the maximum number of iovecs that can
127 * be sent in one go. We define our own version of it here as there are two
128 * possible names, and also define a fall-back value if none of the constants
129 * are defined */
130 #if defined(IOV_MAX)
131 #define GST_IOV_MAX IOV_MAX
132 #elif defined(UIO_MAXIOV)
133 #define GST_IOV_MAX UIO_MAXIOV
134 #elif defined(__APPLE__)
135 /* For osx/ios, UIO_MAXIOV is documented in writev(2), but <sys/uio.h>
136 * only declares it if defined(KERNEL) */
137 #define GST_IOV_MAX 512
138 #else
139 /* 16 is the minimum value required by POSIX */
140 #define GST_IOV_MAX 16
141 #endif
142
143 static gssize
gst_writev(gint fd,const struct iovec * iov,gint iovcnt,gsize total_bytes)144 gst_writev (gint fd, const struct iovec *iov, gint iovcnt, gsize total_bytes)
145 {
146 gssize written;
147
148 #ifdef HAVE_SYS_UIO_H
149 if (iovcnt <= GST_IOV_MAX) {
150 do {
151 written = writev (fd, iov, iovcnt);
152 } while (written < 0 && errno == EINTR);
153 } else
154 #endif
155 {
156 gint i;
157
158 /* We merge the memories here because technically write()/writev() is
159 * supposed to be atomic, which it's not if we do multiple separate
160 * write() calls. It's very doubtful anyone cares though in our use
161 * cases, and it's not clear how that can be reconciled with the
162 * possibility of short writes, so in any case we might want to
163 * simplify this later or just remove it. */
164 if (total_bytes <= FDSINK_MAX_MALLOC_SIZE) {
165 gchar *mem, *p;
166
167 if (total_bytes <= FDSINK_MAX_ALLOCA_SIZE)
168 mem = g_alloca (total_bytes);
169 else
170 mem = g_malloc (total_bytes);
171
172 p = mem;
173 for (i = 0; i < iovcnt; ++i) {
174 memcpy (p, iov[i].iov_base, iov[i].iov_len);
175 p += iov[i].iov_len;
176 }
177
178 do {
179 written = write (fd, mem, total_bytes);
180 } while (written < 0 && errno == EINTR);
181
182 if (total_bytes > FDSINK_MAX_ALLOCA_SIZE)
183 g_free (mem);
184 } else {
185 gssize ret;
186
187 written = 0;
188 for (i = 0; i < iovcnt; ++i) {
189 do {
190 ret = write (fd, iov[i].iov_base, iov[i].iov_len);
191 } while (ret < 0 && errno == EINTR);
192 if (ret > 0)
193 written += ret;
194 if (ret != iov[i].iov_len)
195 break;
196 }
197 }
198 }
199
200 return written;
201 }
202
203 static gsize
fill_vectors(struct iovec * vecs,GstMapInfo * maps,guint n,GstBuffer * buf)204 fill_vectors (struct iovec *vecs, GstMapInfo * maps, guint n, GstBuffer * buf)
205 {
206 GstMemory *mem;
207 gsize size = 0;
208 guint i;
209
210 g_assert (gst_buffer_n_memory (buf) == n);
211
212 for (i = 0; i < n; ++i) {
213 mem = gst_buffer_peek_memory (buf, i);
214 if (gst_memory_map (mem, &maps[i], GST_MAP_READ)) {
215 vecs[i].iov_base = maps[i].data;
216 vecs[i].iov_len = maps[i].size;
217 } else {
218 GST_WARNING ("Failed to map memory %p for reading", mem);
219 vecs[i].iov_base = (void *) "";
220 vecs[i].iov_len = 0;
221 }
222 size += vecs[i].iov_len;
223 }
224
225 return size;
226 }
227
228 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)229 gst_writev_buffers (GstObject * sink, gint fd, GstPoll * fdset,
230 GstBuffer ** buffers, guint num_buffers, guint8 * mem_nums,
231 guint total_mem_num, guint64 * bytes_written, guint64 skip)
232 {
233 struct iovec *vecs;
234 GstMapInfo *map_infos;
235 GstFlowReturn flow_ret;
236 gsize size = 0;
237 guint i, j;
238
239 GST_LOG_OBJECT (sink, "%u buffers, %u memories", num_buffers, total_mem_num);
240
241 vecs = g_newa (struct iovec, total_mem_num);
242 map_infos = g_newa (GstMapInfo, total_mem_num);
243
244 /* populate output vectors */
245 for (i = 0, j = 0; i < num_buffers; ++i) {
246 size += fill_vectors (&vecs[j], &map_infos[j], mem_nums[i], buffers[i]);
247 j += mem_nums[i];
248 }
249
250 /* now write it all out! */
251 {
252 gssize ret, left;
253 guint n_vecs = total_mem_num;
254
255 left = size;
256
257 if (skip) {
258 ret = skip;
259 errno = 0;
260 goto skip_first;
261 }
262
263 do {
264 #ifndef HAVE_WIN32
265 if (fdset != NULL) {
266 do {
267 GST_DEBUG_OBJECT (sink, "going into select, have %" G_GSSIZE_FORMAT
268 " bytes to write", left);
269 ret = gst_poll_wait (fdset, GST_CLOCK_TIME_NONE);
270 } while (ret == -1 && (errno == EINTR || errno == EAGAIN));
271
272 if (ret == -1) {
273 if (errno == EBUSY)
274 goto stopped;
275 else
276 goto select_error;
277 }
278 }
279 #endif
280
281 ret = gst_writev (fd, vecs, n_vecs, left);
282
283 if (ret > 0) {
284 if (bytes_written)
285 *bytes_written += ret;
286 }
287
288 skip_first:
289
290 if (ret == left)
291 break;
292
293 if (ret < 0 && (errno == EAGAIN || errno == EWOULDBLOCK)) {
294 /* do nothing, try again */
295 } else if (ret < 0) {
296 goto write_error;
297 } else if (ret < left) {
298 /* skip vectors that have been written in full */
299 while (ret >= vecs[0].iov_len) {
300 ret -= vecs[0].iov_len;
301 left -= vecs[0].iov_len;
302 ++vecs;
303 --n_vecs;
304 }
305 g_assert (n_vecs > 0);
306 /* skip partially written vector data */
307 if (ret > 0) {
308 vecs[0].iov_len -= ret;
309 vecs[0].iov_base = ((guint8 *) vecs[0].iov_base) + ret;
310 left -= ret;
311 }
312 }
313 #ifdef HAVE_WIN32
314 /* do short sleep on windows where we don't use gst_poll(),
315 * to avoid excessive busy looping */
316 if (fdset != NULL)
317 g_usleep (1000);
318 #endif
319
320 }
321 while (left > 0);
322 }
323
324 flow_ret = GST_FLOW_OK;
325
326 out:
327
328 for (i = 0; i < total_mem_num; ++i)
329 gst_memory_unmap (map_infos[i].memory, &map_infos[i]);
330
331 return flow_ret;
332
333 /* ERRORS */
334 #ifndef HAVE_WIN32
335 select_error:
336 {
337 GST_ELEMENT_ERROR (sink, RESOURCE, READ, (NULL),
338 ("select on file descriptor: %s", g_strerror (errno)));
339 GST_DEBUG_OBJECT (sink, "Error during select: %s", g_strerror (errno));
340 flow_ret = GST_FLOW_ERROR;
341 goto out;
342 }
343 stopped:
344 {
345 GST_DEBUG_OBJECT (sink, "Select stopped");
346 flow_ret = GST_FLOW_FLUSHING;
347 goto out;
348 }
349 #endif
350 write_error:
351 {
352 switch (errno) {
353 case ENOSPC:
354 GST_ELEMENT_ERROR (sink, RESOURCE, NO_SPACE_LEFT, (NULL), (NULL));
355 break;
356 default:{
357 GST_ELEMENT_ERROR (sink, RESOURCE, WRITE, (NULL),
358 ("Error while writing to file descriptor %d: %s",
359 fd, g_strerror (errno)));
360 }
361 }
362 flow_ret = GST_FLOW_ERROR;
363 goto out;
364 }
365 }
366