* Copyright (C) 1999 Erik Walthinsen <omega@cse.ogi.edu>
* Copyright (C) 2004 Wim Taymans <wim.taymans@gmail.com>
* Copyright (C) 2007 Peter Kjellerstedt <pkj@axis.com>
- * Copyright (C) 2008 Ole Andr
é Vadla Ravnås <ole.andre.ravnas@tandberg.com>
+ * Copyright (C) 2008 Ole Andr
é Vadla Ravnås <ole.andre.ravnas@tandberg.com>
*
* gstpoll.c: File descriptor set
*
*/
/**
* SECTION:gstpoll
+ * @title: GstPoll
* @short_description: Keep track of file descriptors and make it possible
* to wait on them in a cancellable way
*
* New file descriptors are added to the set using gst_poll_add_fd(), and
* removed using gst_poll_remove_fd(). Controlling which file descriptors
* should be waited for to become readable and/or writable are done using
- * gst_poll_fd_ctl_read() and gst_poll_fd_ctl_write().
+ * gst_poll_fd_ctl_read(), gst_poll_fd_ctl_write() and gst_poll_fd_ctl_pri().
*
* Use gst_poll_wait() to wait for the file descriptors to actually become
* readable and/or writable, or to timeout if no file descriptor is available
#include <sys/socket.h>
#endif
+#ifdef G_OS_WIN32
+# ifndef EWOULDBLOCK
+# define EWOULDBLOCK EAGAIN /* This is just to placate gcc */
+# endif
+#endif /* G_OS_WIN32 */
+
/* OS/X needs this because of bad headers */
#include <string.h>
GArray *active_fds;
#ifndef G_OS_WIN32
- gchar buf[1];
GstPollFD control_read_fd;
GstPollFD control_write_fd;
#else
#define MARK_REBUILD(s) (g_atomic_int_set(&(s)->rebuild, 1))
#ifndef G_OS_WIN32
-#define WAKE_EVENT(s) (write ((s)->control_write_fd.fd, "W", 1) == 1)
-#define RELEASE_EVENT(s) (read ((s)->control_read_fd.fd, (s)->buf, 1) == 1)
+
+static gboolean
+wake_event (GstPoll * set)
+{
+ ssize_t num_written;
+ while ((num_written = write (set->control_write_fd.fd, "W", 1)) != 1) {
+ if (num_written == -1 && errno != EAGAIN && errno != EINTR) {
+ g_critical ("%p: failed to wake event: %s", set, strerror (errno));
+ return FALSE;
+ }
+ }
+ return TRUE;
+}
+
+static gboolean
+release_event (GstPoll * set)
+{
+ gchar buf[1] = { '\0' };
+ ssize_t num_read;
+ while ((num_read = read (set->control_read_fd.fd, buf, 1)) != 1) {
+ if (num_read == -1 && errno != EAGAIN && errno != EINTR) {
+ g_critical ("%p: failed to release event: %s", set, strerror (errno));
+ return FALSE;
+ }
+ }
+ return TRUE;
+}
+
#else
-#define WAKE_EVENT(s) (SetEvent ((s)->wakeup_event), errno = GetLastError () == NO_ERROR ? 0 : EACCES, errno == 0 ? 1 : 0)
-#define RELEASE_EVENT(s) (ResetEvent ((s)->wakeup_event))
+
+static void
+format_last_error (gchar * buf, size_t buf_len)
+{
+ DWORD flags = FORMAT_MESSAGE_FROM_SYSTEM;
+ LPCVOID src = NULL;
+ DWORD lang = 0;
+ DWORD id;
+ id = GetLastError ();
+ FormatMessage (flags, src, id, lang, buf, (DWORD) buf_len, NULL);
+ SetLastError (id);
+}
+
+static gboolean
+wake_event (GstPoll * set)
+{
+ SetLastError (0);
+ errno = 0;
+ if (!SetEvent (set->wakeup_event)) {
+ gchar msg[1024] = "<unknown>";
+ format_last_error (msg, sizeof (msg));
+ g_critical ("%p: failed to set wakup_event: %s", set, msg);
+ errno = EBADF;
+ return FALSE;
+ }
+
+ return TRUE;
+}
+
+static gboolean
+release_event (GstPoll * set)
+{
+ DWORD status;
+ SetLastError (0);
+ errno = 0;
+
+ status = WaitForSingleObject (set->wakeup_event, INFINITE);
+ if (status) {
+ const gchar *reason = "unknown";
+ gchar msg[1024] = "<unknown>";
+ switch (status) {
+ case WAIT_ABANDONED:
+ reason = "WAIT_ABANDONED";
+ break;
+ case WAIT_TIMEOUT:
+ reason = "WAIT_TIMEOUT";
+ break;
+ case WAIT_FAILED:
+ format_last_error (msg, sizeof (msg));
+ reason = msg;
+ break;
+ default:
+ reason = "other";
+ break;
+ }
+ g_critical ("%p: failed to block on wakup_event: %s", set, reason);
+ errno = EBADF;
+ return FALSE;
+ }
+
+ if (!ResetEvent (set->wakeup_event)) {
+ gchar msg[1024] = "<unknown>";
+ format_last_error (msg, sizeof (msg));
+ g_critical ("%p: failed to reset wakup_event: %s", set, msg);
+ errno = EBADF;
+ return FALSE;
+ }
+
+ return TRUE;
+}
+
#endif
/* the poll/select call is also performed on a control socket, that way
{
gboolean result = TRUE;
- if (g_atomic_int_add (&set->control_pending, 1) == 0) {
+ /* makes testing control_pending and WAKE_EVENT() atomic. */
+ g_mutex_lock (&set->lock);
+
+ if (set->control_pending == 0) {
/* raise when nothing pending */
GST_LOG ("%p: raise", set);
- result = WAKE_EVENT (set);
+ result = wake_event (set);
+ }
+
+ if (result) {
+ set->control_pending++;
}
+
+ g_mutex_unlock (&set->lock);
+
return result;
}
-/* note how bad things can happen when the 2 threads both raise and release the
- * wakeup. This is however not a problem because you must always pair a raise
- * with a release */
static inline gboolean
release_wakeup (GstPoll * set)
{
- gboolean result = TRUE;
+ gboolean result = FALSE;
+
+ /* makes testing/modifying control_pending and RELEASE_EVENT() atomic. */
+ g_mutex_lock (&set->lock);
- if (g_atomic_int_dec_and_test (&set->control_pending)) {
- GST_LOG ("%p: release", set);
- result = RELEASE_EVENT (set);
+ if (set->control_pending > 0) {
+ /* release, only if this was the last pending. */
+ if (set->control_pending == 1) {
+ GST_LOG ("%p: release", set);
+ result = release_event (set);
+ } else {
+ result = TRUE;
+ }
+
+ if (result) {
+ set->control_pending--;
+ }
+ } else {
+ errno = EWOULDBLOCK;
}
+
+ g_mutex_unlock (&set->lock);
+
return result;
}
{
gint old;
- while (TRUE) {
- if (!(old = g_atomic_int_get (&set->control_pending)))
- /* nothing pending, just exit */
- break;
+ /* makes testing control_pending and RELEASE_EVENT() atomic. */
+ g_mutex_lock (&set->lock);
- /* try to remove all pending control messages */
- if (g_atomic_int_compare_and_exchange (&set->control_pending, old, 0)) {
- /* we managed to remove all messages, read the control socket */
- if (RELEASE_EVENT (set))
- break;
- else
- /* retry again until we read it successfully */
- g_atomic_int_add (&set->control_pending, 1);
+ if ((old = set->control_pending) > 0) {
+ GST_LOG ("%p: releasing %d", set, old);
+ if (release_event (set)) {
+ set->control_pending = 0;
+ } else {
+ old = 0;
}
}
+
+ g_mutex_unlock (&set->lock);
+
return old;
}
if (socketpair (PF_UNIX, SOCK_STREAM, 0, control_sock) < 0)
goto no_socket_pair;
- fcntl (control_sock[0], F_SETFL, O_NONBLOCK);
- fcntl (control_sock[1], F_SETFL, O_NONBLOCK);
-
nset->control_read_fd.fd = control_sock[0];
nset->control_write_fd.fd = control_sock[1];
MARK_REBUILD (nset);
nset->controllable = controllable;
+ nset->control_pending = 0;
return nset;
* timeouts.
*
* A timeout is performed with gst_poll_wait(). Multiple timeouts can be
- * performed from different threads.
+ * performed from different threads.
*
* Free-function: gst_poll_free
*
struct pollfd *pfd = &g_array_index (set->fds, struct pollfd, idx);
if (active)
- pfd->events |= (POLLIN | POLLPRI);
+ pfd->events |= POLLIN;
else
- pfd->events &= ~(POLLIN | POLLPRI);
+ pfd->events &= ~POLLIN;
#else
gst_poll_update_winsock_event_mask (set, idx, FD_READ | FD_ACCEPT, active);
#endif
}
/**
+ * gst_poll_fd_ctl_pri:
+ * @set: a file descriptor set.
+ * @fd: a file descriptor.
+ * @active: a new status.
+ *
+ * Control whether the descriptor @fd in @set will be monitored for
+ * exceptional conditions (POLLPRI).
+ *
+ * Not implemented on Windows (will just return %FALSE there).
+ *
+ * Returns: %TRUE if the descriptor was successfully updated.
+ *
+ * Since: 1.16
+ */
+gboolean
+gst_poll_fd_ctl_pri (GstPoll * set, GstPollFD * fd, gboolean active)
+{
+#ifdef G_OS_WIN32
+ return FALSE;
+#else
+ gint idx;
+
+ g_return_val_if_fail (set != NULL, FALSE);
+ g_return_val_if_fail (fd != NULL, FALSE);
+ g_return_val_if_fail (fd->fd >= 0, FALSE);
+
+ GST_DEBUG ("%p: fd (fd:%d, idx:%d), active : %d", set,
+ fd->fd, fd->idx, active);
+
+ g_mutex_lock (&set->lock);
+
+ idx = find_index (set->fds, fd);
+ if (idx >= 0) {
+ struct pollfd *pfd = &g_array_index (set->fds, struct pollfd, idx);
+
+ if (active)
+ pfd->events |= POLLPRI;
+ else
+ pfd->events &= ~POLLPRI;
+
+ GST_LOG ("%p: pfd->events now %d (POLLPRI:%d)", set, pfd->events, POLLOUT);
+ MARK_REBUILD (set);
+ } else {
+ GST_WARNING ("%p: couldn't find fd !", set);
+ }
+
+ g_mutex_unlock (&set->lock);
+
+ return idx >= 0;
+#endif
+}
+
+/**
* gst_poll_fd_ignored:
* @set: a file descriptor set.
* @fd: a file descriptor.
#ifndef G_OS_WIN32
struct pollfd *pfd = &g_array_index (set->active_fds, struct pollfd, idx);
- res = (pfd->revents & (POLLIN | POLLPRI)) != 0;
+ res = (pfd->revents & POLLIN) != 0;
#else
WinsockFd *wfd = &g_array_index (set->active_fds, WinsockFd, idx);
}
/**
+ * gst_poll_fd_has_pri:
+ * @set: a file descriptor set.
+ * @fd: a file descriptor.
+ *
+ * Check if @fd in @set has an exceptional condition (POLLPRI).
+ *
+ * Not implemented on Windows (will just return %FALSE there).
+ *
+ * Returns: %TRUE if the descriptor has an exceptional condition.
+ *
+ * Since: 1.16
+ */
+gboolean
+gst_poll_fd_has_pri (const GstPoll * set, GstPollFD * fd)
+{
+#ifdef G_OS_WIN32
+ return FALSE;
+#else
+ gboolean res = FALSE;
+ gint idx;
+
+ g_return_val_if_fail (set != NULL, FALSE);
+ g_return_val_if_fail (fd != NULL, FALSE);
+ g_return_val_if_fail (fd->fd >= 0, FALSE);
+
+ g_mutex_lock (&((GstPoll *) set)->lock);
+
+ idx = find_index (set->active_fds, fd);
+ if (idx >= 0) {
+ struct pollfd *pfd = &g_array_index (set->active_fds, struct pollfd, idx);
+
+ res = (pfd->revents & POLLPRI) != 0;
+ } else {
+ GST_WARNING ("%p: couldn't find fd !", set);
+ }
+ g_mutex_unlock (&((GstPoll *) set)->lock);
+
+ GST_DEBUG ("%p: fd (fd:%d, idx:%d) %d", set, fd->fd, fd->idx, res);
+
+ return res;
+#endif
+}
+
+/**
* gst_poll_wait:
* @set: a #GstPoll.
* @timeout: a timeout in nanoseconds.
* gst_poll_wait() will be affected by gst_poll_restart() and
* gst_poll_set_flushing().
*
+ * This function only works for non-timer #GstPoll objects created with
+ * gst_poll_new().
+ *
* Returns: %TRUE if the controllability of @set could be updated.
*/
gboolean
gst_poll_set_controllable (GstPoll * set, gboolean controllable)
{
g_return_val_if_fail (set != NULL, FALSE);
+ g_return_val_if_fail (!set->timer, FALSE);
GST_LOG ("%p: controllable : %d", set, controllable);
* used after adding or removing descriptors to @set.
*
* If @set is not controllable, then this call will have no effect.
+ *
+ * This function only works for non-timer #GstPoll objects created with
+ * gst_poll_new().
*/
void
gst_poll_restart (GstPoll * set)
{
g_return_if_fail (set != NULL);
+ g_return_if_fail (!set->timer);
if (set->controllable && GET_WAITING (set) > 0) {
/* we are controllable and waiting, wake up the waiter. The socket will be
* to gst_poll_wait() will return -1, with errno set to EBUSY.
*
* Unsetting the flushing state will restore normal operation of @set.
+ *
+ * This function only works for non-timer #GstPoll objects created with
+ * gst_poll_new().
*/
void
gst_poll_set_flushing (GstPoll * set, gboolean flushing)
{
g_return_if_fail (set != NULL);
+ g_return_if_fail (!set->timer);
GST_LOG ("%p: flushing: %d", set, flushing);
*
* Write a byte to the control socket of the controllable @set.
* This function is mostly useful for timer #GstPoll objects created with
- * gst_poll_new_timer().
+ * gst_poll_new_timer().
*
* It will make any current and future gst_poll_wait() function return with
* 1, meaning the control socket is set. After an equal amount of calls to
* gst_poll_read_control() have been performed, calls to gst_poll_wait() will
* block again until their timeout expired.
*
- * Returns: %TRUE on success. %FALSE when @set is not controllable or when the
- * byte could not be written.
+ * This function only works for timer #GstPoll objects created with
+ * gst_poll_new_timer().
+ *
+ * Returns: %TRUE on success. %FALSE when when the byte could not be written.
+ * errno contains the detailed error code but will never be EAGAIN, EINTR or
+ * EWOULDBLOCK. %FALSE always signals a critical error.
*/
gboolean
gst_poll_write_control (GstPoll * set)
* @set: a #GstPoll.
*
* Read a byte from the control socket of the controllable @set.
- * This function is mostly useful for timer #GstPoll objects created with
- * gst_poll_new_timer().
*
- * Returns: %TRUE on success. %FALSE when @set is not controllable or when there
- * was no byte to read.
+ * This function only works for timer #GstPoll objects created with
+ * gst_poll_new_timer().
+ *
+ * Returns: %TRUE on success. %FALSE when when there was no byte to read or
+ * reading the byte failed. If there was no byte to read, and only then, errno
+ * will contain EWOULDBLOCK or EAGAIN. For all other values of errno this always signals a
+ * critical error.
*/
gboolean
gst_poll_read_control (GstPoll * set)
projects / platform / upstream / gstreamer.git / blobdiff