tport.c

this is simple sip stack.

tport_primary_t *tport_alloc_primary(tport_master_t *mr,
				     tport_vtable_t const *vtable,
				     tp_name_t tpn[1],
				     su_addrinfo_t *ai,
				     tagi_t const *tags,
				     char const **return_culprit)
{
  tport_primary_t *pri, **next;
  tport_t *tp;
  int save_errno;

  for (next = &mr->mr_primaries; *next; next = &(*next)->pri_next)
    ;
  
  assert(vtable->vtp_pri_size >= sizeof *pri);

  if ((pri = su_home_clone(mr->mr_home, vtable->vtp_pri_size))) {
    tport_t *tp = pri->pri_primary;
    pri->pri_vtable = vtable;
    pri->pri_public = vtable->vtp_public;

    tp->tp_master = mr;
    tp->tp_pri = pri;
    tp->tp_socket = SOCKET_ERROR;

    tp->tp_magic = mr->mr_master->tp_magic;

    tp->tp_params = pri->pri_params;
    memcpy(tp->tp_params, mr->mr_params, sizeof (*tp->tp_params));
    tp->tp_reusable = mr->mr_master->tp_reusable;

    if (!pri->pri_public)
      tp->tp_addrinfo->ai_addr = &tp->tp_addr->su_sa;

    SU_DEBUG_5(("%s(%p): new primary tport %p\n", __func__, mr, pri));
  }

  *next = pri; 
  tp = pri->pri_primary;
  
  if (!tp)
    *return_culprit = "alloc";
  else if (tport_set_params(tp, TAG_NEXT(tags)) < 0)
    *return_culprit = "tport_set_params";
  else if (vtable->vtp_init_primary &&
	   vtable->vtp_init_primary(pri, tpn, ai, tags, return_culprit) < 0)
    ;
  else if (tport_setname(tp, vtable->vtp_name, ai, tpn->tpn_canon) == -1) 
    *return_culprit = "tport_setname";
  else if (tpn->tpn_ident && 
	   !(tp->tp_name->tpn_ident = su_strdup(tp->tp_home, tpn->tpn_ident)))
    *return_culprit = "alloc ident";
  else
    return pri;			/* Success */

  save_errno = su_errno();
  tport_zap_primary(pri);
  su_seterrno(save_errno);

  return NULL;
}


/** Destroy a primary transport and its secondary transports. @internal */
static 
void tport_zap_primary(tport_primary_t *pri)
{
  tport_primary_t **prip;

  if (pri == NULL)
    return;

  assert(tport_is_primary(pri->pri_primary));

  if (pri->pri_vtable->vtp_deinit_primary)
    pri->pri_vtable->vtp_deinit_primary(pri);

  while (pri->pri_secondary)
    tport_zap_secondary(pri->pri_secondary);

  /* We have just a single-linked list for primary transports */
  for (prip = &pri->pri_master->mr_primaries;
       *prip != pri;
       prip = &(*prip)->pri_next)
    assert(*prip);

  *prip = pri->pri_next;

  tport_zap_secondary(pri->pri_primary);
}

/**Create a primary transport object with socket.
 *
 * Creates a primary transport object with a server socket, and then
 * registers the socket with suitable events to the root.
 *
 * @param dad   parent (master or primary) transport object
 * @param ai    pointer to addrinfo structure
 * @param canon canonical name of node
 * @param protoname name of the protocol
 */
static
tport_primary_t *tport_listen(tport_master_t *mr,
			      tport_vtable_t const *vtable,
			      tp_name_t tpn[1],
			      su_addrinfo_t *ai, 
			      tagi_t *tags)
{
  tport_primary_t *pri = NULL;

  int err;
  int errlevel = 3;
  char buf[TPORT_HOSTPORTSIZE];

  char const *protoname = vtable->vtp_name;
  char const *culprit = "unknown";

  su_sockaddr_t *su = (void *)ai->ai_addr;

  /* Log an error, return error */
#define TPORT_LISTEN_ERROR(errno, what)				     \
  ((void)(err = errno,						     \
	  ((err == EADDRINUSE || err == EAFNOSUPPORT ||		     \
	    err == ESOCKTNOSUPPORT || err == EPROTONOSUPPORT ||	     \
	    err == ENOPROTOOPT ? 7 : 3) < SU_LOG_LEVEL ?	     \
	     su_llog(tport_log, errlevel,			     \
		     "%s(%p): %s(pf=%d %s/%s): %s\n",		     \
		     __func__, pri ? pri : mr, what, ai->ai_family,  \
		     protoname,					     \
		     tport_hostport(buf, sizeof(buf), su, 2),	     \
		     su_strerror(err)) : (void)0),		     \
	    tport_zap_primary(pri),		                     \
	    su_seterrno(err)),					     \
     (void *)NULL)

  /* Create a primary transport object for another transport. */
  pri = tport_alloc_primary(mr, vtable, tpn, ai, tags, &culprit);
  if (pri == NULL)
    return TPORT_LISTEN_ERROR(errno, culprit);

  if (pri->pri_primary->tp_socket != SOCKET_ERROR) {
    int index = 0;
    tport_t *tp = pri->pri_primary;
    su_wait_t wait[1] = { SU_WAIT_INIT };

    if (su_wait_create(wait, tp->tp_socket, tp->tp_events) == -1)
      return TPORT_LISTEN_ERROR(su_errno(), "su_wait_create");

    /* Register receiving or accepting function with events specified above */
    index = su_root_register(mr->mr_root, wait, tport_wakeup_pri, tp, 0); 
    if (index == -1) {
      su_wait_destroy(wait);
      return TPORT_LISTEN_ERROR(su_errno(), "su_root_register");
    }

    tp->tp_index = index;
  }

  pri->pri_primary->tp_connected = 0;

  SU_DEBUG_5(("%s(%p): %s " TPN_FORMAT "\n", 
	      __func__, pri, "listening at",
	      TPN_ARGS(pri->pri_primary->tp_name)));

  return pri;
}

int tport_bind_socket(int socket,
		      su_addrinfo_t *ai,
		      char const **return_culprit)
{
  su_sockaddr_t *su = (su_sockaddr_t *)ai->ai_addr;
  socklen_t sulen = ai->ai_addrlen;

  if (bind(socket, ai->ai_addr, ai->ai_addrlen) == -1) {
    return *return_culprit = "bind", -1;
  }

  if (getsockname(socket, &su->su_sa, &sulen) == SOCKET_ERROR) {
    return *return_culprit = "getsockname", -1;
  }

  ai->ai_addrlen = sulen;

#if defined (__linux__) && defined (SU_HAVE_IN6)
  if (ai->ai_family == AF_INET6) {
    if (!SU_SOCKADDR_INADDR_ANY(su) &&
	(IN6_IS_ADDR_V4MAPPED(&su->su_sin6.sin6_addr) ||
	 IN6_IS_ADDR_V4COMPAT(&su->su_sin6.sin6_addr))) {
      su_sockaddr_t su0[1];
      
      memcpy(su0, su, sizeof su0);
      
      memset(su, 0, ai->ai_addrlen = sizeof su->su_sin);
      su->su_family = ai->ai_family = AF_INET;
      su->su_port = su0->su_port;
      
#ifndef IN6_V4MAPPED_TO_INADDR
#define IN6_V4MAPPED_TO_INADDR(in6, in4) \
      memcpy((in4), 12 + (uint8_t *)(in6), sizeof(struct in_addr))
#endif
      IN6_V4MAPPED_TO_INADDR(&su0->su_sin6.sin6_addr, &su->su_sin.sin_addr);
    }
  }
#endif

  return 0;
}


/** Indicate stack that a transport has been updated */
void tport_has_been_updated(tport_t *self)
{
  self->tp_pri->pri_updating = 0;

  if (self->tp_master->mr_tpac->tpac_address)
    self->tp_master->mr_tpac->tpac_address(self->tp_master->mr_stack, self);
}


static
int tport_set_events(tport_t *self, int set, int clear)
{
  int events;

  if (self == NULL)
    return -1;

  events = (self->tp_events | set) & ~clear;
  self->tp_events = events;

  if (self->tp_pri->pri_vtable->vtp_set_events)
    return self->tp_pri->pri_vtable->vtp_set_events(self);

  SU_DEBUG_7(("tport_set_events(%p): events%s%s%s\n", self,
	      (events & SU_WAIT_IN) ? " IN" : "",
	      (events & SU_WAIT_OUT) ? " OUT" : "",
	      SU_WAIT_CONNECT != SU_WAIT_OUT &&
	      (events & SU_WAIT_CONNECT) ? " CONNECT" : ""));
  
  return 
    su_root_eventmask(self->tp_master->mr_root, 
		      self->tp_index, 
		      self->tp_socket, 
		      self->tp_events = events);
}

/**Allocate a secondary transport. @internal
 *
 * The function tport_alloc_secondary() creates a secondary transport
 * object. The new transport initally shares parameters structure with the
 * original transport.
 *
 * @param pri    primary transport
 * @param socket socket for transport
 * @parma accepted true if the socket was accepted from server socket
 *
 * @return
 * Pointer to the newly created transport, or NULL upon an error.
 *
 * @note The socket is always closed upon error.
 */
tport_t *tport_alloc_secondary(tport_primary_t *pri, 
			       int socket,
			       int accepted,
			       char const **return_reason)
{
  tport_master_t *mr = pri->pri_master;
  tport_t *self;

  self = su_home_clone(mr->mr_home, pri->pri_vtable->vtp_secondary_size);

  if (self) {
    SU_DEBUG_7(("%s(%p): new secondary tport %p\n", __func__, pri, self));

    self->tp_refs = -1;			/* Freshly allocated  */
    self->tp_master = mr;
    self->tp_pri = pri;
    self->tp_params = pri->pri_params;
    self->tp_accepted = accepted != 0;
    self->tp_reusable = pri->pri_primary->tp_reusable;

    self->tp_magic = pri->pri_primary->tp_magic;

    self->tp_addrinfo->ai_addr = (void *)self->tp_addr;

    self->tp_socket = socket;
    
    if (pri->pri_vtable->vtp_init_secondary &&
	pri->pri_vtable->vtp_init_secondary(self, socket, accepted,
					    return_reason) < 0) {
      if (pri->pri_vtable->vtp_deinit_secondary)
	pri->pri_vtable->vtp_deinit_secondary(self);
      su_home_zap(self->tp_home);
      return NULL;
    }
  }
  else {
    su_close(socket);
    *return_reason = "malloc";
  }

  return self;
}


/** Create a connected transport object with socket.
 *
 * The function tport_connect() creates a secondary transport with a
 * connected socket. It registers the socket with suitable events to the
 * root.
 *
 * @param pri   primary transport object
 * @param ai    pointer to addrinfo structure
 * @param tpn   canonical name of node
 */
static
tport_t *tport_connect(tport_primary_t *pri, 
		       su_addrinfo_t *ai,
		       tp_name_t const *tpn)
{
  if (ai == NULL || ai->ai_addrlen > sizeof (pri->pri_primary->tp_addr))
    return NULL;

  if (pri->pri_vtable->vtp_connect)
    return pri->pri_vtable->vtp_connect(pri, ai, tpn);
  else
    return tport_base_connect(pri, ai, ai, tpn);
}

/**Create a connected transport object with socket.
 *
 * The function tport_connect() creates a secondary transport with a
 * connected socket. It registers the socket with suitable events to the
 * root.
 *
 * @param pri   primary transport object
 * @param ai    pointer to addrinfo structure describing socket
 * @param real_ai  pointer to addrinfo structure describing real target
 * @param tpn   canonical name of node
 */
tport_t *tport_base_connect(tport_primary_t *pri,
			    su_addrinfo_t *ai,
			    su_addrinfo_t *real_ai,
			    tp_name_t const *tpn)
{
  tport_master_t *mr = pri->pri_master;
  tport_t *self = NULL;

  su_socket_t s, server_socket;
  su_wait_t wait[1] = { SU_WAIT_INIT };
  su_wakeup_f wakeup = tport_wakeup;
  int index = 0;
  int events = SU_WAIT_IN | SU_WAIT_ERR;

  int err, errlevel = 3;
  char buf[TPORT_HOSTPORTSIZE];
  char const *what;

  /* Log an error, return error */
#define TPORT_CONNECT_ERROR(errno, what)			     \
  return							     \
    ((void)(err = errno,					     \
	    su_wait_destroy(wait),				     \
	    (SU_LOG_LEVEL >= errlevel ?				     \
	     su_llog(tport_log, errlevel,			     \
		     "%s(%p): %s(pf=%d %s/%s): %s\n",		     \
		     __func__, pri, #what, ai->ai_family,	     \
		     tpn->tpn_proto,				     \
		     tport_hostport(buf, sizeof(buf),		     \
				    (void *)ai->ai_addr, 2),	     \
		     su_strerror(err)) : (void)0),		     \
	    tport_zap_secondary(self),				     \
	    su_seterrno(err)),					     \
     (void *)NULL)

  s = su_socket(ai->ai_family, ai->ai_socktype, ai->ai_protocol);
  if (s == SOCKET_ERROR)
    TPORT_CONNECT_ERROR(su_errno(), "socket");

  what = "tport_alloc_secondary";
  if ((self = tport_alloc_secondary(pri, s, 0, &what)) == NULL)
    TPORT_CONNECT_ERROR(errno, what);

  self->tp_conn_orient = 1;

  if ((server_socket = pri->pri_primary->tp_socket) != SOCKET_ERROR) {
    su_sockaddr_t susa;
    socklen_t susalen = sizeof(susa);

    /* Bind this socket to same IP address as the primary server socket */
    if (getsockname(server_socket, &susa.su_sa, &susalen) < 0) {
      SU_DEBUG_3(("tport_connect: getsockname(): %s\n", 
		  su_strerror(su_errno())));
    }
    else {
      susa.su_port = 0;
      if (bind(s, &susa.su_sa, susalen) < 0) {
	SU_DEBUG_3(("tport_connect: bind(local-ip): %s\n", 
		    su_strerror(su_errno())));
      }
    }
  }

  /* Set sockname for the tport */
  if (tport_setname(self, tpn->tpn_proto, real_ai, tpn->tpn_canon) == -1) 
    TPORT_CONNECT_ERROR(su_errno(), tport_setname);

  if (connect(s, ai->ai_addr, ai->ai_addrlen) == SOCKET_ERROR) {
    err = su_errno();
    if (err != EINPROGRESS && err != EAGAIN && err != EWOULDBLOCK)
      TPORT_CONNECT_ERROR(err, connect);
    events = SU_WAIT_CONNECT | SU_WAIT_ERR;
    wakeup = tport_connected;
    what = "connecting";
  }
  else {
    what = "connected";
  }

  if (su_wait_create(wait, s, self->tp_events = events) == -1)
    TPORT_CONNECT_ERROR(su_errno(), su_wait_create);

  /* Register receiving function with events specified above */
  if ((index = su_root_register(mr->mr_root, wait, wakeup, self, 0)) == -1)
    TPORT_CONNECT_ERROR(su_errno(), su_root_register);

  self->tp_index = index;

  if (ai == real_ai) {
    SU_DEBUG_5(("%s(%p): %s to " TPN_FORMAT "\n", 
		__func__, self, what, TPN_ARGS(self->tp_name)));
  }
  else {
    SU_DEBUG_5(("%s(%p): %s via %s to " TPN_FORMAT "\n",
		__func__, self, what,
		tport_hostport(buf, sizeof(buf), (void *)ai->ai_addr, 2),
		TPN_ARGS(self->tp_name)));
  }
  
  tprb_append(&pri->pri_secondary, self);

  return self;
}

/** Destroy a secondary transport. @internal */
void tport_zap_secondary(tport_t *self)
{
  tport_master_t *mr;

  if (self == NULL)
    return;

  /* Remove from rbtree */
  tprb_remove(&self->tp_pri->pri_secondary, self);

  /* Do not deinit primary as secondary! */
  if (tport_is_secondary(self) &&
      self->tp_pri->pri_vtable->vtp_deinit_secondary)
    self->tp_pri->pri_vtable->vtp_deinit_secondary(self);

  if (self->tp_msg) {
    msg_destroy(self->tp_msg), self->tp_msg = NULL;
    SU_DEBUG_3(("%s(%p): zapped partially received message\n", 
		__func__, self));
  }

  if (self->tp_queue && self->tp_queue[self->tp_qhead]) {
    size_t n = 0, i, N = self->tp_params->tpp_qsize;
    for (i = self->tp_qhead; self->tp_queue[i]; i = (i + 1) % N) {
      msg_destroy(self->tp_queue[i]), self->tp_queue[i] = NULL;
      n++;
    }
    SU_DEBUG_3(("%s(%p): zapped %lu queued messages\n", 
		__func__, self, (LU)n));
  }

  if (self->tp_pused) {
    SU_DEBUG_3(("%s(%p): zapped with pending requests\n", __func__, self));
  }

  mr = self->tp_master;

  tport_stun_server_remove_socket(self);

  if (self->tp_index)
    su_root_deregister(mr->mr_root, self->tp_index);
  self->tp_index = 0;
  if (self->tp_socket != -1)
    su_close(self->tp_socket);
  self->tp_socket = -1;

  su_home_zap(self->tp_home);
}

/** Create a new reference to a transport object. */
tport_t *tport_ref(tport_t *tp)
{
  if (tp) {
    if (tp->tp_refs >= 0)
      tp->tp_refs++;
    else if (tp->tp_refs == -1)
      tp->tp_refs = 1;
  }
  return tp;
}

/** Destroy reference to a transport object. */
void tport_unref(tport_t *tp)
{
  if (tp && tp->tp_refs > 0)
    if (--tp->tp_refs == 0 && tp->tp_params->tpp_idle == 0)
      if (!tport_is_closed(tp))
	tport_close(tp);
}