su_osx_runloop.c

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

      indices[index] = indices[0];
      indices[0] = -index;
      continue;
    }

    if (i != j) {
      indices[index] = j;

      CFSocketInvalidate(self->sup_sockets[j]);
      CFRelease(self->sup_sockets[j]);
      CFRunLoopRemoveSource(rl, sources[j], kCFRunLoopDefaultMode);
      CFRelease(sources[j]);

      reverses[j]   = reverses[i];
      sources[j]    = sources[i];
      sockets[j]    = sockets[i];
      waits[j]      = waits[i];
      wait_cbs[j]   = wait_cbs[i];
      wait_args[j]  = wait_args[i];
      wait_roots[j] = wait_roots[i];
    }
    
    j++;
  }

  /* Prepare for removing CFSources */
  for (i = j; i < N; i++) {
    reverses[i] = -1;

    CFSocketInvalidate(self->sup_sockets[i]);
    CFRelease(self->sup_sockets[i]);
    CFRunLoopRemoveSource(rl, sources[i], kCFRunLoopDefaultMode);
    CFRunLoopSourceInvalidate(sources[i]);

    sources[i] = NULL;
    sockets[i] = NULL;
    wait_cbs[i] = NULL;
    wait_args[i] = NULL;
    wait_roots[i] = NULL;
  }
  memset(&waits[j], 0, (char *)&waits[N] - (char *)&waits[j]);

  /* Tell run loop things have changed */
  CFRunLoopWakeUp(rl);

  self->sup_n_waits = j;
  self->sup_registers++;

  return N - j;
}

/**Set mask for a registered event. @internal
 *
 * The function su_osx_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_osx_port_eventmask(su_port_t *self, int index, int socket, int events)
{
  int n, ret;

  assert(self);
  // XXX - mela assert(SU_OSX_PORT_OWN_THREAD(self));

  if (index <= 0 || index > self->sup_size_waits)
    return su_seterrno(EBADF);
  n = self->sup_indices[index];
  if (n < 0)
    return su_seterrno(EBADF);

  ret = su_wait_mask(&self->sup_waits[n], socket, events);

  CFSocketSetSocketFlags(self->sup_sockets[n],
			 map_poll_event_to_cf_event(events));
  
  return ret;
}

/** @internal
 *
 *  Copies the su_wait_t objects from the port. The number of wait objects
 *  can be found out by calling su_osx_port_query() with @a n_waits as zero.
 * 
 * @note This function is called only by friends.
 *
 * @param self     - pointer to port object
 * @param waits    - pointer to array to which wait objects are copied
 * @param n_waits  - number of wait objects fitting in array waits
 *
 * @return Number of wait objects, or 0 upon an error.
 */
unsigned su_osx_port_query(su_port_t *self, su_wait_t *waits, unsigned n_waits)
{
  unsigned n;

  // XXX - mela assert(SU_OSX_PORT_OWN_THREAD(self));

  n = self->sup_n_waits;

  if (n_waits != 0) {
    if (waits && n_waits >= n)
      memcpy(waits, self->sup_waits, n * sizeof(*waits));
    else
      n = 0;
  }

  return n;
}

/** @internal Enable multishot mode.
 *
 * The function su_osx_port_multishot() 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 disables, 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
 */
int su_osx_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;
}

#if 0
/** @internal Enable threadsafe operation. */
static
int su_osx_port_threadsafe(su_port_t *port)
{
  return su_home_threadsafe(port->sup_home);
}
#endif

/** Prepare root to be run on OSX Run Loop.
 *
 * Sets #su_root_t object to be callable by the application's run loop. This
 * function is to be used instead of su_root_run() for OSX applications
 * using Core Foundation's Run Loop.
 *
 * The function su_root_osx_prepare_run() returns immmediately.
 * 
 * @param root     pointer to root object
 * 
 * @NEW_1_12_4.
 */
void su_root_osx_prepare_run(su_root_t *root)
{
  su_port_t *self = root->sur_task->sut_port;
  CFRunLoopRef rl;
  su_duration_t tout = 0;

  // XXX - mela assert(SU_OSX_PORT_OWN_THREAD(self));

  enter;

  self->sup_base->sup_running = 1;
  rl = CFRunLoopGetCurrent();

  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, su_now());

  if (!self->sup_base->sup_running)
    return;

  CFRetain(rl);
  self->sup_main_loop = rl;

  return;
}

/** @internal Main loop.
 * 
 * The function @c su_osx_port_run() waits for wait objects and the timers
 * associated with the port object.  When any wait object is signaled or
 * timer is expired, it invokes the callbacks, and returns waiting.
 * 
 * The function @c su_osx_port_run() runs until @c su_osx_port_break() is called
 * from a callback.
 * 
 * @param self     pointer to port object
 * 
 */
void su_osx_port_run(su_port_t *self)
{
  CFRunLoopRef rl;
  su_duration_t tout = 0;

  // XXX - mela assert(SU_OSX_PORT_OWN_THREAD(self));

  enter;

  self->sup_base->sup_running = 1;
  rl = CFRunLoopGetCurrent();

  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, su_now());

  if (!self->sup_base->sup_running)
    return;

  CFRetain(rl);
  self->sup_main_loop = rl;

  /* if there are messages do a quick wait */
  if (self->sup_base->sup_head)
    tout = 0;

  CFRunLoopRun();

  self->sup_main_loop = NULL;

}

#if tuning
/* This version can help tuning... */
void su_osx_port_run_tune(su_port_t *self)
{
  int i;
  int timers = 0, messages = 0, events = 0;
  su_duration_t tout = 0, tout0;
  su_time_t started = su_now(), woken = started, bedtime = woken;

  // XXX - mela assert(SU_OSX_PORT_OWN_THREAD(self));

  for (self->sup_base->sup_running = 1; self->sup_base->sup_running;) {
    tout0 = tout, tout = 2000;

    timers = 0, messages = 0;

    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)
      messages = su_port_getmsgs(self);

    if (self->sup_base->sup_timers)
      timers = su_timer_expire(&self->sup_base->sup_timers, &tout, su_now());

    if (!self->sup_base->sup_running)
      break;

    if (self->sup_base->sup_head)      /* if there are messages do a quick wait */
      tout = 0;

    bedtime = su_now();

    events = su_osx_port_wait_events(self, tout);

    woken = su_now();

    if (messages || timers || events)
      SU_DEBUG_1(("su_osx_port_run(%p): %.6f: %u messages %u timers %u "
		  "events slept %.6f/%.3f\n",
		  self, su_time_diff(woken, started), messages, timers, events,
		  su_time_diff(woken, bedtime), tout0 * 1e-3));

    if (!self->sup_base->sup_running)
      break;
  }
}
#endif

/** @internal
 * The function @c su_osx_port_break() is used to terminate execution of @c
 * su_osx_port_run(). It can be called from a callback function.
 * 
 * @param self     pointer to port
 * 
 */
void su_osx_port_break(su_port_t *self)
{
  if (self->sup_main_loop)
    CFRunLoopStop(self->sup_main_loop);

  self->sup_base->sup_running = 0; 
}

/** @internal
 * The function @c su_osx_port_wait_events() is used to poll() for wait objects
 *
 * @param self     pointer to port
 * @param tout     timeout in milliseconds
 *
 * @return number of events handled
 */
static
int su_osx_port_wait_events(su_port_t *self, su_duration_t tout)
{
  int i, events = 0;
  su_wait_t *waits = self->sup_waits;
  unsigned n = self->sup_n_waits;
#if HAVE_POLL
  unsigned version = self->sup_registers;
#endif
  su_root_t *root;

  i = su_wait(waits, n, tout);

  if (i >= 0 && (unsigned)i < n) {
#if HAVE_POLL			
    /* poll() can return events for multiple wait objects */
    if (self->sup_multishot) {
      for (; i < n; i++) {
        if (waits[i].revents) {
          root = self->sup_wait_roots[i];
          self->sup_wait_cbs[i](root ? su_root_magic(root) : NULL,
                                &waits[i],
                                self->sup_wait_args[i]);
          events++;
          /* Callback function used su_register()/su_deregister() */
          if (version != self->sup_registers)
            break;
        }
      }
    }
#else /* !HAVE_POLL */
    if (0) {
    }
#endif
    else {
      root = self->sup_wait_roots[i];
      self->sup_wait_cbs[i](root ? su_root_magic(root) : NULL,
                            &self->sup_waits[i],
                            self->sup_wait_args[i]);
      events++;
    }
  }

  return events;
}

/** @internal Block until wait object is signaled or timeout.
 *
 * This function waits for wait objects and the timers associated with 
 * the root object.  When any wait object is signaled or timer is
 * expired, it invokes the callbacks. 
 * 
 *   This function returns when a callback has been invoked or @c tout
 *   milliseconds is elapsed. 
 *
 * @param self     pointer to port
 * @param tout     timeout in milliseconds
 * 
 * @return
 *   Milliseconds to the next invocation of timer, or @c SU_WAIT_FOREVER if
 *   there are no active timers.
 */
su_duration_t su_osx_port_step(su_port_t *self, su_duration_t tout)
{
  CFRunLoopRef rl;
  su_time_t now = su_now();
  CFAbsoluteTime start;
  int ret, timeout = tout > INT32_MAX ? INT32_MAX : tout;

  rl = CFRunLoopGetCurrent();

  if (!rl)
    return -1;

  CFRunLoopWakeUp(rl);

  if (tout < timeout)
    timeout = tout;

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

  /* if there are messages do a quick wait */
  if (self->sup_base->sup_head)
    tout = 0;

  ret = CFRunLoopRunInMode(kCFRunLoopDefaultMode,
			   tout/1000000.0,
			   true);

  CFRunLoopWakeUp(rl);

  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, su_now());

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

  return tout;
}