/* GStreamer

 * 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>

 *

 * gstpoll.c: File descriptor set

 *

 * 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.

 */

/**

 * SECTION:gstpoll

 * @title: GstPoll

 * @short_description: Keep track of file descriptors and make it possible

 *                     to wait on them in a cancellable way

 *

 * A #GstPoll keeps track of file descriptors much like fd_set (used with

 * select()) or a struct pollfd array (used with poll()). Once created with

 * gst_poll_new(), the set can be used to wait for file descriptors to be

 * readable and/or writable. It is possible to make this wait be controlled

 * by specifying %TRUE for the @controllable flag when creating the set (or

 * later calling gst_poll_set_controllable()).

 *

 * 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().

 *

 * 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

 * in time. The wait can be controlled by calling gst_poll_restart() and

 * gst_poll_set_flushing().

 *

 * Once the file descriptor set has been waited for, one can use

 * gst_poll_fd_has_closed() to see if the file descriptor has been closed,

 * gst_poll_fd_has_error() to see if it has generated an error,

 * gst_poll_fd_can_read() to see if it is possible to read from the file

 * descriptor, and gst_poll_fd_can_write() to see if it is possible to

 * write to it.

 *

 */

#ifdef HAVE_CONFIG_H

#include "config.h"

#endif

#include "gst_private.h"

#include "glib-compat-private.h"

#include <sys/types.h>

#ifdef HAVE_UNISTD_H

#include <unistd.h>

#endif

#include <errno.h>

#include <fcntl.h>

#include <glib.h>

#ifdef G_OS_WIN32

#include <winsock2.h>

#else

#define _GNU_SOURCE 1

#ifdef HAVE_SYS_POLL_H

#include <sys/poll.h>

#endif

#ifdef HAVE_POLL_H

#include <poll.h>

#endif

#include <sys/time.h>

#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>

/* The poll() emulation on OS/X doesn't handle fds=NULL, nfds=0,

 * so we prefer our own poll emulation.

 */

#if defined(BROKEN_POLL)

#undef HAVE_POLL

#endif

#include "gstpoll.h"

#define GST_CAT_DEFAULT GST_CAT_POLL

#ifdef G_OS_WIN32

typedef struct _WinsockFd WinsockFd;

struct _WinsockFd

{

  gint fd;

  glong event_mask;

  WSANETWORKEVENTS events;

  glong ignored_event_mask;

};

#endif

typedef enum

{

  GST_POLL_MODE_AUTO,

  GST_POLL_MODE_SELECT,

  GST_POLL_MODE_PSELECT,

  GST_POLL_MODE_POLL,

  GST_POLL_MODE_PPOLL,

  GST_POLL_MODE_WINDOWS

} GstPollMode;

struct _GstPoll

{

  GstPollMode mode;

  GMutex lock;

  /* array of fds, always written to and read from with lock */

  GArray *fds;

  /* array of active fds, only written to from the waiting thread with the

   * lock and read from with the lock or without the lock from the waiting

   * thread */

  GArray *active_fds;

#ifndef G_OS_WIN32

  GstPollFD control_read_fd;

  GstPollFD control_write_fd;

#else

  GArray *active_fds_ignored;

  GArray *events;

  GArray *active_events;

  HANDLE wakeup_event;

#endif

  gboolean controllable;

  volatile gint waiting;

  volatile gint control_pending;

  volatile gint flushing;

  gboolean timer;

  volatile gint rebuild;

};

static gboolean gst_poll_fd_ctl_read_unlocked (GstPoll * set, GstPollFD * fd,

    gboolean active);

static gboolean gst_poll_add_fd_unlocked (GstPoll * set, GstPollFD * fd);

#define IS_FLUSHING(s)      (g_atomic_int_get(&(s)->flushing))

#define SET_FLUSHING(s,val) (g_atomic_int_set(&(s)->flushing, (val)))

#define INC_WAITING(s)      (g_atomic_int_add(&(s)->waiting, 1))

#define DEC_WAITING(s)      (g_atomic_int_add(&(s)->waiting, -1))

#define GET_WAITING(s)      (g_atomic_int_get(&(s)->waiting))

#define TEST_REBUILD(s)     (g_atomic_int_compare_and_exchange(&(s)->rebuild, 1, 0))

#define MARK_REBUILD(s)     (g_atomic_int_set(&(s)->rebuild, 1))

#ifndef G_OS_WIN32

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

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

 * we can send special commands to control it */

static inline gboolean

raise_wakeup (GstPoll * set)

{

  gboolean result = TRUE;

  /* 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);

  }

  if (result) {

    set->control_pending++;

  }

  g_mutex_unlock (&set->lock);

  return result;

}

static inline gboolean

release_wakeup (GstPoll * set)

{

  gboolean result = FALSE;

  /* makes testing/modifying control_pending and RELEASE_EVENT() atomic. */

  g_mutex_lock (&set->lock);

  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;

}

static inline gint

release_all_wakeup (GstPoll * set)

{

  gint old;

  /* makes testing control_pending and RELEASE_EVENT() atomic. */

  g_mutex_lock (&set->lock);

  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;

}

static gint

find_index (GArray * array, GstPollFD * fd)

{

#ifndef G_OS_WIN32

  struct pollfd *ifd;

#else

  WinsockFd *ifd;

#endif

  guint i;

  /* start by assuming the index found in the fd is still valid */

  if (fd->idx >= 0 && fd->idx < array->len) {

#ifndef G_OS_WIN32

    ifd = &g_array_index (array, struct pollfd, fd->idx);

#else

    ifd = &g_array_index (array, WinsockFd, fd->idx);

#endif

    if (ifd->fd == fd->fd) {

      return fd->idx;

    }

  }

  /* the pollfd array has changed and we need to lookup the fd again */

  for (i = 0; i < array->len; i++) {

#ifndef G_OS_WIN32

    ifd = &g_array_index (array, struct pollfd, i);

#else

    ifd = &g_array_index (array, WinsockFd, i);

#endif

    if (ifd->fd == fd->fd) {

      fd->idx = (gint) i;

      return fd->idx;

    }

  }

  fd->idx = -1;

  return fd->idx;

}

#if !defined(HAVE_PPOLL) && defined(HAVE_POLL)

/* check if all file descriptors will fit in an fd_set */

static gboolean

selectable_fds (GstPoll * set)

{

  guint i;

  g_mutex_lock (&set->lock);

  for (i = 0; i < set->fds->len; i++) {

    struct pollfd *pfd = &g_array_index (set->fds, struct pollfd, i);

    if (pfd->fd >= FD_SETSIZE)

      goto too_many;

  }

  g_mutex_unlock (&set->lock);

  return TRUE;

too_many:

  {

    g_mutex_unlock (&set->lock);

    return FALSE;

  }

}

/* check if the timeout will convert to a timeout value used for poll()

 * without a loss of precision

 */

static gboolean

pollable_timeout (GstClockTime timeout)

{

  if (timeout == GST_CLOCK_TIME_NONE)

    return TRUE;

  /* not a nice multiple of milliseconds */

  if (timeout % 1000000)

    return FALSE;

  return TRUE;

}

#endif

static GstPollMode

choose_mode (GstPoll * set, GstClockTime timeout)

{

  GstPollMode mode;

  if (set->mode == GST_POLL_MODE_AUTO) {

#ifdef HAVE_PPOLL

    mode = GST_POLL_MODE_PPOLL;

#elif defined(HAVE_POLL)

    if (!selectable_fds (set) || pollable_timeout (timeout)) {

      mode = GST_POLL_MODE_POLL;

    } else {

#ifdef HAVE_PSELECT

      mode = GST_POLL_MODE_PSELECT;

#else

      mode = GST_POLL_MODE_SELECT;

#endif

    }

#elif defined(HAVE_PSELECT)

    mode = GST_POLL_MODE_PSELECT;

#else

    mode = GST_POLL_MODE_SELECT;

#endif

  } else {

    mode = set->mode;

  }

  return mode;

}

#ifndef G_OS_WIN32

static gint

pollfd_to_fd_set (GstPoll * set, fd_set * readfds, fd_set * writefds,

    fd_set * errorfds)

{

  gint max_fd = -1;

  guint i;

  FD_ZERO (readfds);

  FD_ZERO (writefds);

  FD_ZERO (errorfds);

  g_mutex_lock (&set->lock);

  for (i = 0; i < set->active_fds->len; i++) {

    struct pollfd *pfd = &g_array_index (set->fds, struct pollfd, i);

    if (pfd->fd < FD_SETSIZE) {

      if (pfd->events & POLLIN)

        FD_SET (pfd->fd, readfds);

      if (pfd->events & POLLOUT)

        FD_SET (pfd->fd, writefds);

      if (pfd->events)

        FD_SET (pfd->fd, errorfds);

      if (pfd->fd > max_fd && (pfd->events & (POLLIN | POLLOUT)))

        max_fd = pfd->fd;

    }

  }

  g_mutex_unlock (&set->lock);

  return max_fd;

}

static void

fd_set_to_pollfd (GstPoll * set, fd_set * readfds, fd_set * writefds,

    fd_set * errorfds)

{

  guint i;

  g_mutex_lock (&set->lock);

  for (i = 0; i < set->active_fds->len; i++) {

    struct pollfd *pfd = &g_array_index (set->active_fds, struct pollfd, i);

    if (pfd->fd < FD_SETSIZE) {

      pfd->revents = 0;

      if (FD_ISSET (pfd->fd, readfds))

        pfd->revents |= POLLIN;

      if (FD_ISSET (pfd->fd, writefds))

        pfd->revents |= POLLOUT;

      if (FD_ISSET (pfd->fd, errorfds))

        pfd->revents |= POLLERR;

    }

  }

  g_mutex_unlock (&set->lock);

}

#else /* G_OS_WIN32 */

/*

 * Translate errors thrown by the Winsock API used by GstPoll:

 *   WSAEventSelect, WSAWaitForMultipleEvents and WSAEnumNetworkEvents

 */

static gint

gst_poll_winsock_error_to_errno (DWORD last_error)

{

  switch (last_error) {

    case WSA_INVALID_HANDLE:

    case WSAEINVAL:

    case WSAENOTSOCK:

      return EBADF;

    case WSA_NOT_ENOUGH_MEMORY:

      return ENOMEM;

      /*

       * Anything else, including:

       *   WSA_INVALID_PARAMETER, WSAEFAULT, WSAEINPROGRESS, WSAENETDOWN,

       *   WSANOTINITIALISED

       */

    default:

      return EINVAL;

  }

}

static void

gst_poll_free_winsock_event (GstPoll * set, gint idx)

{

  WinsockFd *wfd = &g_array_index (set->fds, WinsockFd, idx);

  HANDLE event = g_array_index (set->events, HANDLE, idx);

  WSAEventSelect (wfd->fd, event, 0);

  CloseHandle (event);

}

static void

gst_poll_update_winsock_event_mask (GstPoll * set, gint idx, glong flags,

    gboolean active)

{

  WinsockFd *wfd;

  wfd = &g_array_index (set->fds, WinsockFd, idx);

  if (active)

    wfd->event_mask |= flags;

  else

    wfd->event_mask &= ~flags;

  /* reset ignored state if the new mask doesn't overlap at all */

  if ((wfd->ignored_event_mask & wfd->event_mask) == 0)

    wfd->ignored_event_mask = 0;

}

static gboolean

gst_poll_prepare_winsock_active_sets (GstPoll * set)

{

  guint i;

  g_array_set_size (set->active_fds, 0);

  g_array_set_size (set->active_fds_ignored, 0);

  g_array_set_size (set->active_events, 0);

  g_array_append_val (set->active_events, set->wakeup_event);

  for (i = 0; i < set->fds->len; i++) {

    WinsockFd *wfd = &g_array_index (set->fds, WinsockFd, i);

    HANDLE event = g_array_index (set->events, HANDLE, i);

    if (wfd->ignored_event_mask == 0) {

      gint ret;

      g_array_append_val (set->active_fds, *wfd);

      g_array_append_val (set->active_events, event);

      ret = WSAEventSelect (wfd->fd, event, wfd->event_mask);

      if (G_UNLIKELY (ret != 0)) {

        errno = gst_poll_winsock_error_to_errno (WSAGetLastError ());

        return FALSE;

      }

    } else {

      g_array_append_val (set->active_fds_ignored, wfd);

    }

  }

  return TRUE;

}

static gint

gst_poll_collect_winsock_events (GstPoll * set)

{

  gint res, i;

  /*

   * We need to check which events are signaled, and call

   * WSAEnumNetworkEvents for those that are, which resets

   * the event and clears the internal network event records.

   */

  res = 0;

  for (i = 0; i < set->active_fds->len; i++) {

    WinsockFd *wfd = &g_array_index (set->active_fds, WinsockFd, i);

    HANDLE event = g_array_index (set->active_events, HANDLE, i + 1);

    DWORD wait_ret;

    wait_ret = WaitForSingleObject (event, 0);

    if (wait_ret == WAIT_OBJECT_0) {

      gint enum_ret = WSAEnumNetworkEvents (wfd->fd, event, &wfd->events);

      if (G_UNLIKELY (enum_ret != 0)) {

        res = -1;

        errno = gst_poll_winsock_error_to_errno (WSAGetLastError ());

        break;

      }

      res++;

    } else {

      /* clear any previously stored result */

      memset (&wfd->events, 0, sizeof (wfd->events));

    }

  }

  /* If all went well we also need to reset the ignored fds. */

  if (res >= 0) {

    res += set->active_fds_ignored->len;

    for (i = 0; i < set->active_fds_ignored->len; i++) {

      WinsockFd *wfd = g_array_index (set->active_fds_ignored, WinsockFd *, i);

      wfd->ignored_event_mask = 0;

    }

    g_array_set_size (set->active_fds_ignored, 0);

  }

  return res;

}

#endif

/**

 * gst_poll_new: (skip)

 * @controllable: whether it should be possible to control a wait.

 *

 * Create a new file descriptor set. If @controllable, it

 * is possible to restart or flush a call to gst_poll_wait() with

 * gst_poll_restart() and gst_poll_set_flushing() respectively.

 *

 * Free-function: gst_poll_free

 *

 * Returns: (transfer full) (nullable): a new #GstPoll, or %NULL in

 *     case of an error.  Free with gst_poll_free().

 */

GstPoll *

gst_poll_new (gboolean controllable)

{

  GstPoll *nset;

  nset = g_slice_new0 (GstPoll);

  GST_DEBUG ("%p: new controllable : %d", nset, controllable);

  g_mutex_init (&nset->lock);

#ifndef G_OS_WIN32

  nset->mode = GST_POLL_MODE_AUTO;

  nset->fds = g_array_new (FALSE, FALSE, sizeof (struct pollfd));

  nset->active_fds = g_array_new (FALSE, FALSE, sizeof (struct pollfd));

  nset->control_read_fd.fd = -1;

  nset->control_write_fd.fd = -1;

  {

    gint control_sock[2];

    if (socketpair (PF_UNIX, SOCK_STREAM, 0, control_sock) < 0)

      goto no_socket_pair;

    nset->control_read_fd.fd = control_sock[0];

    nset->control_write_fd.fd = control_sock[1];

    gst_poll_add_fd_unlocked (nset, &nset->control_read_fd);

    gst_poll_fd_ctl_read_unlocked (nset, &nset->control_read_fd, TRUE);

  }

#else

  nset->mode = GST_POLL_MODE_WINDOWS;

  nset->fds = g_array_new (FALSE, FALSE, sizeof (WinsockFd));

  nset->active_fds = g_array_new (FALSE, FALSE, sizeof (WinsockFd));

  nset->active_fds_ignored = g_array_new (FALSE, FALSE, sizeof (WinsockFd *));

  nset->events = g_array_new (FALSE, FALSE, sizeof (HANDLE));

  nset->active_events = g_array_new (FALSE, FALSE, sizeof (HANDLE));

  nset->wakeup_event = CreateEvent (NULL, TRUE, FALSE, NULL);

#endif

  /* ensure (re)build, though already sneakily set in non-windows case */

  MARK_REBUILD (nset);

  nset->controllable = controllable;

  nset->control_pending = 0;

  return nset;

  /* ERRORS */

#ifndef G_OS_WIN32

no_socket_pair:

  {

    GST_WARNING ("%p: can't create socket pair !", nset);

    gst_poll_free (nset);

    return NULL;

  }

#endif

}

/**

 * gst_poll_new_timer: (skip)

 *

 * Create a new poll object that can be used for scheduling cancellable

 * timeouts.

 *

 * A timeout is performed with gst_poll_wait(). Multiple timeouts can be

 * performed from different threads.

 *

 * Free-function: gst_poll_free

 *

 * Returns: (transfer full) (nullable): a new #GstPoll, or %NULL in

 *     case of an error.  Free with gst_poll_free().

 */

GstPoll *

gst_poll_new_timer (void)

{

  GstPoll *poll;

  /* make a new controllable poll set */

  if (!(poll = gst_poll_new (TRUE)))

    goto done;

  /* we are a timer */

  poll->timer = TRUE;

done:

  return poll;

}

/**

 * gst_poll_free:

 * @set: (transfer full): a file descriptor set.

 *

 * Free a file descriptor set.

 */

void

gst_poll_free (GstPoll * set)

{

  g_return_if_fail (set != NULL);

  GST_DEBUG ("%p: freeing", set);

#ifndef G_OS_WIN32

  if (set->control_write_fd.fd >= 0)

    close (set->control_write_fd.fd);

  if (set->control_read_fd.fd >= 0)

    close (set->control_read_fd.fd);

#else

  CloseHandle (set->wakeup_event);

  {

    guint i;

    for (i = 0; i < set->events->len; i++)

      gst_poll_free_winsock_event (set, i);

  }

  g_array_free (set->active_events, TRUE);

  g_array_free (set->events, TRUE);

  g_array_free (set->active_fds_ignored, TRUE);

#endif

  g_array_free (set->active_fds, TRUE);

  g_array_free (set->fds, TRUE);

  g_mutex_clear (&set->lock);

  g_slice_free (GstPoll, set);

}

/**

 * gst_poll_get_read_gpollfd:

 * @set: a #GstPoll

 * @fd: a #GPollFD

 *

 * Get a GPollFD for the reading part of the control socket. This is useful when

 * integrating with a GSource and GMainLoop.

 */

void

gst_poll_get_read_gpollfd (GstPoll * set, GPollFD * fd)

{

  g_return_if_fail (set != NULL);

  g_return_if_fail (fd != NULL);

#ifndef G_OS_WIN32

  fd->fd = set->control_read_fd.fd;

#else

#if GLIB_SIZEOF_VOID_P == 8

  fd->fd = (gint64) set->wakeup_event;

#else

  fd->fd = (gint) set->wakeup_event;

#endif

#endif

  fd->events = G_IO_IN | G_IO_HUP | G_IO_ERR;

  fd->revents = 0;

}

/**

 * gst_poll_fd_init:

 * @fd: a #GstPollFD

 *

 * Initializes @fd. Alternatively you can initialize it with

 * #GST_POLL_FD_INIT.

 */

void

gst_poll_fd_init (GstPollFD * fd)

{

  g_return_if_fail (fd != NULL);

  fd->fd = -1;

  fd->idx = -1;

}

static gboolean

gst_poll_add_fd_unlocked (GstPoll * set, GstPollFD * fd)

{

  gint idx;

  GST_DEBUG ("%p: fd (fd:%d, idx:%d)", set, fd->fd, fd->idx);

  idx = find_index (set->fds, fd);

  if (idx < 0) {

#ifndef G_OS_WIN32

    struct pollfd nfd;

    nfd.fd = fd->fd;

    nfd.events = POLLERR | POLLNVAL | POLLHUP;

    nfd.revents = 0;

    g_array_append_val (set->fds, nfd);

    fd->idx = set->fds->len - 1;

#else

    WinsockFd wfd;

    HANDLE event;

    wfd.fd = fd->fd;

    wfd.event_mask = FD_CLOSE;

    memset (&wfd.events, 0, sizeof (wfd.events));

    wfd.ignored_event_mask = 0;

    event = WSACreateEvent ();

    g_array_append_val (set->fds, wfd);

    g_array_append_val (set->events, event);

    fd->idx = set->fds->len - 1;

#endif

    MARK_REBUILD (set);

  } else {

    GST_WARNING ("%p: fd already added !", set);

  }