PyFCGI/src/responder.c

/*
 * Copyright (C) 2019 Weber Yann
 * 
 * This file is part of PyFCGI.
 * 
 * PyFCGI is free software: you can redistribute it and/or modify
 * it under the terms of the GNU Affero General Public License as published by
 * the Free Software Foundation, either version 3 of the License, or
 * any later version.
 * 
 * PyFCGI 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 Affero General Public License for more details.
 * 
 * You should have received a copy of the GNU Affero General Public License
 * along with PyFCGI.  If not, see <http://www.gnu.org/licenses/>.
 */
#include "responder.h"
int pyfcgi_semid = 0;

void init_context()
{
	pyfcgi_semid = 0;
}


int responder_loop()
{
	unsigned int n_wrk, wanted_n, n;
	pid_t *wrk_pids;
	int semid, err;
	int status;
	pid_t ret;
	struct sembuf sop;
	struct timespec timeout;
	short idle;

	sop.sem_num = 0;
	sop.sem_op = 0;
	sop.sem_flg = 0;
	timeout.tv_sec = 0;
	timeout.tv_nsec = 100000000;
	idle = 0;

	pyfcgi_log(LOG_INFO, "Preparing workers");

	init_context();

	wrk_pids = malloc(sizeof(int) * PyFCGI_conf.max_wrk);
	if(!wrk_pids)
	{
		err = errno;
		pyfcgi_log(	LOG_ALERT,
			"Unable to allocate memory for childs PID : %s",
			strerror(err));
		clean_exit(err);
	}
	bzero(wrk_pids, sizeof(int) * PyFCGI_conf.max_wrk);

	semid = new_semaphore();
	
	wanted_n = PyFCGI_conf.min_wrk;
	n_wrk = 0;
	// prespawning minimum worker count
	for(n_wrk=0; n_wrk < wanted_n; n_wrk++)
	{
		wrk_pids[n_wrk] = spawn(n_wrk, semid);
	}
	//Wait at least for a process to be ready
	while(!semtimedop(semid, &sop, 1, &timeout));

	// main loop, taking care to restart terminated workers, 
	// spawn new one if needed, etc.
	while(1)
	{
		if( (ret = waitpid(0, &status, WNOHANG)) )
		{
			if(ret < 0)
			{
				//TODO : error
			}
			if(!ret)
			{
				continue;
			}
			for(n=0; n<n_wrk; n++)
			{
				if(wrk_pids[n] == ret)
				{
					break;
				}
			}
			if(n == n_wrk)
			{
				pyfcgi_log(LOG_WARNING,
				       "Child %d stopped but was notregistered",
				       ret);
				continue;
			}
			idle=0;
			sop.sem_op = -1;
			ret = semop(semid, &sop, 1);
			sop.sem_op = 0;
			if(ret < 0)
			{
				err = errno;
				pyfcgi_log(LOG_ALERT,
				       "Unable to dec sem after child exit : %s",
				       strerror(err));
				clean_exit(err);
			}
			if(status)
			{
				if(WIFSIGNALED(status))
				{
					if(WTERMSIG(status) == 11)
					{
						pyfcgi_log(LOG_ALERT,
							"Worker[%d] segfault !",
							n);
					}
					else
					{
						pyfcgi_log(LOG_ALERT,
							"Worker[%d] terminated by signal %d",
							n, WTERMSIG(status));
					}
				}
				if(WEXITSTATUS(status) & PYFCGI_FATAL)
				{
					pyfcgi_log(LOG_ALERT,
						"Worker[%d] exited with status FATAL",
						n);
					//TODO : restart ?
				}
				else
				{
					pyfcgi_log(LOG_WARNING,
						"Worker[%d] exited with status %d",
						n, WEXITSTATUS(status));
				}
			}
			else
			{
				pyfcgi_log(LOG_INFO,
				       "Worker[%d] PID %d exited normally",
				       n, wrk_pids[n]);
			}
			// child stopped, looking for it
			if(wanted_n < n_wrk)
			{	// need to shift the list and dec n_wrk
				pyfcgi_log(LOG_DEBUG, "GC Workers");
pyfcgi_log( LOG_DEBUG, "GC want %d have %d", wanted_n, n_wrk);
				memmove(wrk_pids+n, wrk_pids+n+1,
				        sizeof(pid_t) * (n_wrk - n));
				n_wrk--;
			}
			else
			{	// respawn on same slot
				pyfcgi_log(LOG_INFO, "respawn #%d", n);
				wrk_pids[n] = spawn(n, semid);
				continue;
			}
		}
		ret = semtimedop(semid, &sop, 1, &timeout);
//pyfcgi_log( LOG_DEBUG, "semtimeop ret=%d want %d have %d", ret, wanted_n, n_wrk);
		if(ret < 0)
		{
			err = errno;
			if(err == EAGAIN)
			{
//pyfcgi_log(LOG_DEBUG, "IDLE want %d have %d\t min=%d", wanted_n, n_wrk, min_wrk);
				// workers idle
				if(!idle)
				{
					idle = 1;
				}
				else if(wanted_n > PyFCGI_conf.min_wrk
					&& n_wrk - wanted_n < 2)
				{
					wanted_n--;
				}
				continue;
			}
			pyfcgi_log(LOG_ERR, "Unable to read semaphore : %s",
			       strerror(err));
		}
		if(!ret && n_wrk < PyFCGI_conf.max_wrk)
		{
			idle=0;
			pyfcgi_log( LOG_DEBUG,
				"All workers busy, spawning a new one");
			n = n_wrk;
			n_wrk++;
			wanted_n = n_wrk;
			wrk_pids[n] = spawn(n, semid);
		}
	}
	
	//Debug wait & exit
	for(n_wrk=0; n_wrk != PyFCGI_conf.min_wrk; n_wrk++)
	{
		waitpid(wrk_pids[n_wrk], &status, 0);
		pyfcgi_log(LOG_DEBUG, "Child %d stopped with status %d",
		       wrk_pids[n_wrk], status);
	}
		//printf("Content-Type: text/html\r\n\r\nHello world !\n");
	pyfcgi_log(LOG_INFO,"Child workers stoped, stopping responder");
	exit(0);
}

pid_t spawn(int wrk_id, int semid)
{
	pid_t res;
	struct timespec timeout;
	timeout.tv_sec = 0;
	timeout.tv_nsec = 100000000;
	

	res = fork();
	if(res == -1)
	{
		pyfcgi_log(LOG_ERR, "Fork fails for worker #%d : %s",
			wrk_id, strerror(errno));
		return -1;
	}
	else if(!res)
	{
		// Child process
		if(PyFCGI_conf.pep333)
		{
			exit(work333(wrk_id, semid));
		}
		else
		{
			exit(work(wrk_id, semid));
		}
	}
	// Sleep to avoid spawning like hell thinking all workers are
	// busy. Let some time to this one to go up...
	// TODO: find a better way to avoid spawning to max_wrk
	nanosleep(&timeout, NULL);
	pyfcgi_log(	LOG_INFO,
		"Worker #%d spawned with PID %d", wrk_id, res);
	return res;
}

int new_semaphore(key_t semkey)
{
	int semid, err;

	semid = semget(IPC_PRIVATE, 2, 0770);

	if(semid == -1)
	{
		err = errno;
		pyfcgi_log(	LOG_ALERT,
			"Unable to create semaphore : %s",
			strerror(err));
		clean_exit(err);
	}
	if(pyfcgi_semid)
	{
		pyfcgi_log(	LOG_WARNING,
			"The semid context was not zero when calling new_semaphore, attempt to closeing it.");
		semctl(pyfcgi_semid, 0, IPC_RMID);

	}
	pyfcgi_semid = semid;
	return semid;
}

void clean_exit(int status)
{
	if(pyfcgi_semid && semctl(pyfcgi_semid, 0, IPC_RMID) == -1)
	{
		pyfcgi_log(	LOG_CRIT,
			"Unable to delete semaphore before exiting.");
	}
	exit(status);
}