su_kqueue_port.c

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

/*
 * This file is part of the Sofia-SIP package
 *
 * Copyright (C) 2005 Nokia Corporation.
 *
 * Contact: Pekka Pessi <[e-mail]>
 *
 * This library is free software; you can redistribute it and/or
 * modify it under the terms of the GNU Lesser General Public License
 * as published by the Free Software Foundation; either version 2.1 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
 * Lesser General Public License for more details.
 *
 * You should have received a copy of the GNU Lesser 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
 *
 */

/**@ingroup su_wait
 * @CFILE su_kqueue_port.c
 *
 * Port implementation using kqueue()
 *
 * @author Martti Mela <[e-mail]>
 * @author Pekka Pessi <[e-mail]>
 *
 * @date Created: Sun Feb 18 19:55:37 EET 2007 mela
 */

#include "config.h"
#include "sofia-sip/su.h"

#define su_port_s su_kqueue_port_s

#include "su_port.h"

#if HAVE_KQUEUE

#include "sofia-sip/su_alloc.h"

#include <sys/event.h>

#define SU_ENABLE_MULTISHOT_KQUEUE 1

#include <stdlib.h>
#include <assert.h>
#include <stdarg.h>
#include <stdio.h>
#include <string.h>
#include <limits.h>
#include <errno.h>

/** Port based on kqueue(). */

struct su_kqueue_port_s {
  su_socket_port_t sup_base[1];

#define sup_home sup_base->sup_base->sup_base->sup_home

  /** kqueue fd */
  int              sup_kqueue;
  unsigned         sup_multishot; /**< Multishot operation? */

  unsigned         sup_registers; /** Counter incremented by 
				      su_port_register() or 
				      su_port_unregister()
				   */
  int              sup_n_registrations;
  int              sup_max_index; /**< Indexes are equal or smaller than this */
  int              sup_size_indices; /**< Size of allocated index table */

#define INDEX_MAX (0x7fffffff)

  /** Structure containing registration data */
  struct su_register {
    struct su_register *ser_next; /* Next in free list */
    su_wakeup_f     ser_cb; 
    su_wakeup_arg_t*ser_arg; 
    su_root_t      *ser_root; 
    int             ser_id; /** registration identifier */
    su_wait_t       ser_wait[1];
  } **sup_indices;
};

static void su_kqueue_port_decref(su_port_t *, int blocking, char const *who);

static int su_kqueue_port_register(su_port_t *self,
				 su_root_t *root, 
				 su_wait_t *wait, 
				 su_wakeup_f callback,
				 su_wakeup_arg_t *arg,
				 int priority);
static int su_kqueue_port_unregister(su_port_t *port,
				   su_root_t *root, 
				   su_wait_t *wait,	
				   su_wakeup_f callback, 
				   su_wakeup_arg_t *arg);
static int su_kqueue_port_deregister(su_port_t *self, int i);
static int su_kqueue_port_unregister_all(su_port_t *self, su_root_t *root);
static int su_kqueue_port_eventmask(su_port_t *self, 
				  int index,
				  int socket,
				  int events);
static int su_kqueue_port_multishot(su_port_t *self, int multishot);
static int su_kqueue_port_wait_events(su_port_t *self, su_duration_t tout);
static char const *su_kqueue_port_name(su_port_t const *self);

su_port_vtable_t const su_kqueue_port_vtable[1] =
  {{
      /* su_vtable_size: */ sizeof su_kqueue_port_vtable,
      su_pthread_port_lock,
      su_pthread_port_unlock,
      su_base_port_incref,
      su_kqueue_port_decref,
      su_base_port_gsource,
      su_socket_port_send,
      su_kqueue_port_register,
      su_kqueue_port_unregister,
      su_kqueue_port_deregister,
      su_kqueue_port_unregister_all,
      su_kqueue_port_eventmask,
      su_base_port_run,
      su_base_port_break,
      su_base_port_step,
      su_pthread_port_thread,
      su_base_port_add_prepoll,
      su_base_port_remove_prepoll,
      su_base_port_timers,
      su_kqueue_port_multishot,
      su_kqueue_port_wait_events,
      su_base_port_getmsgs,
      su_base_port_getmsgs_from,
      su_kqueue_port_name,
      su_base_port_start_shared,
      su_pthread_port_wait,
      su_pthread_port_execute,
    }};

static char const *su_kqueue_port_name(su_port_t const *self)
{
  return "kqueue";
}

static void su_kqueue_port_deinit(void *arg)
{
  su_port_t *self = arg;

  SU_DEBUG_9(("%s(%p) called\n", "su_kqueue_port_deinit", (void *)self));

  su_socket_port_deinit(self->sup_base);

  close(self->sup_kqueue);
}

static void su_kqueue_port_decref(su_port_t *self, int blocking, char const *who)
{
  su_base_port_decref(self, blocking, who);
}

/** @internal
 *
 *  Register a @c su_wait_t object. The wait object, a callback function and
 *  an argument pointer is stored in the port object.  The callback function
 *  will be called when the wait object is signaled.
 *
 *  Please note if identical wait objects are inserted, only first one is
 *  ever signalled.
 * 
 * @param self	     pointer to port
 * @param root	     pointer to root object
 * @param waits	     pointer to wait object
 * @param callback   callback function pointer
 * @param arg	     argument given to callback function when it is invoked
 * @param priority   relative priority of the wait object 
 *              (0 is normal, 1 important, 2 realtime)
 * 
 * @return
 *   Positive index of the wait object, 
 *   or -1 upon an error.
 */
int su_kqueue_port_register(su_port_t *self,
			    su_root_t *root, 
			    su_wait_t *wait, 
			    su_wakeup_f callback,
			    su_wakeup_arg_t *arg,
			    int priority)
{
  int i, j, n;
  struct su_register *ser;
  struct su_register **indices = self->sup_indices;
  struct kevent ev[1];
  int flags;

  assert(su_port_own_thread(self));

  n = self->sup_size_indices;

  if (n >= SU_WAIT_MAX)
    return su_seterrno(ENOMEM);

  ser = indices[0];

  if (!ser) {
    su_home_t *h = su_port_home(self);

    i = self->sup_max_index, j = i == 0 ? 15 : i + 16;
    
    if (j >= self->sup_size_indices) {
      /* Reallocate index table */
      n = n < 1024 ? 2 * n : n + 1024;
      indices = su_realloc(h, indices, n * sizeof(indices[0]));
      if (!indices)
	return -1;
      self->sup_indices = indices;
      self->sup_size_indices = n;
    }

    /* Allocate registrations */
    ser = su_zalloc(h, (j - i) * (sizeof *ser));
    if (!ser)
      return -1;

    indices[0] = ser;

    for (i++; i <= j; i++) {
      ser->ser_id = i;
      ser->ser_next = i < j ? ser + 1 : NULL;
      indices[i] = ser++;
    }

    self->sup_max_index = j;

    ser = indices[0];
  }

  i = ser->ser_id;

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

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

    EV_SET(ev, wait->fd, EVFILT_READ, EV_DELETE, 0, 0, (void *)i);
    kevent(self->sup_kqueue, ev, 1, NULL, 0, NULL);

    errno = error;
    return -1;
  }
  indices[0] = ser->ser_next;

  ser->ser_next = NULL;
  *ser->ser_wait = *wait;
  ser->ser_cb = callback;
  ser->ser_arg = arg;
  ser->ser_root = root;

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

  return i;			/* return index */
}

/** Deregister a su_wait_t object. */
static int su_kqueue_port_deregister0(su_port_t *self, int i, int destroy_wait)
{
  struct su_register **indices = self->sup_indices;
  struct su_register *ser;
  struct kevent ev[1];
  su_wait_t *wait;

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

  assert(ser->ser_id == i);

  wait = ser->ser_wait;

  EV_SET(ev, wait->fd, EVFILT_READ, EV_DELETE, 0, 0, (void *)i);
  if (kevent(self->sup_kqueue, ev, 1, NULL, 0, NULL) == -1) {
    SU_DEBUG_0(("remove kevent((%u, %s, %s, %p)) failed: %s\n",
		wait->fd, "EVFILT_READ", "EV_DELETE", (void *)i,
		strerror(errno)));
  }

  EV_SET(ev, wait->fd, EVFILT_WRITE, EV_DELETE, 0, 0, (void *)i);
  if (kevent(self->sup_kqueue, ev, 1, NULL, 0, NULL) == -1) {
    SU_DEBUG_0(("remove kevent((%u, %s, %s, %p)) failed: %s\n",
		wait->fd, "EVFILT_WRITE", "EV_DELETE", (void *)i,
		strerror(errno)));
  }

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

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