xjab_wlist.c

性能优秀的SIP Proxy

/*
 * $Id: xjab_wlist.c,v 1.1.1.1 2005/06/13 16:47:39 bogdan_iancu Exp $
 *
 * eXtended JABber module - worker implementation
 *
 * Copyright (C) 2001-2003 FhG Fokus
 *
 * This file is part of openser, a free SIP server.
 *
 * openser 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
 *
 * openser is distributed in the hope that it will be useful,
 * but WITHOUT ANY WARRANTY; without even the implied warranty of
 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 * GNU General Public License for more details.
 *
 * You should have received a copy of the GNU General Public License 
 * along with this program; if not, write to the Free Software
 * Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA  02111-1307  USA
 *
 * ---
 *
 * History
 * -------
 * 2003-02-24  added 'xj_wlist_set_flag' function (dcm)
 * 2003-03-11  major locking changes - uses locking.h (andrei)
 *
 */


#include <string.h>
#include <unistd.h>
#include <stdio.h>

#include "../../dprint.h"
#include "../../mem/mem.h"
#include "../../mem/shm_mem.h"

#include "xjab_worker.h"
#include "mdefines.h"

#define XJ_DEF_JDELIM '*'

/**
 * init a workers list
 * - pipes : communication pipes
 * - size : size of list - number of workers
 * - max : maximum number of jobs per worker
 * #return : pointer to workers list or NULL on error
 */
xj_wlist xj_wlist_init(int **pipes, int size, int max, int cache_time,
		int sleep_time, int delay_time)
{
	int i;
	xj_wlist jwl = NULL;

	if(pipes == NULL || size <= 0 || max <= 0)
		return NULL;
#ifdef XJ_EXTRA_DEBUG
	DBG("XJAB:xj_wlist_init: -----START-----\n");
#endif	
	jwl = (xj_wlist)_M_SHM_MALLOC(sizeof(t_xj_wlist));
	if(jwl == NULL)
		return NULL;
	jwl->len = size;
	jwl->maxj = max;
	
	jwl->cachet = cache_time;
	jwl->delayt = delay_time;
	jwl->sleept = sleep_time;

	jwl->aliases = NULL;
	jwl->sems = NULL;
	i = 0;
	/* alloc locks*/
	if((jwl->sems = lock_set_alloc(size)) == NULL){
		LOG(L_CRIT, "jabber: failed to alloc lock set\n");
		goto clean;
	};
	/* init the locks*/
	if (lock_set_init(jwl->sems)==0){
		LOG(L_CRIT, "jabber: failed to initialize the locks\n");
		goto clean;
	};
	jwl->workers = (xj_worker)_M_SHM_MALLOC(size*sizeof(t_xj_worker));
	if(jwl->workers == NULL){
		lock_set_destroy(jwl->sems);
		goto clean;
	}

	for(i = 0; i < size; i++)
	{
		jwl->workers[i].nr = 0;
		jwl->workers[i].pid = 0;
		jwl->workers[i].wpipe = pipes[i][1];
		jwl->workers[i].rpipe = pipes[i][0];
		if((jwl->workers[i].sip_ids = newtree234(xj_jkey_cmp)) == NULL){
			lock_set_destroy(jwl->sems);
			goto clean;
		}
	}	

	return jwl;

clean:
	DBG("XJAB:xj_wlist_init: error occurred -> cleaning\n");
	if(jwl->sems != NULL)
		lock_set_dealloc(jwl->sems);
	if(jwl->workers != NULL)
	{
		while(i>=0)
		{
			if(jwl->workers[i].sip_ids == NULL)
				free2tree234(jwl->workers[i].sip_ids, xj_jkey_free_p);
			i--;
		}
		_M_SHM_FREE(jwl->workers);
	}
	_M_SHM_FREE(jwl);
	return NULL;

}

/**
 * set the p.id's of the workers
 * - jwl : pointer to the workers list
 * - pids : p.id's array
 * - size : number of pids
 * #return : 0 on success or <0 on error
 */
int xj_wlist_set_pid(xj_wlist jwl, int pid, int idx)
{
	if(jwl == NULL || pid <= 0 || idx < 0 || idx >= jwl->len)
		return -1;
	lock_set_get(jwl->sems, idx);
	jwl->workers[idx].pid = pid;
	lock_set_release(jwl->sems, idx);
	return 0;
}

/**
 * free jab_wlist
 * - jwl : pointer to the workers list
 */
void xj_wlist_free(xj_wlist jwl)
{
	int i;
#ifdef XJ_EXTRA_DEBUG
	DBG("XJAB:xj_wlist_free: freeing 'xj_wlist' memory ...\n");
#endif
	if(jwl == NULL)
		return;

	if(jwl->workers != NULL)
	{
		for(i=0; i<jwl->len; i++)
			free2tree234(jwl->workers[i].sip_ids, xj_jkey_free_p);
		_M_SHM_FREE(jwl->workers);
	}

	if(jwl->aliases != NULL)
	{
		if(jwl->aliases->d)
			_M_SHM_FREE(jwl->aliases->d);

		if(jwl->aliases->jdm != NULL)
		{
			_M_SHM_FREE(jwl->aliases->jdm->s);
			_M_SHM_FREE(jwl->aliases->jdm);
		}
		if(jwl->aliases->proxy != NULL)
		{
			_M_SHM_FREE(jwl->aliases->proxy->s);
			_M_SHM_FREE(jwl->aliases->proxy);
		}
		if(jwl->aliases->size > 0)
		{
			for(i=0; i<jwl->aliases->size; i++)
				_M_SHM_FREE(jwl->aliases->a[i].s);
			_M_SHM_FREE(jwl->aliases->a);
		}
		_M_SHM_FREE(jwl->aliases);
		jwl->aliases = NULL;
	}
	
	if(jwl->sems != NULL){
		lock_set_destroy(jwl->sems);
		lock_set_dealloc(jwl->sems);
	}
	
	_M_SHM_FREE(jwl);
}

/**
 * return communication pipe with the worker that will process the message for
 * 		the id 'sid' only if it exists, or -1 if error
 * - jwl : pointer to the workers list
 * - sid : id of the entity (connection to Jabber - usually SHOULD be FROM
 *   header of the incoming SIP message)
 * - p : will point to the SHM location of the 'sid' in jwl
 */
int xj_wlist_check(xj_wlist jwl, xj_jkey jkey, xj_jkey *p)
{
	int i;
	if(jwl==NULL || jkey==NULL || jkey->id==NULL || jkey->id->s==NULL)
		return -1;
	
	i = 0;
	*p = NULL;
	while(i < jwl->len)
	{
		lock_set_get(jwl->sems, i);
		if(jwl->workers[i].pid <= 0)
		{
			lock_set_release(jwl->sems, i);
			i++;
			continue;
		}
		if((*p = find234(jwl->workers[i].sip_ids, (void*)jkey, NULL)) != NULL)
		{
			lock_set_release(jwl->sems, i);
#ifdef XJ_EXTRA_DEBUG
			DBG("XJAB:xj_wlist_check: entry exists for <%.*s> in the"
				" pool of <%d> [%d]\n",jkey->id->len, jkey->id->s,
				jwl->workers[i].pid,i);
#endif
			return jwl->workers[i].wpipe;
		}
		lock_set_release(jwl->sems, i);
		i++;
	}
#ifdef XJ_EXTRA_DEBUG
	DBG("XJAB:xj_wlist_check: entry does not exist for <%.*s>\n",
			jkey->id->len, jkey->id->s);
#endif
	return -1;
}

/**
 * return communication pipe with the worker that will process the message for
 * 		the id 'sid', or -1 if error
 * - jwl : pointer to the workers list
 * - sid : id of the entity (connection to Jabber - usually SHOULD be FROM
 *   header of the incoming SIP message)
 * - p : will point to the SHM location of the 'sid' in jwl
 */
int xj_wlist_get(xj_wlist jwl, xj_jkey jkey, xj_jkey *p)
{
	int i = 0, pos = -1, min = 100000;
	xj_jkey msid = NULL;
	
	if(jwl==NULL || jkey==NULL || jkey->id==NULL || jkey->id->s==NULL)
		return -1;

	*p = NULL;
	while(i < jwl->len)
	{
		lock_set_get(jwl->sems, i);
		if(jwl->workers[i].pid <= 0)
		{
			lock_set_release(jwl->sems, i);
			i++;
			continue;
		}
		if((*p = find234(jwl->workers[i].sip_ids, (void*)jkey, NULL))!=NULL)
		{
			if(pos >= 0)
				lock_set_release(jwl->sems, pos);
				lock_set_release(jwl->sems, i);
#ifdef XJ_EXTRA_DEBUG
			DBG("XJAB:xj_wlist_get: entry already exists for <%.*s> in the"
				" pool of <%d> [%d]\n",jkey->id->len, jkey->id->s,
				jwl->workers[i].pid,i);
#endif
			return jwl->workers[i].wpipe;
		}
		if(min > jwl->workers[i].nr)
		{
			if(pos >= 0)
				lock_set_release(jwl->sems, pos);
			pos = i;
			min = jwl->workers[i].nr;
		}
		else
			lock_set_release(jwl->sems, i);
		i++;
	}
	if(pos >= 0 && jwl->workers[pos].nr < jwl->maxj)
	{
		jwl->workers[pos].nr++;

		msid = (xj_jkey)_M_SHM_MALLOC(sizeof(t_xj_jkey));
		if(msid == NULL)
			goto error;
		msid->id = (str*)_M_SHM_MALLOC(sizeof(str));
		if(msid->id == NULL)
		{
			_M_SHM_FREE(msid);
			goto error;
		}
		
		msid->id->s = (char*)_M_SHM_MALLOC(jkey->id->len);
		if(msid->id == NULL)
		{
			_M_SHM_FREE(msid->id);
			_M_SHM_FREE(msid);
			goto error;
		}
		
		if((*p = add234(jwl->workers[pos].sip_ids, msid)) != NULL)
		{
			msid->id->len = jkey->id->len;
			memcpy(msid->id->s, jkey->id->s, jkey->id->len);
			msid->hash = jkey->hash;
			msid->flag = XJ_FLAG_OPEN;
			lock_set_release(jwl->sems, pos);
#ifdef XJ_EXTRA_DEBUG
			DBG("XJAB:xj_wlist_get: new entry for <%.*s> in the pool of"
				" <%d> - [%d]\n", jkey->id->len, jkey->id->s,
				jwl->workers[pos].pid, pos);
#endif
			return jwl->workers[pos].wpipe;
		}
		_M_SHM_FREE(msid->id->s);
		_M_SHM_FREE(msid->id);
		_M_SHM_FREE(msid);
	}

error:
	if(pos >= 0)
		lock_set_release(jwl->sems, pos);