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
137 GstPollFD control_read_fd;
138 GstPollFD control_write_fd;
140 GArray *active_fds_ignored;
142 GArray *active_events;
147 gboolean controllable;
148 volatile gint waiting;
149 volatile gint control_pending;
150 volatile gint flushing;
152 volatile gint rebuild;
155 static gboolean gst_poll_fd_ctl_read_unlocked (GstPoll * set, GstPollFD * fd,
157 static gboolean gst_poll_add_fd_unlocked (GstPoll * set, GstPollFD * fd);
159 #define IS_FLUSHING(s) (g_atomic_int_get(&(s)->flushing))
160 #define SET_FLUSHING(s,val) (g_atomic_int_set(&(s)->flushing, (val)))
162 #define INC_WAITING(s) (g_atomic_int_add(&(s)->waiting, 1))
163 #define DEC_WAITING(s) (g_atomic_int_add(&(s)->waiting, -1))
164 #define GET_WAITING(s) (g_atomic_int_get(&(s)->waiting))
166 #define TEST_REBUILD(s) (g_atomic_int_compare_and_exchange(&(s)->rebuild, 1, 0))
167 #define MARK_REBUILD(s) (g_atomic_int_set(&(s)->rebuild, 1))
170 #define WAKE_EVENT(s) (write ((s)->control_write_fd.fd, "W", 1) == 1)
171 #define RELEASE_EVENT(s) (read ((s)->control_read_fd.fd, (s)->buf, 1) == 1)
173 #define WAKE_EVENT(s) (SetEvent ((s)->wakeup_event), errno = GetLastError () == NO_ERROR ? 0 : EACCES, errno == 0 ? 1 : 0)
174 #define RELEASE_EVENT(s) (ResetEvent ((s)->wakeup_event))
177 /* the poll/select call is also performed on a control socket, that way
178 * we can send special commands to control it */
179 static inline gboolean
180 raise_wakeup (GstPoll * set)
182 gboolean result = TRUE;
184 if (g_atomic_int_add (&set->control_pending, 1) == 0) {
185 /* raise when nothing pending */
186 GST_LOG ("%p: raise", set);
187 result = WAKE_EVENT (set);
192 /* note how bad things can happen when the 2 threads both raise and release the
193 * wakeup. This is however not a problem because you must always pair a raise
195 static inline gboolean
196 release_wakeup (GstPoll * set)
198 gboolean result = TRUE;
200 if (g_atomic_int_dec_and_test (&set->control_pending)) {
201 GST_LOG ("%p: release", set);
202 result = RELEASE_EVENT (set);
208 release_all_wakeup (GstPoll * set)
213 if (!(old = g_atomic_int_get (&set->control_pending)))
214 /* nothing pending, just exit */
217 /* try to remove all pending control messages */
218 if (g_atomic_int_compare_and_exchange (&set->control_pending, old, 0)) {
219 /* we managed to remove all messages, read the control socket */
220 if (RELEASE_EVENT (set))
223 /* retry again until we read it successfully */
224 g_atomic_int_add (&set->control_pending, 1);
231 find_index (GArray * array, GstPollFD * fd)
240 /* start by assuming the index found in the fd is still valid */
241 if (fd->idx >= 0 && fd->idx < array->len) {
243 ifd = &g_array_index (array, struct pollfd, fd->idx);
245 ifd = &g_array_index (array, WinsockFd, fd->idx);
248 if (ifd->fd == fd->fd) {
253 /* the pollfd array has changed and we need to lookup the fd again */
254 for (i = 0; i < array->len; i++) {
256 ifd = &g_array_index (array, struct pollfd, i);
258 ifd = &g_array_index (array, WinsockFd, i);
261 if (ifd->fd == fd->fd) {
271 #if !defined(HAVE_PPOLL) && defined(HAVE_POLL)
272 /* check if all file descriptors will fit in an fd_set */
274 selectable_fds (GstPoll * set)
278 g_mutex_lock (&set->lock);
279 for (i = 0; i < set->fds->len; i++) {
280 struct pollfd *pfd = &g_array_index (set->fds, struct pollfd, i);
282 if (pfd->fd >= FD_SETSIZE)
285 g_mutex_unlock (&set->lock);
291 g_mutex_unlock (&set->lock);
296 /* check if the timeout will convert to a timeout value used for poll()
297 * without a loss of precision
300 pollable_timeout (GstClockTime timeout)
302 if (timeout == GST_CLOCK_TIME_NONE)
305 /* not a nice multiple of milliseconds */
306 if (timeout % 1000000)
314 choose_mode (GstPoll * set, GstClockTime timeout)
318 if (set->mode == GST_POLL_MODE_AUTO) {
320 mode = GST_POLL_MODE_PPOLL;
321 #elif defined(HAVE_POLL)
322 if (!selectable_fds (set) || pollable_timeout (timeout)) {
323 mode = GST_POLL_MODE_POLL;
326 mode = GST_POLL_MODE_PSELECT;
328 mode = GST_POLL_MODE_SELECT;
331 #elif defined(HAVE_PSELECT)
332 mode = GST_POLL_MODE_PSELECT;
334 mode = GST_POLL_MODE_SELECT;
344 pollfd_to_fd_set (GstPoll * set, fd_set * readfds, fd_set * writefds,
354 g_mutex_lock (&set->lock);
356 for (i = 0; i < set->active_fds->len; i++) {
357 struct pollfd *pfd = &g_array_index (set->fds, struct pollfd, i);
359 if (pfd->fd < FD_SETSIZE) {
360 if (pfd->events & POLLIN)
361 FD_SET (pfd->fd, readfds);
362 if (pfd->events & POLLOUT)
363 FD_SET (pfd->fd, writefds);
365 FD_SET (pfd->fd, errorfds);
366 if (pfd->fd > max_fd && (pfd->events & (POLLIN | POLLOUT)))
371 g_mutex_unlock (&set->lock);
377 fd_set_to_pollfd (GstPoll * set, fd_set * readfds, fd_set * writefds,
382 g_mutex_lock (&set->lock);
384 for (i = 0; i < set->active_fds->len; i++) {
385 struct pollfd *pfd = &g_array_index (set->active_fds, struct pollfd, i);
387 if (pfd->fd < FD_SETSIZE) {
389 if (FD_ISSET (pfd->fd, readfds))
390 pfd->revents |= POLLIN;
391 if (FD_ISSET (pfd->fd, writefds))
392 pfd->revents |= POLLOUT;
393 if (FD_ISSET (pfd->fd, errorfds))
394 pfd->revents |= POLLERR;
398 g_mutex_unlock (&set->lock);
400 #else /* G_OS_WIN32 */
402 * Translate errors thrown by the Winsock API used by GstPoll:
403 * WSAEventSelect, WSAWaitForMultipleEvents and WSAEnumNetworkEvents
406 gst_poll_winsock_error_to_errno (DWORD last_error)
408 switch (last_error) {
409 case WSA_INVALID_HANDLE:
414 case WSA_NOT_ENOUGH_MEMORY:
418 * Anything else, including:
419 * WSA_INVALID_PARAMETER, WSAEFAULT, WSAEINPROGRESS, WSAENETDOWN,
428 gst_poll_free_winsock_event (GstPoll * set, gint idx)
430 WinsockFd *wfd = &g_array_index (set->fds, WinsockFd, idx);
431 HANDLE event = g_array_index (set->events, HANDLE, idx);
433 WSAEventSelect (wfd->fd, event, 0);
438 gst_poll_update_winsock_event_mask (GstPoll * set, gint idx, glong flags,
443 wfd = &g_array_index (set->fds, WinsockFd, idx);
446 wfd->event_mask |= flags;
448 wfd->event_mask &= ~flags;
450 /* reset ignored state if the new mask doesn't overlap at all */
451 if ((wfd->ignored_event_mask & wfd->event_mask) == 0)
452 wfd->ignored_event_mask = 0;
456 gst_poll_prepare_winsock_active_sets (GstPoll * set)
460 g_array_set_size (set->active_fds, 0);
461 g_array_set_size (set->active_fds_ignored, 0);
462 g_array_set_size (set->active_events, 0);
463 g_array_append_val (set->active_events, set->wakeup_event);
465 for (i = 0; i < set->fds->len; i++) {
466 WinsockFd *wfd = &g_array_index (set->fds, WinsockFd, i);
467 HANDLE event = g_array_index (set->events, HANDLE, i);
469 if (wfd->ignored_event_mask == 0) {
472 g_array_append_val (set->active_fds, *wfd);
473 g_array_append_val (set->active_events, event);
475 ret = WSAEventSelect (wfd->fd, event, wfd->event_mask);
476 if (G_UNLIKELY (ret != 0)) {
477 errno = gst_poll_winsock_error_to_errno (WSAGetLastError ());
481 g_array_append_val (set->active_fds_ignored, wfd);
489 gst_poll_collect_winsock_events (GstPoll * set)
494 * We need to check which events are signaled, and call
495 * WSAEnumNetworkEvents for those that are, which resets
496 * the event and clears the internal network event records.
499 for (i = 0; i < set->active_fds->len; i++) {
500 WinsockFd *wfd = &g_array_index (set->active_fds, WinsockFd, i);
501 HANDLE event = g_array_index (set->active_events, HANDLE, i + 1);
504 wait_ret = WaitForSingleObject (event, 0);
505 if (wait_ret == WAIT_OBJECT_0) {
506 gint enum_ret = WSAEnumNetworkEvents (wfd->fd, event, &wfd->events);
508 if (G_UNLIKELY (enum_ret != 0)) {
510 errno = gst_poll_winsock_error_to_errno (WSAGetLastError ());
516 /* clear any previously stored result */
517 memset (&wfd->events, 0, sizeof (wfd->events));
521 /* If all went well we also need to reset the ignored fds. */
523 res += set->active_fds_ignored->len;
525 for (i = 0; i < set->active_fds_ignored->len; i++) {
526 WinsockFd *wfd = g_array_index (set->active_fds_ignored, WinsockFd *, i);
528 wfd->ignored_event_mask = 0;
531 g_array_set_size (set->active_fds_ignored, 0);
539 * gst_poll_new: (skip)
540 * @controllable: whether it should be possible to control a wait.
542 * Create a new file descriptor set. If @controllable, it
543 * is possible to restart or flush a call to gst_poll_wait() with
544 * gst_poll_restart() and gst_poll_set_flushing() respectively.
546 * Free-function: gst_poll_free
548 * Returns: (transfer full): a new #GstPoll, or %NULL in case of an error.
549 * Free with gst_poll_free().
552 gst_poll_new (gboolean controllable)
556 GST_DEBUG ("controllable : %d", controllable);
558 nset = g_slice_new0 (GstPoll);
559 g_mutex_init (&nset->lock);
561 nset->mode = GST_POLL_MODE_AUTO;
562 nset->fds = g_array_new (FALSE, FALSE, sizeof (struct pollfd));
563 nset->active_fds = g_array_new (FALSE, FALSE, sizeof (struct pollfd));
564 nset->control_read_fd.fd = -1;
565 nset->control_write_fd.fd = -1;
567 gint control_sock[2];
569 if (socketpair (PF_UNIX, SOCK_STREAM, 0, control_sock) < 0)
572 fcntl (control_sock[0], F_SETFL, O_NONBLOCK);
573 fcntl (control_sock[1], F_SETFL, O_NONBLOCK);
575 nset->control_read_fd.fd = control_sock[0];
576 nset->control_write_fd.fd = control_sock[1];
578 gst_poll_add_fd_unlocked (nset, &nset->control_read_fd);
579 gst_poll_fd_ctl_read_unlocked (nset, &nset->control_read_fd, TRUE);
582 nset->mode = GST_POLL_MODE_WINDOWS;
583 nset->fds = g_array_new (FALSE, FALSE, sizeof (WinsockFd));
584 nset->active_fds = g_array_new (FALSE, FALSE, sizeof (WinsockFd));
585 nset->active_fds_ignored = g_array_new (FALSE, FALSE, sizeof (WinsockFd *));
586 nset->events = g_array_new (FALSE, FALSE, sizeof (HANDLE));
587 nset->active_events = g_array_new (FALSE, FALSE, sizeof (HANDLE));
589 nset->wakeup_event = CreateEvent (NULL, TRUE, FALSE, NULL);
592 /* ensure (re)build, though already sneakily set in non-windows case */
595 nset->controllable = controllable;
603 GST_WARNING ("%p: can't create socket pair !", nset);
604 gst_poll_free (nset);
611 * gst_poll_new_timer: (skip)
613 * Create a new poll object that can be used for scheduling cancellable
616 * A timeout is performed with gst_poll_wait(). Multiple timeouts can be
617 * performed from different threads.
619 * Free-function: gst_poll_free
621 * Returns: (transfer full): a new #GstPoll, or %NULL in case of an error.
622 * Free with gst_poll_free().
625 gst_poll_new_timer (void)
629 /* make a new controllable poll set */
630 if (!(poll = gst_poll_new (TRUE)))
642 * @set: (transfer full): a file descriptor set.
644 * Free a file descriptor set.
647 gst_poll_free (GstPoll * set)
649 g_return_if_fail (set != NULL);
651 GST_DEBUG ("%p: freeing", set);
654 if (set->control_write_fd.fd >= 0)
655 close (set->control_write_fd.fd);
656 if (set->control_read_fd.fd >= 0)
657 close (set->control_read_fd.fd);
659 CloseHandle (set->wakeup_event);
664 for (i = 0; i < set->events->len; i++)
665 gst_poll_free_winsock_event (set, i);
668 g_array_free (set->active_events, TRUE);
669 g_array_free (set->events, TRUE);
670 g_array_free (set->active_fds_ignored, TRUE);
673 g_array_free (set->active_fds, TRUE);
674 g_array_free (set->fds, TRUE);
675 g_mutex_clear (&set->lock);
676 g_slice_free (GstPoll, set);
680 * gst_poll_get_read_gpollfd:
684 * Get a GPollFD for the reading part of the control socket. This is useful when
685 * integrating with a GSource and GMainLoop.
688 gst_poll_get_read_gpollfd (GstPoll * set, GPollFD * fd)
690 g_return_if_fail (set != NULL);
691 g_return_if_fail (fd != NULL);
694 fd->fd = set->control_read_fd.fd;
696 #if GLIB_SIZEOF_VOID_P == 8
697 fd->fd = (gint64) set->wakeup_event;
699 fd->fd = (gint) set->wakeup_event;
702 fd->events = G_IO_IN | G_IO_HUP | G_IO_ERR;
710 * Initializes @fd. Alternatively you can initialize it with
714 gst_poll_fd_init (GstPollFD * fd)
716 g_return_if_fail (fd != NULL);
723 gst_poll_add_fd_unlocked (GstPoll * set, GstPollFD * fd)
727 GST_DEBUG ("%p: fd (fd:%d, idx:%d)", set, fd->fd, fd->idx);
729 idx = find_index (set->fds, fd);
735 nfd.events = POLLERR | POLLNVAL | POLLHUP;
738 g_array_append_val (set->fds, nfd);
740 fd->idx = set->fds->len - 1;
746 wfd.event_mask = FD_CLOSE;
747 memset (&wfd.events, 0, sizeof (wfd.events));
748 wfd.ignored_event_mask = 0;
749 event = WSACreateEvent ();
751 g_array_append_val (set->fds, wfd);
752 g_array_append_val (set->events, event);
754 fd->idx = set->fds->len - 1;
758 GST_WARNING ("%p: fd already added !", set);
766 * @set: a file descriptor set.
767 * @fd: a file descriptor.
769 * Add a file descriptor to the file descriptor set.
771 * Returns: %TRUE if the file descriptor was successfully added to the set.
774 gst_poll_add_fd (GstPoll * set, GstPollFD * fd)
778 g_return_val_if_fail (set != NULL, FALSE);
779 g_return_val_if_fail (fd != NULL, FALSE);
780 g_return_val_if_fail (fd->fd >= 0, FALSE);
782 g_mutex_lock (&set->lock);
784 ret = gst_poll_add_fd_unlocked (set, fd);
786 g_mutex_unlock (&set->lock);
792 * gst_poll_remove_fd:
793 * @set: a file descriptor set.
794 * @fd: a file descriptor.
796 * Remove a file descriptor from the file descriptor set.
798 * Returns: %TRUE if the file descriptor was successfully removed from the set.
801 gst_poll_remove_fd (GstPoll * set, GstPollFD * fd)
805 g_return_val_if_fail (set != NULL, FALSE);
806 g_return_val_if_fail (fd != NULL, FALSE);
807 g_return_val_if_fail (fd->fd >= 0, FALSE);
810 GST_DEBUG ("%p: fd (fd:%d, idx:%d)", set, fd->fd, fd->idx);
812 g_mutex_lock (&set->lock);
814 /* get the index, -1 is an fd that is not added */
815 idx = find_index (set->fds, fd);
818 gst_poll_free_winsock_event (set, idx);
819 g_array_remove_index_fast (set->events, idx);
822 /* remove the fd at index, we use _remove_index_fast, which copies the last
823 * element of the array to the freed index */
824 g_array_remove_index_fast (set->fds, idx);
826 /* mark fd as removed by setting the index to -1 */
830 GST_WARNING ("%p: couldn't find fd !", set);
833 g_mutex_unlock (&set->lock);
839 * gst_poll_fd_ctl_write:
840 * @set: a file descriptor set.
841 * @fd: a file descriptor.
842 * @active: a new status.
844 * Control whether the descriptor @fd in @set will be monitored for
847 * Returns: %TRUE if the descriptor was successfully updated.
850 gst_poll_fd_ctl_write (GstPoll * set, GstPollFD * fd, gboolean active)
854 g_return_val_if_fail (set != NULL, FALSE);
855 g_return_val_if_fail (fd != NULL, FALSE);
856 g_return_val_if_fail (fd->fd >= 0, FALSE);
858 GST_DEBUG ("%p: fd (fd:%d, idx:%d), active : %d", set,
859 fd->fd, fd->idx, active);
861 g_mutex_lock (&set->lock);
863 idx = find_index (set->fds, fd);
866 struct pollfd *pfd = &g_array_index (set->fds, struct pollfd, idx);
869 pfd->events |= POLLOUT;
871 pfd->events &= ~POLLOUT;
873 GST_LOG ("pfd->events now %d (POLLOUT:%d)", pfd->events, POLLOUT);
875 gst_poll_update_winsock_event_mask (set, idx, FD_WRITE | FD_CONNECT,
880 GST_WARNING ("%p: couldn't find fd !", set);
883 g_mutex_unlock (&set->lock);
889 gst_poll_fd_ctl_read_unlocked (GstPoll * set, GstPollFD * fd, gboolean active)
893 GST_DEBUG ("%p: fd (fd:%d, idx:%d), active : %d", set,
894 fd->fd, fd->idx, active);
896 idx = find_index (set->fds, fd);
900 struct pollfd *pfd = &g_array_index (set->fds, struct pollfd, idx);
903 pfd->events |= (POLLIN | POLLPRI);
905 pfd->events &= ~(POLLIN | POLLPRI);
907 gst_poll_update_winsock_event_mask (set, idx, FD_READ | FD_ACCEPT, active);
911 GST_WARNING ("%p: couldn't find fd !", set);
918 * gst_poll_fd_ctl_read:
919 * @set: a file descriptor set.
920 * @fd: a file descriptor.
921 * @active: a new status.
923 * Control whether the descriptor @fd in @set will be monitored for
926 * Returns: %TRUE if the descriptor was successfully updated.
929 gst_poll_fd_ctl_read (GstPoll * set, GstPollFD * fd, gboolean active)
933 g_return_val_if_fail (set != NULL, FALSE);
934 g_return_val_if_fail (fd != NULL, FALSE);
935 g_return_val_if_fail (fd->fd >= 0, FALSE);
937 g_mutex_lock (&set->lock);
939 ret = gst_poll_fd_ctl_read_unlocked (set, fd, active);
941 g_mutex_unlock (&set->lock);
947 * gst_poll_fd_ignored:
948 * @set: a file descriptor set.
949 * @fd: a file descriptor.
951 * Mark @fd as ignored so that the next call to gst_poll_wait() will yield
952 * the same result for @fd as last time. This function must be called if no
953 * operation (read/write/recv/send/etc.) will be performed on @fd before
954 * the next call to gst_poll_wait().
956 * The reason why this is needed is because the underlying implementation
957 * might not allow querying the fd more than once between calls to one of
958 * the re-enabling operations.
961 gst_poll_fd_ignored (GstPoll * set, GstPollFD * fd)
966 g_return_if_fail (set != NULL);
967 g_return_if_fail (fd != NULL);
968 g_return_if_fail (fd->fd >= 0);
970 g_mutex_lock (&set->lock);
972 idx = find_index (set->fds, fd);
974 WinsockFd *wfd = &g_array_index (set->fds, WinsockFd, idx);
976 wfd->ignored_event_mask = wfd->event_mask & (FD_READ | FD_WRITE);
980 g_mutex_unlock (&set->lock);
985 * gst_poll_fd_has_closed:
986 * @set: a file descriptor set.
987 * @fd: a file descriptor.
989 * Check if @fd in @set has closed the connection.
991 * Returns: %TRUE if the connection was closed.
994 gst_poll_fd_has_closed (const GstPoll * set, GstPollFD * fd)
996 gboolean res = FALSE;
999 g_return_val_if_fail (set != NULL, FALSE);
1000 g_return_val_if_fail (fd != NULL, FALSE);
1001 g_return_val_if_fail (fd->fd >= 0, FALSE);
1003 g_mutex_lock (&((GstPoll *) set)->lock);
1005 idx = find_index (set->active_fds, fd);
1008 struct pollfd *pfd = &g_array_index (set->active_fds, struct pollfd, idx);
1010 res = (pfd->revents & POLLHUP) != 0;
1012 WinsockFd *wfd = &g_array_index (set->active_fds, WinsockFd, idx);
1014 res = (wfd->events.lNetworkEvents & FD_CLOSE) != 0;
1017 GST_WARNING ("%p: couldn't find fd !", set);
1019 g_mutex_unlock (&((GstPoll *) set)->lock);
1021 GST_DEBUG ("%p: fd (fd:%d, idx:%d) %d", set, fd->fd, fd->idx, res);
1027 * gst_poll_fd_has_error:
1028 * @set: a file descriptor set.
1029 * @fd: a file descriptor.
1031 * Check if @fd in @set has an error.
1033 * Returns: %TRUE if the descriptor has an error.
1036 gst_poll_fd_has_error (const GstPoll * set, GstPollFD * fd)
1038 gboolean res = FALSE;
1041 g_return_val_if_fail (set != NULL, FALSE);
1042 g_return_val_if_fail (fd != NULL, FALSE);
1043 g_return_val_if_fail (fd->fd >= 0, FALSE);
1045 g_mutex_lock (&((GstPoll *) set)->lock);
1047 idx = find_index (set->active_fds, fd);
1050 struct pollfd *pfd = &g_array_index (set->active_fds, struct pollfd, idx);
1052 res = (pfd->revents & (POLLERR | POLLNVAL)) != 0;
1054 WinsockFd *wfd = &g_array_index (set->active_fds, WinsockFd, idx);
1056 res = (wfd->events.iErrorCode[FD_CLOSE_BIT] != 0) ||
1057 (wfd->events.iErrorCode[FD_READ_BIT] != 0) ||
1058 (wfd->events.iErrorCode[FD_WRITE_BIT] != 0) ||
1059 (wfd->events.iErrorCode[FD_ACCEPT_BIT] != 0) ||
1060 (wfd->events.iErrorCode[FD_CONNECT_BIT] != 0);
1063 GST_WARNING ("%p: couldn't find fd !", set);
1065 g_mutex_unlock (&((GstPoll *) set)->lock);
1067 GST_DEBUG ("%p: fd (fd:%d, idx:%d) %d", set, fd->fd, fd->idx, res);
1073 gst_poll_fd_can_read_unlocked (const GstPoll * set, GstPollFD * fd)
1075 gboolean res = FALSE;
1078 idx = find_index (set->active_fds, fd);
1081 struct pollfd *pfd = &g_array_index (set->active_fds, struct pollfd, idx);
1083 res = (pfd->revents & (POLLIN | POLLPRI)) != 0;
1085 WinsockFd *wfd = &g_array_index (set->active_fds, WinsockFd, idx);
1087 res = (wfd->events.lNetworkEvents & (FD_READ | FD_ACCEPT)) != 0;
1090 GST_WARNING ("%p: couldn't find fd !", set);
1092 GST_DEBUG ("%p: fd (fd:%d, idx:%d) %d", set, fd->fd, fd->idx, res);
1098 * gst_poll_fd_can_read:
1099 * @set: a file descriptor set.
1100 * @fd: a file descriptor.
1102 * Check if @fd in @set has data to be read.
1104 * Returns: %TRUE if the descriptor has data to be read.
1107 gst_poll_fd_can_read (const GstPoll * set, GstPollFD * fd)
1109 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 res = gst_poll_fd_can_read_unlocked (set, fd);
1119 g_mutex_unlock (&((GstPoll *) set)->lock);
1125 * gst_poll_fd_can_write:
1126 * @set: a file descriptor set.
1127 * @fd: a file descriptor.
1129 * Check if @fd in @set can be used for writing.
1131 * Returns: %TRUE if the descriptor can be used for writing.
1134 gst_poll_fd_can_write (const GstPoll * set, GstPollFD * fd)
1136 gboolean res = FALSE;
1139 g_return_val_if_fail (set != NULL, FALSE);
1140 g_return_val_if_fail (fd != NULL, FALSE);
1141 g_return_val_if_fail (fd->fd >= 0, FALSE);
1143 g_mutex_lock (&((GstPoll *) set)->lock);
1145 idx = find_index (set->active_fds, fd);
1148 struct pollfd *pfd = &g_array_index (set->active_fds, struct pollfd, idx);
1150 res = (pfd->revents & POLLOUT) != 0;
1152 WinsockFd *wfd = &g_array_index (set->active_fds, WinsockFd, idx);
1154 res = (wfd->events.lNetworkEvents & FD_WRITE) != 0;
1157 GST_WARNING ("%p: couldn't find fd !", set);
1159 g_mutex_unlock (&((GstPoll *) set)->lock);
1161 GST_DEBUG ("%p: fd (fd:%d, idx:%d) %d", set, fd->fd, fd->idx, res);
1169 * @timeout: a timeout in nanoseconds.
1171 * Wait for activity on the file descriptors in @set. This function waits up to
1172 * the specified @timeout. A timeout of #GST_CLOCK_TIME_NONE waits forever.
1174 * For #GstPoll objects created with gst_poll_new(), this function can only be
1175 * called from a single thread at a time. If called from multiple threads,
1176 * -1 will be returned with errno set to EPERM.
1178 * This is not true for timer #GstPoll objects created with
1179 * gst_poll_new_timer(), where it is allowed to have multiple threads waiting
1182 * Returns: The number of #GstPollFD in @set that have activity or 0 when no
1183 * activity was detected after @timeout. If an error occurs, -1 is returned
1187 gst_poll_wait (GstPoll * set, GstClockTime timeout)
1189 gboolean restarting;
1194 g_return_val_if_fail (set != NULL, -1);
1196 GST_DEBUG ("timeout :%" GST_TIME_FORMAT, GST_TIME_ARGS (timeout));
1198 is_timer = set->timer;
1200 /* add one more waiter */
1201 old_waiting = INC_WAITING (set);
1203 /* we cannot wait from multiple threads unless we are a timer */
1204 if (G_UNLIKELY (old_waiting > 0 && !is_timer))
1205 goto already_waiting;
1207 /* flushing, exit immediately */
1208 if (G_UNLIKELY (IS_FLUSHING (set)))
1217 mode = choose_mode (set, timeout);
1219 if (TEST_REBUILD (set)) {
1220 g_mutex_lock (&set->lock);
1222 g_array_set_size (set->active_fds, set->fds->len);
1223 memcpy (set->active_fds->data, set->fds->data,
1224 set->fds->len * sizeof (struct pollfd));
1226 if (!gst_poll_prepare_winsock_active_sets (set))
1229 g_mutex_unlock (&set->lock);
1233 case GST_POLL_MODE_AUTO:
1234 g_assert_not_reached ();
1236 case GST_POLL_MODE_PPOLL:
1240 struct timespec *tsptr;
1242 if (timeout != GST_CLOCK_TIME_NONE) {
1243 GST_TIME_TO_TIMESPEC (timeout, ts);
1250 ppoll ((struct pollfd *) set->active_fds->data,
1251 set->active_fds->len, tsptr, NULL);
1253 g_assert_not_reached ();
1258 case GST_POLL_MODE_POLL:
1263 if (timeout != GST_CLOCK_TIME_NONE) {
1264 t = GST_TIME_AS_MSECONDS (timeout);
1270 poll ((struct pollfd *) set->active_fds->data,
1271 set->active_fds->len, t);
1273 g_assert_not_reached ();
1278 case GST_POLL_MODE_PSELECT:
1279 #ifndef HAVE_PSELECT
1281 g_assert_not_reached ();
1286 case GST_POLL_MODE_SELECT:
1294 max_fd = pollfd_to_fd_set (set, &readfds, &writefds, &errorfds);
1296 if (mode == GST_POLL_MODE_SELECT) {
1298 struct timeval *tvptr;
1300 if (timeout != GST_CLOCK_TIME_NONE) {
1301 GST_TIME_TO_TIMEVAL (timeout, tv);
1307 GST_DEBUG ("Calling select");
1308 res = select (max_fd + 1, &readfds, &writefds, &errorfds, tvptr);
1309 GST_DEBUG ("After select, res:%d", res);
1313 struct timespec *tsptr;
1315 if (timeout != GST_CLOCK_TIME_NONE) {
1316 GST_TIME_TO_TIMESPEC (timeout, ts);
1322 GST_DEBUG ("Calling pselect");
1324 pselect (max_fd + 1, &readfds, &writefds, &errorfds, tsptr, NULL);
1325 GST_DEBUG ("After pselect, res:%d", res);
1330 fd_set_to_pollfd (set, &readfds, &writefds, &errorfds);
1332 #else /* G_OS_WIN32 */
1333 g_assert_not_reached ();
1338 case GST_POLL_MODE_WINDOWS:
1341 gint ignore_count = set->active_fds_ignored->len;
1344 if (G_LIKELY (ignore_count == 0)) {
1345 if (timeout != GST_CLOCK_TIME_NONE)
1346 t = GST_TIME_AS_MSECONDS (timeout);
1350 /* already one or more ignored fds, so we quickly sweep the others */
1354 if (set->active_events->len != 0) {
1355 wait_ret = WSAWaitForMultipleEvents (set->active_events->len,
1356 (HANDLE *) set->active_events->data, FALSE, t, FALSE);
1358 wait_ret = WSA_WAIT_FAILED;
1359 WSASetLastError (WSA_INVALID_PARAMETER);
1362 if (ignore_count == 0 && wait_ret == WSA_WAIT_TIMEOUT) {
1364 } else if (wait_ret == WSA_WAIT_FAILED) {
1366 errno = gst_poll_winsock_error_to_errno (WSAGetLastError ());
1368 /* the first entry is the wakeup event */
1369 if (wait_ret - WSA_WAIT_EVENT_0 >= 1) {
1370 res = gst_poll_collect_winsock_events (set);
1372 res = 1; /* wakeup event */
1376 g_assert_not_reached ();
1384 /* Applications needs to clear the control socket themselves for timer
1386 * For other polls, we need to clear the control socket. If there was only
1387 * one socket with activity and it was the control socket, we need to
1389 if (release_all_wakeup (set) > 0 && res == 1)
1393 /* we got woken up and we are flushing, we need to stop */
1394 if (G_UNLIKELY (IS_FLUSHING (set)))
1397 } while (G_UNLIKELY (restarting));
1406 GST_LOG ("%p: we are already waiting", set);
1413 GST_LOG ("%p: we are flushing", set);
1421 GST_LOG ("%p: winsock error", set);
1422 g_mutex_unlock (&set->lock);
1430 * gst_poll_set_controllable:
1432 * @controllable: new controllable state.
1434 * When @controllable is %TRUE, this function ensures that future calls to
1435 * gst_poll_wait() will be affected by gst_poll_restart() and
1436 * gst_poll_set_flushing().
1438 * Returns: %TRUE if the controllability of @set could be updated.
1441 gst_poll_set_controllable (GstPoll * set, gboolean controllable)
1443 g_return_val_if_fail (set != NULL, FALSE);
1445 GST_LOG ("%p: controllable : %d", set, controllable);
1447 set->controllable = controllable;
1456 * Restart any gst_poll_wait() that is in progress. This function is typically
1457 * used after adding or removing descriptors to @set.
1459 * If @set is not controllable, then this call will have no effect.
1462 gst_poll_restart (GstPoll * set)
1464 g_return_if_fail (set != NULL);
1466 if (set->controllable && GET_WAITING (set) > 0) {
1467 /* we are controllable and waiting, wake up the waiter. The socket will be
1468 * cleared by the _wait() thread and the poll will be restarted */
1474 * gst_poll_set_flushing:
1476 * @flushing: new flushing state.
1478 * When @flushing is %TRUE, this function ensures that current and future calls
1479 * to gst_poll_wait() will return -1, with errno set to EBUSY.
1481 * Unsetting the flushing state will restore normal operation of @set.
1484 gst_poll_set_flushing (GstPoll * set, gboolean flushing)
1486 g_return_if_fail (set != NULL);
1488 GST_LOG ("%p: flushing: %d", set, flushing);
1490 /* update the new state first */
1491 SET_FLUSHING (set, flushing);
1493 if (flushing && set->controllable && GET_WAITING (set) > 0) {
1494 /* we are flushing, controllable and waiting, wake up the waiter. When we
1495 * stop the flushing operation we don't clear the wakeup fd here, this will
1496 * happen in the _wait() thread. */
1502 * gst_poll_write_control:
1505 * Write a byte to the control socket of the controllable @set.
1506 * This function is mostly useful for timer #GstPoll objects created with
1507 * gst_poll_new_timer().
1509 * It will make any current and future gst_poll_wait() function return with
1510 * 1, meaning the control socket is set. After an equal amount of calls to
1511 * gst_poll_read_control() have been performed, calls to gst_poll_wait() will
1512 * block again until their timeout expired.
1514 * Returns: %TRUE on success. %FALSE when @set is not controllable or when the
1515 * byte could not be written.
1518 gst_poll_write_control (GstPoll * set)
1522 g_return_val_if_fail (set != NULL, FALSE);
1523 g_return_val_if_fail (set->timer, FALSE);
1525 res = raise_wakeup (set);
1531 * gst_poll_read_control:
1534 * Read a byte from the control socket of the controllable @set.
1535 * This function is mostly useful for timer #GstPoll objects created with
1536 * gst_poll_new_timer().
1538 * Returns: %TRUE on success. %FALSE when @set is not controllable or when there
1539 * was no byte to read.
1542 gst_poll_read_control (GstPoll * set)
1546 g_return_val_if_fail (set != NULL, FALSE);
1547 g_return_val_if_fail (set->timer, FALSE);
1549 res = release_wakeup (set);