2 * Copyright (C) 1999 Erik Walthinsen <omega@cse.ogi.edu>
3 * Copyright (C) 2004 Wim Taymans <wim.taymans@gmail.com>
4 * Copyright (C) 2007 Peter Kjellerstedt <pkj@axis.com>
5 * Copyright (C) 2008 Ole André Vadla Ravnås <ole.andre.ravnas@tandberg.com>
7 * gstpoll.c: File descriptor set
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.
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.
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.
26 * @short_description: Keep track of file descriptors and make it possible
27 * to wait on them in a cancellable way
29 * A #GstPoll keeps track of file descriptors much like fd_set (used with
30 * select()) or a struct pollfd array (used with poll()). Once created with
31 * gst_poll_new(), the set can be used to wait for file descriptors to be
32 * readable and/or writable. It is possible to make this wait be controlled
33 * by specifying %TRUE for the @controllable flag when creating the set (or
34 * later calling gst_poll_set_controllable()).
36 * New file descriptors are added to the set using gst_poll_add_fd(), and
37 * removed using gst_poll_remove_fd(). Controlling which file descriptors
38 * should be waited for to become readable and/or writable are done using
39 * gst_poll_fd_ctl_read() and gst_poll_fd_ctl_write().
41 * Use gst_poll_wait() to wait for the file descriptors to actually become
42 * readable and/or writable, or to timeout if no file descriptor is available
43 * in time. The wait can be controlled by calling gst_poll_restart() and
44 * gst_poll_set_flushing().
46 * Once the file descriptor set has been waited for, one can use
47 * gst_poll_fd_has_closed() to see if the file descriptor has been closed,
48 * gst_poll_fd_has_error() to see if it has generated an error,
49 * gst_poll_fd_can_read() to see if it is possible to read from the file
50 * descriptor, and gst_poll_fd_can_write() to see if it is possible to
59 #include "gst_private.h"
60 #include "glib-compat-private.h"
62 #include <sys/types.h>
77 #ifdef HAVE_SYS_POLL_H
84 #include <sys/socket.h>
87 /* OS/X needs this because of bad headers */
90 /* The poll() emulation on OS/X doesn't handle fds=NULL, nfds=0,
91 * so we prefer our own poll emulation.
93 #if defined(BROKEN_POLL)
99 #define GST_CAT_DEFAULT GST_CAT_POLL
102 typedef struct _WinsockFd WinsockFd;
108 WSANETWORKEVENTS events;
109 glong ignored_event_mask;
116 GST_POLL_MODE_SELECT,
117 GST_POLL_MODE_PSELECT,
120 GST_POLL_MODE_WINDOWS
128 /* array of fds, always written to and read from with lock */
130 /* array of active fds, only written to from the waiting thread with the
131 * lock and read from with the lock or without the lock from the waiting
136 GstPollFD control_read_fd;
137 GstPollFD control_write_fd;
139 GArray *active_fds_ignored;
141 GArray *active_events;
146 gboolean controllable;
147 volatile gint waiting;
148 volatile gint control_pending;
149 volatile gint flushing;
151 volatile gint rebuild;
154 static gboolean gst_poll_fd_ctl_read_unlocked (GstPoll * set, GstPollFD * fd,
156 static gboolean gst_poll_add_fd_unlocked (GstPoll * set, GstPollFD * fd);
158 #define IS_FLUSHING(s) (g_atomic_int_get(&(s)->flushing))
159 #define SET_FLUSHING(s,val) (g_atomic_int_set(&(s)->flushing, (val)))
161 #define INC_WAITING(s) (g_atomic_int_add(&(s)->waiting, 1))
162 #define DEC_WAITING(s) (g_atomic_int_add(&(s)->waiting, -1))
163 #define GET_WAITING(s) (g_atomic_int_get(&(s)->waiting))
165 #define TEST_REBUILD(s) (g_atomic_int_compare_and_exchange(&(s)->rebuild, 1, 0))
166 #define MARK_REBUILD(s) (g_atomic_int_set(&(s)->rebuild, 1))
171 wake_event (GstPoll * set)
174 while ((num_written = write (set->control_write_fd.fd, "W", 1)) != 1) {
175 if (num_written == -1 && errno != EAGAIN && errno != EINTR) {
176 g_critical ("%p: failed to wake event: %s", set, strerror (errno));
184 release_event (GstPoll * set)
186 gchar buf[1] = { '\0' };
188 while ((num_read = read (set->control_read_fd.fd, buf, 1)) != 1) {
189 if (num_read == -1 && errno != EAGAIN && errno != EINTR) {
190 g_critical ("%p: failed to release event: %s", set, strerror (errno));
200 format_last_error (gchar * buf, size_t buf_len)
202 DWORD flags = FORMAT_MESSAGE_FROM_SYSTEM;
206 id = GetLastError ();
207 FormatMessage (flags, src, id, lang, buf, (DWORD) buf_len, NULL);
212 wake_event (GstPoll * set)
216 if (!SetEvent (set->wakeup_event)) {
217 gchar msg[1024] = "<unknown>";
218 format_last_error (msg, sizeof (msg));
219 g_critical ("%p: failed to set wakup_event: %s", set, msg);
228 release_event (GstPoll * set)
233 if (status = WaitForSingleObject (set->wakeup_event, INFINITE)) {
234 const gchar *reason = "unknown";
235 gchar msg[1024] = "<unknown>";
238 reason = "WAIT_ABANDONED";
241 reason = "WAIT_TIMEOUT";
244 format_last_error (msg, sizeof (msg));
251 g_critical ("%p: failed to block on wakup_event: %s", set, reason);
256 if (!ResetEvent (set->wakeup_event)) {
257 gchar msg[1024] = "<unknown>";
258 format_last_error (msg, sizeof (msg));
259 g_critical ("%p: failed to reset wakup_event: %s", set, msg);
269 /* the poll/select call is also performed on a control socket, that way
270 * we can send special commands to control it */
271 static inline gboolean
272 raise_wakeup (GstPoll * set)
274 gboolean result = TRUE;
276 /* makes testing control_pending and WAKE_EVENT() atomic. */
277 g_mutex_lock (&set->lock);
279 if (set->control_pending == 0) {
280 /* raise when nothing pending */
281 GST_LOG ("%p: raise", set);
282 result = wake_event (set);
286 set->control_pending++;
289 g_mutex_unlock (&set->lock);
294 static inline gboolean
295 release_wakeup (GstPoll * set)
297 gboolean result = FALSE;
299 /* makes testing/modifying control_pending and RELEASE_EVENT() atomic. */
300 g_mutex_lock (&set->lock);
302 if (set->control_pending > 0) {
303 /* release, only if this was the last pending. */
304 if (set->control_pending == 1) {
305 GST_LOG ("%p: release", set);
306 result = release_event (set);
312 set->control_pending--;
318 g_mutex_unlock (&set->lock);
324 release_all_wakeup (GstPoll * set)
328 /* makes testing control_pending and RELEASE_EVENT() atomic. */
329 g_mutex_lock (&set->lock);
331 if ((old = set->control_pending) > 0) {
332 GST_LOG ("%p: releasing %d", set, old);
333 if (release_event (set)) {
334 set->control_pending = 0;
340 g_mutex_unlock (&set->lock);
346 find_index (GArray * array, GstPollFD * fd)
355 /* start by assuming the index found in the fd is still valid */
356 if (fd->idx >= 0 && fd->idx < array->len) {
358 ifd = &g_array_index (array, struct pollfd, fd->idx);
360 ifd = &g_array_index (array, WinsockFd, fd->idx);
363 if (ifd->fd == fd->fd) {
368 /* the pollfd array has changed and we need to lookup the fd again */
369 for (i = 0; i < array->len; i++) {
371 ifd = &g_array_index (array, struct pollfd, i);
373 ifd = &g_array_index (array, WinsockFd, i);
376 if (ifd->fd == fd->fd) {
386 #if !defined(HAVE_PPOLL) && defined(HAVE_POLL)
387 /* check if all file descriptors will fit in an fd_set */
389 selectable_fds (GstPoll * set)
393 g_mutex_lock (&set->lock);
394 for (i = 0; i < set->fds->len; i++) {
395 struct pollfd *pfd = &g_array_index (set->fds, struct pollfd, i);
397 if (pfd->fd >= FD_SETSIZE)
400 g_mutex_unlock (&set->lock);
406 g_mutex_unlock (&set->lock);
411 /* check if the timeout will convert to a timeout value used for poll()
412 * without a loss of precision
415 pollable_timeout (GstClockTime timeout)
417 if (timeout == GST_CLOCK_TIME_NONE)
420 /* not a nice multiple of milliseconds */
421 if (timeout % 1000000)
429 choose_mode (GstPoll * set, GstClockTime timeout)
433 if (set->mode == GST_POLL_MODE_AUTO) {
435 mode = GST_POLL_MODE_PPOLL;
436 #elif defined(HAVE_POLL)
437 if (!selectable_fds (set) || pollable_timeout (timeout)) {
438 mode = GST_POLL_MODE_POLL;
441 mode = GST_POLL_MODE_PSELECT;
443 mode = GST_POLL_MODE_SELECT;
446 #elif defined(HAVE_PSELECT)
447 mode = GST_POLL_MODE_PSELECT;
449 mode = GST_POLL_MODE_SELECT;
459 pollfd_to_fd_set (GstPoll * set, fd_set * readfds, fd_set * writefds,
469 g_mutex_lock (&set->lock);
471 for (i = 0; i < set->active_fds->len; i++) {
472 struct pollfd *pfd = &g_array_index (set->fds, struct pollfd, i);
474 if (pfd->fd < FD_SETSIZE) {
475 if (pfd->events & POLLIN)
476 FD_SET (pfd->fd, readfds);
477 if (pfd->events & POLLOUT)
478 FD_SET (pfd->fd, writefds);
480 FD_SET (pfd->fd, errorfds);
481 if (pfd->fd > max_fd && (pfd->events & (POLLIN | POLLOUT)))
486 g_mutex_unlock (&set->lock);
492 fd_set_to_pollfd (GstPoll * set, fd_set * readfds, fd_set * writefds,
497 g_mutex_lock (&set->lock);
499 for (i = 0; i < set->active_fds->len; i++) {
500 struct pollfd *pfd = &g_array_index (set->active_fds, struct pollfd, i);
502 if (pfd->fd < FD_SETSIZE) {
504 if (FD_ISSET (pfd->fd, readfds))
505 pfd->revents |= POLLIN;
506 if (FD_ISSET (pfd->fd, writefds))
507 pfd->revents |= POLLOUT;
508 if (FD_ISSET (pfd->fd, errorfds))
509 pfd->revents |= POLLERR;
513 g_mutex_unlock (&set->lock);
515 #else /* G_OS_WIN32 */
517 * Translate errors thrown by the Winsock API used by GstPoll:
518 * WSAEventSelect, WSAWaitForMultipleEvents and WSAEnumNetworkEvents
521 gst_poll_winsock_error_to_errno (DWORD last_error)
523 switch (last_error) {
524 case WSA_INVALID_HANDLE:
529 case WSA_NOT_ENOUGH_MEMORY:
533 * Anything else, including:
534 * WSA_INVALID_PARAMETER, WSAEFAULT, WSAEINPROGRESS, WSAENETDOWN,
543 gst_poll_free_winsock_event (GstPoll * set, gint idx)
545 WinsockFd *wfd = &g_array_index (set->fds, WinsockFd, idx);
546 HANDLE event = g_array_index (set->events, HANDLE, idx);
548 WSAEventSelect (wfd->fd, event, 0);
553 gst_poll_update_winsock_event_mask (GstPoll * set, gint idx, glong flags,
558 wfd = &g_array_index (set->fds, WinsockFd, idx);
561 wfd->event_mask |= flags;
563 wfd->event_mask &= ~flags;
565 /* reset ignored state if the new mask doesn't overlap at all */
566 if ((wfd->ignored_event_mask & wfd->event_mask) == 0)
567 wfd->ignored_event_mask = 0;
571 gst_poll_prepare_winsock_active_sets (GstPoll * set)
575 g_array_set_size (set->active_fds, 0);
576 g_array_set_size (set->active_fds_ignored, 0);
577 g_array_set_size (set->active_events, 0);
578 g_array_append_val (set->active_events, set->wakeup_event);
580 for (i = 0; i < set->fds->len; i++) {
581 WinsockFd *wfd = &g_array_index (set->fds, WinsockFd, i);
582 HANDLE event = g_array_index (set->events, HANDLE, i);
584 if (wfd->ignored_event_mask == 0) {
587 g_array_append_val (set->active_fds, *wfd);
588 g_array_append_val (set->active_events, event);
590 ret = WSAEventSelect (wfd->fd, event, wfd->event_mask);
591 if (G_UNLIKELY (ret != 0)) {
592 errno = gst_poll_winsock_error_to_errno (WSAGetLastError ());
596 g_array_append_val (set->active_fds_ignored, wfd);
604 gst_poll_collect_winsock_events (GstPoll * set)
609 * We need to check which events are signaled, and call
610 * WSAEnumNetworkEvents for those that are, which resets
611 * the event and clears the internal network event records.
614 for (i = 0; i < set->active_fds->len; i++) {
615 WinsockFd *wfd = &g_array_index (set->active_fds, WinsockFd, i);
616 HANDLE event = g_array_index (set->active_events, HANDLE, i + 1);
619 wait_ret = WaitForSingleObject (event, 0);
620 if (wait_ret == WAIT_OBJECT_0) {
621 gint enum_ret = WSAEnumNetworkEvents (wfd->fd, event, &wfd->events);
623 if (G_UNLIKELY (enum_ret != 0)) {
625 errno = gst_poll_winsock_error_to_errno (WSAGetLastError ());
631 /* clear any previously stored result */
632 memset (&wfd->events, 0, sizeof (wfd->events));
636 /* If all went well we also need to reset the ignored fds. */
638 res += set->active_fds_ignored->len;
640 for (i = 0; i < set->active_fds_ignored->len; i++) {
641 WinsockFd *wfd = g_array_index (set->active_fds_ignored, WinsockFd *, i);
643 wfd->ignored_event_mask = 0;
646 g_array_set_size (set->active_fds_ignored, 0);
654 * gst_poll_new: (skip)
655 * @controllable: whether it should be possible to control a wait.
657 * Create a new file descriptor set. If @controllable, it
658 * is possible to restart or flush a call to gst_poll_wait() with
659 * gst_poll_restart() and gst_poll_set_flushing() respectively.
661 * Free-function: gst_poll_free
663 * Returns: (transfer full) (nullable): a new #GstPoll, or %NULL in
664 * case of an error. Free with gst_poll_free().
667 gst_poll_new (gboolean controllable)
671 nset = g_slice_new0 (GstPoll);
672 GST_DEBUG ("%p: new controllable : %d", nset, controllable);
673 g_mutex_init (&nset->lock);
675 nset->mode = GST_POLL_MODE_AUTO;
676 nset->fds = g_array_new (FALSE, FALSE, sizeof (struct pollfd));
677 nset->active_fds = g_array_new (FALSE, FALSE, sizeof (struct pollfd));
678 nset->control_read_fd.fd = -1;
679 nset->control_write_fd.fd = -1;
681 gint control_sock[2];
683 if (socketpair (PF_UNIX, SOCK_STREAM, 0, control_sock) < 0)
686 nset->control_read_fd.fd = control_sock[0];
687 nset->control_write_fd.fd = control_sock[1];
689 gst_poll_add_fd_unlocked (nset, &nset->control_read_fd);
690 gst_poll_fd_ctl_read_unlocked (nset, &nset->control_read_fd, TRUE);
693 nset->mode = GST_POLL_MODE_WINDOWS;
694 nset->fds = g_array_new (FALSE, FALSE, sizeof (WinsockFd));
695 nset->active_fds = g_array_new (FALSE, FALSE, sizeof (WinsockFd));
696 nset->active_fds_ignored = g_array_new (FALSE, FALSE, sizeof (WinsockFd *));
697 nset->events = g_array_new (FALSE, FALSE, sizeof (HANDLE));
698 nset->active_events = g_array_new (FALSE, FALSE, sizeof (HANDLE));
700 nset->wakeup_event = CreateEvent (NULL, TRUE, FALSE, NULL);
703 /* ensure (re)build, though already sneakily set in non-windows case */
706 nset->controllable = controllable;
707 nset->control_pending = 0;
715 GST_WARNING ("%p: can't create socket pair !", nset);
716 gst_poll_free (nset);
723 * gst_poll_new_timer: (skip)
725 * Create a new poll object that can be used for scheduling cancellable
728 * A timeout is performed with gst_poll_wait(). Multiple timeouts can be
729 * performed from different threads.
731 * Free-function: gst_poll_free
733 * Returns: (transfer full) (nullable): a new #GstPoll, or %NULL in
734 * case of an error. Free with gst_poll_free().
737 gst_poll_new_timer (void)
741 /* make a new controllable poll set */
742 if (!(poll = gst_poll_new (TRUE)))
754 * @set: (transfer full): a file descriptor set.
756 * Free a file descriptor set.
759 gst_poll_free (GstPoll * set)
761 g_return_if_fail (set != NULL);
763 GST_DEBUG ("%p: freeing", set);
766 if (set->control_write_fd.fd >= 0)
767 close (set->control_write_fd.fd);
768 if (set->control_read_fd.fd >= 0)
769 close (set->control_read_fd.fd);
771 CloseHandle (set->wakeup_event);
776 for (i = 0; i < set->events->len; i++)
777 gst_poll_free_winsock_event (set, i);
780 g_array_free (set->active_events, TRUE);
781 g_array_free (set->events, TRUE);
782 g_array_free (set->active_fds_ignored, TRUE);
785 g_array_free (set->active_fds, TRUE);
786 g_array_free (set->fds, TRUE);
787 g_mutex_clear (&set->lock);
788 g_slice_free (GstPoll, set);
792 * gst_poll_get_read_gpollfd:
796 * Get a GPollFD for the reading part of the control socket. This is useful when
797 * integrating with a GSource and GMainLoop.
800 gst_poll_get_read_gpollfd (GstPoll * set, GPollFD * fd)
802 g_return_if_fail (set != NULL);
803 g_return_if_fail (fd != NULL);
806 fd->fd = set->control_read_fd.fd;
808 #if GLIB_SIZEOF_VOID_P == 8
809 fd->fd = (gint64) set->wakeup_event;
811 fd->fd = (gint) set->wakeup_event;
814 fd->events = G_IO_IN | G_IO_HUP | G_IO_ERR;
822 * Initializes @fd. Alternatively you can initialize it with
826 gst_poll_fd_init (GstPollFD * fd)
828 g_return_if_fail (fd != NULL);
835 gst_poll_add_fd_unlocked (GstPoll * set, GstPollFD * fd)
839 GST_DEBUG ("%p: fd (fd:%d, idx:%d)", set, fd->fd, fd->idx);
841 idx = find_index (set->fds, fd);
847 nfd.events = POLLERR | POLLNVAL | POLLHUP;
850 g_array_append_val (set->fds, nfd);
852 fd->idx = set->fds->len - 1;
858 wfd.event_mask = FD_CLOSE;
859 memset (&wfd.events, 0, sizeof (wfd.events));
860 wfd.ignored_event_mask = 0;
861 event = WSACreateEvent ();
863 g_array_append_val (set->fds, wfd);
864 g_array_append_val (set->events, event);
866 fd->idx = set->fds->len - 1;
870 GST_WARNING ("%p: fd already added !", set);
878 * @set: a file descriptor set.
879 * @fd: a file descriptor.
881 * Add a file descriptor to the file descriptor set.
883 * Returns: %TRUE if the file descriptor was successfully added to the set.
886 gst_poll_add_fd (GstPoll * set, GstPollFD * fd)
890 g_return_val_if_fail (set != NULL, FALSE);
891 g_return_val_if_fail (fd != NULL, FALSE);
892 g_return_val_if_fail (fd->fd >= 0, FALSE);
894 g_mutex_lock (&set->lock);
896 ret = gst_poll_add_fd_unlocked (set, fd);
898 g_mutex_unlock (&set->lock);
904 * gst_poll_remove_fd:
905 * @set: a file descriptor set.
906 * @fd: a file descriptor.
908 * Remove a file descriptor from the file descriptor set.
910 * Returns: %TRUE if the file descriptor was successfully removed from the set.
913 gst_poll_remove_fd (GstPoll * set, GstPollFD * fd)
917 g_return_val_if_fail (set != NULL, FALSE);
918 g_return_val_if_fail (fd != NULL, FALSE);
919 g_return_val_if_fail (fd->fd >= 0, FALSE);
922 GST_DEBUG ("%p: fd (fd:%d, idx:%d)", set, fd->fd, fd->idx);
924 g_mutex_lock (&set->lock);
926 /* get the index, -1 is an fd that is not added */
927 idx = find_index (set->fds, fd);
930 gst_poll_free_winsock_event (set, idx);
931 g_array_remove_index_fast (set->events, idx);
934 /* remove the fd at index, we use _remove_index_fast, which copies the last
935 * element of the array to the freed index */
936 g_array_remove_index_fast (set->fds, idx);
938 /* mark fd as removed by setting the index to -1 */
942 GST_WARNING ("%p: couldn't find fd !", set);
945 g_mutex_unlock (&set->lock);
951 * gst_poll_fd_ctl_write:
952 * @set: a file descriptor set.
953 * @fd: a file descriptor.
954 * @active: a new status.
956 * Control whether the descriptor @fd in @set will be monitored for
959 * Returns: %TRUE if the descriptor was successfully updated.
962 gst_poll_fd_ctl_write (GstPoll * set, GstPollFD * fd, gboolean active)
966 g_return_val_if_fail (set != NULL, FALSE);
967 g_return_val_if_fail (fd != NULL, FALSE);
968 g_return_val_if_fail (fd->fd >= 0, FALSE);
970 GST_DEBUG ("%p: fd (fd:%d, idx:%d), active : %d", set,
971 fd->fd, fd->idx, active);
973 g_mutex_lock (&set->lock);
975 idx = find_index (set->fds, fd);
978 struct pollfd *pfd = &g_array_index (set->fds, struct pollfd, idx);
981 pfd->events |= POLLOUT;
983 pfd->events &= ~POLLOUT;
985 GST_LOG ("%p: pfd->events now %d (POLLOUT:%d)", set, pfd->events, POLLOUT);
987 gst_poll_update_winsock_event_mask (set, idx, FD_WRITE | FD_CONNECT,
992 GST_WARNING ("%p: couldn't find fd !", set);
995 g_mutex_unlock (&set->lock);
1001 gst_poll_fd_ctl_read_unlocked (GstPoll * set, GstPollFD * fd, gboolean active)
1005 GST_DEBUG ("%p: fd (fd:%d, idx:%d), active : %d", set,
1006 fd->fd, fd->idx, active);
1008 idx = find_index (set->fds, fd);
1012 struct pollfd *pfd = &g_array_index (set->fds, struct pollfd, idx);
1015 pfd->events |= (POLLIN | POLLPRI);
1017 pfd->events &= ~(POLLIN | POLLPRI);
1019 gst_poll_update_winsock_event_mask (set, idx, FD_READ | FD_ACCEPT, active);
1023 GST_WARNING ("%p: couldn't find fd !", set);
1030 * gst_poll_fd_ctl_read:
1031 * @set: a file descriptor set.
1032 * @fd: a file descriptor.
1033 * @active: a new status.
1035 * Control whether the descriptor @fd in @set will be monitored for
1038 * Returns: %TRUE if the descriptor was successfully updated.
1041 gst_poll_fd_ctl_read (GstPoll * set, GstPollFD * fd, gboolean active)
1045 g_return_val_if_fail (set != NULL, FALSE);
1046 g_return_val_if_fail (fd != NULL, FALSE);
1047 g_return_val_if_fail (fd->fd >= 0, FALSE);
1049 g_mutex_lock (&set->lock);
1051 ret = gst_poll_fd_ctl_read_unlocked (set, fd, active);
1053 g_mutex_unlock (&set->lock);
1059 * gst_poll_fd_ignored:
1060 * @set: a file descriptor set.
1061 * @fd: a file descriptor.
1063 * Mark @fd as ignored so that the next call to gst_poll_wait() will yield
1064 * the same result for @fd as last time. This function must be called if no
1065 * operation (read/write/recv/send/etc.) will be performed on @fd before
1066 * the next call to gst_poll_wait().
1068 * The reason why this is needed is because the underlying implementation
1069 * might not allow querying the fd more than once between calls to one of
1070 * the re-enabling operations.
1073 gst_poll_fd_ignored (GstPoll * set, GstPollFD * fd)
1078 g_return_if_fail (set != NULL);
1079 g_return_if_fail (fd != NULL);
1080 g_return_if_fail (fd->fd >= 0);
1082 g_mutex_lock (&set->lock);
1084 idx = find_index (set->fds, fd);
1086 WinsockFd *wfd = &g_array_index (set->fds, WinsockFd, idx);
1088 wfd->ignored_event_mask = wfd->event_mask & (FD_READ | FD_WRITE);
1092 g_mutex_unlock (&set->lock);
1097 * gst_poll_fd_has_closed:
1098 * @set: a file descriptor set.
1099 * @fd: a file descriptor.
1101 * Check if @fd in @set has closed the connection.
1103 * Returns: %TRUE if the connection was closed.
1106 gst_poll_fd_has_closed (const GstPoll * set, GstPollFD * fd)
1108 gboolean res = FALSE;
1111 g_return_val_if_fail (set != NULL, FALSE);
1112 g_return_val_if_fail (fd != NULL, FALSE);
1113 g_return_val_if_fail (fd->fd >= 0, FALSE);
1115 g_mutex_lock (&((GstPoll *) set)->lock);
1117 idx = find_index (set->active_fds, fd);
1120 struct pollfd *pfd = &g_array_index (set->active_fds, struct pollfd, idx);
1122 res = (pfd->revents & POLLHUP) != 0;
1124 WinsockFd *wfd = &g_array_index (set->active_fds, WinsockFd, idx);
1126 res = (wfd->events.lNetworkEvents & FD_CLOSE) != 0;
1129 GST_WARNING ("%p: couldn't find fd !", set);
1131 g_mutex_unlock (&((GstPoll *) set)->lock);
1133 GST_DEBUG ("%p: fd (fd:%d, idx:%d) %d", set, fd->fd, fd->idx, res);
1139 * gst_poll_fd_has_error:
1140 * @set: a file descriptor set.
1141 * @fd: a file descriptor.
1143 * Check if @fd in @set has an error.
1145 * Returns: %TRUE if the descriptor has an error.
1148 gst_poll_fd_has_error (const GstPoll * set, GstPollFD * fd)
1150 gboolean res = FALSE;
1153 g_return_val_if_fail (set != NULL, FALSE);
1154 g_return_val_if_fail (fd != NULL, FALSE);
1155 g_return_val_if_fail (fd->fd >= 0, FALSE);
1157 g_mutex_lock (&((GstPoll *) set)->lock);
1159 idx = find_index (set->active_fds, fd);
1162 struct pollfd *pfd = &g_array_index (set->active_fds, struct pollfd, idx);
1164 res = (pfd->revents & (POLLERR | POLLNVAL)) != 0;
1166 WinsockFd *wfd = &g_array_index (set->active_fds, WinsockFd, idx);
1168 res = (wfd->events.iErrorCode[FD_CLOSE_BIT] != 0) ||
1169 (wfd->events.iErrorCode[FD_READ_BIT] != 0) ||
1170 (wfd->events.iErrorCode[FD_WRITE_BIT] != 0) ||
1171 (wfd->events.iErrorCode[FD_ACCEPT_BIT] != 0) ||
1172 (wfd->events.iErrorCode[FD_CONNECT_BIT] != 0);
1175 GST_WARNING ("%p: couldn't find fd !", set);
1177 g_mutex_unlock (&((GstPoll *) set)->lock);
1179 GST_DEBUG ("%p: fd (fd:%d, idx:%d) %d", set, fd->fd, fd->idx, res);
1185 gst_poll_fd_can_read_unlocked (const GstPoll * set, GstPollFD * fd)
1187 gboolean res = FALSE;
1190 idx = find_index (set->active_fds, fd);
1193 struct pollfd *pfd = &g_array_index (set->active_fds, struct pollfd, idx);
1195 res = (pfd->revents & (POLLIN | POLLPRI)) != 0;
1197 WinsockFd *wfd = &g_array_index (set->active_fds, WinsockFd, idx);
1199 res = (wfd->events.lNetworkEvents & (FD_READ | FD_ACCEPT)) != 0;
1202 GST_WARNING ("%p: couldn't find fd !", set);
1204 GST_DEBUG ("%p: fd (fd:%d, idx:%d) %d", set, fd->fd, fd->idx, res);
1210 * gst_poll_fd_can_read:
1211 * @set: a file descriptor set.
1212 * @fd: a file descriptor.
1214 * Check if @fd in @set has data to be read.
1216 * Returns: %TRUE if the descriptor has data to be read.
1219 gst_poll_fd_can_read (const GstPoll * set, GstPollFD * fd)
1221 gboolean res = FALSE;
1223 g_return_val_if_fail (set != NULL, FALSE);
1224 g_return_val_if_fail (fd != NULL, FALSE);
1225 g_return_val_if_fail (fd->fd >= 0, FALSE);
1227 g_mutex_lock (&((GstPoll *) set)->lock);
1229 res = gst_poll_fd_can_read_unlocked (set, fd);
1231 g_mutex_unlock (&((GstPoll *) set)->lock);
1237 * gst_poll_fd_can_write:
1238 * @set: a file descriptor set.
1239 * @fd: a file descriptor.
1241 * Check if @fd in @set can be used for writing.
1243 * Returns: %TRUE if the descriptor can be used for writing.
1246 gst_poll_fd_can_write (const GstPoll * set, GstPollFD * fd)
1248 gboolean res = FALSE;
1251 g_return_val_if_fail (set != NULL, FALSE);
1252 g_return_val_if_fail (fd != NULL, FALSE);
1253 g_return_val_if_fail (fd->fd >= 0, FALSE);
1255 g_mutex_lock (&((GstPoll *) set)->lock);
1257 idx = find_index (set->active_fds, fd);
1260 struct pollfd *pfd = &g_array_index (set->active_fds, struct pollfd, idx);
1262 res = (pfd->revents & POLLOUT) != 0;
1264 WinsockFd *wfd = &g_array_index (set->active_fds, WinsockFd, idx);
1266 res = (wfd->events.lNetworkEvents & FD_WRITE) != 0;
1269 GST_WARNING ("%p: couldn't find fd !", set);
1271 g_mutex_unlock (&((GstPoll *) set)->lock);
1273 GST_DEBUG ("%p: fd (fd:%d, idx:%d) %d", set, fd->fd, fd->idx, res);
1281 * @timeout: a timeout in nanoseconds.
1283 * Wait for activity on the file descriptors in @set. This function waits up to
1284 * the specified @timeout. A timeout of #GST_CLOCK_TIME_NONE waits forever.
1286 * For #GstPoll objects created with gst_poll_new(), this function can only be
1287 * called from a single thread at a time. If called from multiple threads,
1288 * -1 will be returned with errno set to EPERM.
1290 * This is not true for timer #GstPoll objects created with
1291 * gst_poll_new_timer(), where it is allowed to have multiple threads waiting
1294 * Returns: The number of #GstPollFD in @set that have activity or 0 when no
1295 * activity was detected after @timeout. If an error occurs, -1 is returned
1299 gst_poll_wait (GstPoll * set, GstClockTime timeout)
1301 gboolean restarting;
1306 g_return_val_if_fail (set != NULL, -1);
1308 GST_DEBUG ("%p: timeout :%" GST_TIME_FORMAT, set, GST_TIME_ARGS (timeout));
1310 is_timer = set->timer;
1312 /* add one more waiter */
1313 old_waiting = INC_WAITING (set);
1315 /* we cannot wait from multiple threads unless we are a timer */
1316 if (G_UNLIKELY (old_waiting > 0 && !is_timer))
1317 goto already_waiting;
1319 /* flushing, exit immediately */
1320 if (G_UNLIKELY (IS_FLUSHING (set)))
1329 mode = choose_mode (set, timeout);
1331 if (TEST_REBUILD (set)) {
1332 g_mutex_lock (&set->lock);
1334 g_array_set_size (set->active_fds, set->fds->len);
1335 memcpy (set->active_fds->data, set->fds->data,
1336 set->fds->len * sizeof (struct pollfd));
1338 if (!gst_poll_prepare_winsock_active_sets (set))
1341 g_mutex_unlock (&set->lock);
1345 case GST_POLL_MODE_AUTO:
1346 g_assert_not_reached ();
1348 case GST_POLL_MODE_PPOLL:
1352 struct timespec *tsptr;
1354 if (timeout != GST_CLOCK_TIME_NONE) {
1355 GST_TIME_TO_TIMESPEC (timeout, ts);
1362 ppoll ((struct pollfd *) set->active_fds->data,
1363 set->active_fds->len, tsptr, NULL);
1365 g_assert_not_reached ();
1370 case GST_POLL_MODE_POLL:
1375 if (timeout != GST_CLOCK_TIME_NONE) {
1376 t = GST_TIME_AS_MSECONDS (timeout);
1382 poll ((struct pollfd *) set->active_fds->data,
1383 set->active_fds->len, t);
1385 g_assert_not_reached ();
1390 case GST_POLL_MODE_PSELECT:
1391 #ifndef HAVE_PSELECT
1393 g_assert_not_reached ();
1398 case GST_POLL_MODE_SELECT:
1406 max_fd = pollfd_to_fd_set (set, &readfds, &writefds, &errorfds);
1408 if (mode == GST_POLL_MODE_SELECT) {
1410 struct timeval *tvptr;
1412 if (timeout != GST_CLOCK_TIME_NONE) {
1413 GST_TIME_TO_TIMEVAL (timeout, tv);
1419 GST_DEBUG ("%p: Calling select", set);
1420 res = select (max_fd + 1, &readfds, &writefds, &errorfds, tvptr);
1421 GST_DEBUG ("%p: After select, res:%d", set, res);
1425 struct timespec *tsptr;
1427 if (timeout != GST_CLOCK_TIME_NONE) {
1428 GST_TIME_TO_TIMESPEC (timeout, ts);
1434 GST_DEBUG ("%p: Calling pselect", set);
1436 pselect (max_fd + 1, &readfds, &writefds, &errorfds, tsptr, NULL);
1437 GST_DEBUG ("%p: After pselect, res:%d", set, res);
1442 fd_set_to_pollfd (set, &readfds, &writefds, &errorfds);
1444 #else /* G_OS_WIN32 */
1445 g_assert_not_reached ();
1450 case GST_POLL_MODE_WINDOWS:
1453 gint ignore_count = set->active_fds_ignored->len;
1456 if (G_LIKELY (ignore_count == 0)) {
1457 if (timeout != GST_CLOCK_TIME_NONE)
1458 t = GST_TIME_AS_MSECONDS (timeout);
1462 /* already one or more ignored fds, so we quickly sweep the others */
1466 if (set->active_events->len != 0) {
1467 wait_ret = WSAWaitForMultipleEvents (set->active_events->len,
1468 (HANDLE *) set->active_events->data, FALSE, t, FALSE);
1470 wait_ret = WSA_WAIT_FAILED;
1471 WSASetLastError (WSA_INVALID_PARAMETER);
1474 if (ignore_count == 0 && wait_ret == WSA_WAIT_TIMEOUT) {
1476 } else if (wait_ret == WSA_WAIT_FAILED) {
1478 errno = gst_poll_winsock_error_to_errno (WSAGetLastError ());
1480 /* the first entry is the wakeup event */
1481 if (wait_ret - WSA_WAIT_EVENT_0 >= 1) {
1482 res = gst_poll_collect_winsock_events (set);
1484 res = 1; /* wakeup event */
1488 g_assert_not_reached ();
1496 /* Applications needs to clear the control socket themselves for timer
1498 * For other polls, we need to clear the control socket. If there was only
1499 * one socket with activity and it was the control socket, we need to
1501 if (release_all_wakeup (set) > 0 && res == 1)
1505 /* we got woken up and we are flushing, we need to stop */
1506 if (G_UNLIKELY (IS_FLUSHING (set)))
1509 } while (G_UNLIKELY (restarting));
1518 GST_LOG ("%p: we are already waiting", set);
1525 GST_LOG ("%p: we are flushing", set);
1533 GST_LOG ("%p: winsock error", set);
1534 g_mutex_unlock (&set->lock);
1542 * gst_poll_set_controllable:
1544 * @controllable: new controllable state.
1546 * When @controllable is %TRUE, this function ensures that future calls to
1547 * gst_poll_wait() will be affected by gst_poll_restart() and
1548 * gst_poll_set_flushing().
1550 * This function only works for non-timer #GstPoll objects created with
1553 * Returns: %TRUE if the controllability of @set could be updated.
1556 gst_poll_set_controllable (GstPoll * set, gboolean controllable)
1558 g_return_val_if_fail (set != NULL, FALSE);
1559 g_return_val_if_fail (!set->timer, FALSE);
1561 GST_LOG ("%p: controllable : %d", set, controllable);
1563 set->controllable = controllable;
1572 * Restart any gst_poll_wait() that is in progress. This function is typically
1573 * used after adding or removing descriptors to @set.
1575 * If @set is not controllable, then this call will have no effect.
1577 * This function only works for non-timer #GstPoll objects created with
1581 gst_poll_restart (GstPoll * set)
1583 g_return_if_fail (set != NULL);
1584 g_return_if_fail (!set->timer);
1586 if (set->controllable && GET_WAITING (set) > 0) {
1587 /* we are controllable and waiting, wake up the waiter. The socket will be
1588 * cleared by the _wait() thread and the poll will be restarted */
1594 * gst_poll_set_flushing:
1596 * @flushing: new flushing state.
1598 * When @flushing is %TRUE, this function ensures that current and future calls
1599 * to gst_poll_wait() will return -1, with errno set to EBUSY.
1601 * Unsetting the flushing state will restore normal operation of @set.
1603 * This function only works for non-timer #GstPoll objects created with
1607 gst_poll_set_flushing (GstPoll * set, gboolean flushing)
1609 g_return_if_fail (set != NULL);
1610 g_return_if_fail (!set->timer);
1612 GST_LOG ("%p: flushing: %d", set, flushing);
1614 /* update the new state first */
1615 SET_FLUSHING (set, flushing);
1617 if (flushing && set->controllable && GET_WAITING (set) > 0) {
1618 /* we are flushing, controllable and waiting, wake up the waiter. When we
1619 * stop the flushing operation we don't clear the wakeup fd here, this will
1620 * happen in the _wait() thread. */
1626 * gst_poll_write_control:
1629 * Write a byte to the control socket of the controllable @set.
1630 * This function is mostly useful for timer #GstPoll objects created with
1631 * gst_poll_new_timer().
1633 * It will make any current and future gst_poll_wait() function return with
1634 * 1, meaning the control socket is set. After an equal amount of calls to
1635 * gst_poll_read_control() have been performed, calls to gst_poll_wait() will
1636 * block again until their timeout expired.
1638 * This function only works for timer #GstPoll objects created with
1639 * gst_poll_new_timer().
1641 * Returns: %TRUE on success. %FALSE when @set is not controllable or when the
1642 * byte could not be written.
1645 gst_poll_write_control (GstPoll * set)
1649 g_return_val_if_fail (set != NULL, FALSE);
1650 g_return_val_if_fail (set->timer, FALSE);
1652 res = raise_wakeup (set);
1658 * gst_poll_read_control:
1661 * Read a byte from the control socket of the controllable @set.
1663 * This function only works for timer #GstPoll objects created with
1664 * gst_poll_new_timer().
1666 * Returns: %TRUE on success. %FALSE when @set is not controllable or when there
1667 * was no byte to read.
1670 gst_poll_read_control (GstPoll * set)
1674 g_return_val_if_fail (set != NULL, FALSE);
1675 g_return_val_if_fail (set->timer, FALSE);
1677 res = release_wakeup (set);