event_dispatcher.c

Go to the documentation of this file.

/* $Id: event_dispatcher.c 4518 2008-07-28 15:39:28Z henningw $
 *
 * Copyright (C) 2006-2007 VozTelecom Sistemas S.L
 *
 * This file is part of Kamailio, a free SIP server.
 *
 * Kamailio is free software; you can redistribute it and/or modify
 * it under the terms of the GNU General Public License as published by
 * the Free Software Foundation; either version 2 of the License, or
 * (at your option) any later version
 *
 * Kamailio 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 General Public License for more details.
 *
 * You should have received a copy of the GNU General Public License 
 * along with this program; if not, write to the Free Software 
 * Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA  02111-1307  USA
 */

#include <sys/types.h>/*setsockopt,bind,accept,fork,pid_t*/
#include <sys/socket.h>/*setsockopt,bind,accept,listen*/
#include <netinet/tcp.h>/*TCP_NODELAY*/
#include <string.h>/*strcmp,memset*/
#include <errno.h>/*errno*/
#include <unistd.h>/*close(),read(),pipe,fork,pid_t*/
#include <sys/poll.h>/*poll*/
#include <signal.h>/*signal*/
#include <time.h>/*time*/
#include <string.h>/*memcmp*/
#include <sys/types.h>/*waitpid*/
#include <sys/wait.h>/*waitpid*/

#include "../../ip_addr.h" /*sockaddr_union, ip_addr*/
#include "../../hash_func.h" /*T_TABLE_POWER*/
#include "../../mem/mem.h" /*pkg_malloc*/
#include "../../mem/shm_mem.h" /*shm_malloc*/
#include "../../dprint.h" /*LM_**/
#include "../../locking.h"

#include "seas.h"
#include "ha.h"
#include "cluster.h"
#include "seas_action.h"
#include "statistics.h"
#include "event_dispatcher.h"

#define PING_OVER_FACTOR 2
#define MAX_WRITE_TRIES 10

char *action_names[]={"NONE",
     "PROVISIONAL_REPLY",
     "FINAL_REPLY",
     "REPLY_FIN_DLG",
     "UAC_REQ",
     "AC_RES_FAIL",
     "STATELESS_MSG",
     "AC_CANCEL",
     "JAIN_PONG"};


struct unc_as unc_as_t[2*MAX_UNC_AS_NR];

/*this is for the Action Dispatcher Process */
struct as_entry *my_as;
extern int process_no;
extern int sig_flag;

static int process_event_reply(as_p as);
static int handle_as_data(int fd);
static inline int print_sock_info(char *buffer,int wheremax,int *idx,struct socket_info *s,enum sip_protos type);
static inline int send_sockinfo(int fd);
static inline int add_new_as(int event_idx,int action_idx,struct as_entry *as);
static int dispatch_relay();
static int new_as_connect(int fd,char which);
static inline int read_name(int sock,char *dst,int dstlen);
static int handle_unc_as_data(int fd);
static int open_server_sockets(struct ip_addr *address,unsigned short port,int *fd);



/** Main loop for the Event Dispatcher process.
 * 
 */
int dispatcher_main_loop(void)
{
   struct pollfd poll_fds[3+MAX_AS_NR],*poll_tmp;
   int clean_index,i,j,k,fd,poll_events=0,socks[2],chld_status;
   int as_nr,unc_as_nr;
   pid_t chld;
   struct timeval last_ping,now;
   struct as_entry *as;

   sig_flag=0;
   is_dispatcher=1;
   as_nr=0;

   timerclear(&last_ping);
   timerclear(&now);
   signal(SIGCHLD,seas_sighandler);
   signal(SIGTERM,seas_sighandler);
   signal(SIGUSR1,seas_sighandler);
   signal(SIGINT, seas_sighandler);
   signal(SIGKILL,seas_sighandler);

   strcpy(whoami,"Seas Event Dispatcher process");
   /*I set process_no to -1 because otherwise, the logging process confuses this process with another from SER
    * (see LM_*() and dprint() and my_pid())*/
   process_no = -1;

   if(open_server_sockets(seas_listen_ip,seas_listen_port,socks)==-1){
      LM_ERR("unable to open server sockets on dispatcher\n");
      return -1;
   }
   for(i=0;i<2;i++){
      poll_fds[i].fd=socks[i];
      poll_fds[i].revents=0;
      poll_fds[i].events=POLLIN;
   }
   poll_fds[2].fd=read_pipe;
   poll_fds[2].revents=0;
   poll_fds[2].events=POLLIN;/*pollhup ?*/

   poll_events=0;
   unc_as_nr=0;

   if(use_ha)
      spawn_pinger();

   while(1){
      if(sig_flag==SIGCHLD){
    while ((chld=waitpid( -1, &chld_status, WNOHANG ))>0) {
       if (WIFEXITED(chld_status)){
          LM_INFO("child process %d exited normally, status=%d\n",
               chld,WEXITSTATUS(chld_status));
       }else if (WIFSIGNALED(chld_status)) {
          LM_INFO("child process %d exited by a signal %d\n",
               chld,WTERMSIG(chld_status));
       }else if (WIFSTOPPED(chld_status)) 
          LM_INFO("child process %d stopped by a signal %d\n",
               chld,WSTOPSIG(chld_status));
       for (as=as_list;as;as=as->next) {
          if(as->type!=AS_TYPE)
        continue;
          if(as->u.as.action_pid==chld){
        for(i=0;i<as_nr && ((poll_fds[3+i].fd)!=(as->u.as.event_fd));i++)
           ;
        if(i==as_nr){
           LM_ERR("Either the pinger has died or BUG found..\n");
           continue;
        }
        /*overwrite the obsolete 'i' position with the next position*/
        for(j=3+i;j<(as_nr+unc_as_nr+3-1);i++){
           poll_fds[j].fd=poll_fds[j+1].fd;
           poll_fds[j].events=poll_fds[j+1].events;
           poll_fds[j].revents=poll_fds[j+1].revents;
        }
        close(as->u.as.event_fd);/*close the socket fd*/
        if (as->u.as.ev_buffer.s) {
           pkg_free(as->u.as.ev_buffer.s);
           as->u.as.ev_buffer.s=(char *)0;
           as->u.as.ev_buffer.len=0;
        }
        as->u.as.event_fd=as->u.as.action_fd=-1;
        as->connected=0;
        destroy_pingtable(&as->u.as.jain_pings);
        destroy_pingtable(&as->u.as.servlet_pings);
        as_nr--;
        LM_WARN("client [%.*s] leaving (Action Dispatcher Process died !)\n",
              as->name.len,as->name.s);
        break;
          }/*if(action_pid==chld)*/
       }/*for(as=as_list;as;as=as->next)*/
    }/*while(waitpid(-1)>0)*/
      }else if (sig_flag) {
    LM_WARN("received signal != sigchld(%d)\n",sig_flag);
     }
      sig_flag=0;
      clean_index=0;
      LM_INFO("polling [2 ServSock] [1 pipe] [%d App Servers]"
           " [%d Uncomplete AS]\n",as_nr,unc_as_nr);
      poll_events = poll(poll_fds,3+unc_as_nr+as_nr,-1);
      if (poll_events == -1) {
    if(errno==EINTR){
       /*handle the case a child has died.
        * It will be done in the next iteration in if(seas_sigchld_received)*/
       continue;
    }
    if(errno==EBADF){
       LM_ERR("invalid file descriptor pased to poll (%s)\n",
            strerror(errno));
       return -1;/*??*/
    }
    /* errors */
    LM_ERR("poll'ing:%s\n",strerror(errno));
    poll_events=0;
    continue;
      } else if (poll_events == 0) {/*timeout*/
    continue;
      } else {/*there are events !*/
    /*handle connections from server sockets*/
    for(i=0;i<2;i++){
       if(poll_fds[i].revents)
          poll_events--;
       if(poll_fds[i].revents & POLLIN){
          poll_fds[i].revents &= (~POLLIN);
          if((fd=new_as_connect(socks[i],i==0?'e':'a'))>=0){
        poll_tmp=&poll_fds[3+as_nr+unc_as_nr];
        poll_tmp->fd=fd;
        poll_tmp->events=POLLIN|POLLHUP;
        unc_as_nr++;
        LM_DBG("Have new %s client\n",i==0?"event":"action");
          }else{
        LM_ERR("accepting connection from AS\n");
          }
       }
    }
    /*handle data from pipe*/
    if(poll_fds[2].revents & POLLIN){
       poll_fds[2].revents &= (~POLLIN);
       poll_events--;
       if(dispatch_relay()<0){
          LM_ERR("dispatch_relay returned -1"
           "should clean-up table\n");
       }
    }
    /*now handle receive data from completed AS*/
    clean_index=0;
    LM_DBG("Scanning data from %d AS\n",as_nr);
    for(i=0;(i<as_nr) && poll_events;i++){
       clean_index=0;
       poll_tmp=&poll_fds[3+i];
       if(poll_tmp->revents)
          poll_events--;
       if(poll_tmp->revents & POLLIN){
          LM_DBG("POLLIN found in AS #%i\n",i);
          poll_tmp->revents &= (~POLLIN);
          switch(handle_as_data(poll_tmp->fd)){
        case -2:/*read returned 0 bytes, an AS client is leaving*/
           clean_index=1;
           break;
        case -1:/*shouldnt happen*/
           LM_ERR("reading from AS socket\n");
           break;
        case 0:/* event_response received and processed*/
           break;
        default:
           LM_WARN("unknown return type from handle_as_data\n");
          }
       }
       if(clean_index || (poll_tmp->revents & POLLHUP)){
          LM_DBG("POLHUP or read==0 found in %i AS \n",i);
          clean_index=0;
          poll_tmp->revents = 0;
          for(as=as_list;as;as=as->next){
        if(as->type==CLUSTER_TYPE)
           continue;
        if(as->connected && (as->u.as.event_fd == poll_tmp->fd)){
           close(poll_tmp->fd);/*close the socket fd*/
           /*TODO  we should send a signal to the Action Dispatcher !!!*/
           as->connected=0;
           as_nr--;
           /*overwrite the obsolete 'i' position with the next position*/
           for(k=i;k<(as_nr+unc_as_nr);k++){
         j=3+k;
         poll_fds[j].fd=poll_fds[j+1].fd;
         poll_fds[j].events=poll_fds[j+1].events;
         poll_fds[j].revents=poll_fds[j+1].revents;
           }
           --i;
           LM_WARN("client %.*s leaving !!!\n",as->name.len,as->name.s);
           break;
        }
          }
          if (!as) {
        LM_ERR("the leaving client was not found in the as_list\n");
         }
       }
    }
    /*now handle data sent from uncompleted AS*/
    LM_DBG("Scanning data from %d uncomplete AS \n",unc_as_nr);
    clean_index=0;
    for(i=0;i<unc_as_nr && poll_events;i++){
       poll_tmp=&poll_fds[3+as_nr+i];
       if(poll_tmp->revents)
          poll_events--;
       if(poll_tmp->revents & POLLIN){
          LM_DBG("POLLIN found in %d uncomplete AS \n",i);
          poll_tmp->revents &= (~POLLIN);
          fd=handle_unc_as_data(poll_tmp->fd);
          if(fd>0){
        /* there's a new AS, push the uncomplete poll_fds up and set the AS */
        for(k=i;k>0;k--){
           j=3+as_nr+k;
           poll_fds[j].fd=poll_fds[j-1].fd;
           poll_fds[j].events=poll_fds[j-1].events;
           poll_fds[j].revents=poll_fds[j-1].revents;
        }
        poll_fds[3+as_nr].fd=fd;
        poll_fds[3+as_nr].events=POLLIN|POLLHUP;
        poll_fds[3+as_nr].revents=0;
        as_nr++;/*not very sure if this is thread-safe*/
        unc_as_nr--;
          }else if(fd<=0){/* pull the upper set of uncomplete AS down and take this one out*/
        poll_tmp->revents=0;
        for(k=i;k<(unc_as_nr-1);k++){
           j=3+as_nr+k;
           poll_fds[j].fd=poll_fds[j+1].fd;
           poll_fds[j].events=poll_fds[j+1].events;
           poll_fds[j].revents=poll_fds[j+1].revents;
        }
        unc_as_nr--;
        /** we decrement i so that pulling down the upper part of the unc_as array so that
         * it doesn't affect our for loop */
        i--;
          }
       }
       if(poll_tmp->revents & POLLHUP){
          LM_DBG("POLLHUP found in %d uncomplete AS \n",i);
          close(poll_tmp->fd);
          for(k=i;k<(unc_as_nr-1);k++){
        j=3+as_nr+k;
        poll_fds[j].fd=poll_fds[j+1].fd;
        poll_fds[j].events=poll_fds[j+1].events;
        poll_fds[j].revents=poll_fds[j+1].revents;
          }
          unc_as_nr--;
          i--;
          poll_tmp->revents = 0;
       }
    }/*for*/
      }/*else ...(poll_events>0)*/
   }/*while(1)*/
}


/**
 * opens the server socket, which attends (accepts) the clients, that is: 
 * params:
 * address:
 *    address to which to listen
 * port:
 *    base port to which to listen. then port+1 will be the socket
 *    for action's delivery.
 * fds:
 *    in fd[0] the action socket will be put.
 *    in fd[1] the event socket will be put.
 *
 * returns 0 on exit, <0 on fail
 *
 */
static int open_server_sockets(struct ip_addr *address,unsigned short port,int *fd)
{

   /*using sockaddr_union enables ipv6..*/
   union sockaddr_union su;
   int i,optval;

   fd[0]=fd[1]=-1;

   if(address->af!=AF_INET && address->af!=AF_INET6){
      LM_ERR("Only ip and ipv6 allowed socket types\n");
      return -1;
   }

   for(i=0;i<2;i++){
      if(init_su(&su,address,port+i)<0){
    LM_ERR("unable to init sockaddr_union\n");
    return -1;
      }
      if((fd[i]=socket(AF2PF(su.s.sa_family), SOCK_STREAM, 0))==-1){
    LM_ERR("trying to open server %s socket (%s)\n",i==0?"event":"action",strerror(errno));
    goto error;
      }
      optval=1;
      if (setsockopt(fd[i], SOL_SOCKET, SO_REUSEADDR, (void*)&optval, sizeof(optval))==-1) {
    LM_ERR("setsockopt (%s)\n",strerror(errno));
    goto error;
      }
      if ((bind(fd[i], &su.s,sizeof(union sockaddr_union)))==-1){
    LM_ERR( "bind (%s)\n",strerror(errno));
    goto error;
      }
      if (listen(fd[i], 10)==-1){
    LM_ERR( "listen (%s)\n",strerror(errno));
    goto error;
      }
   }
   return 0;

error:
   for(i=0;i<2;i++)
      if(fd[i]!=-1){
    close(fd[i]);
    fd[i]=-1;
      }
   return -1;
}


union helper{
   as_msg_p ptr;
   char bytes[sizeof(as_msg_p)];
};

/**
 * Sends event 
 *
 * returns
 *     0  OK
 *    -1 couldn't read the event from the pipe
 *    -2 couldn't send the event
 * TODO this should be FAR more generic... for example, there might be events
 * which are not related to any transaction (finish event, or error event...)
 * we should separate event-specific handling in different functions...
 */
static int dispatch_relay(void)
{
   int i,j,retval,tries;
   union helper thepointer;

   i=j=0;
   retval=0;
read_again:
   i=read(read_pipe,thepointer.bytes+j,sizeof(as_msg_p)-j);
   if(i<0){
      if(errno==EINTR){
    goto read_again;
      }else{
    LM_ERR("Dispatcher Process received unknown error"
          " reading from pipe (%s)\n",strerror(errno));
    retval=-1;
    goto error;
      }
   }else if(i==0){
    LM_ERR("Dispatcher Process "
          "received 0 while reading from pipe\n");
    goto error;
   }else{
      j+=i;
      if(j<sizeof(as_msg_p))
    goto read_again;
   }

   if (!thepointer.ptr) {
      LM_ERR("Received Corrupted pointer to event !!\n");
      retval=0;
      goto error;
   }
   /*the message*/
   if(use_stats && thepointer.ptr->transaction)
      event_stat(thepointer.ptr->transaction);
   if(thepointer.ptr->as == NULL || !thepointer.ptr->as->connected || thepointer.ptr->as->type==CLUSTER_TYPE){
      LM_WARN("tryied to send an event to an App Server"
           " that is scheduled to die!!\n");
      retval=-2;
      goto error;
   }
   j=0;
   tries=0;
write_again:
   i=write(thepointer.ptr->as->u.as.event_fd,thepointer.ptr->msg+j,thepointer.ptr->len-j);
   if(i==-1){
      switch(errno){
    case EINTR:
       if(!thepointer.ptr->as->connected){
          LM_WARN("tryied to send an event to an App Server"
               " that is scheduled to die!!\n");
          retval=-2;
          goto error;
       }
       goto write_again;
    case EPIPE:
       LM_ERR("AS [%.*s] closed "
        "the socket !\n",thepointer.ptr->as->u.as.name.len,thepointer.ptr->as->u.as.name.s);
       retval=-2;
       goto error;
    default:
       LM_ERR("unknown error while trying to write to AS socket(%s)\n",
            strerror(errno));
       retval=-2;
       goto error;
      }
   }else if(i>0){
      j+=i;
      if(j<thepointer.ptr->len)
    goto write_again;
   }else if(i==0){
      if (tries++ > MAX_WRITE_TRIES) { 
    LM_ERR("MAX WRITE TRIES !!!\n");
    goto error;
      }else
    goto write_again;
   }
   LM_DBG("Event relaied to %.*s AS\n",thepointer.ptr->as->u.as.name.len,
         thepointer.ptr->as->u.as.name.s);
   LM_DBG("Event type %s \n",action_names[thepointer.ptr->type]);
   retval=0;
error:
   if(thepointer.ptr){
      if(thepointer.ptr->msg)
    shm_free(thepointer.ptr->msg);
      shm_free(thepointer.ptr);
   }
   return retval;
}

/**
 * receives 2 indexes in unc_as_t which correspond one to
 * the events socket and the other to the actions socket
 *
 * returns
 *    0 on success
 *    -1 on error
 */
static inline int add_new_as(int event_idx,int action_idx,struct as_entry *as)
{
   struct unc_as *ev,*ac;
   int j;
   as_p the_as=0;
   struct as_entry *tmp;

   ev=&unc_as_t[event_idx];
   ac=&unc_as_t[action_idx];

   the_as=&(as->u.as);

   the_as->action_fd=ac->fd;
   the_as->event_fd=ev->fd;
   the_as->name.len = strlen(ev->name);
   if(use_ha){
      if(jain_ping_timeout){
    if (0>init_pingtable(&the_as->jain_pings,jain_ping_timeout,(jain_ping_timeout/jain_ping_period+1)*PING_OVER_FACTOR)){
       LM_ERR("Unable to init jain pinging table...\n");
       goto error;
    }
      }
      if(servlet_ping_timeout){
    if (0>init_pingtable(&the_as->servlet_pings,servlet_ping_timeout,(servlet_ping_timeout/servlet_ping_period+1)*PING_OVER_FACTOR)){
       LM_ERR("Unable to init servlet pinging table...\n");
       goto error;
    }
      }
   }
   /*TODO attention, this is pkg_malloc because only the Event_Dispatcher process 
    * has to use it !!*/
   if(!(the_as->ev_buffer.s = pkg_malloc(AS_BUF_SIZE))){
      LM_ERR("unable to alloc pkg mem for the event buffer\n");
      goto error;
   }
   the_as->ev_buffer.len=0;
   as->connected=1;
   the_as->action_pid=0;
   for(tmp=as_list;tmp;tmp=tmp->next){
      if(tmp->type==AS_TYPE)
    continue;
      for (j=0;j<tmp->u.cs.num;j++) {
    if (tmp->u.cs.as_names[j].len == the_as->name.len && 
          !memcmp(tmp->u.cs.as_names[j].s,the_as->name.s,the_as->name.len)) { 
       if(tmp->u.cs.num==tmp->u.cs.registered){
          LM_ERR("AS %.*s belongs to cluster %.*s which is already completed\n",
           the_as->name.len,the_as->name.s,tmp->name.len,tmp->name.s);
          break;
       }
       tmp->u.cs.registered++;
       break;
    }
      }
   }
   if(0>spawn_action_dispatcher(as)){
      LM_ERR("Unable to spawn Action Dispatcher for as %s\n",ev->name);
      goto error;
   }
   if(send_sockinfo(the_as->event_fd)==-1){
      LM_ERR("Unable to send socket info to as %s\n",ev->name);
      goto error;
   }
   return 0;
error:
   if(the_as->ev_buffer.s){
      pkg_free(the_as->ev_buffer.s);
      the_as->ev_buffer.s=(char*)0;
   }
   if(the_as->action_pid)
      kill(the_as->action_pid,SIGTERM);
   if(jain_ping_timeout)
      destroy_pingtable(&the_as->jain_pings);
   if(servlet_ping_timeout)
      destroy_pingtable(&the_as->servlet_pings);
   return -1;
}






/**prints available sockets in SER to the App Server.
 * format is:
 * 1: transport identifier (u for UDP, t for TCP, s for TLS)
 * 1: length of socket name (sip.voztele.com or whatever)
 * N: name
 * 1: length of IP address (192.168.1.2)
 * N: ip address in ascii
 * 2: port nubmer in NBO
 *
 * returns
 *    -1 on error
 *    0  on success
 */
static inline int send_sockinfo(int fd)
{
   struct socket_info *s;
   unsigned char i;
   char buffer[300];
   int k=0,j;
   buffer[k++]=16;/*This used to be T_TABLE_POWER in openser 1.0.1, now its hardcoded in config.h*/
   for(i=0,s=udp_listen;s;s=s->next,i++);
#ifdef USE_TCP
   for(s=tcp_listen;s;s=s->next,i++);
#endif
#ifdef USE_TLS
   for(s=tls_listen;s;s=s->next,i++);
#endif
   if(i==0){
      LM_ERR("no udp|tcp|tls sockets ?!!\n");
      return -1;
   }
   buffer[k++]=i;
   for(s=udp_listen;s;s=s->next){
      if(print_sock_info(buffer,300,&k,s,PROTO_UDP)==-1)
    return -1;
   }
#ifdef USE_TCP
   for(s=tcp_listen;s;s=s->next){
      if(print_sock_info(buffer,300,&k,s,PROTO_TCP)==-1)
    return -1;
   }
#endif
#ifdef USE_TLS
   for(s=tls_listen;s;s=s->next){
      if(print_sock_info(buffer,300,&k,s,PROTO_TLS)==-1)
    return -1;
   }
#endif
write_again:
   j=write(fd,buffer,k);
   if(j==-1){
      if(errno==EINTR)
    goto write_again;
      else
    return -1;
   }
   return 0;
}

/* prints sock info into the byte array where
 * returns 0 on success, -1 on err
 * the message sent is as follows:
 * 1: protocol type (0=NONE,1=UDP, 2=TCP, 3=TLS)
 * 1: name length
 * N: name
 * 1: address string length
 * N: address
 * 2: NBO unsigned shor int port number
 *
 * TODO buffer overflow risk
 */
static inline int print_sock_info(char *buffer,int wheremax,int *idx,struct socket_info *s,enum sip_protos type)
{
   int k;
   unsigned char i;
   unsigned short int j;
   if((wheremax-*idx)<49)/*31*name+17*ipv6+2*port+1*type*/
      return -1;
   k=*idx;
   buffer[k++]=(char)type;
   if((i=(unsigned char)s->name.len)>30){
      LM_ERR("name too long\n");
      return -1;
   }
   buffer[k++]=i;
   memcpy(&buffer[k],s->name.s,i);
   k+=i;
   i=(unsigned char)s->address_str.len;
   buffer[k++]=i;
   memcpy(&buffer[k],s->address_str.s,i);
   k+=i;
   j=htons(s->port_no);
   memcpy(&buffer[k],&j,2);
   k+=2;
   *idx=k;
   return 0;
}

/**
 * Handles data from an AppServer. First searches in the AS table which was the AS
 * that sent the data (we dont already know it because this comes from a poll_fd
 * struct). When the one is found, it calls process_event_reply, which in turn
 * looks if there's a complete event in the buffer, and if there is, processes it.
 *
 * returns
 *    -1 on error
 *    -2 on read()==0 (the socket has been closed by the other end)
 *    0 on success
 */
static int handle_as_data(int fd)
{
   int j,k;
   struct as_entry *as;
   for(as=as_list;as;as=as->next)
      if(as->type == AS_TYPE && as->connected && (as->u.as.event_fd==fd))
    break;
   if(!as){
      LM_ERR("AS not found\n");
      return -1;
   }
   k=AS_BUF_SIZE-(as->u.as.ev_buffer.len);
again:
   if((j=read(fd,as->u.as.ev_buffer.s+as->u.as.ev_buffer.len,k))<0){
      LM_ERR("reading data for as %.*s\n",as->name.len,as->name.s);
      if(errno==EINTR)
    goto again;
      else
    return -1;
   }else if(j==0){
      LM_ERR("AS client leaving (%.*s)\n",as->name.len,as->name.s);
      return -2;
   }
   as->u.as.ev_buffer.len+=j;
   LM_DBG("read %d bytes from AS (total = %d)\n",j,as->u.as.ev_buffer.len);
   if(as->u.as.ev_buffer.len>5)
      process_event_reply(&as->u.as);
   return 0;
}

/**
 * This function processess the Application Server buffer. We do buffered
 * processing because it increases performance quite a bit. Any message
 * sent from the AS comes with the first 2 bytes as an NBO unsigned short int 
 * which says the length of the following message (header and payload).
 * This way, we avoid multiple small reads() to the socket, which (as we know), consumes
 * far more processor because of the kernel read(2) system call. The drawback
 * is the added complexity of mantaining a buffer, the bytes read, and looking
 * if there is a complete message already prepared.
 *
 * Actions are supposed to be small, that's why BUF_SIZE is 2000 bytes length. 
 * Most of the actions will be that size or less. That is why the 4 bytes telling the
 * length of the Action payload are included in its size. This way you can use a fixed size
 * buffer to receive the Actions and not need to be pkb_malloc'ing for each new event.
 * If there is a particular bigger packet, for example one carrying a picture (a JPG can
 * easily surpass the 2000 byte limit) then a pkg_malloc will be required. This is left TODO
 *
 * returns
 *    -1 on error (packet too big)
 *    0  on success
 */
static int process_event_reply(as_p as)
{
   unsigned int ev_len;
   ev_len=(as->ev_buffer.s[0]<<24)|(as->ev_buffer.s[1]<<16)|(as->ev_buffer.s[2]<<8)|(as->ev_buffer.s[3]);/*yeah, it comes in network byte order*/
   /*if ev_len > BUF_SIZE then a flag should be put on the AS so that the whole length
    * of the action is skipped, until a mechanism for handling big packets is implemented*/
   if(ev_len>AS_BUF_SIZE){
      LM_WARN("Packet too big (%d)!!! should be skipped"
           " and an error returned!\n",ev_len);
      return -1;
   }
   if((as->ev_buffer.len<ev_len) || as->ev_buffer.len<4)
      return 0;
   switch(as->ev_buffer.s[4]){
      case BIND_AC:
    LM_DBG("Processing a BIND action from AS (length=%d): %.*s\n",
    ev_len,as->name.len,as->name.s);
    process_bind_action(as,&as->ev_buffer.s[5],ev_len-5);
    break;
      case UNBIND_AC:
    LM_DBG("Processing a UNBIND action from AS (length=%d): %.*s\n",
          ev_len,as->name.len,as->name.s);
    process_unbind_action(as,&as->ev_buffer.s[5],ev_len-5);
    break;
      default:
    return 0;
   }
   memmove(as->ev_buffer.s,&(as->ev_buffer.s[ev_len]),(as->ev_buffer.len)-ev_len);
   (as->ev_buffer.len)-=ev_len;
   return 0;
}


/**
 * processes a BIND event type from the AS.
 * Bind events follow this form:
 * 4:flags
 * 1:processor_id
 * 1:Address Family
 * 1:address length in bytes (16 for ipv6, 4 for ipv4) in NETWORK BYTE ORDER (fortunately, ip_addr struct stores it in NBO)
 * [16|4]:the IP address
 * 1:protocol used (UDP,TCP or TLS);
 * 2:NBO port
 *
 */
int process_bind_action(as_p as,char *payload,int len)
{
   struct socket_info *si,*xxx_listen;
   struct ip_addr my_addr;
   int i,k,proto;
   unsigned int flags;
   unsigned short port;
   char processor_id,buffer[300],*proto_s;
   k=0;
   *buffer=0;
   proto_s="NONE";
   net2hostL(flags,payload,k);
   processor_id=payload[k++];
   for(i=0;i<MAX_BINDS;i++){
      if(as->bound_processor[i]==0)
    break;
   }
   if(i==MAX_BINDS){
      LM_ERR("No more bindings allowed. Ignoring bind request for processor %d\n",processor_id);
      return -1;
   }
   memset(&my_addr,0,sizeof(struct ip_addr));
   my_addr.af=payload[k++];
   my_addr.len=payload[k++];
   memcpy(my_addr.u.addr,payload+k,my_addr.len);
   k+=my_addr.len;
   proto=payload[k++];
   memcpy(&port,payload+k,2);
   k+=2;
   port=ntohs(port);
   print_ip_buf(&my_addr,buffer,300);
   switch(proto){
      case PROTO_UDP:
    proto_s="UDP";
    xxx_listen=udp_listen;
    break;
#ifdef USE_TCP
      case PROTO_TCP:
    proto_s="TCP";
    xxx_listen=tcp_listen;
    break;
#endif
#ifdef USE_TLS
      case PROTO_TLS:
    proto_s="TLS";
    xxx_listen=tls_listen;
    break;
#endif
      default:
    goto error;
   }
   for(si=xxx_listen;si;si=si->next){
      if(my_addr.af==si->address.af && 
       my_addr.len==si->address.len && 
       !memcmp(si->address.u.addr,my_addr.u.addr,my_addr.len) && 
       port == si->port_no){
    as->binds[i]=si;
    as->bound_processor[i]=processor_id;
    as->num_binds++;
    LM_DBG("AS processor with id: %d bound to %s %s %d\n",processor_id,proto_s,buffer,port);
    return 0;
      }
   }
error:
   LM_ERR("Cannot bind to %s %s %d !!!\n",proto_s,buffer,port);
   return -1;
}

/**
 * processes a UNBIND event type from the AS.
 * Bind events follow this form:
 * 1:processor_id
 *
 */
int process_unbind_action(as_p as,char *payload,int len)
{
   int i,k;
   unsigned int flags;
   char processor_id;
   k=0;
   net2hostL(flags,payload,k);
   processor_id=payload[k++];
   for(i=0;i<as->num_binds;i++){
      if(as->bound_processor[i] == processor_id)
    break;
   }
   if(i==MAX_BINDS){
      LM_ERR("tried to unbind a processor which is not registered (id=%d)!\n",processor_id);
      return 0;
   }
   as->bound_processor[i]=0;
   as->num_binds--;
   LM_DBG("AS processor un-bound with id: %d\n",processor_id);
   return 0;
}

/**
 * params:
 *   the filedes where the data was received.
 * returns:
 *   0 if this fd should be taken out of the poll_fd array bcause it already has AS name.
 *   fd if an AS was completed (returns the fd of the events socket)
 *   -1 if there was an error
 *   -2 if client disconnected and should be closed and taken outside the poll_fd array
 */
static int handle_unc_as_data(int fd)
{
   int i,j,k,len;
   char *name1;
   struct as_entry *as;
   /*first, we see if the data to read is from any of the uncompleted as's*/
   for(i=0;i<2*MAX_UNC_AS_NR ;i++)
      if(unc_as_t[i].valid && unc_as_t[i].fd==fd)
    break;
   if(i==2*MAX_UNC_AS_NR){
      LM_ERR("has received an fd which is not in uncompleted AS array\n");
      return -1;
   }
   if(unc_as_t[i].flags & HAS_NAME){/*shouldn't happen, if it has a name, it shouldnt be in fdset[]*/
      LM_WARN("this shouldn't happen\n");
      return 0;/*already have a name, please take me out the uncompleted AS array*/
   }
   LM_DBG("Reading client name\n");

   if(-1==(len=read_name(fd,unc_as_t[i].name,MAX_AS_NAME))){
      /*this guy should be disconnected, it sent an AS_NAME too long*/
      LM_ERR("Bad name passed from fd\n");
      unc_as_t[i].valid=0;
      unc_as_t[i].flags=0;
      return -2;
   }else if(len==-2){
      LM_WARN("client disconnected\n");
      return -2;
   }
   name1=unc_as_t[i].name;

   /* Check the name isn't already taken */
   for(as=as_list;as;as=as->next){
      if(as->name.len==len && !memcmp(name1,as->name.s,len)){
    if(as->connected){
       LM_WARN("AppServer trying to connect with a name already taken (%.*s)\n",len,name1);
       unc_as_t[i].valid=0;
       unc_as_t[i].flags=0;
       return -2;
    }
    break;
      }
   }
   if (!as) {
      LM_ERR("a client tried to connect which is not declared in config. script(%.*s)\n",len,name1);
      unc_as_t[i].valid=0;
      unc_as_t[i].flags=0;
      return -2;
   }
   unc_as_t[i].flags |= HAS_NAME;
   /* the loop's upper bound, 
    * if 'i' is in the lower part, then look for an unc_as in the upper part*/
   k=(i>=MAX_UNC_AS_NR?MAX_UNC_AS_NR:2*MAX_UNC_AS_NR);
   /* the loop's lower bound */
   for(j=(i>=MAX_UNC_AS_NR?0:MAX_UNC_AS_NR);j<k;j++)
      if(unc_as_t[j].valid && 
       (unc_as_t[j].flags & HAS_NAME) && 
       !strcmp(unc_as_t[i].name,unc_as_t[j].name))
    break;
   LM_INFO("Fantastic, we have a new client: %s\n",unc_as_t[i].name);
   if(j==k)/* the unc_as peer's socket hasn't been found, just take this one out of fdset because it already has its name */
      return 0;/*take me out from fdset[]*/
   LM_INFO("EUREKA, we have a new completed AS: %s\n",unc_as_t[i].name);
   /* EUREKA ! we have a sweet pair of AS sockets, with the same name !!*/
   if(add_new_as(i<j?i:j,i<j?j:i,as)==-1){
      close(unc_as_t[j].fd);
      close(unc_as_t[i].fd);
      unc_as_t[j].valid=unc_as_t[i].valid=0;
      unc_as_t[j].flags=unc_as_t[i].flags=0;
      return -1;
   }
   unc_as_t[j].valid=unc_as_t[i].valid=0;
   unc_as_t[j].flags=unc_as_t[i].flags=0;
   return unc_as_t[i<j?i:j].fd;
}


/* This bloated function reads the name of the AS from the
 * socket and copies it into dst, then returns strlen(dst)
 * or -1 if error.
 */
static inline int read_name(int sock,char *dst,int dstlen)
{
   int n,namelen;
   namelen=0;
try_again1:
   if((n=read(sock,&namelen,1))<0){
      if(errno==EINTR)
    goto try_again1;
      else{
    LM_ERR("trying to read length from fd=%d (%s)\n",sock,strerror(errno));
    return -1;
      }
   }else if(n==0){
      LM_WARN("uncomplete AS has disconnected before giving its name\n");
      return -2;
   }
   if(namelen>dstlen || namelen==0){
      LM_ERR("name too long to fit in dst (%d > %d)\n",namelen,dstlen);
      return -1;
   }
try_again2:
   if((n=read(sock,dst,namelen))<0){
      if(errno==EINTR)
    goto try_again2;
      else{
    LM_ERR("trying to read %d chars into %p from fd=%d (%s)\n",namelen,dst,sock,strerror(errno));
    return -1;
      }
   }else if(n==0){
      LM_WARN("uncomplete AS has disconnected before giving its name\n");
      return -2;
   }
   dst[namelen]=0;
   return namelen;
}

/* handle new App Server connect. 
 * params:
 * fd:
 *    fd on which to accept.
 * which:
 *    if the fd is the event one, which='e', if is action, which='a'
 *
 * TODO: not very reliable, because if someone connects() to one of the serversockets 
 * but not to the other one, then synchronization would be lost, and any subsequent connect
 * attempts would fail (remember, we receive a connect in event[] and wait for a connect in action)
 * the point is, the connects must allways come in pairs, if one comes alone, we lost sync.
 * we should put kind of timeout in connects or something...
 */
static int new_as_connect(int fd,char which)
{
   union sockaddr_union su;
   int sock,i,flags;
   socklen_t su_len;

   su_len = sizeof(union sockaddr_union);
   sock=-1;
again:
   sock=accept(fd, &su.s, &su_len);
   if(sock==-1){
      if(errno==EINTR){
    goto again;
      }else{
       LM_ERR("while accepting connection: %s\n", strerror(errno));
    return -1;
      }
   }
   switch(which){
      case 'e':
    for(i=0;i<MAX_UNC_AS_NR && unc_as_t[i].valid;i++);
    if(i==MAX_UNC_AS_NR){
       LM_WARN("no more uncomplete connections allowed\n");
       goto error;
    }
    unc_as_t[i].fd=sock;
    unc_as_t[i].valid=1;
    unc_as_t[i].flags=HAS_FD;
    memcpy(&unc_as_t[i].su,&su,su_len);
    break;
      case 'a':
    for(i=MAX_UNC_AS_NR;(i<(2*MAX_UNC_AS_NR)) && unc_as_t[i].valid;i++);
    if(i==2*MAX_UNC_AS_NR){
       LM_WARN("no more uncomplete connections allowed\n");
       goto error;
    }
    unc_as_t[i].fd=sock;
    unc_as_t[i].valid=1;
    unc_as_t[i].flags=HAS_FD;
    memcpy(&unc_as_t[i].su,&su,su_len);
    break;
      default:
    break;
   }
   flags=1;
   if ((setsockopt(sock, IPPROTO_TCP , TCP_NODELAY,&flags, sizeof(flags))<0) ){
      LM_WARN("could not disable Nagle: %s\n",
       strerror(errno));
   }

   return sock;
error:
   if(sock!=-1)
      close(sock);
   return -1;
}

int spawn_action_dispatcher(struct as_entry *the_as)
{
   pid_t pid;
   pid=fork();
   if(pid<0){
      LM_ERR("unable to fork an action dispatcher for %.*s\n",the_as->name.len,the_as->name.s);
      return -1;
   }
   if(pid==0){/*child*/
      my_as = the_as;
      is_dispatcher=0;
      dispatch_actions();
      exit(0);
   }else{
      the_as->u.as.action_pid=pid;
   }
   return 0;
}