/**
 * Procedure to calculate the virtual queue ID from an incoming call.
 * This procedure also sets the 'arrival time' of the call.
 * The formula is id = md5sum { "REQ/RES:Method:from:to" }
 *
 * @param struct sip_msg *msg incoming SIP message
 * @return pointer to the new allocated call information
 */
queueID_t *
vq_get_call_id (struct sip_msg *msg)
{
  queueID_t *newcall_ID;
  char type[5];
  char string[256];
  char method[12];
  str from = {NULL, 0};
  str to = {NULL, 0};
  struct msg_start *fl;
  //struct hdr_field *hf;
  int isReq;
      
  MD5_CTX Md5Ctx;
  char HA1[HASHLEN];
  
  DBG ("Generating a queue ID...\n");
  
  memset (type, 0, sizeof(type));
  memset (string, 0, sizeof(string));
  memset (method, 0, sizeof(method));
  
  fl = &msg->first_line;
  
  isReq = fl->type;
  if (isReq != SIP_REQUEST) {
    DBG ("Not a SIP request");
    return NULL;
  }
  
  newcall_ID = shm_malloc (sizeof(queueID_t));
  if (!newcall_ID) {
    ERR ("Could not allocate memory !\n");
    return NULL;
  }

  // init time in the call_ID
  gettimeofday (&newcall_ID->time, NULL);
  memset (HA1, 0, HASHLEN);
  
  //DBG ("set call time\n");
  
  // calculate the hash
  // Use md5sum { "REQ/RES:Method:from:to" }
  MD5Init(&Md5Ctx);
  
  // add type REQ/RES
  snprintf (type, sizeof(type), "%d", fl->type);

  // add method
  memcpy (method, msg->first_line.u.request.method.s, msg->first_line.u.request.method.len);

  // add from
  cscf_get_from_uri (msg, &from);
  //DBG ("From 'uri': %.*s\n", from.len, from.s);
   
  // add to
  cscf_get_to_uri (msg, &to);
  //DBG ("To 'uri': %.*s\n", to.len, to.s);
  
  // put it together
  snprintf (string, sizeof(string), "%s:%s:%.*s:%.*s", type, method, from.len, from.s, to.len, to.s);
  LOG (L_INFO, "Calculated ID for \"%s\"\n", string);
    
  MD5Update(&Md5Ctx, string, strlen(string));
  
  MD5Final(HA1, &Md5Ctx);

  memcpy (newcall_ID->id, HA1, HASHLEN);
  print_hex ((void *)newcall_ID->id, (void *)newcall_ID->strid, HASHLEN);
  
  LOG (L_INFO, " is %s\n", newcall_ID->strid);

  return newcall_ID;

}

int
add_rt_info(
	ptree_node_t *pn,
	rt_info_t* r,
	unsigned int rgid
	)
{
	rg_entry_t    *trg=NULL;
	rt_info_wrp_t *rtl_wrp=NULL;
	rt_info_wrp_t *rtlw=NULL;
	int i=0;

	if((NULL == pn) || (NULL == r))
		goto err_exit;

	if (NULL == (rtl_wrp = (rt_info_wrp_t*)shm_malloc(sizeof(rt_info_wrp_t)))) {
		LM_ERR("no more shm mem\n");
		goto err_exit;
	}
	memset( rtl_wrp, 0, sizeof(rt_info_wrp_t));
	rtl_wrp->rtl = r;

	if(NULL==pn->rg) {
		/* allocate the routing groups array */
		pn->rg_len = RG_INIT_LEN;
		if(NULL == (pn->rg = (rg_entry_t*)shm_malloc(
						pn->rg_len*sizeof(rg_entry_t)))) {
			/* recover the old pointer to be able to shm_free mem */
			goto err_exit;
		}
		memset( pn->rg, 0, pn->rg_len*sizeof(rg_entry_t));
		pn->rg_pos=0;
	}
	/* search for the rgid up to the rg_pos */
	for(i=0; (i<pn->rg_pos) && (pn->rg[i].rgid!=rgid); i++);
	if((i==pn->rg_len-1)&&(pn->rg[i].rgid!=rgid)) {
		/* realloc & copy the old rg */
		trg = pn->rg;
		if(NULL == (pn->rg = (rg_entry_t*)shm_malloc(
						2*pn->rg_len*sizeof(rg_entry_t)))) {
			/* recover the old pointer to be able to shm_free mem */
			pn->rg = trg;
			goto err_exit;
		}
		memset(pn->rg+pn->rg_len, 0, pn->rg_len*sizeof(rg_entry_t));
		memcpy(pn->rg, trg, pn->rg_len*sizeof(rg_entry_t));
		pn->rg_len*=2;
		shm_free( trg );
	}
	/* insert into list */
	r->ref_cnt++;
	if(NULL==pn->rg[i].rtlw){
		pn->rg[i].rtlw = rtl_wrp;
		pn->rg[i].rgid = rgid;
		pn->rg_pos++;
		goto ok_exit;
	}
	if( r->priority > pn->rg[i].rtlw->rtl->priority) {
		/* change the head of the list */
		rtl_wrp->next = pn->rg[i].rtlw;
		pn->rg[i].rtlw = rtl_wrp;
		goto ok_exit;
	}
	rtlw = pn->rg[i].rtlw;
	while( rtlw->next !=NULL) {
		if(r->priority > rtlw->next->rtl->priority) {
			rtl_wrp->next = rtlw->next;
			rtlw->next = rtl_wrp;
			goto ok_exit;
		}
		rtlw = rtlw->next;
	}
	/* the smallest priority is linked at the end */
	rtl_wrp->next=NULL;
	rtlw->next=rtl_wrp;
ok_exit:
	return 0;

err_exit:
	if (rtl_wrp) shm_free(rtl_wrp);
	return -1;
}

rt_info_t*
build_rt_info(
	int priority,
	tmrec_t *trec,
	/* script routing table index */
	int route_idx,
	/* list of destinations indexes */
	char* dstlst,
	pgw_t* pgw_l
	) 
{
	char *tmp=NULL;
	char *ep=NULL;
	rt_info_t* rt = NULL;
	int *idx = NULL, *t_idx=NULL;
	int n=0, idx_size=0,i, grp_idx=0;
	long t=0;
	pgw_t *pgw=NULL;

	if(NULL == (rt = (rt_info_t*)shm_malloc(sizeof(rt_info_t)))) {
		LM_ERR("no more shm mem(1)\n");
		goto err_exit;
	}
	memset(rt, 0, sizeof(rt_info_t));

	idx_size = IDX_SIZE;
	if( NULL == (idx = (int*)shm_malloc(2*idx_size*sizeof(int)))) {
		LM_ERR("no more shm mem(2)\n");
		goto err_exit;
	}
	memset(idx, 0, 2*idx_size*sizeof(int));

	rt->priority = priority;
	rt->time_rec = trec;
	rt->route_idx = route_idx;
	tmp=dstlst;
	n=0;
	/* parse the dstlst */
	while(tmp && (*tmp!=0)) {
		errno = 0;
		t = strtol(tmp, &ep, 10);
		if (ep == tmp) {
			LM_ERR("bad id '%c' (%d)[%s]\n",
					*ep, (int)(ep-dstlst), dstlst);
			goto err_exit;
		}
		if ((!IS_SPACE(*ep)) && (*ep != SEP) && (*ep != SEP1)
				&& (*ep != SEP_GRP) && (*ep!=0)) {
			LM_ERR("bad char %c (%d) [%s]\n",
					*ep, (int)(ep-dstlst), dstlst);
			goto err_exit;
		}
		if (errno == ERANGE && (t== LONG_MAX || t== LONG_MIN)) {
			LM_ERR("out of bounds\n");
			goto err_exit;
		}
		idx[2*n]=t;
		idx[2*n+1]=grp_idx;
		if(*ep == SEP_GRP)
			grp_idx++;
		n++;
		/* reallocate the array which keeps the parsed indexes */
		if(n>=idx_size){
			if(NULL==((t_idx)=(int*)shm_malloc((idx_size*2*2)*sizeof(int)))) {
				LM_ERR("out of shm\n");
				goto err_exit;
			}
			memset(t_idx+(2*idx_size), 0, 2*idx_size*sizeof(int));
			memcpy(t_idx, idx, 2*idx_size*sizeof(int));
			shm_free(idx);
			idx_size*=2;
			idx=t_idx;
		}
		if(IS_SPACE(*ep))
			EAT_SPACE(ep);
		if(ep && (*ep == SEP || *ep == SEP1 || *ep == SEP_GRP))
			ep++;
		tmp = ep;
	}
	if(n==0) {
		LM_ERR("invalid n\n");
		goto err_exit;
	}
	/* create the pgwl */
	rt->pgwa_len = n;
	if(NULL ==
		(rt->pgwl=(pgw_list_t*)shm_malloc(rt->pgwa_len*sizeof(pgw_list_t)))) {
		goto err_exit;
	}
	memset(rt->pgwl, 0, rt->pgwa_len*sizeof(pgw_list_t));
	/* translate GW ids to GW pointers */
	for(i=0;i<n; i++){
		if ( NULL == (pgw = get_pgw(pgw_l, idx[2*i]))) {
			LM_ERR("invalid GW id %d\n",
				idx[2*i]);
			goto err_exit;
		}
		rt->pgwl[i].pgw=pgw;
		rt->pgwl[i].grpid=idx[2*i+1];
		/* LM_DBG("added to gwlist [%d/%d/%p]\n",
//.........这里部分代码省略.........

char* create_as_action_reply(struct cell *c,struct tmcb_params *params,int uac_id,char processor_id,int *evt_len)
{
   int i;
   unsigned int code,flags;
   unsigned short int port;
   unsigned int k,len;
   char *buffer;
   struct sip_msg *msg;
   if(!(buffer=shm_malloc(ENCODED_MSG_SIZE))){
      LM_ERR("create_as_action_reply Out Of Memory !!\n");
      return 0;
   }
   msg=0;
   *evt_len=0;
   flags=0;
   if(params->rpl==FAKED_REPLY)
      flags=FAKED_REPLY_FLAG;
   /*length*/
   k=4;
   /*type*/
   buffer[k++]=(unsigned char)RES_IN;
   /*processor id*/
   buffer[k++]=processor_id;
   /*flags (by now, not used)*/
   flags=htonl(flags);
   memcpy(buffer+k,&flags,4);
   k+=4;
   /*recv info*/
   if(!(params->rpl == FAKED_REPLY)) {
      msg=params->rpl;
      /*protocol should be UDP,TCP,TLS or whatever*/
      buffer[k++]=(unsigned char)msg->rcv.proto;
      /*src ip len + src ip*/
      len=msg->rcv.src_ip.len;
      buffer[k++]=(unsigned char)len;
      memcpy(buffer+k,&(msg->rcv.src_ip.u),len);
      k+=len;
      /*dst ip len + dst ip*/
      len=msg->rcv.dst_ip.len;
      buffer[k++]=(unsigned char)len;
      memcpy(buffer+k,&(msg->rcv.dst_ip.u),len);
      k+=len;
      /*src port */
      port=htons(msg->rcv.src_port);
      memcpy(buffer+k,&port,2);
      k+=2;
      /*dst port */
      port=htons(msg->rcv.dst_port);
      memcpy(buffer+k,&port,2);
      k+=2;
   }else{
      /*protocol*/
      buffer[k++]=0;
      /*src ip len*/
      buffer[k++]=0;
      /*dst ip len*/
      buffer[k++]=0;
      /*skip src port and dst port*/
      buffer[k++]=0;
      buffer[k++]=0;
      buffer[k++]=0;
      buffer[k++]=0;
   }
   /*hash_index*/
   i=htonl(c->hash_index);
   memcpy(buffer+k,&i,4);
   k+=4;
   /*label*/
   i=(!strncmp(c->method.s,"CANCEL",6)) ? \
     htonl(((struct as_uac_param*)*params->param)->label) : \
	htonl(c->label);
   memcpy(buffer+k,&i,4);
   k+=4;
   /*uac_id*/
   uac_id=htonl(uac_id);
   memcpy(buffer+k,&uac_id,4);
   k+=4;
   /*code*/
   code=htonl(params->code);
   memcpy(buffer+k,&code,4);
   k+=4;
   /*length of event (hdr+payload-4), copied at the beginning*/
   if(params->rpl != FAKED_REPLY) {
      if((i=encode_msg(msg,buffer+k,ENCODED_MSG_SIZE-k))<0){
	 LM_ERR("failed to encode msg\n");
	 goto error;
      }
      k+=i;
   }
   *evt_len=k;
   k=htonl(k);
   memcpy(buffer,&k,4);
   return buffer;
error:
   return 0;
}

int register_dlgcb(struct dlg_cell *dlg, int types, dialog_cb f,
										void *param, param_free_cb ff )
{
	struct dlg_callback *cb;

	if ( types&DLGCB_LOADED ) {
		if (types!=DLGCB_LOADED) {
			LM_CRIT("DLGCB_LOADED type must be register alone!\n");
			return -1;
		}
	} else if ( types&DLGCB_CREATED ) {
		if (types!=DLGCB_CREATED) {
			LM_CRIT("DLGCB_CREATED type must be register alone!\n");
			return -1;
		}
	} else {
		if (dlg==0) {
			LM_CRIT("non-DLGCB_CREATED type "
				"must be register to a dialog (dlg missing)!\n");
			return -1;
		}
	}
	cb = (struct dlg_callback*)shm_malloc(sizeof(struct dlg_callback));
	if (cb==0) {
		LM_ERR("no more shm mem\n");
		return -1;
	}

	cb->types = types;
	cb->callback = f;
	cb->param = param;
	cb->callback_param_free = ff;

	if ( types==DLGCB_CREATED ) {
		if (load_cbs==POINTER_CLOSED_MARKER) {
			LM_CRIT("DLGCB_CREATED type registered after shutdown!?!\n");
			goto error;
		}
		if (create_cbs==0) {
			/* not initialized yet */
			if ( (create_cbs=init_dlg_callback())==NULL ) {
				LM_ERR("no more shm mem\n");
				goto error;
			}
		}
		cb->next = create_cbs->first;
		create_cbs->first = cb;
		create_cbs->types |= types;
	} else if (types==DLGCB_LOADED) {
		if (load_cbs==POINTER_CLOSED_MARKER) {
			/* run the callback on the spot */
			run_load_callback(cb);
			destroy_dlg_callbacks_list(cb);
			return 0;
		}
		if (load_cbs==0) {
			/* not initialized yet */
			if ( (load_cbs=init_dlg_callback())==NULL ) {
				LM_ERR("no more shm mem\n");
				goto error;
			}
		}
		cb->next = load_cbs->first;
		load_cbs->first = cb;
		load_cbs->types |= types;
	} else {
		cb->next = dlg->cbs.first;
		dlg->cbs.first = cb;
		dlg->cbs.types |= types;
	}

	return 0;
error:
	shm_free(cb);
	return -1;
}

int new_request(str *query, http_m_params_t *query_params, http_multi_cbe_t cb, void *param)
{

	LM_DBG("received query %.*s with timeout %d, tls_verify_peer %d, tls_verify_host %d (param=%p)\n", 
			query->len, query->s, query_params->timeout, query_params->tls_verify_peer, query_params->tls_verify_host, param);
	
	CURL *easy;
	CURLMcode rc;

	struct http_m_cell *cell;

	update_stat(requests, 1);

	easy = NULL;
	cell = NULL;

	easy = curl_easy_init();
	if (!easy) {
		LM_ERR("curl_easy_init() failed!\n");
		update_stat(errors, 1);
		return -1;
	}

	cell = build_http_m_cell(easy);
	if (!cell) {
		LM_ERR("cannot create cell!\n");
		update_stat(errors, 1);
		LM_DBG("cleaning up curl handler %p\n", easy);
		curl_easy_cleanup(easy);
		return -1;
	}

	link_http_m_cell(cell);

	cell->global = g;
	cell->easy=easy;
	cell->error[0] = '\0';
	cell->params = *query_params;
	cell->param = param;
	cell->cb = cb;
	cell->url = (char*)shm_malloc(query->len + 1);
	if (cell->url==0) {
		LM_ERR("no more shm mem\n");
		goto error;
	}
	strncpy(cell->url, query->s, query->len);
	cell->url[query->len] = '\0';

	curl_easy_setopt(cell->easy, CURLOPT_URL, cell->url);
	curl_easy_setopt(cell->easy, CURLOPT_WRITEFUNCTION, write_cb);
	curl_easy_setopt(cell->easy, CURLOPT_WRITEDATA, easy);
	if (curl_verbose) {
		curl_easy_setopt(cell->easy, CURLOPT_VERBOSE, 1L);
		curl_easy_setopt(cell->easy, CURLOPT_DEBUGFUNCTION, debug_cb);
	}
	curl_easy_setopt(cell->easy, CURLOPT_ERRORBUFFER, cell->error);
	curl_easy_setopt(cell->easy, CURLOPT_PRIVATE, cell);
	curl_easy_setopt(cell->easy, CURLOPT_SSL_VERIFYPEER, cell->params.tls_verify_peer);
	curl_easy_setopt(cell->easy, CURLOPT_SSL_VERIFYHOST, cell->params.tls_verify_host?2:0);
	curl_easy_setopt(cell->easy, CURLOPT_SSLVERSION, tls_version);

	if (cell->params.tls_client_cert) {
		curl_easy_setopt(cell->easy, CURLOPT_SSLCERT, cell->params.tls_client_cert);
	}

	if (cell->params.tls_client_key) {
		curl_easy_setopt(cell->easy, CURLOPT_SSLKEY, cell->params.tls_client_key);
	}

	if (cell->params.tls_ca_path) {
		curl_easy_setopt(cell->easy, CURLOPT_CAPATH, cell->params.tls_ca_path);
	}

	curl_easy_setopt(cell->easy, CURLOPT_HEADER, 1);
	if (cell->params.headers) {
		curl_easy_setopt(cell->easy, CURLOPT_HTTPHEADER, cell->params.headers);
	}

	if (cell->params.body.s && cell->params.body.len) {
		curl_easy_setopt(cell->easy, CURLOPT_POSTFIELDSIZE, (long)cell->params.body.len);
		curl_easy_setopt(cell->easy, CURLOPT_COPYPOSTFIELDS, cell->params.body.s);
	}

	switch (cell->params.method) {
	case 1:
		curl_easy_setopt(cell->easy, CURLOPT_CUSTOMREQUEST, "GET");
		break;
	case 2:
		curl_easy_setopt(cell->easy, CURLOPT_CUSTOMREQUEST, "POST");
		break;
	case 3:
		curl_easy_setopt(cell->easy, CURLOPT_CUSTOMREQUEST, "PUT");
		break;
	case 4:
		curl_easy_setopt(cell->easy, CURLOPT_CUSTOMREQUEST, "DELETE");
		break;
	default:
		break;
	}

//.........这里部分代码省略.........

/* initialize ratelimit module */
static int mod_init(void)
{
	int i;

	LM_DBG("initializing ...\n");

	rl_lock = lock_alloc();
	if (! rl_lock) {
		LM_ERR("oom in lock_alloc()\n");
		return -1;
	}

	if (lock_init(rl_lock)==0) {
		LM_ERR("failed to init lock\n");
		return -1;
	}

	/* register timer to reset counters */
	if (register_timer_process(rl_timer, NULL, timer_interval,
	TIMER_PROC_INIT_FLAG) == NULL) {
		LM_ERR("could not register timer function\n");
		return -1;
	}

	/* load the SIGNALLING API */
	if(load_sig_api(&sigb)< 0) {
		LM_ERR("can't load signaling functions\n");
		return -1;
	}

	network_load_value = shm_malloc(sizeof(int));
	if (network_load_value==NULL) {
		LM_ERR("oom for network_load_value\n");
		return -1;
	}
	check_network_load = shm_malloc(sizeof(int));
	if (check_network_load==NULL) {
		LM_ERR("oom for check_network_load\n");
		return -1;
	}

	load_value = shm_malloc(sizeof(double));
	if (load_value==NULL) {
		LM_ERR("oom for load_value\n");
		return -1;
	}
	load_source = shm_malloc(sizeof(int));
	if (load_source==NULL) {
		LM_ERR("oom for load_source\n");
		return -1;
	}
	pid_kp = shm_malloc(sizeof(double));
	if (pid_kp==NULL) {
		LM_ERR("oom for pid_kp\n");
		return -1;
	}
	pid_ki = shm_malloc(sizeof(double));
	if (pid_ki==NULL) {
		LM_ERR("oom for pid_ki\n");
		return -1;
	}
	pid_kd = shm_malloc(sizeof(double));
	if (pid_kd==NULL) {
		LM_ERR("oom for pid_kd\n");
		return -1;
	}
	pid_setpoint = shm_malloc(sizeof(double));
	if (pid_setpoint==NULL) {
		LM_ERR("oom for pid_setpoint\n");
		return -1;
	}
	drop_rate = shm_malloc(sizeof(int));
	if (drop_rate==NULL) {
		LM_ERR("oom for drop_rate\n");
		return -1;
	}
	nqueues = shm_malloc(sizeof(int));
	if (nqueues==NULL) {
		LM_ERR("oom for nqueues\n");
		return -1;
	}
	rl_dbg_str = shm_malloc(sizeof(str));
	if (rl_dbg_str==NULL) {
		LM_ERR("oom for rl_dbg_str\n");
		return -1;
	}

	*network_load_value = 0;
	*check_network_load = 0;
	*load_value = 0.0;
	*load_source = load_source_mp;
	*pid_kp = 0.0;
	*pid_ki = -25.0;
	*pid_kd = 0.0;
	*pid_setpoint = 0.01 * (double)cfg_setpoint;
	*drop_rate      = 0;
	*nqueues = nqueues_mp;
	rl_dbg_str->s = NULL;
	rl_dbg_str->len = 0;
//.........这里部分代码省略.........

/* adds a proto ip:port combination to the blacklist
 * returns 0 on success, -1 on error (blacklist full -- would use more then
 *  blst:_max_mem, or out of shm. mem.)
 */
inline static int dst_blacklist_add_ip(unsigned char err_flags,
									unsigned char proto,
									struct ip_addr* ip, unsigned short port,
									ticks_t timeout)
{
	int size;
	struct dst_blst_entry* e;
	unsigned short hash;
	ticks_t now;
	int ret;

	ret=0;
	if (ip->af==AF_INET){
		err_flags&=~BLST_IS_IPV6; /* make sure the ipv6 flag is reset */
		size=sizeof(struct dst_blst_entry);
	}else{
		err_flags|=BLST_IS_IPV6;
		size=sizeof(struct dst_blst_entry)+12 /* ipv6 addr - 4 */;
	}
	now=get_ticks_raw();
	hash=dst_blst_hash_no(proto, ip, port);
	/* check if the entry already exists */
	LOCK_BLST(hash);
		e=_dst_blacklist_lst_find(hash, ip, proto, port, now);
		if (e){
			e->flags|=err_flags;
			e->expire=now+timeout; /* update the timeout */
		}else{
			if (unlikely((*blst_mem_used+size) >=
					cfg_get(core, core_cfg, blst_max_mem))){
#ifdef USE_DST_BLACKLIST_STATS
				dst_blacklist_stats[process_no].bkl_lru_cnt++;
#endif
				UNLOCK_BLST(hash);
				/* first try to free some memory  (~ 12%), but don't
				 * spend more then 250 ms*/
				dst_blacklist_clean_expired(*blst_mem_used/16*14, 0,
															MS_TO_TICKS(250));
				if (unlikely(*blst_mem_used+size >=
						cfg_get(core, core_cfg, blst_max_mem))){
					ret=-1;
					goto error;
				}
				LOCK_BLST(hash);
			}
			e=shm_malloc(size);
			if (e==0){
				UNLOCK_BLST(hash);
				ret=E_OUT_OF_MEM;
				goto error;
			}
			*blst_mem_used+=size;
			e->flags=err_flags;
			e->proto=proto;
			e->port=port;
			memcpy(e->ip, ip->u.addr, ip->len);
			e->expire=now+timeout; /* update the timeout */
			e->next=0;
			dst_blacklist_lst_add(&dst_blst_hash[hash].first, e);
			BLST_HASH_STATS_INC(hash);
		}
	UNLOCK_BLST(hash);
error:
	return ret;
}

static int mod_init(void)
{
	if(register_mi_mod(exports.name, mi_cmds)!=0)
	{
		LM_ERR("failed to register MI commands\n");
		return -1;
	}

	dp_db_url.len = dp_db_url.s ? strlen(dp_db_url.s) : 0;
	LM_DBG("db_url=%s/%d/%p\n", ZSW(dp_db_url.s), dp_db_url.len,dp_db_url.s);
	dp_table_name.len   = strlen(dp_table_name.s);
	dpid_column.len     = strlen( dpid_column.s);
	pr_column.len       = strlen(pr_column.s);
	match_op_column.len = strlen(match_op_column.s);
	match_exp_column.len= strlen(match_exp_column.s);
	match_len_column.len= strlen(match_len_column.s);
	subst_exp_column.len= strlen(subst_exp_column.s);
	repl_exp_column.len = strlen(repl_exp_column.s);
	attrs_column.len    = strlen(attrs_column.s);

	if(attr_pvar_s.s) {
		attr_pvar = (pv_spec_t *)shm_malloc(sizeof(pv_spec_t));
		if(!attr_pvar){
			LM_ERR("out of shm memory\n");
			return -1;
		}

		attr_pvar_s.len = strlen(attr_pvar_s.s);
		if( (pv_parse_spec(&attr_pvar_s, attr_pvar)==NULL) ||
		((attr_pvar->type != PVT_AVP) && (attr_pvar->type!=PVT_SCRIPTVAR))) {
				LM_ERR("invalid pvar name\n");
				return -1;
			}
	}

	default_par2 = (dp_param_p)shm_malloc(sizeof(dp_param_t));
	if(default_par2 == NULL){
		LM_ERR("no shm more memory\n");
		return -1;
	}
	memset(default_par2, 0, sizeof(dp_param_t));

	default_param_s.len = strlen(default_param_s.s);
	if (pv_parse_spec( &default_param_s, &default_par2->v.sp[0])==NULL) {
		LM_ERR("input pv is invalid\n");
		return -1;
	}

	default_param_s.len = strlen(default_param_s.s);
	if (pv_parse_spec( &default_param_s, &default_par2->v.sp[1])==NULL) {
		LM_ERR("output pv is invalid\n");
		return -1;
	}

	if(dp_fetch_rows<=0)
		dp_fetch_rows = 1000;

	if(init_data() != 0) {
		LM_ERR("could not initialize data\n");
		return -1;
	}

	return 0;
}

int update_phtable(presentity_t* presentity, str pres_uri, str body)
{
	char* sphere= NULL;
	unsigned int hash_code;
	pres_entry_t* p;
	int ret= 0;
	str* xcap_doc= NULL;

	/* get new sphere */
	sphere= extract_sphere(body);
	if(sphere==NULL)
	{
		LM_DBG("no sphere defined in new body\n");
		return 0;
	}

	/* search for record in hash table */
	hash_code= core_case_hash(&pres_uri, NULL, phtable_size);
	
	lock_get(&pres_htable[hash_code].lock);

	p= search_phtable(&pres_uri, presentity->event->evp->type, hash_code);
	if(p== NULL)
	{
		lock_release(&pres_htable[hash_code].lock);
		goto done;
	}
	
	if(p->sphere)
	{
		if(strcmp(p->sphere, sphere)!= 0)
		{
			/* new sphere definition */
			shm_free(p->sphere);
		}
		else
		{
			/* no change in sphere definition */
			lock_release(&pres_htable[hash_code].lock);
			pkg_free(sphere);
			return 0;
		}
	
	}


	p->sphere= (char*)shm_malloc((strlen(sphere)+ 1)*sizeof(char));
	if(p->sphere== NULL)
	{
		lock_release(&pres_htable[hash_code].lock);
		ret= -1;
		goto done;
	}
	strcpy(p->sphere, sphere);
		
	lock_release(&pres_htable[hash_code].lock);

	/* call for watchers status update */

	if(presentity->event->get_rules_doc(&presentity->user, &presentity->domain,
				&xcap_doc)< 0)
	{
		LM_ERR("failed to retrieve xcap document\n");
		ret= -1;
		goto done;
	}

	update_watchers_status(pres_uri, presentity->event, xcap_doc);


done:

	if(xcap_doc)
	{
		if(xcap_doc->s)
			pkg_free(xcap_doc->s);
		pkg_free(xcap_doc);
	}

	if(sphere)
		pkg_free(sphere);
	return ret;
}

int init_dst_blacklist()
{
	int ret;
#ifdef BLST_LOCK_PER_BUCKET
	int r;
#endif

	if (dst_blacklist_init==0) {
		/* the dst blacklist is turned off */
		default_core_cfg.use_dst_blacklist=0;
		return 0;
	}

	ret=-1;
#ifdef DST_BLACKLIST_HOOKS
	if (init_blacklist_hooks()!=0){
		ret=E_OUT_OF_MEM;
		goto error;
	}
#endif
	blst_mem_used=shm_malloc(sizeof(*blst_mem_used));
	if (blst_mem_used==0){
		ret=E_OUT_OF_MEM;
		goto error;
	}
	*blst_mem_used=0;
	dst_blst_hash=shm_malloc(sizeof(struct dst_blst_lst_head) *
											DST_BLST_HASH_SIZE);
	if (dst_blst_hash==0){
		ret=E_OUT_OF_MEM;
		goto error;
	}
	memset(dst_blst_hash, 0, sizeof(struct dst_blst_lst_head) *
								DST_BLST_HASH_SIZE);
#ifdef BLST_LOCK_PER_BUCKET
	for (r=0; r<DST_BLST_HASH_SIZE; r++){
		if (lock_init(&dst_blst_hash[r].lock)==0){
			ret=-1;
			goto error;
		}
	}
#elif defined BLST_LOCK_SET
	blst_lock_set=lock_set_alloc(DST_BLST_HASH_SIZE);
	if (blst_lock_set==0){
		ret=E_OUT_OF_MEM;
		goto error;
	}
	if (lock_set_init(blst_lock_set)==0){
		lock_set_dealloc(blst_lock_set);
		blst_lock_set=0;
		ret=-1;
		goto error;
	}
#else /* BLST_ONE_LOCK */
	blst_lock=lock_alloc();
	if (blst_lock==0){
		ret=E_OUT_OF_MEM;
		goto error;
	}
	if (lock_init(blst_lock)==0){
		lock_dealloc(blst_lock);
		blst_lock=0;
		ret=-1;
		goto error;
	}
#endif /* BLST*LOCK*/
	blst_timer_h=timer_alloc();
	if (blst_timer_h==0){
		ret=E_OUT_OF_MEM;
		goto error;
	}
	/* fix options */
	default_core_cfg.blst_max_mem<<=10; /* in Kb */ /* TODO: test with 0 */
	if (blst_timer_interval){
		timer_init(blst_timer_h, blst_timer, 0 ,0); /* slow timer */
		if (timer_add(blst_timer_h, S_TO_TICKS(blst_timer_interval))<0){
			LOG(L_CRIT, "BUG: init_dst_blacklist: failed to add the timer\n");
			timer_free(blst_timer_h);
			blst_timer_h=0;
			goto error;
		}
	}
	return 0;
error:
	destroy_dst_blacklist();
	return ret;
}

subs_t* mem_copy_subs_noc(subs_t* s)
{
	int size;
	subs_t* dest;

	size= sizeof(subs_t)+ (s->pres_uri.len+ s->to_user.len
		+ s->to_domain.len+ s->from_user.len+ s->from_domain.len+ s->callid.len
		+ s->to_tag.len+ s->from_tag.len+s->sockinfo_str.len+s->event_id.len
		+ s->local_contact.len + s->record_route.len+
		+ s->reason.len+ s->watcher_user.len+ s->watcher_domain.len
		+ 1)*sizeof(char);

	dest= (subs_t*)shm_malloc(size);
	if(dest== NULL)
	{
		ERR_MEM(SHARE_MEM);
	}
	memset(dest, 0, size);
	size= sizeof(subs_t);

	CONT_COPY(dest, dest->pres_uri, s->pres_uri)
	CONT_COPY(dest, dest->to_user, s->to_user)
	CONT_COPY(dest, dest->to_domain, s->to_domain)
	CONT_COPY(dest, dest->from_user, s->from_user)
	CONT_COPY(dest, dest->from_domain, s->from_domain)
	CONT_COPY(dest, dest->watcher_user, s->watcher_user)
	CONT_COPY(dest, dest->watcher_domain, s->watcher_domain)
	CONT_COPY(dest, dest->to_tag, s->to_tag)
	CONT_COPY(dest, dest->from_tag, s->from_tag)
	CONT_COPY(dest, dest->callid, s->callid)
	CONT_COPY(dest, dest->sockinfo_str, s->sockinfo_str)
	CONT_COPY(dest, dest->local_contact, s->local_contact)
	CONT_COPY(dest, dest->record_route, s->record_route)
	if(s->event_id.s)
		CONT_COPY(dest, dest->event_id, s->event_id)
	if(s->reason.s)
		CONT_COPY(dest, dest->reason, s->reason)

	dest->event= s->event;
	dest->local_cseq= s->local_cseq;
	dest->remote_cseq= s->remote_cseq;
	dest->status= s->status;
	dest->version= s->version;
	dest->send_on_cback= s->send_on_cback;
	dest->expires= s->expires;
	dest->db_flag= s->db_flag;

	dest->contact.s= (char*)shm_malloc(s->contact.len* sizeof(char));
	if(dest->contact.s== NULL)
	{
		ERR_MEM(SHARE_MEM);
	}
	memcpy(dest->contact.s, s->contact.s, s->contact.len);
	dest->contact.len= s->contact.len;

	return dest;

error:
	if(dest)
			shm_free(dest);
	return NULL;
}

subs_t* mem_copy_subs(subs_t* s, int mem_type)
{
	int size;
	subs_t* dest;

	size= sizeof(subs_t)+ (s->pres_uri.len+ s->to_user.len
		+ s->to_domain.len+ s->from_user.len+ s->from_domain.len+ s->callid.len
		+ s->to_tag.len+ s->from_tag.len+s->sockinfo_str.len+s->event_id.len
		+ s->local_contact.len+ s->contact.len+ s->record_route.len
		+ s->reason.len+ s->watcher_user.len+ s->watcher_domain.len
		+ 1)*sizeof(char);

	if(mem_type & PKG_MEM_TYPE)
		dest= (subs_t*)pkg_malloc(size);
	else
		dest= (subs_t*)shm_malloc(size);

	if(dest== NULL)
	{
		ERR_MEM((mem_type==PKG_MEM_TYPE)?PKG_MEM_STR:SHARE_MEM);
	}
	memset(dest, 0, size);
	size= sizeof(subs_t);

	CONT_COPY(dest, dest->pres_uri, s->pres_uri)
	CONT_COPY(dest, dest->to_user, s->to_user)
	CONT_COPY(dest, dest->to_domain, s->to_domain)
	CONT_COPY(dest, dest->from_user, s->from_user)
	CONT_COPY(dest, dest->from_domain, s->from_domain)
	CONT_COPY(dest, dest->watcher_user, s->watcher_user)
	CONT_COPY(dest, dest->watcher_domain, s->watcher_domain)
	CONT_COPY(dest, dest->to_tag, s->to_tag)
	CONT_COPY(dest, dest->from_tag, s->from_tag)
	CONT_COPY(dest, dest->callid, s->callid)
	CONT_COPY(dest, dest->sockinfo_str, s->sockinfo_str)
	CONT_COPY(dest, dest->local_contact, s->local_contact)
	CONT_COPY(dest, dest->contact, s->contact)
	CONT_COPY(dest, dest->record_route, s->record_route)
	if(s->event_id.s)
		CONT_COPY(dest, dest->event_id, s->event_id)
	if(s->reason.s)
		CONT_COPY(dest, dest->reason, s->reason)

	dest->event= s->event;
	dest->local_cseq= s->local_cseq;
	dest->remote_cseq= s->remote_cseq;
	dest->status= s->status;
	dest->version= s->version;
	dest->send_on_cback= s->send_on_cback;
	dest->expires= s->expires;
	dest->db_flag= s->db_flag;

	return dest;

error:
	if(dest)
	{
		if(mem_type & PKG_MEM_TYPE)
			pkg_free(dest);
		else
			shm_free(dest);
	}
	return NULL;
}

/**
 * Loads the routing data from the database given in global
 * variable db_url and stores it in routing tree rd.
 *
 * @param rd Pointer to the route data tree where the routing data
 * shall be loaded into
 *
 * @return 0 means ok, -1 means an error occured
 *
 */
int load_route_data_db(struct route_data_t * rd) {
	db1_res_t * res = NULL;
	db_row_t * row = NULL;
	int i, ret;
	struct carrier_data_t * tmp_carrier_data;
	static str query_str;
	str tmp_scan_prefix, tmp_rewrite_host, tmp_rewrite_prefix,
		tmp_rewrite_suffix, tmp_host_name, tmp_reply_code, tmp_comment;
	str *p_tmp_comment;

	if( (strlen("SELECT DISTINCT  FROM  WHERE = ")
			+ carrierroute_table.len + columns[COL_DOMAIN]->len
			+ columns[COL_CARRIER]->len + 20) >  QUERY_LEN) {
		LM_ERR("query too long\n");
		return -1;
	}

	if((rd->carrier_num = load_carrier_map(rd)) <= 0){
		LM_ERR("error while retrieving carriers\n");
		goto errout;
	}

	if((rd->domain_num = load_domain_map(rd)) <= 0){
		LM_ERR("error while retrieving domains\n");
		goto errout;
	}

	if ((rd->carriers = shm_malloc(sizeof(struct carrier_data_t *) * rd->carrier_num)) == NULL) {
		SHM_MEM_ERROR;
		goto errout;
	}
	memset(rd->carriers, 0, sizeof(struct carrier_data_t *) * rd->carrier_num);

	for (i=0; i<rd->carrier_num; i++) {
		memset(query, 0, QUERY_LEN);
		ret = snprintf(query, QUERY_LEN, "SELECT DISTINCT %.*s FROM %.*s WHERE %.*s=%i",
		columns[COL_DOMAIN]->len, columns[COL_DOMAIN]->s, carrierroute_table.len,
		carrierroute_table.s, columns[COL_CARRIER]->len, columns[COL_CARRIER]->s, rd->carrier_map[i].id);
		if (ret < 0) {
			LM_ERR("error in snprintf");
			goto errout;
		}
		query_str.s = query;
		query_str.len = ret;

		if (carrierroute_dbf.raw_query(carrierroute_dbh, &query_str, &res) < 0) {
			LM_ERR("Failed to query database.\n");
			goto errout;
		}
		LM_INFO("carrier '%.*s' (id %i) has %i domains\n", rd->carrier_map[i].name.len, rd->carrier_map[i].name.s, rd->carrier_map[i].id, RES_ROW_N(res));
		tmp_carrier_data = create_carrier_data(rd->carrier_map[i].id, &rd->carrier_map[i].name, RES_ROW_N(res));
		if (tmp_carrier_data == NULL) {
			LM_ERR("can't create new carrier '%.*s'\n", rd->carrier_map[i].name.len, rd->carrier_map[i].name.s);
			goto errout;
		}
		if (add_carrier_data(rd, tmp_carrier_data) < 0) {
			LM_ERR("can't add carrier '%.*s'\n", rd->carrier_map[i].name.len, rd->carrier_map[i].name.s);
			destroy_carrier_data(tmp_carrier_data);
			goto errout;
		}
		carrierroute_dbf.free_result(carrierroute_dbh, res);
		res = NULL;
	}

	if (carrierroute_dbf.use_table(carrierroute_dbh, &carrierroute_table) < 0) {
		LM_ERR("Cannot set database table '%.*s'.\n", carrierroute_table.len, carrierroute_table.s);
		return -1;
	}

	if (DB_CAPABILITY(carrierroute_dbf, DB_CAP_FETCH)) {
		if (carrierroute_dbf.query(carrierroute_dbh, NULL, NULL, NULL, (db_key_t *) columns, 0,
				columns_load_num, NULL, NULL) < 0) {
			LM_ERR("Failed to query database to prepare fetch row.\n");
			return -1;
		}
		if(carrierroute_dbf.fetch_result(carrierroute_dbh, &res, cfg_get(carrierroute, carrierroute_cfg, fetch_rows)) < 0) {
			LM_ERR("Fetching rows failed\n");
			return -1;
		}
	} else {
		if (carrierroute_dbf.query(carrierroute_dbh, NULL, NULL, NULL, (db_key_t *) columns, 0,
				columns_load_num, NULL, &res) < 0) {
			LM_ERR("Failed to query database.\n");
			return -1;
		}
	}
	int n = 0;
	do {
		LM_DBG("loading, cycle %d", n++);
		for (i = 0; i < RES_ROW_N(res); ++i) {
//.........这里部分代码省略.........

static int mod_init(void)
{
	LM_INFO("Load-Balancer module - initializing\n");

	init_db_url( db_url , 0 /*cannot be null*/);

	/* Load dialog API */
	if (load_dlg_api(&lb_dlg_binds) != 0) {
		LM_ERR("Can't load dialog hooks");
		return -1;
	}

	/* data pointer in shm */
	curr_data = (struct lb_data**)shm_malloc( sizeof(struct lb_data*) );
	if (curr_data==0) {
		LM_CRIT("failed to get shm mem for data ptr\n");
		return -1;
	}
	*curr_data = 0;

	/* create & init lock */
	if ((ref_lock = lock_init_rw()) == NULL) {
		LM_CRIT("failed to init lock\n");
		return -1;
	}

	if (init_lb_bls()) {
		LM_ERR("BL INIT failed\n");
		return -1;
	}

	/* init and open DB connection */
	if (init_lb_db(&db_url, table_name)!=0) {
		LM_ERR("failed to initialize the DB support\n");
		return -1;
	}

	/* load data */
	if ( lb_reload_data()!=0 ) {
		LM_CRIT("failed to load load-balancing data\n");
		return -1;
	}

	/* close DB connection */
	lb_close_db();

	/* arm a function for probing */
	if (lb_prob_interval) {
		/* load TM API */
		if (load_tm_api(&lb_tmb)!=0) {
			LM_ERR("can't load TM API\n");
			return -1;
		}

		/* probing method */
		lb_probe_method.len = strlen(lb_probe_method.s);
		lb_probe_from.len = strlen(lb_probe_from.s);
		if (lb_probe_replies.s)
			lb_probe_replies.len = strlen(lb_probe_replies.s);

		/* register pinger function */
		if (register_timer( "lb-pinger", lb_prob_handler , NULL,
		lb_prob_interval)<0) {
			LM_ERR("failed to register probing handler\n");
			return -1;
		}

		if (lb_probe_replies.s) {
			lb_probe_replies.len = strlen(lb_probe_replies.s);
			if(parse_reply_codes( &lb_probe_replies, &probing_reply_codes,
			&probing_codes_no )< 0) {
				LM_ERR("Bad format for options_reply_code parameter"
					" - Need a code list separated by commas\n");
				return -1;
			}
		}

	}

	/* parse avps */
	if (parse_avp_spec(&grp_avp_name_s, &grp_avp_name)) {
		LM_ERR("cannot parse group avp\n");
		return -1;
	}
	if (parse_avp_spec(&mask_avp_name_s, &mask_avp_name)) {
		LM_ERR("cannot parse mask avp\n");
		return -1;
	}
	if (parse_avp_spec(&id_avp_name_s, &id_avp_name)) {
		LM_ERR("cannot parse id avp\n");
		return -1;
	}

	return 0;
}

//.........这里部分代码省略.........
		goto error;
	}
	etag= hdr->body;
		
	LM_DBG("completed with status %d [contact:%.*s]\n",
			ps->code, hentity->pres_uri->len, hentity->pres_uri->s);

	if (lexpire == 0)
	{
		find_and_delete_record(hentity, hash_code);
		goto done;
	}

	if (pua_dbf.affected_rows != NULL || dbmode != PUA_DB_ONLY)
	{
		if (find_and_update_record(hentity, hash_code, lexpire, &etag) > 0)
			goto done;
	}
	else
	{
		if ((db_presentity = get_record_puadb(hentity->id, &hentity->etag, &dbpres, &res)) != NULL)
		{
			update_record_puadb(hentity, lexpire, &etag);
			goto done;
		}
	}

	size= sizeof(ua_pres_t)+ sizeof(str)+ 
		(hentity->pres_uri->len+ hentity->tuple_id.len + 
		 hentity->id.len)* sizeof(char);
	if(hentity->extra_headers)
		size+= sizeof(str)+ hentity->extra_headers->len* sizeof(char);

	presentity= (ua_pres_t*)shm_malloc(size);
	if(presentity== NULL)
	{
		LM_ERR("no more share memory\n");
		goto error;
	}	
	memset(presentity, 0, size);

	size= sizeof(ua_pres_t);
	presentity->pres_uri= (str*)((char*)presentity+ size);
	size+= sizeof(str);

	presentity->pres_uri->s= (char*)presentity+ size;
	memcpy(presentity->pres_uri->s, hentity->pres_uri->s, 
			hentity->pres_uri->len);
	presentity->pres_uri->len= hentity->pres_uri->len;
	size+= hentity->pres_uri->len;
	
	presentity->tuple_id.s= (char*)presentity+ size;
	memcpy(presentity->tuple_id.s, hentity->tuple_id.s,
			hentity->tuple_id.len);
	presentity->tuple_id.len= hentity->tuple_id.len;
	size+= presentity->tuple_id.len;

	presentity->id.s=(char*)presentity+ size;
	memcpy(presentity->id.s, hentity->id.s, 
			hentity->id.len);
	presentity->id.len= hentity->id.len; 
	size+= presentity->id.len;
		
	if(hentity->extra_headers)
	{
		presentity->extra_headers= (str*)((char*)presentity+ size);

static void *curl_shm_malloc(size_t size)
{
    void *p = shm_malloc(size);
    return p;
}

ua_pres_t* publish_cbparam(publ_info_t* publ,str* body,str* tuple_id,
		int ua_flag)
{
	int size;
	ua_pres_t* cb_param= NULL;

	size= sizeof(ua_pres_t)+ sizeof(str)+ (publ->pres_uri->len+ 
		+ publ->content_type.len+ publ->id.len+ 1)*sizeof(char);
	if(body && body->s && body->len)
		size+= sizeof(str)+ body->len* sizeof(char);
	if(publ->etag)
		size+= publ->etag->len* sizeof(char);
	if(publ->extra_headers)
		size+= sizeof(str)+ publ->extra_headers->len* sizeof(char);
	if(tuple_id )
		size+= tuple_id->len* sizeof(char);

	cb_param= (ua_pres_t*)shm_malloc(size);
	if(cb_param== NULL)
	{
		LM_ERR("ERROR no more share memory while allocating cb_param"
				" - size= %d\n", size);
		return NULL;
	}
	memset(cb_param, 0, size);
	
	size =  sizeof(ua_pres_t);

	cb_param->pres_uri = (str*)((char*)cb_param + size);
	size+= sizeof(str);
	cb_param->pres_uri->s = (char*)cb_param + size;
	memcpy(cb_param->pres_uri->s, publ->pres_uri->s ,
			publ->pres_uri->len ) ;
	cb_param->pres_uri->len= publ->pres_uri->len;
	size+= publ->pres_uri->len;

	if(publ->id.s && publ->id.len)
	{	
		cb_param->id.s = ((char*)cb_param+ size);
		memcpy(cb_param->id.s, publ->id.s, publ->id.len);
		cb_param->id.len= publ->id.len;
		size+= publ->id.len;
	}

	if(body && body->s && body->len)
	{
		cb_param->body = (str*)((char*)cb_param  + size);
		size+= sizeof(str);
		
		cb_param->body->s = (char*)cb_param + size;
		memcpy(cb_param->body->s, body->s ,
			body->len ) ;
		cb_param->body->len= body->len;
		size+= body->len;
	}
	if(publ->etag)
	{
		cb_param->etag.s = (char*)cb_param + size;
		memcpy(cb_param->etag.s, publ->etag->s ,
			publ->etag->len ) ;
		cb_param->etag.len= publ->etag->len;
		size+= publ->etag->len;
	}
	if(publ->extra_headers)
	{
		cb_param->extra_headers = (str*)((char*)cb_param  + size);
		size+= sizeof(str);
		cb_param->extra_headers->s = (char*)cb_param + size;
		memcpy(cb_param->extra_headers->s, publ->extra_headers->s ,
			publ->extra_headers->len ) ;
		cb_param->extra_headers->len= publ->extra_headers->len;
		size+= publ->extra_headers->len;
	}	

	if(publ->content_type.s && publ->content_type.len)
	{
		cb_param->content_type.s= (char*)cb_param + size;
		memcpy(cb_param->content_type.s, publ->content_type.s, publ->content_type.len);
		cb_param->content_type.len= publ->content_type.len;
		size+=  publ->content_type.len;
	}	
	if(tuple_id)
	{	
		cb_param->tuple_id.s = (char*)cb_param+ size;
		memcpy(cb_param->tuple_id.s, tuple_id->s ,tuple_id->len);
		cb_param->tuple_id.len=