su_kqueue_port.c

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

  return i;
}


/** Unregister a su_wait_t object.
 *  
 *  The function su_kqueue_port_unregister() unregisters a su_wait_t object. The
 *  wait object, a callback function and a argument 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_kqueue_port_deregister() instead. 
 *
 * @return Nonzero index of the wait object, or -1 upon an error.
 */
int su_kqueue_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_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_kqueue_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_kqueue_port_deregister(su_port_t *self, int i)
{
  struct su_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_kqueue_port_deregister0(self, i, 1);
}

/** @internal
 * Unregister all su_wait_t objects belonging to a root.
 *
 * The function su_kqueue_port_unregister_all() unregisters all su_wait_t objects
 * and destroys all queued timers 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_kqueue_port_unregister_all(su_port_t *self, 
				su_root_t *root)
{
  int i, I, n;

  struct su_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_kqueue_port_deregister0(self, ser->ser_id, 0);
    n++;
  }

  return n;
}

/**Set mask for a registered event. @internal
 *
 * The function su_kqueue_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_kqueue_port_eventmask(su_port_t *self, int index, int socket, int events)
{
  struct su_register *ser;
  struct kevent ev[1];
  su_wait_t *wait;
  int flags;

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

  wait = ser->ser_wait;

  assert(socket == wait->fd);

  wait->events = events;

  flags = (wait->events & SU_WAIT_IN) ? EV_ADD | EV_ENABLE : EV_ADD | EV_DISABLE;
  EV_SET(ev, wait->fd, EVFILT_READ, flags, 0, 0, (void *)index);
  if (kevent(self->sup_kqueue, ev, 1, NULL, 0, NULL) == -1) {
    SU_DEBUG_0(("modify kevent((%u, %s, %s, %p)) failed: %s\n",
		wait->fd, "EVFILT_READ",
		(events & SU_WAIT_IN) ? "EV_ENABLE" : "EV_DISABLE",
		(void *)index, strerror(errno)));
  }

  flags = (wait->events & SU_WAIT_OUT) ? EV_ADD | EV_ENABLE : EV_ADD | EV_DISABLE;
  EV_SET(ev, wait->fd, EVFILT_WRITE, flags, 0, 0, (void *)index);
  if (kevent(self->sup_kqueue, ev, 1, NULL, 0, NULL) == -1) {
    SU_DEBUG_0(("modify kevent((%u, %s, %s, %p)) failed: %s\n",
		wait->fd, "EVFILT_WRITE",
		(events & SU_WAIT_OUT) ? "EV_ENABLE" : "EV_DISABLE",
		(void *)index, strerror(errno)));
  }

  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_kqueue_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 (kqueue()) for wait objects in port.
 *
 * @param self     pointer to port
 * @param tout     timeout in milliseconds
 *
 * @return number of events handled
 */
static
int su_kqueue_port_wait_events(su_port_t *self, su_duration_t tout)
{
  int j, n, events = 0, index;
  unsigned version = self->sup_registers;

  int const M = 4;
  struct kevent ev[M];
  
  struct timespec ts;

  ts.tv_sec = tout / 1000;
  ts.tv_nsec = tout % 1000 * 1000000;

  n = kevent(self->sup_kqueue, NULL, 0,
	     ev, self->sup_multishot ? M : 1,
	     tout < SU_DURATION_MAX ? &ts : NULL);

  assert(n <= M);
  
  for (j = 0; j < n; j++) {
    struct su_register *ser;
    su_root_magic_t *magic;

    index = (int)ev[j].udata;
    if (index <= 0 || self->sup_max_index < index)
      continue;
    ser = self->sup_indices[index];

    magic = ser->ser_root ? su_root_magic(ser->ser_root) : NULL;
    ser->ser_wait->revents =
      (ser->ser_wait->events | SU_WAIT_HUP) &
      (
       ((ev[j].filter == EVFILT_READ) ? SU_WAIT_IN : 0) |
       ((ev[j].filter == EVFILT_WRITE) ? SU_WAIT_OUT : 0) |
       ((ev[j].flags & EV_EOF) ? SU_WAIT_HUP : 0)
       );
    if (ser->ser_wait->revents) {
      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;
    }
  }    

  return n;
}


/** Create a port using kqueue() (or poll()/select(), if kqueue() fails).
 */
su_port_t *su_kqueue_port_create(void)
{
  su_port_t *self = NULL;
  int kq = kqueue();

  if (kq < 0) {
#if HAVE_POLL
    return su_poll_port_create();
#else
    return su_select_port_create();
#endif
  }

  self = su_home_new(sizeof *self);
  if (!self)
    goto failed;

  if (su_home_destructor(su_port_home(self), su_kqueue_port_deinit) < 0)
    goto failed;

  self->sup_kqueue = kq, kq = -1;
  self->sup_indices = su_zalloc(su_port_home(self),
				(sizeof self->sup_indices[0]) * 
				(self->sup_size_indices = 64));
  if (!self->sup_indices)
    goto failed;

  if (su_socket_port_init(self->sup_base, su_kqueue_port_vtable) < 0)
    goto failed;

  self->sup_multishot = SU_ENABLE_MULTISHOT_KQUEUE;

  return self;

 failed:
  if (kq != -1)
    close(kq);
  su_home_unref(su_port_home(self));
  return NULL;
}

int su_kqueue_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_kqueue_port_create, 
				 parent, return_clone, magic, init, deinit);
}

#else

su_port_t *su_kqueue_port_create(void)
{
  return su_default_port_create();
}

int su_kqueue_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_default_clone_start(parent, return_clone, magic, init, deinit);
}

#endif  /* HAVE_KQUEUE */