su_root.c

sip协议栈

/*
 * 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_root.c 
 * OS-independent synchronization interface. 
 * @internal
 *
 * @author Pekka Pessi <[e-mail]>
 * 
 * @date Created: Tue Sep 14 15:51:04 1999 ppessi
 */

#include "config.h"

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

#include "sofia-sip/su.h"

#if SU_HAVE_PTHREADS
#include <pthread.h>
#endif

struct su_root_s;

typedef struct su_cloned_s {
  struct su_root_s *sc_root;
  int *sc_wait;
#if SU_HAVE_PTHREADS
  pthread_t  sc_tid;
  pthread_mutex_t sc_pause[1];
  pthread_cond_t sc_resume[1];
  int sc_paused;
#endif  
} su_cloned_t;

#define SU_ROOT_MAGIC_T struct su_root_magic_s
#define SU_WAKEUP_ARG_T struct su_wakeup_arg_s
#define SU_TIMER_ARG_T  struct su_timer_arg_s
#define SU_CLONE_T      su_msg_t
#define SU_MSG_ARG_T    struct su_cloned_s

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

/**@ingroup su_wait
 * 
 * @page su_root_t Tasks and root objects
 *
 * A task is the basic execution unit for the Sofia event-driven programming
 * model. According to the model, the program can ask that the event loop
 * invokes a callback function when a certain event occurs. Such events
 * include @ref su_root_register "I/O activity", @ref su_timer_t "timers" or
 * a @ref su_msg_t "message" from other task. The event loop is run with
 * function su_root_run() or su_root_step().
 *
 * Root object gives access to the task control. The root object represents
 * the task to the code running within task. Through the root, the task code
 * can access its context object (magic) and thread-synchronization features
 * like wait objects, timers, and messages.
 *
 * When a message is sent between tasks, a task reference #su_task_r is used
 * to reprent the task address. Reference counting is used to make sure that
 * the task references stay valid.
 *
 * The public API contains following functions:
 *    - su_root_create() [Do not call from cloned task]
 *    - su_root_destroy() [Do not call from cloned task]
 *    - su_root_magic()
 *    - su_root_register()
 *    - su_root_deregister()
 *    - su_root_unregister()
 *    - su_root_threading()
 *    - su_root_run() [Do not call from cloned task]
 *    - su_root_break() [Do not call from cloned task]
 *    - su_root_step() [Do not call from cloned task]
 *    - su_root_task()
 *
 * New tasks can be created via su_clone_start() function.
 */

/**@ingroup su_wait 
 * 
 * @page su_root_register Registering Wait Objects
 *
 * When application expects I/O events, it can create a wait object and
 * register it, a callback function and a context pointer to the #su_root_t
 * object using the su_root_register() function. Whenever the wait object
 * receives an event, the registered @link ::su_wakeup_f callback function
 * @endlink is invoked.
 *
 * When successful, the su_root_register() returns an small non-negative
 * integer representing the registration. The registration can be
 * manipulated with su_root_eventmask() function, for instance, when sending
 * through a socket block, the application can add SU_WAIT_OUT event to the
 * mask.
 *
 * The registration can be removed using su_root_deregister() function.
 */

/**@ingroup su_wait
 * 
 * Contains hint of number of sockets supported by su_root_t */ 
int su_root_size_hint = 64;

/* =========================================================================
 * Tasks
 */

su_task_r const su_task_null = SU_TASK_R_INIT;

#define SU_TASK_ZAP(t, f) \
  while (t->sut_port) { \
   SU_PORT_DECREF(t->sut_port, f); t->sut_port = NULL; break; }

#define SU_TASK_ZAPP(t, f) \
  do { if (t->sut_port) { \
   SU_PORT_DECREF(t->sut_port, f); t->sut_port = NULL; } \
   t->sut_root = NULL; } while(0)

/**
 * Initialize a task handle with su_task_null.
 *
 * @param task task handle
 *
 * @return A reference to the initialized task handle.
 */
_su_task_r su_task_init(su_task_r task)
{
  assert(task);

  memset(task, 0, sizeof(task));
  return task;
}

/**
 * Destroy a task handle
 *
 * @param task task handle
 */
void su_task_deinit(su_task_r task)
{
  assert(task);

  SU_TASK_ZAP(task, su_task_deinit);
  task->sut_root = NULL;
}

/**
 * Create a new task handle.
 *
 * @param task task reference
 * @param root pointer to root object
 * @param port pointer to port object
 *
 * @return New task handle.
 */
_su_task_r su_task_new(su_task_r task, su_root_t *root, su_port_t *port)
{
  assert(task);

  task->sut_root = root;
  if ((task->sut_port = port)) {
    SU_PORT_INCREF(port, su_task_new);
  }
  return task;
}

/**
 * Duplicates a task handle.
 *
 * @param  dst      destination task reference
 * @param  src      source task reference
 */
void su_task_copy(su_task_r dst, su_task_r const src)
{
  su_port_t *port;
  
  assert(src); assert(dst);

  SU_TASK_ZAP(dst, su_task_copy);

  port = src->sut_port;
  if (port) {
    SU_PORT_INCREF(port, su_task_copy);
  }

  dst[0] = src[0];
}

#define SU_TASK_COPY(d, s, by) (void)((d)[0]=(s)[0], \
  (s)->sut_port?(void)SU_PORT_INCREF(s->sut_port, by):(void)0)

/**
 * Moves a task handle.
 *
 * @param  dst      destination task reference
 * @param  src      source task reference
 */
void su_task_move(su_task_r dst, su_task_r src)
{
  SU_TASK_ZAP(dst, su_task_move);
  dst[0] = src[0];
  src->sut_port = 0;
  src->sut_root = 0;
}

/**
 * Compare two tasks with each other.
 * 
 * @param a  First task
 * @param b  Second task
 * 
 * @return 
 *   The function @c su_task_cmp() returns negative number, if a < b,
 *   positive number, if a > b, and 0, if a == b.
 */
int  su_task_cmp(su_task_r const a, su_task_r const b)
{
  int retval = a->sut_port - b->sut_port;

  return retval ? retval : (char *)a->sut_root - (char *)b->sut_root;
}

/**
 * Tests if a task is running.
 *
 * @param task  task handle
 *
 * @retval true (nonzero) if task is not stopped, 
 * @retval zero if it is null or stopped.
 */
int su_task_is_running(su_task_r const task)
{
  return 
    task && 
    task->sut_port && 
    task->sut_root;
}

/** @internal
 * Attach a root object to the task handle.
 *
 * @param self task handle
 * @param root pointer to the root object
 * 
 * @retval 0 if successful,
 * @retval -1 otherwise.
 */
int su_task_attach(su_task_r self, su_root_t *root)
{
  if (self->sut_port) {
    self->sut_root = root;
    return 0;
  }
  else 
    return -1;
}

/** 
 * Get root pointer attached to a task handle.
 *
 * @param self task handle
 *
 * @return 
 * The function @c su_task_root() returns a pointer to root object attached
 * to the task handle, or NULL if no root object has been attached.
 */
su_root_t *su_task_root(su_task_r const self)
{
  if (self->sut_port) return self->sut_root; else return NULL;
}

#if 0
/** @internal
 * Detach a root pointer from task handle.
 * @bug Not used anymore.
 */
int su_task_detach(su_task_r self)
{
  self->sut_root = NULL;
  return 0;
}
#endif

/**
 * Return the timer list associated with given task.
 * 
 * @param task task handle
 *
 * @return The function @c su_task_timers() returns a timer list of the
 * task. If there are no timers, it returns NULL.
 */
su_timer_t **su_task_timers(su_task_r const task)
{
  return task ? su_port_timers(task->sut_port) : NULL;
}

#if SU_HAVE_PTHREADS

struct su_task_execute
{
  pthread_mutex_t mutex[1];
  pthread_cond_t cond[1];
  int (*function)(void *);
  void *arg;
  int *return_value;
};

static void _su_task_execute(su_root_magic_t *m,
			     su_msg_r msg,
			     su_msg_arg_t *a)
{
  struct su_task_execute *frame = (void *)a;
  pthread_mutex_lock(frame->mutex);
  *frame->return_value = frame->function(frame->arg);
  pthread_cond_signal(frame->cond);
  pthread_mutex_unlock(frame->mutex);
}

#endif

/** Execute by task thread
 *
 * @retval 0 if successful
 * @retval -1 upon an error
 */
int su_task_execute(su_task_r const task,
		    int (*function)(void *), void *arg,
		    int *return_value)
{
  int value;

  if (!su_port_own_thread(task->sut_port)) {
#if SU_HAVE_PTHREADS
    su_msg_r m = SU_MSG_R_INIT;
    struct su_task_execute *frame;

    if (su_msg_create(m, task, su_task_null,
		      _su_task_execute, (sizeof *frame)) < 0)
      return -1;

    frame = (void *)su_msg_data(m);
    pthread_mutex_init(frame->mutex, NULL);
    pthread_cond_init(frame->cond, NULL);
    frame->function = function;
    frame->arg = arg;
    frame->return_value = &value;

    pthread_mutex_lock(frame->mutex);

    if (su_msg_send(m) < 0) {
      su_msg_destroy(m);
      return -1;
    }

    pthread_cond_wait(frame->cond, frame->mutex);
#else
    return -1;
#endif
  }
  else
    value = function(arg);
  if (return_value)
    *return_value = value;

  return 0;
}

_su_task_r su_task_new(su_task_r task, su_root_t *root, su_port_t *port);
int su_task_attach(su_task_r self, su_root_t *root);
int su_task_detach(su_task_r self);

int su_timer_reset_all(su_timer_t **t0, su_task_r);

/**@ingroup su_wait
 * 
 * @page su_clone_t Clone Objects
 *
 * The process may be divided into many tasks via cloning. Several tasks may
 * run in context of one thread, or each task may be run by its own thread. 
 * However, only a single thread can execute code within a task. There can
 * be a 1-to-N mapping from thread to tasks. Thus, software using tasks can
 * be executed by multiple threads in a multithreaded environment and by a
 * single thread in a singlethreaded environment.
 * 
 * The clones are useful for handling tasks that can be executed by a
 * separate threads, but which do not block excessively. When threads are
 * not available or they are not needed, clones can also be run in a
 * single-threaded mode. Running in single-threaded mode is especially
 * useful while debugging.
 * 
 * A clone task is created with function su_clone_start(). Each clone has
 * its own root object (su_root_t), which holds a context pointer
 * (su_root_magic_t *). The context object can be different from that of 
 * parent task.
 *
 * When a clone is started, the clone initialization function is called. The
 * initialization function should do whatever initialization there is to be
 * performed, register I/O events and timers, and then return. If the
 * initialization is successful, the clone task reverts to run the event
 * loop and invoking the event callbacks until its parent stops it by
 * calling su_clone_wait() which invokes the deinit function. The clone task
 * is destroyed when the deinit function returns. 
 *
 * The public API consists of following functions:
 *    - su_clone_start()
 *    - su_clone_task()
 *    - su_clone_wait()
 *    - su_clone_forget()
 *
 * @note 
 * There is only one event loop for each thread which can be shared by
 * multiple clone tasks. Therefore, the clone tasks can not explicitly run
 * or step the event loop, but they are limited to event callbacks. A clone
 * task may not call su_root_break(), su_root_run() or su_root_step().
 */

static void su_root_deinit(su_root_t *self);

/* Note that is *not* necessary same as su_root_t,
 * as su_root_t can be extended */

#define sur_port sur_task->sut_port
#define sur_root sur_task->sut_root

#define SU_ROOT_OWN_THREAD(r) (su_port_own_thread(r->sur_port))

/** 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.
 */
su_root_t *su_root_create(su_root_magic_t *magic)
{
  return su_root_create_with_port(magic, su_port_create());
}

/** Create a reactor object using given message port.
 *
 * Allocate and initialize the instance of su_root_t.
 *
 * @param magic     pointer to user data
 * @param port      pointer to a message port
 *
 * @return A pointer to allocated su_root_t instance, NULL on error.
 */
su_root_t *su_root_create_with_port(su_root_magic_t *magic,
				    su_port_t *port)
{
  su_root_t *self;

  if (!port)
    return NULL;

  self = su_salloc(NULL, sizeof(struct su_root_s));

  if (self) {
    self->sur_magic = magic;
#if SU_HAVE_PTHREADS
    self->sur_threading = SU_HAVE_PTHREADS;
#else
    self->sur_threading = 0;
#endif
    su_task_new(self->sur_task, self, port);
  } else {
    su_port_decref(port, "su_root_create");
  }

  return self;
}

/** Destroy a synchronization object.
 * 
 *  Stop and free an instance of su_root_t
 *
 * @param self     pointer to a root object.
 */
void su_root_destroy(su_root_t *self)
{
  if (self) {
    assert(SU_ROOT_OWN_THREAD(self));
    su_root_deinit(self);
    su_free(NULL, self);
  }
}

/** @internal Deinitialize a synchronization object.
 *
 *  Deinitialize an instance of su_root_t
 *
 * @param self     pointer to a root object.
 */
static void su_root_deinit(su_root_t *self)
{
  self->sur_deiniting = 1;