su_osx_runloop.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_osx_runloop.c
 *
 * OS-Independent Socket Syncronization Interface.
 *
 * This looks like nth reincarnation of "reactor".  It implements the
 * poll/select/WaitForMultipleObjects and message passing functionality.
 *
 * @author Pekka Pessi <[e-mail]>
 * @author Martti Mela <[e-mail]>
 *
 * @date Created: Tue Sep 14 15:51:04 1999 ppessi
 */

#include "config.h"

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

#define su_port_s su_osx_port_s 

#include "su_port.h"
#include "sofia-sip/su_osx_runloop.h"
#include "sofia-sip/su_alloc.h"
#include "sofia-sip/su_debug.h"

#if HAVE_FUNC
#define enter (void)SU_DEBUG_9(("%s: entering\n", __func__))
#elif HAVE_FUNCTION
#define enter (void)SU_DEBUG_9(("%s: entering\n", __FUNCTION__))
#else
#define enter (void)0
#endif

static su_port_t *su_osx_runloop_create(void) __attribute__((__malloc__));

/* Callback for CFObserver and CFSocket */
static void cf_observer_cb(CFRunLoopObserverRef observer, 
			   CFRunLoopActivity activity, 
			   void *info);

static void su_osx_port_socket_cb(CFSocketRef s, 
				  CFSocketCallBackType callbackType, 
				  CFDataRef address, 
				  const void *data, 
				  void *info);

static void su_osx_port_deinit(void *arg);

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


static CFSocketCallBackType map_poll_event_to_cf_event(int events);

static int su_osx_port_send(su_port_t *self, su_msg_r rmsg);

static int su_osx_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_osx_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_osx_port_deregister(su_port_t *self, int i);

static int su_osx_port_unregister_all(su_port_t *self,
			   su_root_t *root);

static int su_osx_port_eventmask(su_port_t *, int , int, int );
static void su_osx_port_run(su_port_t *self);
static void su_osx_port_break(su_port_t *self);
static su_duration_t su_osx_port_step(su_port_t *self, su_duration_t tout);

static int su_osx_port_multishot(su_port_t *port, int multishot);

static int su_osx_port_wait_events(su_port_t *self, su_duration_t tout);

static char const *su_osx_port_name(su_port_t const *self)
{
  return "CFRunLoop";
}

/*
 * Port is a per-thread reactor.  
 *
 * Multiple root objects executed by single thread share a su_port_t object. 
 */
struct su_osx_port_s {
  su_socket_port_t sup_socket[1];

#define sup_pthread sup_socket->sup_base
#define sup_base sup_socket->sup_base->sup_base
#define sup_home sup_socket->sup_base->sup_base->sup_home

  unsigned         sup_source_fired;

  CFRunLoopRef        sup_main_loop;
  CFRunLoopSourceRef *sup_sources;
  CFSocketRef        *sup_sockets;

  CFRunLoopObserverRef sup_observer;
  CFRunLoopObserverContext sup_observer_cntx[1];
  /* Struct for CFSocket callbacks; contains current CFSource index */
  struct osx_magic {
    su_port_t *o_port;
    int        o_current;
    int        o_count;
  } osx_magic[1];
  
  unsigned         sup_multishot; /**< Multishot operation? */

  unsigned         sup_registers; /** Counter incremented by 
				      su_port_register() or 
				      su_port_unregister()
				   */
  int              sup_n_waits; /**< Active su_wait_t in su_waits */
  int              sup_size_waits; /**< Size of allocate su_waits */

  int              sup_pri_offset; /**< Offset to prioritized waits */

#define INDEX_MAX (0x7fffffff)

  /** Indices from index returned by su_root_register() to tables below. 
   *
   * Free elements are negative. Free elements form a list, value of free
   * element is (0 - index of next free element).
   *
   * First element sup_indices[0] points to first free element. 
   */
  int             *sup_indices;

  int             *sup_reverses; /** Reverse index */
  su_wakeup_f     *sup_wait_cbs; 
  su_wakeup_arg_t**sup_wait_args; 
  su_root_t      **sup_wait_roots; 

  su_wait_t       *sup_waits; 
};


su_port_vtable_t const su_osx_port_vtable[1] =
  {{
      /* su_vtable_size: */ sizeof su_osx_port_vtable,
      su_pthread_port_lock,
      su_pthread_port_unlock,
      su_base_port_incref,
      su_osx_port_decref,
      su_base_port_gsource,
      su_osx_port_send,
      su_osx_port_register,
      su_osx_port_unregister,
      su_osx_port_deregister,
      su_osx_port_unregister_all,
      su_osx_port_eventmask,
      su_osx_port_run,
      su_osx_port_break,
      su_osx_port_step,
      su_pthread_port_thread,
      su_base_port_add_prepoll,
      su_base_port_remove_prepoll,
      su_base_port_timers,
      su_osx_port_multishot,
      su_osx_port_wait_events,
      su_base_port_getmsgs,
      su_base_port_getmsgs_from,
      su_osx_port_name,
      su_base_port_start_shared,
      su_pthread_port_wait,
      su_pthread_port_execute,
    }};

/* XXX - mela static void su_osx_port_destroy(su_port_t *self); */

/** Create a reactor object.
 *
 * Allocate and initialize the instance of su_root_t.
 *
 * @param magic     pointer to user data
 *
 * @return A pointer to allocated su_root_t instance, NULL on error.
 *
 * @NEW_1_12_4.
 */
su_root_t *su_root_osx_runloop_create(su_root_magic_t *magic)
{
  return su_root_create_with_port(magic, su_osx_runloop_create());
}

void osx_enabler_cb(CFSocketRef s, 
		    CFSocketCallBackType type, 
		    CFDataRef address, 
		    const void *data, 
		    void *info)
{
  CFRunLoopRef  rl;
  struct osx_magic  *magic = (struct osx_magic *) info;
  su_port_t    *self = magic->o_port;
  su_duration_t tout = 0;
  su_time_t     now = su_now();
  
  rl = CFRunLoopGetCurrent();

  if (self->sup_base->sup_running) {
    
    if (self->sup_base->sup_prepoll)
      self->sup_base->sup_prepoll(self->sup_base->sup_pp_magic, self->sup_base->sup_pp_root);
    
    if (self->sup_base->sup_head)
      su_base_port_getmsgs(self);
    
    if (self->sup_base->sup_timers)
      su_timer_expire(&self->sup_base->sup_timers, &tout, now);
  }
  
  CFRunLoopWakeUp(rl);
}


/**@internal
 *
 * Allocates and initializes a message port.
 *
 * @return
 *   If successful a pointer to the new message port is returned, otherwise
 *   NULL is returned.  
 */
su_port_t *su_osx_runloop_create(void)
{
  su_port_t *self = su_home_new(sizeof *self);

  if (!self)
    return self;

  enter;

  if (su_home_destructor(su_port_home(self), su_osx_port_deinit) < 0)
    return su_home_unref(su_port_home(self)), NULL;

  self->sup_multishot = SU_ENABLE_MULTISHOT_POLL;

  if (su_socket_port_init(self->sup_base, su_osx_port_vtable) == 0) {
    self->osx_magic->o_port = self;
    self->sup_observer_cntx->info = self->osx_magic;
    self->sup_observer =
      CFRunLoopObserverCreate(NULL, 
			      kCFRunLoopAfterWaiting | kCFRunLoopBeforeWaiting,
			      TRUE, 0, cf_observer_cb, self->sup_observer_cntx);
#if 0
    CFRunLoopAddObserver(CFRunLoopGetCurrent(),
			 self->sup_observer,
			 kCFRunLoopDefaultMode);
#endif
  }
  else
    return su_home_unref(su_port_home(self)), NULL;

  return self;
}

static
void cf_observer_cb(CFRunLoopObserverRef observer, 
		    CFRunLoopActivity activity, 
		    void *info)
{
  CFRunLoopRef  rl;
  struct osx_magic  *magic = (struct osx_magic *) info;
  su_port_t    *self = magic->o_port;
  su_duration_t tout = 0;
  su_time_t     now = su_now();

  rl = CFRunLoopGetCurrent();

  if (self->sup_base->sup_running) {

    if (self->sup_base->sup_prepoll)
      self->sup_base->sup_prepoll(self->sup_base->sup_pp_magic, self->sup_base->sup_pp_root);
    
    if (self->sup_base->sup_head)
      su_port_getmsgs(self);
    
    if (self->sup_base->sup_timers)
      su_timer_expire(&self->sup_base->sup_timers, &tout, now);
  } else
    SU_DEBUG_9(("cf_observer_cb(): PORT IS NOT RUNNING!\n"));

  CFRunLoopWakeUp(rl);
  
  return;
}

/** @internal Destroy a port. */
static void su_osx_port_deinit(void *arg)
{
  su_port_t *self = arg;

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

  su_socket_port_deinit(self->sup_base);
}

static
CFSocketCallBackType map_poll_event_to_cf_event(int events)
{
  CFSocketCallBackType type = 0;

  if (events & SU_WAIT_IN)
    type |= kCFSocketReadCallBack;
  
  if (events & SU_WAIT_OUT)
    type |= kCFSocketWriteCallBack;
  
#if 0
  if (events & SU_WAIT_CONNECT)
    type |= kCFSocketConnectCallBack;
  
  if (events & SU_WAIT_ACCEPT)
    type |= kCFSocketAcceptCallBack;
#endif

  return type;
}


#if 0
static
int map_cf_event_to_poll_event(CFSocketCallBackType type)
{
  int event = 0;

  if (type & kCFSocketReadCallBack)
    event |= SU_WAIT_IN;
  
  if (type & kCFSocketWriteCallBack)
    event |= SU_WAIT_OUT;
  
  if (type & kCFSocketConnectCallBack)
    event |= SU_WAIT_CONNECT;
  
  if (type & kCFSocketAcceptCallBack)
    event |= SU_WAIT_ACCEPT;

  return event;
}
#endif

static
void su_osx_port_socket_cb(CFSocketRef s, 
			   CFSocketCallBackType type, 
			   CFDataRef address, 
			   const void *data, 
			   void *info)
{
  struct osx_magic *magic = (struct osx_magic *) info;
  su_port_t        *self = magic->o_port;
  int               curr = magic->o_current;
  su_duration_t tout = 0;
  
#if SU_HAVE_POLL
  {
    su_root_t *root;
    su_wait_t *waits = self->sup_waits;
    int n = self->sup_indices[curr];
    
    assert(self->sup_reverses[n] == curr);
    
    SU_DEBUG_9(("socket_cb(%p): count %u index %d\n", self->sup_sources[n], magic->o_count, curr));
    
    waits[n].revents = map_poll_event_to_cf_event(type);

    root = self->sup_wait_roots[n];
    self->sup_wait_cbs[n](root ? su_root_magic(root) : NULL, 
			  &waits[n], 
			  self->sup_wait_args[n]);
    
    if (self->sup_base->sup_running) {
      su_port_getmsgs(self);
      
      if (self->sup_base->sup_timers)
	su_timer_expire(&self->sup_base->sup_timers, &tout, su_now());

      if (self->sup_base->sup_head)
	tout = 0;

      /* CFRunLoopWakeUp(CFRunLoopGetCurrent()); */
    }
    
    /* Tell to run loop an su socket fired */
    self->sup_source_fired = 1;
  }