pt.c

Go to the documentation of this file.

/*
 * $Id$
 *
 * Process Table
 *
 * Copyright (C) 2001-2003 FhG Fokus
 *
 * Permission to use, copy, modify, and distribute this software for any
 * purpose with or without fee is hereby granted, provided that the above
 * copyright notice and this permission notice appear in all copies.
 *
 * THE SOFTWARE IS PROVIDED "AS IS" AND THE AUTHOR DISCLAIMS ALL WARRANTIES
 * WITH REGARD TO THIS SOFTWARE INCLUDING ALL IMPLIED WARRANTIES OF
 * MERCHANTABILITY AND FITNESS. IN NO EVENT SHALL THE AUTHOR BE LIABLE FOR
 * ANY SPECIAL, DIRECT, INDIRECT, OR CONSEQUENTIAL DAMAGES OR ANY DAMAGES
 * WHATSOEVER RESULTING FROM LOSS OF USE, DATA OR PROFITS, WHETHER IN AN
 * ACTION OF CONTRACT, NEGLIGENCE OR OTHER TORTIOUS ACTION, ARISING OUT OF
 * OR IN CONNECTION WITH THE USE OR PERFORMANCE OF THIS SOFTWARE.
 */
/*
 * History:
 * --------
 *  2006-06-14  added process table in shared mem (dragos)
 *  2006-09-20  added profile support (-DPROFILING) (hscholz)
 *  2006-10-25  sanity check before allowing forking w/ tcp support (is_main
 *               & tcp not started yet); set is_main=0 in childs (andrei)
 *  2007-07-04  added register_fds() and get_max_open_fds(() (andrei)
 *  2010-08-19  use daemon_status_on_fork_cleanup() (andrei)
 */
#include "pt.h"
#include "tcp_init.h"
#include "sr_module.h"
#include "socket_info.h"
#include "rand/fastrand.h"
#ifdef PKG_MALLOC
#include "mem/mem.h"
#endif
#ifdef SHM_MEM
#include "mem/shm_mem.h"
#endif
#if defined PKG_MALLOC || defined SHM_MEM
#include "cfg_core.h"
#endif
#include "daemonize.h"

#include <stdio.h>
#include <time.h> /* time(), used to initialize random numbers */

#define FORK_DONT_WAIT  /* child doesn't wait for parent before starting 
                                                   => faster startup, but the child should not assume
                                                   the parent fixed the pt[] entry for it */


#ifdef PROFILING
#include <sys/gmon.h>

        extern void _start(void);
        extern void etext(void);
#endif


static int estimated_proc_no=0;
static int estimated_fds_no=0;

/* number of usec to wait before forking a process */
static unsigned int fork_delay = 0;

unsigned int set_fork_delay(unsigned int v)
{
        unsigned int r;
        r =  fork_delay;
        fork_delay = v;
        return r;
}

/* number of known "common" used fds */
static int calc_common_open_fds_no(void)
{
        int max_fds_no;
        
        /* 1 tcp send unix socket/all_proc, 
         *  + 1 udp sock/udp proc + 1 possible dns comm. socket + 
         *  + 1 temporary tcp send sock.
         * Per process:
         *  + 1 if mhomed (now mhomed keeps a "cached socket" per process)
         *  + 1 if mhomed & ipv6
         */
        max_fds_no=estimated_proc_no*4 /* udp|sctp + tcp unix sock +
                                                                          tmp. tcp send +
                                                                          tmp dns.*/
                                -1 /* timer (no udp)*/ + 3 /* stdin/out/err */ +
#ifdef USE_IPV6
                                2*mhomed
#else
                                mhomed
#endif /* USE_IPV6*/
                                ;
        return max_fds_no;
}



/* returns 0 on success, -1 on error */
int init_pt(int proc_no)
{
#ifdef USE_TCP
        int r;
#endif
        
        estimated_proc_no+=proc_no;
        estimated_fds_no+=calc_common_open_fds_no();
        /*alloc pids*/
#ifdef SHM_MEM
        pt=shm_malloc(sizeof(struct process_table)*estimated_proc_no);
        process_count = shm_malloc(sizeof(int));
#else
        pt=pkg_malloc(sizeof(struct process_table)*estimated_proc_no);
        process_count = pkg_malloc(sizeof(int));
#endif
        process_lock = lock_alloc();
        process_lock = lock_init(process_lock);
        if (pt==0||process_count==0||process_lock==0){
                LOG(L_ERR, "ERROR: out  of memory\n");
                return -1;
        }
        memset(pt, 0, sizeof(struct process_table)*estimated_proc_no);
#ifdef USE_TCP
        for (r=0; r<estimated_proc_no; r++){
                pt[r].unix_sock=-1;
                pt[r].idx=-1;
        }
#endif
        process_no=0; /*main process number*/
        pt[process_no].pid=getpid();
        memcpy(pt[process_no].desc,"main",5);
        *process_count=1;
        return 0;
}


/* register no processes, used from mod_init when processes will be forked
 *  from mod_child 
 *  returns 0 on success, -1 on error
 */
int register_procs(int no)
{
        if (pt){
                LOG(L_CRIT, "BUG: register_procs(%d) called at runtime\n", no);
                return -1;
        }
        estimated_proc_no+=no;
        return 0;
}



/* returns the maximum number of processes */
int get_max_procs()
{
        if (pt==0){
                LOG(L_CRIT, "BUG: get_max_procs() called too early "
                                "(it must _not_ be called from mod_init())\n");
                abort(); /* crash to quickly catch offenders */
        }
        return estimated_proc_no;
}


/* register no fds, used from mod_init when modules will open more global
 *  fds (from mod_init or child_init(PROC_INIT)
 *  or from child_init(rank) when the module will open fds local to the
 *   process "rank".
 *   (this is needed because some other parts of ser code rely on knowing
 *    the maximum open fd number in a process)
 *  returns 0 on success, -1 on error
 */
int register_fds(int no)
{
        /* can be called at runtime, but should be called from child_init() */
        estimated_fds_no+=no;
        return 0;
}



/* returns the maximum open fd number */
int get_max_open_fds()
{
        if (pt==0){
                LOG(L_CRIT, "BUG: get_max_open_fds() called too early "
                                "(it must _not_ be called from mod_init())\n");
                abort(); /* crash to quickly catch offenders */
        }
        return estimated_fds_no;
}


/* return processes pid */
int my_pid()
{
        return pt ? pt[process_no].pid : getpid();
}



/* close unneeded sockets */
int close_extra_socks(int child_id, int proc_no)
{
#ifdef USE_TCP
        int r;
        struct socket_info* si;
        
        if (child_id!=PROC_TCP_MAIN){
                for (r=0; r<proc_no; r++){
                        if (pt[r].unix_sock>=0){
                                /* we can't change the value in pt[] because it's
                                 * shared so we only close it */
                                close(pt[r].unix_sock);
                        }
                }
                /* close all listen sockets (needed only in tcp_main */
                if (!tcp_disable){
                        for(si=tcp_listen; si; si=si->next){
                                close(si->socket);
                                /* safe to change since this is a per process copy */
                                si->socket=-1;
                        }
#ifdef USE_TLS
                        if (!tls_disable){
                                for(si=tls_listen; si; si=si->next){
                                        close(si->socket);
                                        /* safe to change since this is a per process copy */
                                        si->socket=-1;
                                }
                        }
#endif /* USE_TLS */
                }
                /* we still need the udp sockets (for sending) so we don't close them
                 * too */
        }
#endif /* USE_TCP */
        return 0;
}



int fork_process(int child_id, char *desc, int make_sock)
{
        int pid, child_process_no;
        int ret;
        unsigned int new_seed1;
        unsigned int new_seed2;
#ifdef USE_TCP
        int sockfd[2];
#endif

        if(unlikely(fork_delay>0))
                sleep_us(fork_delay);

        ret=-1;
        #ifdef USE_TCP
                sockfd[0]=sockfd[1]=-1;
                if(make_sock && !tcp_disable){
                         if (!is_main){
                                 LOG(L_CRIT, "BUG: fork_process(..., 1) called from a non "
                                                 "\"main\" process! If forking from a module's "
                                                 "child_init() fork only if rank==PROC_MAIN or"
                                                 " give up tcp send support (use 0 for make_sock)\n");
                                 goto error;
                         }
                         if (tcp_main_pid){
                                 LOG(L_CRIT, "BUG: fork_process(..., 1) called, but tcp main "
                                                 " is already started\n");
                                 goto error;
                         }
                         if (socketpair(AF_UNIX, SOCK_STREAM, 0, sockfd)<0){
                                LOG(L_ERR, "ERROR: fork_process(): socketpair failed: %s\n",
                                                        strerror(errno));
                                goto error;
                        }
                }
        #endif
        lock_get(process_lock);
        if (*process_count>=estimated_proc_no) {
                LOG(L_CRIT, "ERROR: fork_process(): Process limit of %d exceeded."
                                        " Will simulate fork fail.\n", estimated_proc_no);
                lock_release(process_lock);
                goto error;
        }       
        
        
        child_process_no = *process_count;
        new_seed1=rand();
        new_seed2=random();
        pid = fork();
        if (pid<0) {
                lock_release(process_lock);
                ret=pid;
                goto error;
        }else if (pid==0){
                /* child */
                is_main=0; /* a forked process cannot be the "main" one */
                process_no=child_process_no;
                daemon_status_on_fork_cleanup();
                /* close tcp unix sockets if this is not tcp main */
#ifdef USE_TCP
                close_extra_socks(child_id, process_no);
#endif /* USE_TCP */
                srand(new_seed1);
                fastrand_seed(rand());
                srandom(new_seed2+time(0));
                shm_malloc_on_fork();
#ifdef PROFILING
                monstartup((u_long) &_start, (u_long) &etext);
#endif
#ifdef FORK_DONT_WAIT
                /* record pid twice to avoid the child using it, before
                 * parent gets a chance to set it*/
                pt[process_no].pid=getpid();
#else
                /* wait for parent to get out of critical zone.
                 * this is actually relevant as the parent updates
                 * the pt & process_count. */
                lock_get(process_lock);
                lock_release(process_lock);
#endif
                #ifdef USE_TCP
                        if (make_sock && !tcp_disable){
                                close(sockfd[0]);
                                unix_tcp_sock=sockfd[1];
                        }
                #endif          
                if ((child_id!=PROC_NOCHLDINIT) && (init_child(child_id) < 0)) {
                        LOG(L_ERR, "ERROR: fork_process(): init_child failed for "
                                        " process %d, pid %d, \"%s\"\n", process_no,
                                        pt[process_no].pid, pt[process_no].desc);
                        return -1;
                }
                return pid;
        } else {
                /* parent */
                (*process_count)++;
#ifdef FORK_DONT_WAIT
                lock_release(process_lock);
#endif
                /* add the process to the list in shm */
                pt[child_process_no].pid=pid;
                if (desc){
                        strncpy(pt[child_process_no].desc, desc, MAX_PT_DESC);
                }
                #ifdef USE_TCP
                        if (make_sock && !tcp_disable){
                                close(sockfd[1]);
                                pt[child_process_no].unix_sock=sockfd[0];
                                pt[child_process_no].idx=-1; /* this is not a "tcp" process*/
                        }
                #endif
#ifdef FORK_DONT_WAIT
#else
                lock_release(process_lock);
#endif
                ret=pid;
                goto end;
        }
error:
#ifdef USE_TCP
        if (sockfd[0]!=-1) close(sockfd[0]);
        if (sockfd[1]!=-1) close(sockfd[1]);
#endif
end:
        return ret;
}

#ifdef USE_TCP
int fork_tcp_process(int child_id, char *desc, int r, int *reader_fd_1)
{
        int pid, child_process_no;
        int sockfd[2];
        int reader_fd[2]; /* for comm. with the tcp children read  */
        int ret;
        int i;
        unsigned int new_seed1;
        unsigned int new_seed2;
        
        /* init */
        sockfd[0]=sockfd[1]=-1;
        reader_fd[0]=reader_fd[1]=-1;
        ret=-1;
        
        if (!is_main){
                 LOG(L_CRIT, "BUG: fork_tcp_process() called from a non \"main\" "
                                        "process\n");
                 goto error;
         }
         if (tcp_main_pid){
                 LOG(L_CRIT, "BUG: fork_tcp_process(..., 1) called _after_ starting"
                                        " tcp main\n");
                 goto error;
         }
        if (socketpair(AF_UNIX, SOCK_STREAM, 0, sockfd)<0){
                LOG(L_ERR, "ERROR: fork_tcp_process(): socketpair failed: %s\n",
                                        strerror(errno));
                goto error;
        }
        if (socketpair(AF_UNIX, SOCK_STREAM, 0, reader_fd)<0){
                LOG(L_ERR, "ERROR: fork_tcp_process(): socketpair failed: %s\n",
                                        strerror(errno));
                goto error;
        }
        if (tcp_fix_child_sockets(reader_fd)<0){
                LOG(L_ERR, "ERROR: fork_tcp_process(): failed to set non blocking"
                                        "on child sockets\n");
                /* continue, it's not critical (it will go slower under
                 * very high connection rates) */
        }
        lock_get(process_lock);
        /* set the local process_no */
        if (*process_count>=estimated_proc_no) {
                LOG(L_CRIT, "ERROR: fork_tcp_process(): Process limit of %d exceeded."
                                        " Simulating fork fail\n", estimated_proc_no);
                lock_release(process_lock);
                goto error;
        }
        
        
        child_process_no = *process_count;
        new_seed1=rand();
        new_seed2=random();
        pid = fork();
        if (pid<0) {
                lock_release(process_lock);
                ret=pid;
                goto end;
        }
        if (pid==0){
                is_main=0; /* a forked process cannot be the "main" one */
                process_no=child_process_no;
                /* close unneeded unix sockets */
                close_extra_socks(child_id, process_no);
                /* same for unneeded tcp_children <-> tcp_main unix socks */
                for (i=0; i<r; i++){
                        if (tcp_children[i].unix_sock>=0){
                                close(tcp_children[i].unix_sock);
                                /* tcp_children is per process, so it's safe to change
                                 * the unix_sock to -1 */
                                tcp_children[i].unix_sock=-1;
                        }
                }
                daemon_status_on_fork_cleanup();
                srand(new_seed1);
                fastrand_seed(rand());
                srandom(new_seed2+time(0));
                shm_malloc_on_fork();
#ifdef PROFILING
                monstartup((u_long) &_start, (u_long) &etext);
#endif
#ifdef FORK_DONT_WAIT
                /* record pid twice to avoid the child using it, before
-                * parent gets a chance to set it*/
                pt[process_no].pid=getpid();
#else
                /* wait for parent to get out of critical zone */
                lock_get(process_lock);
                lock_release(process_lock);
#endif
                close(sockfd[0]);
                unix_tcp_sock=sockfd[1];
                close(reader_fd[0]);
                if (reader_fd_1) *reader_fd_1=reader_fd[1];
                if ((child_id!=PROC_NOCHLDINIT) && (init_child(child_id) < 0)) {
                        LOG(L_ERR, "ERROR: fork_tcp_process(): init_child failed for "
                                        "process %d, pid %d, \"%s\"\n", process_no, 
                                        pt[process_no].pid, pt[process_no].desc);
                        return -1;
                }
                return pid;
        } else {
                /* parent */
                (*process_count)++;
#ifdef FORK_DONT_WAIT
                lock_release(process_lock);
#endif
                /* add the process to the list in shm */
                pt[child_process_no].pid=pid;
                pt[child_process_no].unix_sock=sockfd[0];
                pt[child_process_no].idx=r;
                if (desc){
                        snprintf(pt[child_process_no].desc, MAX_PT_DESC, "%s child=%d", 
                                                desc, r);
                }
#ifdef FORK_DONT_WAIT
#else
                lock_release(process_lock);
#endif
                
                close(sockfd[1]);
                close(reader_fd[1]);
                
                tcp_children[r].pid=pid;
                tcp_children[r].proc_no=child_process_no;
                tcp_children[r].busy=0;
                tcp_children[r].n_reqs=0;
                tcp_children[r].unix_sock=reader_fd[0];
                
                ret=pid;
                goto end;
        }
error:
        if (sockfd[0]!=-1) close(sockfd[0]);
        if (sockfd[1]!=-1) close(sockfd[1]);
        if (reader_fd[0]!=-1) close(reader_fd[0]);
        if (reader_fd[1]!=-1) close(reader_fd[1]);
end:
        return ret;
}
#endif

#ifdef PKG_MALLOC
/* Dumps pkg memory status.
 * Per-child process callback that is called
 * when mem_dump_pkg cfg var is changed.
 */
void mem_dump_pkg_cb(str *gname, str *name)
{
        int     old_memlog;
        int memlog;

        if (cfg_get(core, core_cfg, mem_dump_pkg) == my_pid()) {
                /* set memlog to ALERT level to force
                printing the log messages */
                old_memlog = cfg_get(core, core_cfg, memlog);
                memlog = L_ALERT;
                /* ugly hack to temporarily switch memlog to something visible,
                   possible race with a parallel cfg_set */
                ((struct cfg_group_core*)core_cfg)->memlog=memlog;

                LOG(memlog, "Memory status (pkg) of process %d:\n", my_pid());
                pkg_status();

                ((struct cfg_group_core*)core_cfg)->memlog=old_memlog;
        }
}
#endif

#ifdef SHM_MEM
/* Dumps shm memory status.
 * fixup function that is called
 * when mem_dump_shm cfg var is set.
 */
int mem_dump_shm_fixup(void *handle, str *gname, str *name, void **val)
{
        int     old_memlog;
        int memlog;

        if ((long)(void*)(*val)) {
                /* set memlog to ALERT level to force
                printing the log messages */
                old_memlog = cfg_get(core, core_cfg, memlog);
                memlog = L_ALERT;
                /* ugly hack to temporarily switch memlog to something visible,
                   possible race with a parallel cfg_set */
                ((struct cfg_group_core*)core_cfg)->memlog=memlog;

                LOG(memlog, "Memory status (shm)\n");
                shm_status();

                ((struct cfg_group_core*)core_cfg)->memlog=old_memlog;
                *val = (void*)(long)0;
        }
        return 0;
}
#endif