gsource.c

linux集群服务器软件代码包

	g_assert(IS_CHSOURCE(chp));
	
	if (chp->pausenow){
		return FALSE;
	}
	
	
	return (chp->infd.revents != 0
		||	(!chp->fd_fdx && chp->outfd.revents != 0)
		||	chp->ch->ops->is_message_pending(chp->ch));
}

/*
 *	Some kind of event occurred - notify the user.
 */
static gboolean
G_CH_dispatch(GSource * source,
	      GSourceFunc callback,
	      gpointer user_data)
{
	GCHSource* chp = (GCHSource*)source;

	g_assert(IS_CHSOURCE(chp));
	/* Is output now unblocked? 
	 *
	 * If so, turn off OUTPUT_EVENTS to avoid going into
	 * a tight poll(2) loop.
	 */
	if (chp->fd_fdx) {
		if (chp->infd.revents & OUTPUT_EVENTS) {
			chp->infd.events &= ~OUTPUT_EVENTS;
		}
	}else if (chp->outfd.revents & OUTPUT_EVENTS) {
		chp->outfd.events &= ~OUTPUT_EVENTS;
	}
#if 0
	/* If we got a HUP then mark channel as disconnected */
	if ((apend->infd.revents|chp->outfd.revents) & G_IO_HUP) {
		/* CHEAT!! */
		chp->ch->ch_status = IPC_DISCONNECT;
	}else{
		chp->ch->ops->resume_io(chp->ch);
	}
#else
	chp->ch->ops->resume_io(chp->ch);
#endif

	if(chp->dispatch) {
		if(!(chp->dispatch(chp->ch, chp->udata))){
			g_source_remove_poll(source, &chp->infd);
			if (!chp->fd_fdx) {
				g_source_remove_poll(source, &chp->outfd);
			}
			g_source_unref(source);
			return FALSE;
		}
	}
	
	return TRUE;
}

/*
 *	Free up our data, and notify the user process...
 */
static void
G_CH_destroy(GSource* source)
{
	GCHSource* chp = (GCHSource*)source;
	

	g_assert(IS_CHSOURCE(chp));
	
	if (chp->dnotify) {
		chp->dnotify(chp->udata);
	}	
	chp->ch->ops->destroy(chp->ch);
	
	g_source_destroy(source);
}


/************************************************************
 *		Functions for IPC_WaitConnections
 ***********************************************************/
static gboolean G_WC_prepare(GSource * source,
			     gint* timeout);
static gboolean G_WC_check(GSource* source);
static gboolean G_WC_dispatch(GSource* source, 
			      GSourceFunc callback,
			      gpointer user_data);
static void G_WC_destroy(GSource* source);

static GSourceFuncs G_WC_SourceFuncs = {
	G_WC_prepare,
	G_WC_check,
	G_WC_dispatch,
	G_WC_destroy,
};

#define GET_WC_SOURCE(src)	(GWCSource*)(src +1)

/*
 *	Add an IPC_WaitConnection to the gmainloop world...
 */
GWCSource*
G_main_add_IPC_WaitConnection(int priority
,	IPC_WaitConnection* wch
,	IPC_Auth* auth_info
,	gboolean can_recurse
,	gboolean (*dispatch)(IPC_Channel* wch
,		gpointer        user_data)
,	gpointer userdata
,	GDestroyNotify notify)
{

	GWCSource* wcp;
	GSource * source = g_source_new(&G_WC_SourceFuncs, 
					sizeof(GSource)
					+ sizeof(GWCSource));
	
	wcp = GET_WC_SOURCE(source);
	
	memset(wcp, 0, sizeof(GWCSource));
	wcp->magno = MAG_GWCSOURCE;
	wcp->udata = userdata;
	wcp->gpfd.fd = wch->ops->get_select_fd(wch);
	wcp->gpfd.events = DEF_EVENTS;
	wcp->gpfd.revents = 0;
	wcp->wch = wch;
	wcp->dnotify = notify;
	wcp->auth_info = auth_info;
	wcp->dispatch = dispatch;
	
	g_source_add_poll(source, &wcp->gpfd);
	
	g_source_set_priority(source, priority);
	
	g_source_set_can_recurse(source, can_recurse);
	
	wcp->gsourceid = g_source_attach(source, NULL);
	
	if (wcp->gsourceid == 0) {
		g_source_remove_poll(source, &wcp->gpfd);
		g_source_unref(source);
		source = NULL;
		wcp = NULL;
	}
	return wcp;
}


/* Delete the given IPC_WaitConnection from the gmainloop world */
gboolean G_main_del_IPC_WaitConnection(GWCSource* wcp)
{

	GSource* source = g_main_context_find_source_by_id(NULL, wcp->gsourceid);
	if (source == NULL){
		cl_log(LOG_ERR, "G_main_del_IPC_WaitConnection: Cannot find source using source id");
		return FALSE;
	}

	g_source_unref(source);
	
	wcp->gsourceid = 0;

	return TRUE;
}



/*
 *	For IPC_WaitConnection events, return FALSE because we
 *	have to poll to get events.
 *
 *	We don't modify 'timeout' either.
 */
static gboolean
G_WC_prepare(GSource* source,
	     gint* timeout)
{
	GWCSource*  wcp = GET_WC_SOURCE(source);
	
	g_assert(IS_WCSOURCE(wcp));
	return FALSE;
}

/*
 *	Did we notice any I/O (connection pending) events?
 */

static gboolean
G_WC_check(GSource * source)
{
	GWCSource*  wcp = GET_WC_SOURCE(source);
	
	g_assert(IS_WCSOURCE(wcp));

	return wcp->gpfd.revents != 0;
}

/*
 *	Someone is trying to connect.
 *	Try to accept the connection and notify the user.
 */
static gboolean
G_WC_dispatch(GSource* source,
	      GSourceFunc callback,
	      gpointer user_data)
{
	GWCSource*  wcp = GET_WC_SOURCE(source);
	IPC_Channel*	ch;
	gboolean	rc = TRUE;
	int		count = 0;
	
	g_assert(IS_WCSOURCE(wcp));
	
        while(1) {
		ch = wcp->wch->ops->accept_connection(wcp->wch, wcp->auth_info);
		if (ch == NULL) {
			break;
	  	}
		++count;
		
		if(!wcp->dispatch) {
			continue;
		}

		rc = wcp->dispatch(ch, wcp->udata);
		if(!rc) {
			g_source_remove_poll(source, &wcp->gpfd);
			g_source_unref(source);
			break;
		}
	}
	return rc;
}

/*
 *	Free up our data, and notify the user process...
 */
static void
G_WC_destroy(GSource* source)
{
	
	GWCSource*  wcp = GET_WC_SOURCE(source);
	
	g_assert(IS_WCSOURCE(wcp));
	wcp->wch->ops->destroy(wcp->wch);
	if (wcp->dnotify) {
		wcp->dnotify(wcp->udata);
	}
	g_source_destroy(source);
}


/************************************************************
 *		Functions for Signals
 ***********************************************************/
static gboolean G_SIG_prepare(GSource* source,
			     gint* timeout);
static gboolean G_SIG_check(GSource* source);

static gboolean G_SIG_dispatch(GSource* source,
			      GSourceFunc callback,
			      gpointer user_data);
static void G_SIG_destroy(GSource* source);

static void G_main_signal_handler(int nsig);

static GSourceFuncs G_SIG_SourceFuncs = {
	G_SIG_prepare,
	G_SIG_check,
	G_SIG_dispatch,
	G_SIG_destroy,
};

static GSIGSource *G_main_signal_list[_NSIG];

void
set_SignalHandler_dnotify(GSIGSource* sig_src, GDestroyNotify notify)
{
	sig_src->dnotify = notify;	
}

/*
 *	Add an Signal to the gmainloop world...
 */
GSIGSource*
G_main_add_SignalHandler(int priority, int signal,
			 gboolean (*dispatch)(int nsig, gpointer user_data),
			 gpointer userdata, GDestroyNotify notify)
{
	GSIGSource* sig_src;
	GSource * source = g_source_new(&G_SIG_SourceFuncs, sizeof(GSIGSource));
	gboolean failed = FALSE;
	
	sig_src = (GSIGSource*)source;
	
	sig_src->magno		= MAG_GSIGSOURCE;
	sig_src->signal		= signal;
	sig_src->dispatch	= dispatch;
	sig_src->udata		= userdata;
	sig_src->dnotify	= notify;

	sig_src->signal_triggered = FALSE;

	g_source_set_priority(source, priority);
	g_source_set_can_recurse(source, FALSE);

	if(G_main_signal_list[signal] != NULL) {
		cl_log(LOG_ERR,
		       "G_main_add_SignalHandler: Handler already present for signal %d",
		       signal);
		failed = TRUE;
	}
	if(!failed) {
		sig_src->gsourceid = g_source_attach(source, NULL);
		if (sig_src->gsourceid < 1) {
			cl_log(LOG_ERR, "G_main_add_SignalHandler: Could not attach source for signal %d (%d)",
			       signal, sig_src->gsourceid);
			failed = TRUE;
		}
	}
	
	if(failed) {
		cl_log(LOG_ERR, "G_main_add_SignalHandler: Signal handler for signal %d NOT added",
			signal);
		g_source_remove(sig_src->gsourceid);
		g_source_unref(source);
		source = NULL;
		sig_src = NULL;
	} else {
		cl_log(LOG_INFO, "G_main_add_SignalHandler: Added signal handler for signal %d",
			signal);
		G_main_signal_list[signal] = sig_src;
		CL_SIGNAL(signal, G_main_signal_handler);
	}
	
	return sig_src;
}


/*
 *	Delete a Signal from the gmainloop world...
 */
gboolean 
G_main_del_SignalHandler(GSIGSource* sig_src)
{
	if (sig_src->gsourceid <= 0) {
		cl_log(LOG_CRIT, "Bad gsource in G_main_del_IPC_channel");
		return FALSE;
	}
	g_assert(_NSIG > sig_src->signal);

	CL_SIGNAL(sig_src->signal, NULL);

	sig_src->gsourceid = 0;
	sig_src->signal_triggered = FALSE;
	g_source_remove(sig_src->gsourceid);

	G_main_signal_list[sig_src->signal] = NULL;

	return TRUE;
}

static gboolean
G_SIG_prepare(GSource* source, gint* timeout)
{
	GSIGSource* sig_src = (GSIGSource*)source;
	
	g_assert(IS_SIGSOURCE(sig_src));
	
	return sig_src->signal_triggered;
}

/*
 *	Did we notice any I/O events?
 */

static gboolean
G_SIG_check(GSource* source)
{

	GSIGSource* sig_src = (GSIGSource*)source;

	g_assert(IS_SIGSOURCE(sig_src));
	
	return sig_src->signal_triggered;
}

/*
 *	Some kind of event occurred - notify the user.
 */
static gboolean
G_SIG_dispatch(GSource * source,
	      GSourceFunc callback,
	      gpointer user_data)
{
	GSIGSource* sig_src = (GSIGSource*)source;

	g_assert(IS_SIGSOURCE(sig_src));

	sig_src->signal_triggered = FALSE;

	if(sig_src->dispatch) {
		if(!(sig_src->dispatch(sig_src->signal, sig_src->udata))){
			G_main_del_SignalHandler(sig_src);
			return FALSE;
		}
	}
	
	return TRUE;
}

/*
 *	Free up our data, and notify the user process...
 */
static void
G_SIG_destroy(GSource* source)
{
	GSIGSource* sig_src = (GSIGSource*)source;
	
	g_assert(IS_SIGSOURCE(sig_src));
	
	if (sig_src->dnotify) {
		sig_src->dnotify(sig_src->udata);
	}	
	g_source_destroy(source);
}

/* Find and set the correct mainloop input */

static void
G_main_signal_handler(int nsig)
{
	GSIGSource* sig_src = NULL;

	g_assert(G_main_signal_list != NULL);
	g_assert(_NSIG > nsig);
	
	sig_src = G_main_signal_list[nsig];

/* 	g_assert(sig_src != NULL); */
	if(sig_src == NULL) {
		cl_log(LOG_CRIT, "No handler for signal -%d", nsig);
		return;
	}
	
	g_assert(IS_SIGSOURCE(sig_src));
	sig_src->signal_triggered = TRUE;
}