su_root.c

sip协议栈

  if (self->sur_deinit) {
    su_root_deinit_f deinit = self->sur_deinit;
    su_root_magic_t *magic = self->sur_magic;
    self->sur_deinit = NULL;
    deinit(self, magic);
  }

  if (self->sur_port) {
    int n_w = su_port_unregister_all(self->sur_port, self);
    int n_t = su_timer_reset_all(su_task_timers(self->sur_task), self->sur_task);

    if (n_w || n_t)
      SU_DEBUG_1(("su_root_deinit: "
		  "%u registered waits, %u timers\n", n_w, n_t));
  }

  SU_TASK_ZAP(self->sur_parent, su_root_deinit);
  SU_TASK_ZAP(self->sur_task, su_root_deinit);
}

/** Set the context pointer.
 *
 *  Set the context pointer (magic) of a root object.
 *
 * @param self      pointer to a root object
 * @param magic     pointer to user data
 *
 * @retval 0  when successful,
 * @retval -1 upon error.
 */
int su_root_set_magic(su_root_t *self, su_root_magic_t *magic)
{
  assert(SU_ROOT_OWN_THREAD(self));

  if (self) {
    self->sur_magic = magic;
  }

  return 0;
}

/** Set threading option.
 *
 *   Controls whether su_clone_start() creates a new thread.
 *
 * @param self      pointer to a root object
 * @param enable    if true, enable threading, if false, disable threading
 *
 * @return True if threading is enabled.
 */
int su_root_threading(su_root_t *self, int enable)
{
  if (self) {
    assert(SU_ROOT_OWN_THREAD(self));

#if SU_HAVE_PTHREADS
    self->sur_threading = enable = enable != 0;
    return enable;
#endif
  }

  return 0;
}

/** Get context pointer.
 *
 *  The function su_root_magic() returns the user context pointer that was
 *  given input to su_root_create() or su_root_set_magic().
 *
 * @param self      pointer to a root object
 *
 * @return A pointer to user data
 */
su_root_magic_t *su_root_magic(su_root_t *self)
{
  return self ? self->sur_magic : NULL;
}

/** Get a GSource */
struct _GSource *su_root_gsource(su_root_t *self)
{
  return self ? su_port_gsource(self->sur_port) : NULL;
}

/** Register a su_wait_t object. 
 *
 *  The function su_root_register() registers a su_wait_t object. The wait
 *  object, a callback function and a argument are stored to the root
 *  object. The callback function is 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 root object
 * @param wait      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 Nonzero index of the wait object, or -1 upon an error.
 */
int su_root_register(su_root_t *self,
		     su_wait_t *wait,
		     su_wakeup_f callback,
		     su_wakeup_arg_t *arg,
		     int priority)
{
  assert(self && self->sur_port);

  if (!self || !self->sur_port)
    return -1;

  return su_port_register(self->sur_port, self, wait, callback, arg, priority);
}

/** Unregister a su_wait_t object.
 *
 *  The function su_root_unregister() unregisters a su_wait_t object. The
 *  wait object, a callback function and a argument are removed from the
 *  root object.
 *
 * @param self      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)
 *
 * @return Nonzero index of the wait object, or -1 upon an error.
 */
int su_root_unregister(su_root_t *self,
		       su_wait_t *wait,
		       su_wakeup_f callback, /* XXX - ignored */
		       su_wakeup_arg_t *arg)
{
  assert(self && self->sur_port);

  if (!self || !self->sur_port)
    return -1;

  return su_port_unregister(self->sur_port, self, wait, callback, arg);
}

/** Remove a su_wait_t registration.
 *
 *  The function su_root_deregister() deregisters a su_wait_t object. The
 *  wait object, a callback function and a argument are removed from the
 *  root object. The wait object is destroyed.
 *
 * @param self      pointer to root object
 * @param i         registration index
 *
 * @return Index of the wait object, or -1 upon an error.
 */
int su_root_deregister(su_root_t *self, int i)
{
  if (i == 0 || i == -1)
    return -1;

  assert(self && self->sur_port);

  if (!self || !self->sur_port)
    return -1;

  return su_port_deregister(self->sur_port, i);
}

/** Set mask for a registered event.
 *
 * The function su_root_eventmask() sets the mask describing events that can
 * signal the registered callback.
 *
 * @param self   pointer to root object
 * @param index  registration index
 * @param socket socket
 * @param events new event mask
 *
 * @retval 0 when successful,
 * @retval -1 upon an error.
 */
int su_root_eventmask(su_root_t *self, int index, int socket, int events)
{
  assert(self && self->sur_port);

  if (!self || !self->sur_port)
    return -1;

  return su_port_eventmask(self->sur_port, index, socket, events);
}

/** Set multishot mode.
 *
 * The function su_root_multishot() enables, disables or queries the
 * multishot mode for the root. The multishot mode determines how the events
 * are scheduled by root. If multishot mode is enabled, root serves all the
 * sockets that have received network events. If it is disables, only first
 * socket event is served.
 *
 * @param self      pointer to root 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_root_multishot(su_root_t *self, int multishot)
{
  if (self && self->sur_port) {
    return su_port_multishot(self->sur_port, multishot);
  } else {
    return (errno = EINVAL), -1;
  }
}


/** Run event and message loop.
 *
 * The function su_root_run() runs the root main loop. The root loop 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,
 * and returns waiting.
 *
 * This function returns when su_root_break() is called from a callback.
 *
 * @param self      pointer to root object
 *
 */
void su_root_run(su_root_t *self)
{
  assert(self && self->sur_port);

  if (self && self->sur_port)
    su_port_run(self->sur_port);
}

/** Terminate event loop.
 *
 *   The function su_root_break() is used to terminate execution of
 *   su_root_run(). It can be called from a callback function.
 *
 * @param self      pointer to root object
 */
void su_root_break(su_root_t *self)
{
  assert(self && self->sur_port);

  if (self && self->sur_port)
    su_port_break(self->sur_port);
}

/** Process events, timers and messages.
 *
 *   The function su_root_step() 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 tout
 *   milliseconds is elapsed.
 *
 * @param self      pointer to root object
 * @param tout      timeout in milliseconds
 *
 * @return Milliseconds to the next invocation of timer, or SU_WAIT_FOREVER
 *         if there are no active timers.
 */
su_duration_t su_root_step(su_root_t *self, su_duration_t tout)
{
  assert(self && self->sur_port);

  return su_port_step(self->sur_port, tout);
}

/**Run event and message loop for given duration.
 *
 * The function su_root_sleep() runs event loop for @a duration milliseconds.
 * The event loop waits for wait objects and the timers associated with the
 * @a root object.  When any wait object is signaled, timer is expired, or
 * message is received, it invokes the callbacks and returns waiting.
 *
 * @param self      pointer to root object
 * @param duration  milliseconds to run event loop
 */
su_duration_t su_root_sleep(su_root_t *self, su_duration_t duration)
{
  su_duration_t retval, accrued = 0;
  su_time_t started = su_now();

  assert(self && self->sur_port);

  do {
    retval = su_port_step(self->sur_port, duration - accrued);
    accrued = su_duration(su_now(), started);
  } while (accrued < duration);

  return retval;
}

/** Get task reference.
 *
 *   The function su_root_task() is used to retrieve the task reference
 *   (PId) related with the root object.
 *
 * @param self      a pointer to a root object
 *
 * @return The function su_root_task() returns a reference to the task
 *         object.
 */
_su_task_r su_root_task(su_root_t const *self)
{
  if (self)
    return self->sur_task;
  else
    return su_task_null;
}

/** Get parent task reference.
 *
 *   The function su_root_parent() is used to retrieve the task reference
 *   (PId) of the parent task.
 *
 * @param self      a pointer to a root object
 *
 * @return The function su_root_parent() returns a reference to the parent
 *         task object.
 */
_su_task_r su_root_parent(su_root_t const *self)
{
  if (self)
    return self->sur_parent;
  else
    return su_task_null;
}

/** Add a pre-poll callback. */
int su_root_add_prepoll(su_root_t *root, 
			su_prepoll_f *callback, 
			su_prepoll_magic_t *magic)
{
  if (root == NULL || root->sur_port == NULL)
    return -1;

  return su_port_add_prepoll(root->sur_port, root, callback, magic);
}

/** Remove a pre-poll callback */
int su_root_remove_prepoll(su_root_t *root)
{
  if (root == NULL || root->sur_port == NULL)
    return -1;

  return su_port_remove_prepoll(root->sur_port, root);
}

/* ========================================================================
 * su_clone_t
 */

/* - su_clone_forget() */

#if SU_HAVE_PTHREADS
struct clone_args
{
  su_root_t      * self;
  su_root_init_f   init;
  su_root_deinit_f deinit;
  pthread_mutex_t  mutex;
  pthread_cond_t   cv;
  int              retval;
  su_msg_r         clone;
  su_root_t const *parent;
};

static void su_clone_report2(su_root_magic_t *m,
			     su_msg_r msg,
			     su_cloned_t *sc);

static void su_clone_signal_parent(void *varg)
{
  struct clone_args *arg = (struct clone_args *)varg;

  pthread_mutex_lock(&arg->mutex);
  pthread_cond_signal(&arg->cv);
  pthread_mutex_unlock(&arg->mutex);
}

/** Message function for clone message.
 *
 * This calls the clone task deinitialization function, which should make
 * sure that no more messages are sent by clone task.
 *
 * @sa su_clone_wait()
 */
static void su_clone_break(su_root_magic_t *m,
			   su_msg_r msg,
			   su_cloned_t *sc)
{
  su_root_t *root = sc->sc_root;

  root->sur_deiniting = 1;

  if (root->sur_deinit) {
    su_root_deinit_f deinit = root->sur_deinit;
    su_root_magic_t *magic = root->sur_magic;
    root->sur_deinit = NULL;
    deinit(root, magic);
  }
}

/** Delivery report function for clone message.
 *
 * This is executed by parent task. This is the last message sent by clone task.
 */
static void su_clone_report(su_root_magic_t *m,
			    su_msg_r msg,
			    su_cloned_t *sc)
{
  su_msg_report(msg, su_clone_report2);
}

/** Back delivery report function for clone message.
 *
 * This is executed by clone task. It completes the three way handshake and
 * it is used to signal clone that it can destroy its port.
 */
static void su_clone_report2(su_root_magic_t *m,
			    su_msg_r msg,
			    su_cloned_t *sc)
{
  su_root_break(sc->sc_root);
  if (sc->sc_wait)
    *sc->sc_wait = 0;
}

static void *su_clone_main(void *varg)
{
  struct clone_args *arg = (struct clone_args *)varg;
  su_root_t *self = arg->self;
  su_port_t *port;
  su_cloned_t *sc;

  pthread_cleanup_push(su_clone_signal_parent, varg);

#if SU_HAVE_WINSOCK
  su_init();
#endif

  port = su_port_create();
  if (!port)
    pthread_exit(NULL);
  su_port_threadsafe(port);
  SU_PORT_INCREF(port, su_clone_main);

  /* Change task ownership */
  SU_PORT_INCREF(self->sur_task->sut_port = port, su_clone_main);
  self->sur_task->sut_root = self;

  if (su_msg_create(arg->clone,
		    self->sur_task, su_root_task(arg->parent),
		    su_clone_break, sizeof(self)) != 0) {
    su_port_decref(self->sur_port, "su_clone_main");
    self->sur_port = NULL;
    pthread_exit(NULL);
  }

  su_msg_report(arg->clone, su_clone_report);

  sc = su_msg_data(arg->clone);
  sc->sc_root = self;
  sc->sc_tid = pthread_self();

  pthread_mutex_init(sc->sc_pause, NULL);
  pthread_cond_init(sc->sc_resume, NULL);
  pthread_mutex_lock(sc->sc_pause);

  if (arg->init && arg->init(self, self->sur_magic) != 0) {
    if (arg->deinit)
      arg->deinit(self, self->sur_magic);
    su_msg_destroy(arg->clone);
    su_port_decref(self->sur_port, "su_clone_main");
    self->sur_port = NULL;
    pthread_exit(NULL);
  }

  arg->retval = 0;

  pthread_cleanup_pop(1);  /* signal change of ownership */

  su_root_run(self);   /* Do the work */

  su_root_destroy(self);   /* Cleanup root */   

  SU_PORT_ZAPREF(port, su_clone_main);

#if SU_HAVE_WINSOCK
  su_deinit();
#endif

  return NULL;
}
#endif

static void su_clone_xyzzy(su_root_magic_t *m,
			   su_msg_r msg,
			   su_cloned_t *sc)
{
  su_root_destroy(sc->sc_root);
  if (sc->sc_wait)
    *sc->sc_wait = 0;
}

/** Start a clone task.
 *
 * The function su_clone_start() allocates and initializes a sub-task. 
 * Depending on the settings, a separate thread may be created to execute
 * the sub-task. The sub-task is represented by clone handle to the rest of
 * the application. The function su_clone_start() returns the clone handle
 * in @a return_clone. The clone handle is used to communicate with the
 * newly created clone task using messages.
 *
 * A new #su_root_t object is created for the sub-task with the @a magic as
 * the root context pointer. Because the sub-task may or may not have its
 * own thread, all its activity must be scheduled via this root object. In