su_select_port.c

Sofia SIP is an open-source SIP User-Agent library, compliant with the IETF RFC3261 specification.

static int su_select_port_deregister0(su_port_t *self, int i, int destroy_wait)
{
  struct su_select_register **indices = self->sup_indices;
  struct su_select_register *ser;

  ser = self->sup_indices[i];
  if (ser == NULL || ser->ser_cb == NULL) {
    su_seterrno(ENOENT);
    return -1;
  }

  assert(ser->ser_id == i);

  FD_CLR(ser->ser_wait->fd, self->sup_readfds);
  FD_CLR(ser->ser_wait->fd, self->sup_writefds);

  if (ser->ser_wait->fd + 1 >= self->sup_maxfd)
    self->sup_maxfd = 0;

  memset(ser, 0, sizeof *ser);
  ser->ser_id = i;
  ser->ser_next = indices[0], indices[0] = ser;

  self->sup_n_registrations--;
  self->sup_registers++;

  return i;
}


/** Unregister a su_wait_t object.
 *  
 * The function su_select_port_unregister() unregisters a su_wait_t object. 
 * The registration defined by the wait object, the callback function and
 * the argument pointer are removed from the port object.
 * 
 * @param self     - pointer to port object
 * @param root     - pointer to root object
 * @param wait     - pointer to wait object
 * @param callback - callback function pointer (may be NULL)
 * @param arg      - argument given to callback function when it is invoked 
 *                   (may be NULL)
 *
 * @deprecated Use su_select_port_deregister() instead. 
 *
 * @return Nonzero index of the wait object, or -1 upon an error.
 */
int su_select_port_unregister(su_port_t *self,
			     su_root_t *root, 
			     su_wait_t *wait,	
			     su_wakeup_f callback, /* XXX - ignored */
			     su_wakeup_arg_t *arg)
{
  int i, I;

  struct su_select_register *ser;

  assert(self);
  assert(su_port_own_thread(self));

  I = self->sup_max_index;

  for (i = 1; i <= I; i++) {
    ser = self->sup_indices[i];

    if (ser->ser_cb &&
	arg == ser->ser_arg &&
	SU_WAIT_CMP(wait[0], ser->ser_wait[0]) == 0)
      return su_select_port_deregister0(self, ser->ser_id, 0);
  }

  su_seterrno(ENOENT);

  return -1;
}

/** Deregister a su_wait_t object.
 *  
 *  Deregisters a registration by index. The wait object, a callback
 *  function and a argument are removed from the port object. The wait
 *  object is destroyed.
 * 
 * @param self     - pointer to port object
 * @param i        - registration index
 * 
 * @return Index of the wait object, or -1 upon an error.
 */
int su_select_port_deregister(su_port_t *self, int i)
{
  struct su_select_register *ser;

  if (i <= 0 || i > self->sup_max_index)
    return su_seterrno(EBADF);

  ser = self->sup_indices[i];
  if (!ser->ser_cb)
    return su_seterrno(EBADF);

  return su_select_port_deregister0(self, i, 1);
}


/** @internal
 * Unregister all su_wait_t objects of given su_root_t instance.
 *
 * The function su_select_port_unregister_all() unregisters all su_wait_t
 * objects associated with given root object.
 * 
 * @param  self     - pointer to port object
 * @param  root     - pointer to root object
 * 
 * @return Number of wait objects removed.
 */
int su_select_port_unregister_all(su_port_t *self, su_root_t *root)
{
  int i, I, n;

  struct su_select_register *ser;

  assert(self); assert(root);
  assert(su_port_own_thread(self));

  I = self->sup_max_index;

  for (i = 1, n = 0; i <= I; i++) {
    ser = self->sup_indices[i];
    if (ser->ser_root != root)
      continue;
    su_select_port_deregister0(self, ser->ser_id, 0);
    n++;
  }

  return n;
}

/**Set mask for a registered event. @internal
 *
 * The function su_select_port_eventmask() sets the mask describing events
 * that can signal the registered callback.
 *
 * @param port   pointer to port object
 * @param index  registration index
 * @param socket socket
 * @param events new event mask
 *
 * @retval 0 when successful,
 * @retval -1 upon an error.
 */
int su_select_port_eventmask(su_port_t *self,
			     int index,
			     int socket, int events)
{
  struct su_select_register *ser;

  if (index <= 0 || index > self->sup_max_index)
    return su_seterrno(EBADF);

  ser = self->sup_indices[index];
  if (!ser->ser_cb)
    return su_seterrno(EBADF);

  if (self->sup_maxfd == 0)
    su_select_port_update_maxfd(self);

  if (socket >= self->sup_maxfd)
    return su_seterrno(EBADF);

  if (su_wait_mask(ser->ser_wait, socket, events) < 0)
    return -1;

  assert(socket < self->sup_maxfd);
  
  if (events & SU_WAIT_IN)
    FD_SET(socket, self->sup_readfds);
  else
    FD_CLR(socket, self->sup_readfds);

  if (events & SU_WAIT_OUT)
    FD_SET(socket, self->sup_writefds);
  else
    FD_CLR(socket, self->sup_writefds);

  return 0;
}

/** @internal Enable multishot mode.
 *
 * Enables, disables or queries the multishot mode for the port. The
 * multishot mode determines how the events are scheduled by port. If
 * multishot mode is enabled, port serves all the sockets that have received
 * network events. If it is disabled, only first socket event is served.
 *
 * @param self      pointer to port object
 * @param multishot multishot mode (0 => disables, 1 => enables, -1 => query)
 * 
 * @retval 0 multishot mode is disabled
 * @retval 1 multishot mode is enabled
 * @retval -1 an error occurred
 */
static
int su_select_port_multishot(su_port_t *self, int multishot)
{
  if (multishot < 0)
    return self->sup_multishot;
  else if (multishot == 0 || multishot == 1)
    return self->sup_multishot = multishot;
  else 
    return (errno = EINVAL), -1;
}


/** @internal
 * Wait (poll()) for wait objects in port.
 *
 * @param self     pointer to port
 * @param tout     timeout in milliseconds
 *
 * @return number of events handled
 */
static
int su_select_port_wait_events(su_port_t *self, su_duration_t tout)
{
  int j, n, events = 0;
  unsigned version = self->sup_registers;
  size_t bytes;
  struct timeval tv;
  fd_set *rset = NULL, *wset = NULL;

  if (self->sup_maxfd == 0)
    su_select_port_update_maxfd(self);

  bytes = FDSETSIZE(self->sup_maxfd);

  if (bytes) {
    rset = memcpy(self->sup_readfds2, self->sup_readfds, bytes);
    wset = memcpy(self->sup_writefds2, self->sup_writefds, bytes);
  }  

  tv.tv_sec = tout / 1000;
  tv.tv_usec = (tout % 1000) * 1000;

  n = select(self->sup_maxfd, rset, wset, NULL, &tv);

  if (n < 0) {
    SU_DEBUG_0(("su_select_port_wait_events(%p): %s (%d)\n",
		(void *)self, su_strerror(su_errno()), su_errno()));
    return 0;
  }
  else if (n == 0)
    return 0;

  for (j = 1; j <= self->sup_max_index; j++) {
    struct su_select_register *ser;
    su_root_magic_t *magic;
    int fd;

    ser = self->sup_indices[j];
    if (!ser->ser_cb)
      continue;

    fd = ser->ser_wait->fd;
    ser->ser_wait->revents = 0;

    if (ser->ser_wait->events & SU_WAIT_IN)
      if (FD_ISSET(fd, rset)) ser->ser_wait->revents |= SU_WAIT_IN, n--;
    if (ser->ser_wait->events & SU_WAIT_OUT)
      if (FD_ISSET(fd, wset)) ser->ser_wait->revents |= SU_WAIT_OUT, n--;

    if (ser->ser_wait->revents) {
      magic = ser->ser_root ? su_root_magic(ser->ser_root) : NULL;
      ser->ser_cb(magic, ser->ser_wait, ser->ser_arg);
      events++;
      if (version != self->sup_registers)
	/* Callback function used su_register()/su_deregister() */
	return events;
      if (!self->sup_multishot)
	/* Callback function used su_register()/su_deregister() */
	return events;
    }

    if (n == 0)
      break;
  }    

  assert(n == 0);

  return events;
}

/** Create a port using epoll() or poll().
 */
su_port_t *su_select_port_create(void)
{
  su_port_t *self;

  self = su_home_new(sizeof *self);
  if (!self)
    return NULL;

  if (su_home_destructor(su_port_home(self), su_select_port_deinit) < 0 ||
      !(self->sup_indices = 
	su_zalloc(su_port_home(self),
		  (sizeof self->sup_indices[0]) * 
		  (self->sup_size_indices = __NFDBITS)))) {
    su_home_unref(su_port_home(self));
    return NULL;
  }

  self->sup_multishot = SU_ENABLE_MULTISHOT_POLL;

  if (su_socket_port_init(self->sup_base, su_select_port_vtable) < 0)
    return su_home_unref(su_port_home(self)), NULL;

  return self;
}

int su_select_clone_start(su_root_t *parent,
			 su_clone_r return_clone,
			 su_root_magic_t *magic,
			 su_root_init_f init,
			 su_root_deinit_f deinit)
{
  return su_pthreaded_port_start(su_select_port_create, 
				 parent, return_clone, magic, init, deinit);
}


#endif