自学教程：C++ DCondVar_Signal函数代码示例

void DaoCallServer_AddTimedWait( DaoProcess *wait, DaoTaskEvent *event, double timeout ){	DaoCallServer *server = daoCallServer;	/*	// The "wait" process may not be running in the thread pool,	// so it may have not been added to active process list.	// It is necessary to add it to the active list now,	// to avoid it being activated immediately after it is blocked.	// Activating it immediately may cause a race condition,	// because it may have not been blocked completely	// (namely, it may be still running).	*/	DaoCallServer_MarkActiveProcess( wait, 1 );	DMutex_Lock( & server->mutex );	if( timeout >= 1E-27 ){		server->timestamp.real = timeout + Dao_GetCurrentTime();		server->timestamp.imag += 1;		event->expiring = server->timestamp.real;		DMap_Insert( server->waitings, & server->timestamp, event );		DMap_Insert( server->pending, event, NULL );		DCondVar_Signal( & server->condv2 );	}else{		event->expiring = -1.0;		DaoCallServer_AddEvent( event );		DCondVar_Signal( & server->condv );	}	DMutex_Unlock( & server->mutex );}

void DaoCallServer_AddTimedWait( DaoProcess *wait, DaoTaskEvent *event, double timeout ){	DaoCallServer *server;	if( daoCallServer == NULL ) DaoCallServer_Init( mainVmSpace );	server = daoCallServer;	DMutex_Lock( & server->mutex );	if( timeout >= 1E-27 ){		server->timestamp.value.real = timeout + Dao_GetCurrentTime();		server->timestamp.value.imag += 1;		event->expiring = server->timestamp.value.real;		DMap_Insert( server->waitings, & server->timestamp, event );		DMap_Insert( server->pending, event, NULL );		DCondVar_Signal( & server->condv2 );	}else{		event->expiring = -1.0;		DaoCallServer_AddEvent( event );		DCondVar_Signal( & server->condv );	}	if( wait->condv ){		/*		// Need to suspend the native thread, for suspending inside code sections		// for functional methods such as std.iterate(), mt.iterate() etc:		*/		wait->pauseType = DAO_PAUSE_NATIVE_THREAD;		if( timeout > 0 ){			DCondVar_TimedWait( wait->condv, & server->mutex, timeout );		}else{			DCondVar_Wait( wait->condv, & server->mutex );		}		wait->status = DAO_PROCESS_RUNNING;	}	DMutex_Unlock( & server->mutex );}

static int DaoCallServer_CheckEvent( DaoTaskEvent *event, DaoFuture *fut, DaoChannel *chan ){	DaoTaskEvent event2 = *event;	daoint i, move = 0, closed = 0;	switch( event->type ){	case DAO_EVENT_WAIT_TASKLET :		move = event->future->precond == fut;		if( event->future->process->condv ) DCondVar_Signal( event->future->process->condv );		break;	case DAO_EVENT_WAIT_RECEIVING :		if( event->channel == chan ){			move = chan->buffer->size > 0;			move |= chan->cap <= 0 && chan->buffer->size == 0;		}		break;	case DAO_EVENT_WAIT_SENDING :		move = event->channel == chan && chan->buffer->size < chan->cap;		break;	case DAO_EVENT_WAIT_SELECT :		if( event->channels == NULL ) return 0;		for(i=0; i<event->channels->size; ++i){			DaoChannel *chan = (DaoChannel*) event->channels->items.pValue[i];			if( chan->buffer->size ){				return 1;			}else if( chan->cap <= 0 ){				closed += 1;			}		}		move = closed == event->channels->size;		break;	default: break;	}	return move;}

static void CHANNEL_Send( DaoProcess *proc, DaoValue *par[], int N ){	DaoValue *data;	DaoFuture *future = DaoProcess_GetInitFuture( proc );	DaoChannel *self = (DaoChannel*) par[0];	float timeout = par[2]->xFloat.value;	if( DaoProcess_CheckCB( proc, "cannot send/block inside code section method" ) ) return;	if( self->cap <= 0 ){		DaoProcess_RaiseException( proc, DAO_ERROR_PARAM, "channel is closed" );		return;	}	data = DaoValue_DeepCopy( par[1] );	if( data == NULL ){		DaoProcess_RaiseException( proc, DAO_ERROR_PARAM, "invalid data for the channel" );		return;	}	//printf( "CHANNEL_Send: %p/n", event );	DMutex_Lock( & daoCallServer->mutex );	DArray_Append( self->buffer, data );	DaoChannel_ActivateEvent( self, DAO_EVENT_WAIT_RECEIVING );	DaoChannel_ActivateEvent( self, DAO_EVENT_WAIT_SELECT );	DCondVar_Signal( & daoCallServer->condv );	DMutex_Unlock( & daoCallServer->mutex );	if( self->buffer->size >= self->cap ){		DaoTaskEvent *event = DaoCallServer_MakeEvent();		DaoTaskEvent_Init( event, DAO_EVENT_WAIT_SENDING, DAO_EVENT_WAIT, future, self );		proc->status = DAO_PROCESS_SUSPENDED;		proc->pauseType = DAO_PAUSE_CHANNEL_SEND;		DaoCallServer_AddTimedWait( proc, event, timeout );	}}

void DThread_ResumeVM( DThread *another ){	another->vmpause = 0;	DMutex_Lock( & another->mutex );	DCondVar_Signal( & another->condv );	DMutex_Unlock( & another->mutex );}

void CHANNEL_Select( DaoProcess *proc, DaoValue *par[], int n ){	DaoTaskEvent *event = NULL;	DaoFuture *future = DaoProcess_GetInitFuture( proc );	DaoList *channels = (DaoList*) par[0];	float timeout = par[1]->xFloat.value;	daoint i, size = DaoList_Size( channels );	for(i=0; i<size; ++i){		DaoValue *value = DaoList_GetItem( channels, i );		if( DaoValue_CheckCtype( value, dao_type_channel ) == 0 ){			DaoProcess_RaiseException( proc, DAO_ERROR_PARAM, "invalid type selection" );			return;		}	}	event = DaoCallServer_MakeEvent();	DaoTaskEvent_Init( event, DAO_EVENT_WAIT_SELECT, DAO_EVENT_WAIT, future, NULL );	event->channels = DArray_Copy( & channels->items );	proc->status = DAO_PROCESS_SUSPENDED;	proc->pauseType = DAO_PAUSE_CHANFUT_SELECT;	DaoCallServer_AddTimedWait( proc, event, timeout );	/* Message may have been sent before this call: */	DMutex_Lock( & daoCallServer->mutex );	DaoChannel_ActivateEvent( NULL, DAO_EVENT_WAIT_SELECT );	DCondVar_Signal( & daoCallServer->condv );	DMutex_Unlock( & daoCallServer->mutex );}

void DaoMT_Select( DaoProcess *proc, DaoValue *par[], int n ){	DNode *it;	DaoTaskEvent *event = NULL;	DaoFuture *future = DaoProcess_GetInitFuture( proc );	DaoMap *selects = (DaoMap*) par[0];	float timeout = par[1]->xFloat.value;	for(it=DaoMap_First(selects); it; it=DaoMap_Next(selects,it)){		DaoValue *value = it->key.pValue;		int isfut = DaoValue_CheckCtype( value, dao_type_future );		int ischan = DaoValue_CheckCtype( value, dao_type_channel );		if( isfut == 0 && ischan == 0 ){			DaoProcess_RaiseError( proc, "Param", "invalid type selection" );			return;		}	}	event = DaoCallServer_MakeEvent();	DaoTaskEvent_Init( event, DAO_EVENT_WAIT_SELECT, DAO_EVENT_WAIT, future, NULL );	GC_Assign( & event->selects, selects );	proc->status = DAO_PROCESS_SUSPENDED;	proc->pauseType = DAO_PAUSE_CHANFUT_SELECT;	DaoCallServer_AddTimedWait( proc, event, timeout );	/* Message may have been sent before this call: */	DMutex_Lock( & daoCallServer->mutex );	DaoChannel_ActivateEvent( NULL, DAO_EVENT_WAIT_SELECT );	DCondVar_Signal( & daoCallServer->condv );	DMutex_Unlock( & daoCallServer->mutex );}

/*// Activate all events waiting on a future value:*/void DaoFuture_ActivateEvent( DaoFuture *self ){	DaoCallServer *server = daoCallServer;	DList *array = DList_New(0);	DNode *node;	daoint i;	DMutex_Lock( & server->mutex );	for(i=0; i<server->events2->size; ++i){		DaoTaskEvent *event = (DaoTaskEvent*) server->events2->items.pVoid[i];		if( DaoCallServer_CheckEvent( event, self, NULL ) ){			event->state = DAO_EVENT_RESUME;			DList_Append( server->events, event );			DList_Erase( server->events2, i, 1 );			i -= 1;		}	}	for(node=DMap_First(server->waitings); node; node=DMap_Next(server->waitings,node)){		DaoTaskEvent *event = (DaoTaskEvent*) node->value.pValue;		/* remove from timed waiting list: */		if( DaoCallServer_CheckEvent( event, self, NULL ) ){			event->state = DAO_EVENT_RESUME;			DList_Append( server->events, event );			DList_Append( array, node->key.pVoid );		}	}	for(i=0; i<array->size; i++) DMap_Erase( server->waitings, array->items.pVoid[i] );	DCondVar_Signal( & server->condv );	DMutex_Unlock( & server->mutex );	DList_Delete( array );}

void DThread_Exit( DThread *thd ){	thd->running = 0;	DCondVar_Signal( & thd->condv );	if( thd->cleaner ) (*(thd->cleaner))( thd->taskArg );	thd->myThread = NULL; /* it will be closed by _endthread() */	_endthread();}

static void DaoCallServer_Add( DaoTaskEvent *event ){	DaoCallServer *server = daoCallServer;	DMutex_Lock( & server->mutex );	DaoCallServer_AddEvent( event );	DCondVar_Signal( & server->condv );	DMutex_Unlock( & server->mutex );	DaoCallServer_TryAddThread( NULL, NULL, server->pending->size );}

void DaoChannel_Send( DaoChannel *self, DaoValue *data ){	DMutex_Lock( & daoCallServer->mutex );	DList_Append( self->buffer, data );	DaoChannel_ActivateEvent( self, DAO_EVENT_WAIT_RECEIVING );	DaoChannel_ActivateEvent( self, DAO_EVENT_WAIT_SELECT );	DCondVar_Signal( & daoCallServer->condv );	DMutex_Unlock( & daoCallServer->mutex );}

static void DaoCallServer_ActivateEvents(){	DaoCallServer *server = daoCallServer;	char message[128];	daoint i, j, count = 0;	if( server->finishing == 0 ) return;	if( server->idle != server->total ) return;	if( server->events->size != 0 ) return;	if( server->events2->size == 0 ) return;#ifdef DEBUG	sprintf( message, "WARNING: try activating events (%i,%i,%i,%i)!/n", server->total,			server->idle, (int)server->events->size, (int)server->events2->size );	DaoStream_WriteChars( mainVmSpace->errorStream, message );#endif	for(i=0; i<server->events2->size; ++i){		DaoTaskEvent *event = (DaoTaskEvent*) server->events2->items.pVoid[i];		DaoChannel *chan = event->channel;		DaoFuture *fut = event->future;		int move = 0, closed = 0;		switch( event->type ){		case DAO_EVENT_WAIT_TASKLET :			move = fut->precond == NULL || fut->precond->state == DAO_CALL_FINISHED;			break;		case DAO_EVENT_WAIT_RECEIVING :			move = chan->buffer->size > 0;			if( chan->cap <= 0 && chan->buffer->size == 0 ) move = 1;			break;		case DAO_EVENT_WAIT_SENDING :			move = chan->buffer->size < chan->cap;			break;		case DAO_EVENT_WAIT_SELECT :			if( event->selects == NULL ) continue;			move = DaoTaskEvent_CheckSelect( event );			break;		default: break;		}		if( move ){			DList_Append( server->events, event );			DList_Erase( server->events2, i, 1 );			count += 1;			i -= 1;		}	}	DCondVar_Signal( & server->condv );	if( count == 0 ){		DaoStream *stream = mainVmSpace->errorStream;		DaoStream_WriteChars( stream, "ERROR: All tasklets are suspended - deadlock!/n" );#if DEBUG		fprintf( stderr, "ERROR: All tasklets are suspended - deadlock!/n" );#endif		exit(1);	}}

static void DaoCallServer_ActivateEvents(){	DaoCallServer *server = daoCallServer;	daoint i, j;	if( server->idle != server->total ) return;	if( server->events->size != 0 ) return;	if( server->events2->size == 0 ) return;#ifdef DEBUG	DaoStream_WriteMBS( mainVmSpace->errorStream, "WARNING: activating events!/n" );#endif	for(i=0; i<server->events2->size; ++i){		DaoTaskEvent *event = (DaoTaskEvent*) server->events2->items.pVoid[i];		DaoChannel *chan = event->channel;		DaoFuture *fut = event->future;		int move = 0, closed = 0;		switch( event->type ){		case DAO_EVENT_WAIT_TASKLET :			move = fut->precond == NULL || fut->precond->state == DAO_CALL_FINISHED;			break;		case DAO_EVENT_WAIT_RECEIVING :			move = chan->buffer->size > 0;			if( chan->cap <= 0 && chan->buffer->size == 0 ) move = 1;			break;		case DAO_EVENT_WAIT_SENDING :			move = chan->buffer->size < chan->cap;			break;		case DAO_EVENT_WAIT_SELECT :			if( event->channels == NULL ) continue;			for(j=0; j<event->channels->size; ++j){				DaoChannel *chan = (DaoChannel*) event->channels->items.pValue[j];				closed += chan->cap <= 0;				move = chan->buffer->size > 0;				if( move ) break;			}			if( closed == event->channels->size ) move = 1;			break;		default: break;		}		if( move ){			DArray_Append( server->events, event );			DArray_Erase( server->events2, i, 1 );			i -= 1;		}	}	DCondVar_Signal( & server->condv );}

static void CHANNEL_Cap( DaoProcess *proc, DaoValue *par[], int N ){	DaoChannel *self = (DaoChannel*) par[0];	daoint i;	DaoProcess_PutInteger( proc, self->cap );	if( N == 1 ) return;	/* Closing the channel: */	DMutex_Lock( & daoCallServer->mutex );	self->cap = par[1]->xInteger.value;	if( self->cap == 0 ){		DaoChannel_ActivateEvent( self, DAO_EVENT_WAIT_RECEIVING );		DaoChannel_ActivateEvent( self, DAO_EVENT_WAIT_SELECT );		DCondVar_Signal( & daoCallServer->condv );	}	DMutex_Unlock( & daoCallServer->mutex );}

static void DaoCallServer_Timer( void *p ){	DaoCallServer *server = daoCallServer;	double time = 0.0;	daoint i, timeout;	server->timing = 1;	while( server->finishing == 0 || server->stopped != server->total ){		DMutex_Lock( & server->mutex );		while( server->waitings->size == 0 ){			if( server->idle == server->total && server->events2->size ){				DaoCallServer_ActivateEvents();			}			if( server->finishing && server->stopped == server->total ) break;			DCondVar_TimedWait( & server->condv2, & server->mutex, 0.01 );		}		if( server->waitings->size ){			DNode *node = DMap_First( server->waitings );			time = node->key.pValue->xComplex.value.real;			time -= Dao_GetCurrentTime();			/* wait the right amount of time for the closest arriving timeout: */			if( time > 0 ) DCondVar_TimedWait( & server->condv2, & server->mutex, time );		}		DMutex_Unlock( & server->mutex );		if( server->finishing && server->stopped == server->total ) break;		DMutex_Lock( & server->mutex );		if( server->waitings->size ){ /* a new wait timed out: */			DNode *node = DMap_First( server->waitings );			time = Dao_GetCurrentTime();			if( node->key.pValue->xComplex.value.real < time ){				DaoTaskEvent *event = (DaoTaskEvent*) node->value.pVoid;				event->state = DAO_EVENT_RESUME;				event->timeout = 1;				event->expiring = MIN_TIME;				DArray_Append( server->events, node->value.pVoid );				DMap_EraseNode( server->waitings, node );			}		}		DCondVar_Signal( & server->condv );		DMutex_Unlock( & server->mutex );	}	server->timing = 0;}

static void DaoMT_RunFunctional( void *p ){	DaoTaskData *self = (DaoTaskData*)p;	DaoProcess *clone = self->clone;	switch( self->param->type ){	case DAO_INTEGER : DaoMT_RunIterateFunctional( p ); break;	case DAO_LIST  : DaoMT_RunListFunctional( p ); break;	case DAO_MAP   : DaoMT_RunMapFunctional( p ); break;#ifdef DAO_WITH_NUMARRAY	case DAO_ARRAY : DaoMT_RunArrayFunctional( p ); break;#endif	}	self->status |= clone->status != DAO_PROCESS_FINISHED;	DMutex_Lock( self->mutex );	*self->joined += 1;	if( clone->exceptions->size ) DaoProcess_PrintException( clone, NULL, 1 );	DCondVar_Signal( self->condv );	DMutex_Unlock( self->mutex );}

void DaoCallServer_AddTask( DThreadTask func, void *param, int now ){	int scheduled = 0;	DaoCallServer *server = DaoCallServer_TryInit( mainVmSpace );	DMutex_Lock( & server->mutex );	if( server->idle > server->parameters->size || now == 0 ){		scheduled = 1;		DList_Append( server->functions, func );		DList_Append( server->parameters, param );		DMap_Insert( server->pending, param, NULL );		DCondVar_Signal( & server->condv );	}	DMutex_Unlock( & server->mutex );	if( scheduled ){		if( now == 0 ) DaoCallServer_TryAddThread( NULL, NULL, server->parameters->size );	}else{		DaoCallServer_AddThread( func, param );	}}

static void CHANNEL_Receive( DaoProcess *proc, DaoValue *par[], int N ){	DaoTaskEvent *event = NULL;	DaoFuture *future = DaoProcess_GetInitFuture( proc );	DaoChannel *self = (DaoChannel*) par[0];	float timeout = par[1]->xFloat.value;	event = DaoCallServer_MakeEvent();	DaoTaskEvent_Init( event, DAO_EVENT_WAIT_RECEIVING, DAO_EVENT_WAIT, future, self );	proc->status = DAO_PROCESS_SUSPENDED;	proc->pauseType = DAO_PAUSE_CHANNEL_RECEIVE;	DaoCallServer_AddTimedWait( proc, event, timeout );	/* Message may have been sent before this call: */	if( self->buffer->size ){		DMutex_Lock( & daoCallServer->mutex );		DaoChannel_ActivateEvent( self, DAO_EVENT_WAIT_RECEIVING );		DCondVar_Signal( & daoCallServer->condv );		DMutex_Unlock( & daoCallServer->mutex );	}}

static DaoFuture* DaoCallServer_GetNextFuture(){	DaoCallServer *server = daoCallServer;	DaoFuture *first, *future, *precond;	DArray *events = server->events;	DMap *pending = server->pending;	DMap *active = server->active;	daoint i, j;	for(i=0; i<events->size; i++){		DaoTaskEvent *event = (DaoTaskEvent*) events->items.pVoid[i];		DaoFuture *future = event->future;		DaoChannel *channel = event->channel;		DaoChannel *selected = NULL;		DaoValue *message = NULL;		int closed = 0;		switch( event->type ){		case DAO_EVENT_WAIT_SENDING :			if( channel->buffer->size >= channel->cap ){				if( event->state == DAO_EVENT_WAIT ) goto MoveToWaiting;			}			break;		case DAO_EVENT_WAIT_RECEIVING :			if( channel->buffer->size == 0 ){				if( channel->cap > 0 && event->state == DAO_EVENT_WAIT ) goto MoveToWaiting;				message = dao_none_value;			}else{				message = channel->buffer->items.pValue[0];			}			GC_ShiftRC( message, event->future->message );			event->future->message = message;			event->future->aux1 = channel->cap <= 0 && channel->buffer->size == 0;			DArray_PopFront( channel->buffer );			if( channel->buffer->size < channel->cap )				DaoChannel_ActivateEvent( channel, DAO_EVENT_WAIT_SENDING );			if( channel->buffer->size )				DaoChannel_ActivateEvent( channel, DAO_EVENT_WAIT_RECEIVING );			break;		case DAO_EVENT_WAIT_SELECT :			for(j=0; j<event->channels->size; ++j){				DaoChannel *chan = (DaoChannel*) event->channels->items.pValue[j];				if( chan->buffer->size > 0 ){					selected = chan;					break;				}else if( chan->cap <= 0 ){					closed += 1;				}			}			if( selected == NULL && closed < event->channels->size ){				if( event->state == DAO_EVENT_WAIT ) goto MoveToWaiting;			}			if( selected == NULL || selected->buffer->size == 0 ){				if( event->state == DAO_EVENT_WAIT ) goto MoveToWaiting;				message = dao_none_value;			}else{				message = selected->buffer->items.pValue[0];			}			GC_ShiftRC( message, event->future->message );			GC_ShiftRC( selected, event->future->selected );			event->future->message = message;			event->future->selected = (DaoValue*) selected;			event->future->aux1 = closed == event->channels->size;			if( selected ){				DArray_PopFront( selected->buffer );				if( selected->buffer->size < selected->cap )					DaoChannel_ActivateEvent( selected, DAO_EVENT_WAIT_SENDING );				if( selected->buffer->size )					DaoChannel_ActivateEvent( selected, DAO_EVENT_WAIT_SELECT );			}			break;		default: break;		}		if( event->state == DAO_EVENT_WAIT && future->precond != NULL ){			if( future->precond->state != DAO_CALL_FINISHED ) goto MoveToWaiting;		}		if( future->actor && DMap_Find( active, future->actor->rootObject ) ) continue;		if( future->process && DMap_Find( active, future->process ) ) continue;		DArray_Erase( events, i, 1 );		DMap_Erase( pending, event );		if( future->actor ) DMap_Insert( active, future->actor->rootObject, NULL );		if( future->process ) DMap_Insert( active, future->process, NULL );		GC_IncRC( future ); /* To be decreased at the end of tasklet; */		future->timeout = event->timeout;		DaoCallServer_CacheEvent( event );		return future;MoveToWaiting:		if( event->expiring >= 0.0 && event->expiring < MIN_TIME ) continue;		if( event->expiring >= MIN_TIME ){			DaoComplex com = {DAO_COMPLEX,0,0,0,1,{0.0,0.0}};			com.value.real = event->expiring;			DMap_Insert( server->waitings, & com, event );			DCondVar_Signal( & server->condv2 );		}else{			DArray_Append( server->events2, event );		}		DArray_Erase( server->events, i, 1 );		i -= 1;	}	return NULL;//.........这里部分代码省略.........

void DThread_Detach( DThread *self ){	DCondVar_Signal( & self->condv );}

static void DaoCondV_Lib_Signal( DaoProcess *proc, DaoValue *par[], int N ){	DaoCondVar *self = (DaoCondVar*) par[0];	DCondVar_Signal( & self->myCondVar );}

void DThread_Cancel( DThread *self ){	self->thdSpecData->state |= DTHREAD_CANCELED;	DCondVar_Signal( & self->condv );}

void DaoCondVar_Signal( DaoCondVar *self ){	DCondVar_Signal( & self->myCondVar );}

//.........这里部分代码省略.........				DaoChannel_ActivateEvent( channel, DAO_EVENT_WAIT_RECEIVING );			break;		case DAO_EVENT_WAIT_SELECT :			message = dao_none_value;			for(it=DaoMap_First(event->selects); it; it=DaoMap_Next(event->selects,it)){				if( DaoValue_CheckCtype( it->key.pValue, dao_type_channel ) ){					DaoChannel *chan = (DaoChannel*) it->key.pValue;					if( chan->buffer->size > 0 ){						chselect = chan;						selected = it->key.pValue;						message = chan->buffer->items.pValue[0];						closed = NULL;						break;					}else if( chan->cap == 0 ){						closed = chan;					}				}else{					DaoFuture *fut = (DaoFuture*) it->key.pValue;					if( fut->state == DAO_CALL_FINISHED ){						futselect = fut;						selected = it->key.pValue;						message = fut->value;						break;					}				}			}			if( selected == NULL ) selected = (DaoValue*) closed;			if( event->state == DAO_EVENT_WAIT && event->selects->value->size ){				if( selected == NULL ) goto MoveToWaiting;			}			GC_Assign( & event->message, message );			GC_Assign( & event->selected, selected );			event->auxiliary = event->selects->value->size == 0;			event->type = DAO_EVENT_RESUME_TASKLET;			/* change status to not finished: */			if( chselect != NULL || futselect != NULL ) event->auxiliary = 0;			if( chselect ){				DList_PopFront( chselect->buffer );				if( chselect->buffer->size < chselect->cap )					DaoChannel_ActivateEvent( chselect, DAO_EVENT_WAIT_SENDING );				if( chselect->buffer->size )					DaoChannel_ActivateEvent( chselect, DAO_EVENT_WAIT_SELECT );			}			if( futselect != NULL || closed != NULL ){				void *key = futselect ? (void*)futselect : (void*)closed;				DMap_Erase( event->selects->value, key );			}			break;		default: break;		}		if( actor ){			DNode *it = DMap_Find( active, actor->rootObject );			if( actor->rootObject->isAsync ){				if( it && it->value.pVoid != (void*) future ) continue;			}else if( it ){				continue;			}		}		if( future->process && DMap_Find( active, future->process ) ) continue;		DList_Erase( events, i, 1 );		DMap_Erase( pending, event );		if( actor ){			void *value = actor->rootObject->isAsync ? future : NULL;			DMap_Insert( active, actor->rootObject, value );		}		if( future->process ){			DMap_Insert( active, future->process, NULL );			future->process->active = 1;		}		/*		// DaoValue_Move() should be used instead of GC_Assign() for thread safety.		// Because using GC_Assign() here, may caused "future->message" of primitive		// type being deleted, right after DaoFuture_GetGCFields() has retrieved it		// for GC scanning.		 */		DaoValue_Move( event->message, & future->message, NULL );		DaoValue_Move( event->selected, & future->selected, NULL );		future->aux1 = event->auxiliary;		future->timeout = event->timeout;		GC_IncRC( future ); /* To be decreased at the end of tasklet; */		DaoCallServer_CacheEvent( event );		return future;MoveToWaiting:		if( event->expiring >= 0.0 && event->expiring < MIN_TIME ) continue;		if( event->expiring >= MIN_TIME ){			dao_complex com = {0.0,0.0};			com.real = event->expiring;			DMap_Insert( server->waitings, & com, event );			DCondVar_Signal( & server->condv2 );		}else{			DList_Append( server->events2, event );		}		DList_Erase( server->events, i, 1 );		i -= 1;	}	return NULL;}

static void DaoCallThread_Run( DaoCallThread *self ){	DaoCallServer *server = daoCallServer;	double wt = 0.001;	daoint i, timeout;	self->thdData = DThread_GetSpecific();	if( self->taskFunc ) self->taskFunc( self->taskParam );	while(1){		DaoProcess *process = NULL;		DaoFuture *future = NULL;		DThreadTask function = NULL;		void *parameter = NULL;		self->thdData->state = 0;		DMutex_Lock( & server->mutex );		server->idle += 1;		while( server->pending->size == (server->events2->size + server->waitings->size) ){			//printf( "%p %i %i %i %i/n", self, server->events->size, server->pending->size, server->events2->size, server->waitings->size );			if( server->finishing && server->idle == server->total ){				if( (server->events2->size + server->waitings->size) == 0 ) break;			}			timeout = DCondVar_TimedWait( & server->condv, & server->mutex, wt );		}		for(i=0; i<server->parameters->size; ++i){			void *param = server->parameters->items.pVoid[i];			if( DMap_Find( server->active, param ) ) continue;			DMap_Insert( server->active, param, NULL );			function = (DThreadTask) server->functions->items.pVoid[i];			parameter = param;			DArray_Erase( server->functions, i, 1 );			DArray_Erase( server->parameters, i, 1 );			DMap_Erase( server->pending, parameter );			server->idle -= 1;			break;		}		DMutex_Unlock( & server->mutex );		if( function ){			(*function)( parameter );			DMutex_Lock( & server->mutex );			DMap_Erase( server->active, parameter );			DMutex_Unlock( & server->mutex );			continue;		}		if( server->pending->size == 0 && server->finishing && server->idle == server->total ) break;		DMutex_Lock( & server->mutex );		server->idle -= 1;		future = DaoCallServer_GetNextFuture();		DMutex_Unlock( & server->mutex );		if( future == NULL ) continue;		process = future->process;		if( process == NULL ){			GC_DecRC( future );			continue;		}		if( process->pauseType == DAO_PAUSE_NATIVE_THREAD ){			DMutex_Lock( & server->mutex );			process->pauseType = 0;			/* TODO: demo/concurrent/future.dao */			if( process->condv ) DCondVar_Signal( process->condv );			DMutex_Unlock( & server->mutex );		}else{			int count = process->exceptions->size;			future->state = DAO_CALL_RUNNING;			DaoProcess_InterceptReturnValue( process );			DaoProcess_Execute( process );			if( process->exceptions->size > count ) DaoProcess_PrintException( process, 1 );		}		if( future->actor ){			DMutex_Lock( & server->mutex );			DMap_Erase( server->active, future->actor->rootObject );			DMutex_Unlock( & server->mutex );		}		DMutex_Lock( & server->mutex );		DMap_Erase( server->active, process );		DMutex_Unlock( & server->mutex );		DaoProcess_ReturnFutureValue( process, future );		if( future->state == DAO_CALL_FINISHED ) DaoFuture_ActivateEvent( future );		GC_DecRC( future );	}	DMutex_Lock( & server->mutex );	server->stopped += 1;	DMutex_Unlock( & server->mutex );}