/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"


#define UDP_SELECT_TIMEOUT  1

/* 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;
    setsockopt(si->socket, IPPROTO_IP, IP_MTU_DISCOVER, (void*)&optval, sizeof(optval));
#else
#if defined(IP_DONTFRAG)
    optval = 1;
    setsockopt(si->socket, IPPROTO_IP, IP_DONTFRAG, (void*)&optval, sizeof(optval));
#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(UDP_SELECT_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(UDP_SELECT_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-07-16 06:51

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