/src/opensips/net/net_udp.c

Source
/*
 * Copyright (C) 2014-2015 OpenSIPS Foundation
 * Copyright (C) 2001-2003 FhG Fokus
 *
 * This file is part of opensips, a free SIP server.
 *
 * opensips 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
 *
 * opensips 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., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301  USA
 *
 *
 * History:
 * -------
 *  2015-02-09  first version (bogdan)
 */


#include <unistd.h>

#include "../ipc.h"
#include "../daemonize.h"
#include "../reactor.h"
#include "../timer.h"
#include "../pt_load.h"
#include "../cfg_reload.h"
#include "../profiling.h"
#include "net_udp.h"


/* if the UDP network layer is used or not by some protos */
static int udp_disabled = 1;

extern void handle_sigs(void);

/* initializes the UDP network layer */
int udp_init(void)
{
  unsigned int i;

  /* first we do auto-detection to see if there are any UDP based
   * protocols loaded */
  for ( i=PROTO_FIRST ; i<PROTO_LAST ; i++ )
    if (is_udp_based_proto(i)) {udp_disabled=0;break;}

  return 0;
}

/* destroys the UDP network layer */
void udp_destroy(void)
{
  return;
}

/* tells how many processes the UDP layer will create */
int udp_count_processes(unsigned int *extra)
{
  struct socket_info_full *sif;
  unsigned int n, e, i;

  if (udp_disabled) {
    if (extra) *extra = 0;
    return 0;
  }

  for( i=0,n=0,e=0 ; i<PROTO_LAST ; i++)
    if (protos[i].id!=PROTO_NONE && is_udp_based_proto(i))
      for( sif=protos[i].listeners ; sif; sif=sif->next) {
        const struct socket_info *si = &sif->socket_info;
        n+=si->workers;
        if (si->s_profile)
          if (si->s_profile->max_procs > si->workers)
            e+=si->s_profile->max_procs-si->workers;
      }

  if (extra) *extra = e;
  return n;
}

#ifdef USE_MCAST
/**
 * Setup a multicast receiver socket, supports IPv4 and IPv6.
 * \param sock socket
 * \param addr receiver address
 * \return zero on success, -1 otherwise
 */
static int setup_mcast_rcvr(int sock, union sockaddr_union* addr)
{
  struct ip_mreq mreq;
  struct ipv6_mreq mreq6;

  if (addr->s.sa_family==AF_INET){
    memcpy(&mreq.imr_multiaddr, &addr->sin.sin_addr,
           sizeof(struct in_addr));
    mreq.imr_interface.s_addr = htonl(INADDR_ANY);

    if (setsockopt(sock, IPPROTO_IP, IP_ADD_MEMBERSHIP,&mreq,
             sizeof(mreq))==-1){
      LM_ERR("setsockopt: %s\n", strerror(errno));
      return -1;
    }
  } else if (addr->s.sa_family==AF_INET6){
    memcpy(&mreq6.ipv6mr_multiaddr, &addr->sin6.sin6_addr,
           sizeof(struct in6_addr));
    mreq6.ipv6mr_interface = 0;
#ifdef __OS_linux
    if (setsockopt(sock, IPPROTO_IPV6, IPV6_ADD_MEMBERSHIP, &mreq6,
#else
    if (setsockopt(sock, IPPROTO_IPV6, IPV6_JOIN_GROUP, &mreq6,
#endif
             sizeof(mreq6))==-1){
      LM_ERR("setsockopt:%s\n",  strerror(errno));
      return -1;
    }
  } else {
    LM_ERR("unsupported protocol family\n");
    return -1;
  }
  return 0;
}

#endif /* USE_MCAST */


/**
 * Initialize a UDP socket, supports multicast, IPv4 and IPv6.
 * \param si socket that should be bind
 * \return zero on success, -1 otherwise
 *
 * @status_flags - extra status flags to be set for the socket fd
 */
int udp_init_listener(struct socket_info *si, int status_flags)
{
  union sockaddr_union* addr;
  int optval;
#ifdef USE_MCAST
  unsigned char m_optval;
#endif

  addr=&si->su;
  if (init_su(addr, &si->address, si->port_no)<0){
    LM_ERR("could not init sockaddr_union\n");
    goto error;
  }

  si->socket = socket(AF2PF(addr->s.sa_family), SOCK_DGRAM, 0);
  if (si->socket==-1){
    LM_ERR("socket: %s\n", strerror(errno));
    goto error;
  }

  /* make socket non-blocking */
  if (status_flags) {
    optval=fcntl(si->socket, F_GETFL);
    if (optval==-1){
      LM_ERR("fcntl failed: (%d) %s\n", errno, strerror(errno));
      goto error;
    }
    if (fcntl(si->socket,F_SETFL,optval|status_flags)==-1){
      LM_ERR("set non-blocking failed: (%d) %s\n",
        errno, strerror(errno));
      goto error;
    }
  }

  /* set sock opts? */
  optval=1;
  if (setsockopt(si->socket, SOL_SOCKET, SO_REUSEADDR ,
          (void*)&optval, sizeof(optval)) ==-1){
    LM_ERR("setsockopt: %s\n", strerror(errno));
    goto error;
  }

  if (si->flags & SI_REUSEPORT) {
    optval=1;
    if (setsockopt(si->socket, SOL_SOCKET, SO_REUSEPORT ,
            (void*)&optval, sizeof(optval)) ==-1){
      LM_ERR("setsockopt: %s\n", strerror(errno));
      goto error;
    }
  }

  if (si->flags & SI_FRAG) {
    /* no DF */
#if defined(IP_MTU_DISCOVER)
    optval = IP_PMTUDISC_DONT;
    if (setsockopt(si->socket, IPPROTO_IP, IP_MTU_DISCOVER,
        (void*)&optval, sizeof(optval)) == -1) {
      LM_ERR("setsockopt: %s\n", strerror(errno));
      goto error;
    }
#else
#if defined(IP_DONTFRAG)
    optval = 1;
    if (setsockopt(si->socket, IPPROTO_IP, IP_DONTFRAG,
        (void*)&optval, sizeof(optval)) == -1) {
      LM_ERR("setsockopt: %s\n", strerror(errno));
      goto error;
    }
#else
    LM_ERR("DF flag is not supported by your system\n");
    goto error;
#endif
#endif
  }

  /* tos */
  optval = (si->tos > 0) ? si->tos : tos;
  if (optval > 0) {
    if (addr->s.sa_family==AF_INET6){
      if (setsockopt(si->socket,  IPPROTO_IPV6, IPV6_TCLASS, (void*)&optval, sizeof(optval)) ==-1){
        LM_WARN("setsockopt tos for IPV6: %s\n", strerror(errno));
        /* continue since this is not critical */
      }
    } else {
      if (setsockopt(si->socket, IPPROTO_IP, IP_TOS, (void*)&optval, sizeof(optval)) ==-1){
        LM_WARN("setsockopt tos: %s\n", strerror(errno));
        /* continue since this is not critical */
      }
    }
  }
#if defined (__linux__) && defined(UDP_ERRORS)
  optval=1;
  /* enable error receiving on unconnected sockets */
  if(setsockopt(si->socket, SOL_IP, IP_RECVERR,
          (void*)&optval, sizeof(optval)) ==-1){
    LM_ERR("setsockopt: %s\n", strerror(errno));
    goto error;
  }
#endif

#ifdef USE_MCAST
  if ((si->flags & SI_IS_MCAST)
      && (setup_mcast_rcvr(si->socket, addr)<0)){
      goto error;
  }
  /* set the multicast options */
  if (addr->s.sa_family==AF_INET){
    m_optval = mcast_loopback;
    if (setsockopt(si->socket, IPPROTO_IP, IP_MULTICAST_LOOP,
            &m_optval, sizeof(m_optval))==-1){
      LM_WARN("setsockopt(IP_MULTICAST_LOOP): %s\n", strerror(errno));
      /* it's only a warning because we might get this error if the
        network interface doesn't support multicasting */
    }
    if (mcast_ttl>=0){
      m_optval = mcast_ttl;
      if (setsockopt(si->socket, IPPROTO_IP, IP_MULTICAST_TTL,
            &m_optval, sizeof(m_optval))==-1){
        LM_ERR("setsockopt (IP_MULTICAST_TTL): %s\n", strerror(errno));
        goto error;
      }
    }
  } else if (addr->s.sa_family==AF_INET6){
    if (setsockopt(si->socket, IPPROTO_IPV6, IPV6_MULTICAST_LOOP,
            &mcast_loopback, sizeof(mcast_loopback))==-1){
      LM_WARN("setsockopt (IPV6_MULTICAST_LOOP): %s\n", strerror(errno));
      /* it's only a warning because we might get this error if the
        network interface doesn't support multicasting */
    }
    if (mcast_ttl>=0){
      if (setsockopt(si->socket, IPPROTO_IP, IPV6_MULTICAST_HOPS,
            &mcast_ttl, sizeof(mcast_ttl))==-1){
        LM_ERR("setssckopt (IPV6_MULTICAST_HOPS): %s\n",
            strerror(errno));
        goto error;
      }
    }
  } else {
    LM_ERR("unsupported protocol family %d\n", addr->s.sa_family);
    goto error;
  }
#endif /* USE_MCAST */

  if (probe_max_sock_buff(si->socket,0,MAX_RECV_BUFFER_SIZE,
        BUFFER_INCREMENT)==-1) goto error;

  return 0;

error:
  return -1;
}


int udp_bind_listener(struct socket_info *si)
{
  union sockaddr_union* addr = &si->su;
  if (bind(si->socket,  &addr->s, sockaddru_len(*addr))==-1){
    LM_ERR("bind(%x, %p, %d) on %s: %s\n", si->socket, &addr->s,
        (unsigned)sockaddru_len(*addr),  si->address_str.s,
        strerror(errno));
    if (addr->s.sa_family==AF_INET6)
      LM_ERR("might be caused by using a link "
          " local address, try site local or global\n");
    return -1;
  }
  return 0;
}


inline static int handle_io(struct fd_map* fm, int idx,int event_type)
{
  int n = 0;
  int read;

  pt_become_active();

  pre_run_handle_script_reload(fm->app_flags);

  profiling_proc_start( LEVEL_SIP, 1);

  switch(fm->type){
    case F_UDP_READ:
      profiling_proc_enter( LEVEL_SIP,
        ss_merge256(
          protos[((struct socket_info*)fm->data)->proto].name,
          " proto reading"),
        1 );
      n = protos[((struct socket_info*)fm->data)->proto].net.
        dgram.read( fm->data /*si*/, &read);
      profiling_proc_exit( LEVEL_SIP, "reading done", n );
      break;
    case F_TIMER_JOB:
      profiling_proc_enter( LEVEL_FULL, "timer_job", 1 );
      handle_timer_job();
      profiling_proc_exit( LEVEL_FULL, "timer_job", n);
      break;
    case F_SCRIPT_ASYNC:
      profiling_proc_enter( LEVEL_SIP, "async_script", 0 );
      n = async_script_resume_f( fm->fd, fm->data,
        (event_type==IO_WATCH_TIMEOUT)?1:0 );
      profiling_proc_exit( LEVEL_SIP, "async_script", n);
      break;
    case F_FD_ASYNC:
      profiling_proc_enter( LEVEL_SIP, "async_fd", 0 );
      n = async_fd_resume( fm->fd, fm->data);
      profiling_proc_exit( LEVEL_SIP, "async_fd", n);
      break;
    case F_LAUNCH_ASYNC:
      profiling_proc_enter( LEVEL_SIP, "async_launch", 0 );
      n = async_launch_resume( fm->fd, fm->data);
      profiling_proc_exit( LEVEL_SIP, "async_launch", n);
      break;
    case F_IPC:
      profiling_proc_enter( LEVEL_SIP, "ipc_job", 1 );
      ipc_handle_job(fm->fd);
      profiling_proc_exit( LEVEL_SIP, "ipc_job", n);
      break;
    default:
      LM_CRIT("unknown fd type %d in UDP worker\n", fm->type);
      n = -1;
      break;
  }

  if (reactor_is_empty() && _termination_in_progress==1) {
    LM_WARN("reactor got empty while termination in progress\n");
    ipc_handle_all_pending_jobs(IPC_FD_READ_SELF);
    if (reactor_is_empty())
      dynamic_process_final_exit();
  }

  profiling_proc_end( LEVEL_SIP, n );

  post_run_handle_script_reload();

  pt_become_idle();
  return n;
}


int udp_proc_reactor_init( struct socket_info *si )
{

  /* create the reactor for UDP proc */
  if ( init_worker_reactor( "UDP_worker", RCT_PRIO_MAX)<0 ) {
    LM_ERR("failed to init reactor\n");
    goto error;
  }

  /* init: start watching for the timer jobs */
  if (reactor_add_reader( timer_fd_out, F_TIMER_JOB, RCT_PRIO_TIMER,NULL)<0){
    LM_CRIT("failed to add timer pipe_out to reactor\n");
    goto error;
  }

  /* init: start watching for the IPC jobs */
  if (reactor_add_reader(IPC_FD_READ_SELF, F_IPC, RCT_PRIO_ASYNC, NULL)<0){
    LM_CRIT("failed to add IPC pipe to reactor\n");
    goto error;
  }

  /* init: start watching for IPC "dispatched" jobs */
  if (reactor_add_reader(IPC_FD_READ_SHARED, F_IPC, RCT_PRIO_ASYNC, NULL)<0){
    LM_CRIT("failed to add IPC shared pipe to reactor\n");
    return -1;
  }

  /* init: start watching the SIP UDP fd */
  if (reactor_add_reader( si->socket, F_UDP_READ, RCT_PRIO_NET, si)<0) {
    LM_CRIT("failed to add UDP listen socket to reactor\n");
    goto error;
  }

  return 0;
error:
  destroy_worker_reactor();
  return -1;
}


static int fork_dynamic_udp_process(void *si_filter)
{
  struct socket_info *si = (struct socket_info*)si_filter;
  int p_id;
  const struct internal_fork_params ifp_udp_rcv = {
    .proc_desc = "UDP receiver",
    .flags = OSS_PROC_DYNAMIC|OSS_PROC_NEEDS_SCRIPT,
    .type = TYPE_UDP,
  };

  if ((p_id=internal_fork(&ifp_udp_rcv))<0) {
    LM_CRIT("cannot fork UDP process\n");
    return(-1);
  } else if (p_id==0) {
    /* new UDP process */
    /* set a more detailed description */
    set_proc_attrs("SIP receiver %.*s",
      si->sock_str.len, si->sock_str.s);
    pt[process_no].pg_filter = si;
    bind_address=si; /* shortcut */
    /* we first need to init the reactor to be able to add fd
     * into it in child_init routines */
    if (udp_proc_reactor_init(si) < 0 ||
    init_child(10000/*FIXME*/) < 0 ||
    self_update_routing_script() < 0) {
      goto error;
    }
    report_conditional_status( 1, 0); /*report success*/
    /* the child proc is done read&write) dealing with the status pipe */
    clean_read_pipeend();

    reactor_main_loop( worker_reactor_timeout, error, );
    destroy_worker_reactor();
error:
    report_failure_status();
    LM_ERR("Initializing new process failed, exiting with error \n");
    pt[process_no].flags |= OSS_PROC_SELFEXIT;
    exit( -1);
  } else {
    /*parent/main*/
    return p_id;
  }
}


static void udp_process_graceful_terminate(int sender, void *param)
{
  /* we accept this only from the main proccess */
  if (sender!=0) {
    LM_BUG("graceful terminate received from a non-main process!!\n");
    return;
  }
  LM_NOTICE("process %d received RPC to terminate from Main\n",process_no);

  /*remove from reactor all the shared fds, so we stop reading from them */

  /*remove timer jobs pipe */
  reactor_del_reader( timer_fd_out, -1, 0);

  /*remove IPC dispatcher pipe */
  reactor_del_reader( IPC_FD_READ_SHARED, -1, 0);

  /*remove network interface */
  reactor_del_reader( bind_address->socket, -1, 0);

  /*remove private IPC pipe */
  reactor_del_reader( IPC_FD_READ_SELF, -1, 0);

  /* let's drain the private IPC */
  ipc_handle_all_pending_jobs(IPC_FD_READ_SELF);

  /* what is left now is the reactor are async fd's, so we need to 
   * wait to complete all of them */
  if (reactor_is_empty())
    dynamic_process_final_exit();

  /* the exit will be triggered by the reactor, when empty */
  _termination_in_progress = 1;
  LM_INFO("reactor not empty, waiting for pending async\n");
}


/* starts all UDP related processes */
int udp_start_processes(int *chd_rank, int *startup_done)
{
  struct socket_info_full *sif;
  int p_id;
  int i,p;
  const struct internal_fork_params ifp_udp_rcv = {
    .proc_desc = "UDP receiver",
    .flags = OSS_PROC_NEEDS_SCRIPT,
    .type = TYPE_UDP,
  };

  if (udp_disabled)
    return 0;

  for( p=PROTO_FIRST ; p<PROTO_LAST ; p++ ) {
    if ( !is_udp_based_proto(p) )
      continue;

    for( sif=protos[p].listeners; sif ; sif=sif->next ) {
      struct socket_info* si = &sif->socket_info;

      if ( auto_scaling_enabled && si->s_profile &&
      create_process_group( TYPE_UDP, si, si->s_profile,
      fork_dynamic_udp_process, udp_process_graceful_terminate)!=0)
        LM_ERR("failed to create group of UDP processes for <%.*s>, "
          "auto forking will not be possible\n",
          si->name.len, si->name.s);

      for (i=0;i<si->workers;i++) {
        (*chd_rank)++;
        if ( (p_id=internal_fork(&ifp_udp_rcv))<0 ) {
          LM_CRIT("cannot fork UDP process\n");
          goto error;
        } else if (p_id==0) {
          /* new UDP process */
          /* set a more detailed description */
          set_proc_attrs("SIP receiver %.*s",
            si->sock_str.len, si->sock_str.s);
          pt[process_no].pg_filter = si;
          bind_address=si; /* shortcut */
          /* we first need to init the reactor to be able to add fd
           * into it in child_init routines */
          if (udp_proc_reactor_init(si) < 0 ||
              init_child(*chd_rank) < 0) {
            report_failure_status();
            if (*chd_rank == 1 && startup_done)
              *startup_done = -1;
            exit(-1);
          }

          /* first UDP proc runs statup_route (if defined) */
          if(*chd_rank == 1 && startup_done!=NULL) {
            LM_DBG("running startup for first UDP\n");
            if(run_startup_route()< 0) {
              report_failure_status();
              *startup_done = -1;
              LM_ERR("Startup route processing failed\n");
              exit(-1);
            }
            *startup_done = 1;
          }

          report_conditional_status( (!no_daemon_mode), 0);

          /**
           * Main UDP receiver loop, processes data from the
           * network, does some error checking and save it in an
           * allocated buffer. This data is then forwarded to the
           * receive_msg function. If an dynamic buffer is used, the
           * buffer must be freed in later steps.
           * \see receive_msg
           * \see main_loop
           */
          reactor_main_loop( worker_reactor_timeout, error, );
          destroy_worker_reactor();
          exit(-1);
        } else {
          /*parent*/
          /* wait for first proc to finish the startup route */
          if (*chd_rank == 1 && startup_done)
            while(!(*startup_done)) {
              usleep(5);
              handle_sigs();
            }
        }
      } /* procs per listener */
    } /* looping through the listeners per proto */
  } /* looping through the available protos */

  return 0;
error:
  return -1;
}

Coverage Report

Created: 2026-08-31 06:36

Line	Count	Source
1		/*
2		* Copyright (C) 2014-2015 OpenSIPS Foundation
3		* Copyright (C) 2001-2003 FhG Fokus
4		*
5		* This file is part of opensips, a free SIP server.
6		*
7		* opensips is free software; you can redistribute it and/or modify
8		* it under the terms of the GNU General Public License as published by
9		* the Free Software Foundation; either version 2 of the License, or
10		* (at your option) any later version
11		*
12		* opensips is distributed in the hope that it will be useful,
13		* but WITHOUT ANY WARRANTY; without even the implied warranty of
14		* MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
15		* GNU General Public License for more details.
16		*
17		* You should have received a copy of the GNU General Public License
18		* along with this program; if not, write to the Free Software
19		* Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA
20		*
21		*
22		* History:
23		* -------
24		* 2015-02-09 first version (bogdan)
25		*/
26
27
28		#include <unistd.h>
29
30		#include "../ipc.h"
31		#include "../daemonize.h"
32		#include "../reactor.h"
33		#include "../timer.h"
34		#include "../pt_load.h"
35		#include "../cfg_reload.h"
36		#include "../profiling.h"
37		#include "net_udp.h"
38
39
40		/* if the UDP network layer is used or not by some protos */
41		static int udp_disabled = 1;
42
43		extern void handle_sigs(void);
44
45		/* initializes the UDP network layer */
46		int udp_init(void)
47	0	{
48	0	unsigned int i;
49
50		/* first we do auto-detection to see if there are any UDP based
51		* protocols loaded */
52	0	for ( i=PROTO_FIRST ; i<PROTO_LAST ; i++ )
53	0	if (is_udp_based_proto(i)) {udp_disabled=0;break;}
54
55	0	return 0;
56	0	}
57
58		/* destroys the UDP network layer */
59		void udp_destroy(void)
60	0	{
61	0	return;
62	0	}
63
64		/* tells how many processes the UDP layer will create */
65		int udp_count_processes(unsigned int *extra)
66	0	{
67	0	struct socket_info_full *sif;
68	0	unsigned int n, e, i;
69
70	0	if (udp_disabled) {
71	0	if (extra) *extra = 0;
72	0	return 0;
73	0	}
74
75	0	for( i=0,n=0,e=0 ; i<PROTO_LAST ; i++)
76	0	if (protos[i].id!=PROTO_NONE && is_udp_based_proto(i))
77	0	for( sif=protos[i].listeners ; sif; sif=sif->next) {
78	0	const struct socket_info *si = &sif->socket_info;
79	0	n+=si->workers;
80	0	if (si->s_profile)
81	0	if (si->s_profile->max_procs > si->workers)
82	0	e+=si->s_profile->max_procs-si->workers;
83	0	}
84
85	0	if (extra) *extra = e;
86	0	return n;
87	0	}
88
89		#ifdef USE_MCAST
90		/**
91		* Setup a multicast receiver socket, supports IPv4 and IPv6.
92		* \param sock socket
93		* \param addr receiver address
94		* \return zero on success, -1 otherwise
95		*/
96		static int setup_mcast_rcvr(int sock, union sockaddr_union* addr)
97		{
98		struct ip_mreq mreq;
99		struct ipv6_mreq mreq6;
100
101		if (addr->s.sa_family==AF_INET){
102		memcpy(&mreq.imr_multiaddr, &addr->sin.sin_addr,
103		sizeof(struct in_addr));
104		mreq.imr_interface.s_addr = htonl(INADDR_ANY);
105
106		if (setsockopt(sock, IPPROTO_IP, IP_ADD_MEMBERSHIP,&mreq,
107		sizeof(mreq))==-1){
108		LM_ERR("setsockopt: %s\n", strerror(errno));
109		return -1;
110		}
111		} else if (addr->s.sa_family==AF_INET6){
112		memcpy(&mreq6.ipv6mr_multiaddr, &addr->sin6.sin6_addr,
113		sizeof(struct in6_addr));
114		mreq6.ipv6mr_interface = 0;
115		#ifdef __OS_linux
116		if (setsockopt(sock, IPPROTO_IPV6, IPV6_ADD_MEMBERSHIP, &mreq6,
117		#else
118		if (setsockopt(sock, IPPROTO_IPV6, IPV6_JOIN_GROUP, &mreq6,
119		#endif
120		sizeof(mreq6))==-1){
121		LM_ERR("setsockopt:%s\n", strerror(errno));
122		return -1;
123		}
124		} else {
125		LM_ERR("unsupported protocol family\n");
126		return -1;
127		}
128		return 0;
129		}
130
131		#endif /* USE_MCAST */
132
133
134		/**
135		* Initialize a UDP socket, supports multicast, IPv4 and IPv6.
136		* \param si socket that should be bind
137		* \return zero on success, -1 otherwise
138		*
139		* @status_flags - extra status flags to be set for the socket fd
140		*/
141		int udp_init_listener(struct socket_info *si, int status_flags)
142	0	{
143	0	union sockaddr_union* addr;
144	0	int optval;
145		#ifdef USE_MCAST
146		unsigned char m_optval;
147		#endif
148
149	0	addr=&si->su;
150	0	if (init_su(addr, &si->address, si->port_no)<0){
151	0	LM_ERR("could not init sockaddr_union\n");
152	0	goto error;
153	0	}
154
155	0	si->socket = socket(AF2PF(addr->s.sa_family), SOCK_DGRAM, 0);
156	0	if (si->socket==-1){
157	0	LM_ERR("socket: %s\n", strerror(errno));
158	0	goto error;
159	0	}
160
161		/* make socket non-blocking */
162	0	if (status_flags) {
163	0	optval=fcntl(si->socket, F_GETFL);
164	0	if (optval==-1){
165	0	LM_ERR("fcntl failed: (%d) %s\n", errno, strerror(errno));
166	0	goto error;
167	0	}
168	0	if (fcntl(si->socket,F_SETFL,optval\|status_flags)==-1){
169	0	LM_ERR("set non-blocking failed: (%d) %s\n",
170	0	errno, strerror(errno));
171	0	goto error;
172	0	}
173	0	}
174
175		/* set sock opts? */
176	0	optval=1;
177	0	if (setsockopt(si->socket, SOL_SOCKET, SO_REUSEADDR ,
178	0	(void*)&optval, sizeof(optval)) ==-1){
179	0	LM_ERR("setsockopt: %s\n", strerror(errno));
180	0	goto error;
181	0	}
182
183	0	if (si->flags & SI_REUSEPORT) {
184	0	optval=1;
185	0	if (setsockopt(si->socket, SOL_SOCKET, SO_REUSEPORT ,
186	0	(void*)&optval, sizeof(optval)) ==-1){
187	0	LM_ERR("setsockopt: %s\n", strerror(errno));
188	0	goto error;
189	0	}
190	0	}
191
192	0	if (si->flags & SI_FRAG) {
193		/* no DF */
194	0	#if defined(IP_MTU_DISCOVER)
195	0	optval = IP_PMTUDISC_DONT;
196	0	if (setsockopt(si->socket, IPPROTO_IP, IP_MTU_DISCOVER,
197	0	(void*)&optval, sizeof(optval)) == -1) {
198	0	LM_ERR("setsockopt: %s\n", strerror(errno));
199	0	goto error;
200	0	}
201		#else
202		#if defined(IP_DONTFRAG)
203		optval = 1;
204		if (setsockopt(si->socket, IPPROTO_IP, IP_DONTFRAG,
205		(void*)&optval, sizeof(optval)) == -1) {
206		LM_ERR("setsockopt: %s\n", strerror(errno));
207		goto error;
208		}
209		#else
210		LM_ERR("DF flag is not supported by your system\n");
211		goto error;
212		#endif
213		#endif
214	0	}
215
216		/* tos */
217	0	optval = (si->tos > 0) ? si->tos : tos;
218	0	if (optval > 0) {
219	0	if (addr->s.sa_family==AF_INET6){
220	0	if (setsockopt(si->socket, IPPROTO_IPV6, IPV6_TCLASS, (void*)&optval, sizeof(optval)) ==-1){
221	0	LM_WARN("setsockopt tos for IPV6: %s\n", strerror(errno));
222		/* continue since this is not critical */
223	0	}
224	0	} else {
225	0	if (setsockopt(si->socket, IPPROTO_IP, IP_TOS, (void*)&optval, sizeof(optval)) ==-1){
226	0	LM_WARN("setsockopt tos: %s\n", strerror(errno));
227		/* continue since this is not critical */
228	0	}
229	0	}
230	0	}
231		#if defined (__linux__) && defined(UDP_ERRORS)
232		optval=1;
233		/* enable error receiving on unconnected sockets */
234		if(setsockopt(si->socket, SOL_IP, IP_RECVERR,
235		(void*)&optval, sizeof(optval)) ==-1){
236		LM_ERR("setsockopt: %s\n", strerror(errno));
237		goto error;
238		}
239		#endif
240
241		#ifdef USE_MCAST
242		if ((si->flags & SI_IS_MCAST)
243		&& (setup_mcast_rcvr(si->socket, addr)<0)){
244		goto error;
245		}
246		/* set the multicast options */
247		if (addr->s.sa_family==AF_INET){
248		m_optval = mcast_loopback;
249		if (setsockopt(si->socket, IPPROTO_IP, IP_MULTICAST_LOOP,
250		&m_optval, sizeof(m_optval))==-1){
251		LM_WARN("setsockopt(IP_MULTICAST_LOOP): %s\n", strerror(errno));
252		/* it's only a warning because we might get this error if the
253		network interface doesn't support multicasting */
254		}
255		if (mcast_ttl>=0){
256		m_optval = mcast_ttl;
257		if (setsockopt(si->socket, IPPROTO_IP, IP_MULTICAST_TTL,
258		&m_optval, sizeof(m_optval))==-1){
259		LM_ERR("setsockopt (IP_MULTICAST_TTL): %s\n", strerror(errno));
260		goto error;
261		}
262		}
263		} else if (addr->s.sa_family==AF_INET6){
264		if (setsockopt(si->socket, IPPROTO_IPV6, IPV6_MULTICAST_LOOP,
265		&mcast_loopback, sizeof(mcast_loopback))==-1){
266		LM_WARN("setsockopt (IPV6_MULTICAST_LOOP): %s\n", strerror(errno));
267		/* it's only a warning because we might get this error if the
268		network interface doesn't support multicasting */
269		}
270		if (mcast_ttl>=0){
271		if (setsockopt(si->socket, IPPROTO_IP, IPV6_MULTICAST_HOPS,
272		&mcast_ttl, sizeof(mcast_ttl))==-1){
273		LM_ERR("setssckopt (IPV6_MULTICAST_HOPS): %s\n",
274		strerror(errno));
275		goto error;
276		}
277		}
278		} else {
279		LM_ERR("unsupported protocol family %d\n", addr->s.sa_family);
280		goto error;
281		}
282		#endif /* USE_MCAST */
283
284	0	if (probe_max_sock_buff(si->socket,0,MAX_RECV_BUFFER_SIZE,
285	0	BUFFER_INCREMENT)==-1) goto error;
286
287	0	return 0;
288
289	0	error:
290	0	return -1;
291	0	}
292
293
294		int udp_bind_listener(struct socket_info *si)
295	0	{
296	0	union sockaddr_union* addr = &si->su;
297	0	if (bind(si->socket, &addr->s, sockaddru_len(*addr))==-1){
298	0	LM_ERR("bind(%x, %p, %d) on %s: %s\n", si->socket, &addr->s,
299	0	(unsigned)sockaddru_len(*addr), si->address_str.s,
300	0	strerror(errno));
301	0	if (addr->s.sa_family==AF_INET6)
302	0	LM_ERR("might be caused by using a link "
303	0	" local address, try site local or global\n");
304	0	return -1;
305	0	}
306	0	return 0;
307	0	}
308
309
310		inline static int handle_io(struct fd_map* fm, int idx,int event_type)
311	0	{
312	0	int n = 0;
313	0	int read;
314
315	0	pt_become_active();
316
317	0	pre_run_handle_script_reload(fm->app_flags);
318
319	0	profiling_proc_start( LEVEL_SIP, 1);
320
321	0	switch(fm->type){
322	0	case F_UDP_READ:
323	0	profiling_proc_enter( LEVEL_SIP,
324	0	ss_merge256(
325	0	protos[((struct socket_info*)fm->data)->proto].name,
326	0	" proto reading"),
327	0	1 );
328	0	n = protos[((struct socket_info*)fm->data)->proto].net.
329	0	dgram.read( fm->data /si/, &read);
330	0	profiling_proc_exit( LEVEL_SIP, "reading done", n );
331	0	break;
332	0	case F_TIMER_JOB:
333	0	profiling_proc_enter( LEVEL_FULL, "timer_job", 1 );
334	0	handle_timer_job();
335	0	profiling_proc_exit( LEVEL_FULL, "timer_job", n);
336	0	break;
337	0	case F_SCRIPT_ASYNC:
338	0	profiling_proc_enter( LEVEL_SIP, "async_script", 0 );
339	0	n = async_script_resume_f( fm->fd, fm->data,
340	0	(event_type==IO_WATCH_TIMEOUT)?1:0 );
341	0	profiling_proc_exit( LEVEL_SIP, "async_script", n);
342	0	break;
343	0	case F_FD_ASYNC:
344	0	profiling_proc_enter( LEVEL_SIP, "async_fd", 0 );
345	0	n = async_fd_resume( fm->fd, fm->data);
346	0	profiling_proc_exit( LEVEL_SIP, "async_fd", n);
347	0	break;
348	0	case F_LAUNCH_ASYNC:
349	0	profiling_proc_enter( LEVEL_SIP, "async_launch", 0 );
350	0	n = async_launch_resume( fm->fd, fm->data);
351	0	profiling_proc_exit( LEVEL_SIP, "async_launch", n);
352	0	break;
353	0	case F_IPC:
354	0	profiling_proc_enter( LEVEL_SIP, "ipc_job", 1 );
355	0	ipc_handle_job(fm->fd);
356	0	profiling_proc_exit( LEVEL_SIP, "ipc_job", n);
357	0	break;
358	0	default:
359	0	LM_CRIT("unknown fd type %d in UDP worker\n", fm->type);
360	0	n = -1;
361	0	break;
362	0	}
363
364	0	if (reactor_is_empty() && _termination_in_progress==1) {
365	0	LM_WARN("reactor got empty while termination in progress\n");
366	0	ipc_handle_all_pending_jobs(IPC_FD_READ_SELF);
367	0	if (reactor_is_empty())
368	0	dynamic_process_final_exit();
369	0	}
370
371	0	profiling_proc_end( LEVEL_SIP, n );
372
373	0	post_run_handle_script_reload();
374
375	0	pt_become_idle();
376	0	return n;
377	0	}
378
379
380		int udp_proc_reactor_init( struct socket_info *si )
381	0	{
382
383		/* create the reactor for UDP proc */
384	0	if ( init_worker_reactor( "UDP_worker", RCT_PRIO_MAX)<0 ) {
385	0	LM_ERR("failed to init reactor\n");
386	0	goto error;
387	0	}
388
389		/* init: start watching for the timer jobs */
390	0	if (reactor_add_reader( timer_fd_out, F_TIMER_JOB, RCT_PRIO_TIMER,NULL)<0){
391	0	LM_CRIT("failed to add timer pipe_out to reactor\n");
392	0	goto error;
393	0	}
394
395		/* init: start watching for the IPC jobs */
396	0	if (reactor_add_reader(IPC_FD_READ_SELF, F_IPC, RCT_PRIO_ASYNC, NULL)<0){
397	0	LM_CRIT("failed to add IPC pipe to reactor\n");
398	0	goto error;
399	0	}
400
401		/* init: start watching for IPC "dispatched" jobs */
402	0	if (reactor_add_reader(IPC_FD_READ_SHARED, F_IPC, RCT_PRIO_ASYNC, NULL)<0){
403	0	LM_CRIT("failed to add IPC shared pipe to reactor\n");
404	0	return -1;
405	0	}
406
407		/* init: start watching the SIP UDP fd */
408	0	if (reactor_add_reader( si->socket, F_UDP_READ, RCT_PRIO_NET, si)<0) {
409	0	LM_CRIT("failed to add UDP listen socket to reactor\n");
410	0	goto error;
411	0	}
412
413	0	return 0;
414	0	error:
415	0	destroy_worker_reactor();
416	0	return -1;
417	0	}
418
419
420		static int fork_dynamic_udp_process(void *si_filter)
421	0	{
422	0	struct socket_info si = (struct socket_info)si_filter;
423	0	int p_id;
424	0	const struct internal_fork_params ifp_udp_rcv = {
425	0	.proc_desc = "UDP receiver",
426	0	.flags = OSS_PROC_DYNAMIC\|OSS_PROC_NEEDS_SCRIPT,
427	0	.type = TYPE_UDP,
428	0	};
429
430	0	if ((p_id=internal_fork(&ifp_udp_rcv))<0) {
431	0	LM_CRIT("cannot fork UDP process\n");
432	0	return(-1);
433	0	} else if (p_id==0) {
434		/* new UDP process */
435		/* set a more detailed description */
436	0	set_proc_attrs("SIP receiver %.*s",
437	0	si->sock_str.len, si->sock_str.s);
438	0	pt[process_no].pg_filter = si;
439	0	bind_address=si; /* shortcut */
440		/* we first need to init the reactor to be able to add fd
441		* into it in child_init routines */
442	0	if (udp_proc_reactor_init(si) < 0 \|\|
443	0	init_child(10000/FIXME/) < 0 \|\|
444	0	self_update_routing_script() < 0) {
445	0	goto error;
446	0	}
447	0	report_conditional_status( 1, 0); /report success/
448		/* the child proc is done read&write) dealing with the status pipe */
449	0	clean_read_pipeend();
450
451	0	reactor_main_loop( worker_reactor_timeout, error, );
452	0	destroy_worker_reactor();
453	0	error:
454	0	report_failure_status();
455	0	LM_ERR("Initializing new process failed, exiting with error \n");
456	0	pt[process_no].flags \|= OSS_PROC_SELFEXIT;
457	0	exit( -1);
458	0	} else {
459		/parent/main/
460	0	return p_id;
461	0	}
462	0	}
463
464
465		static void udp_process_graceful_terminate(int sender, void *param)
466	0	{
467		/* we accept this only from the main proccess */
468	0	if (sender!=0) {
469	0	LM_BUG("graceful terminate received from a non-main process!!\n");
470	0	return;
471	0	}
472	0	LM_NOTICE("process %d received RPC to terminate from Main\n",process_no);
473
474		/remove from reactor all the shared fds, so we stop reading from them /
475
476		/remove timer jobs pipe /
477	0	reactor_del_reader( timer_fd_out, -1, 0);
478
479		/remove IPC dispatcher pipe /
480	0	reactor_del_reader( IPC_FD_READ_SHARED, -1, 0);
481
482		/remove network interface /
483	0	reactor_del_reader( bind_address->socket, -1, 0);
484
485		/remove private IPC pipe /
486	0	reactor_del_reader( IPC_FD_READ_SELF, -1, 0);
487
488		/* let's drain the private IPC */
489	0	ipc_handle_all_pending_jobs(IPC_FD_READ_SELF);
490
491		/* what is left now is the reactor are async fd's, so we need to
492		* wait to complete all of them */
493	0	if (reactor_is_empty())
494	0	dynamic_process_final_exit();
495
496		/* the exit will be triggered by the reactor, when empty */
497	0	_termination_in_progress = 1;
498	0	LM_INFO("reactor not empty, waiting for pending async\n");
499	0	}
500
501
502		/* starts all UDP related processes */
503		int udp_start_processes(int chd_rank, int startup_done)
504	0	{
505	0	struct socket_info_full *sif;
506	0	int p_id;
507	0	int i,p;
508	0	const struct internal_fork_params ifp_udp_rcv = {
509	0	.proc_desc = "UDP receiver",
510	0	.flags = OSS_PROC_NEEDS_SCRIPT,
511	0	.type = TYPE_UDP,
512	0	};
513
514	0	if (udp_disabled)
515	0	return 0;
516
517	0	for( p=PROTO_FIRST ; p<PROTO_LAST ; p++ ) {
518	0	if ( !is_udp_based_proto(p) )
519	0	continue;
520
521	0	for( sif=protos[p].listeners; sif ; sif=sif->next ) {
522	0	struct socket_info* si = &sif->socket_info;
523
524	0	if ( auto_scaling_enabled && si->s_profile &&
525	0	create_process_group( TYPE_UDP, si, si->s_profile,
526	0	fork_dynamic_udp_process, udp_process_graceful_terminate)!=0)
527	0	LM_ERR("failed to create group of UDP processes for <%.*s>, "
528	0	"auto forking will not be possible\n",
529	0	si->name.len, si->name.s);
530
531	0	for (i=0;i<si->workers;i++) {
532	0	(*chd_rank)++;
533	0	if ( (p_id=internal_fork(&ifp_udp_rcv))<0 ) {
534	0	LM_CRIT("cannot fork UDP process\n");
535	0	goto error;
536	0	} else if (p_id==0) {
537		/* new UDP process */
538		/* set a more detailed description */
539	0	set_proc_attrs("SIP receiver %.*s",
540	0	si->sock_str.len, si->sock_str.s);
541	0	pt[process_no].pg_filter = si;
542	0	bind_address=si; /* shortcut */
543		/* we first need to init the reactor to be able to add fd
544		* into it in child_init routines */
545	0	if (udp_proc_reactor_init(si) < 0 \|\|
546	0	init_child(*chd_rank) < 0) {
547	0	report_failure_status();
548	0	if (*chd_rank == 1 && startup_done)
549	0	*startup_done = -1;
550	0	exit(-1);
551	0	}
552
553		/* first UDP proc runs statup_route (if defined) */
554	0	if(*chd_rank == 1 && startup_done!=NULL) {
555	0	LM_DBG("running startup for first UDP\n");
556	0	if(run_startup_route()< 0) {
557	0	report_failure_status();
558	0	*startup_done = -1;
559	0	LM_ERR("Startup route processing failed\n");
560	0	exit(-1);
561	0	}
562	0	*startup_done = 1;
563	0	}
564
565	0	report_conditional_status( (!no_daemon_mode), 0);
566
567		/**
568		* Main UDP receiver loop, processes data from the
569		* network, does some error checking and save it in an
570		* allocated buffer. This data is then forwarded to the
571		* receive_msg function. If an dynamic buffer is used, the
572		* buffer must be freed in later steps.
573		* \see receive_msg
574		* \see main_loop
575		*/
576	0	reactor_main_loop( worker_reactor_timeout, error, );
577	0	destroy_worker_reactor();
578	0	exit(-1);
579	0	} else {
580		/parent/
581		/* wait for first proc to finish the startup route */
582	0	if (*chd_rank == 1 && startup_done)
583	0	while(!(*startup_done)) {
584	0	usleep(5);
585	0	handle_sigs();
586	0	}
587	0	}
588	0	} /* procs per listener */
589	0	} /* looping through the listeners per proto */
590	0	} /* looping through the available protos */
591
592	0	return 0;
593	0	error:
594	0	return -1;
595	0	}