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

void DaoCallServer_Stop(){	DCondVar condv;	DaoCallThread *calth;	if( daoCallServer == NULL ) return;	DCondVar_Init( & condv );	daoCallServer->finishing = 1;	calth = DaoCallThread_New( NULL, NULL );	DMutex_Lock( & daoCallServer->mutex );	daoCallServer->total += 1;	DMutex_Unlock( & daoCallServer->mutex );	DaoCallThread_Run( calth );  /* process tasks in the main thread; */	DMutex_Lock( & daoCallServer->mutex );	while( daoCallServer->stopped != daoCallServer->total || daoCallServer->timing ){		DCondVar_TimedWait( & condv, & daoCallServer->mutex, 0.01 );	}	DMutex_Unlock( & daoCallServer->mutex );	DCondVar_Destroy( & condv );	DaoCallThread_Delete( calth );	DaoCallServer_Delete( daoCallServer );	daoCallServer = NULL;}

void DCondVar_Wait( DCondVar *self, DMutex *mtx ){	DThreadData *p = (DThreadData*)TlsGetValue( thdSpecKey );	DMutex_Lock( & self->thdMutex );	DArray_PushBack( self->thdWaiting, (void*) p->thdObject );	DMutex_Unlock( & self->thdMutex );	if( mtx ) DMutex_Unlock( mtx );	WaitForSingleObject( p->thdObject->condv.myCondVar, INFINITE );	ResetEvent( p->thdObject->condv.myCondVar );	if( mtx ) DMutex_Lock( mtx );	if( p->state & DTHREAD_CANCELED ) DThread_Exit( p->thdObject );}

static void DaoMT_Critical( DaoProcess *proc, DaoValue *p[], int n ){	void *key;	DNode *it;	DMap *cache = (DMap*) DaoProcess_GetAuxData( proc, DaoMT_ProcMutexCache );	DaoVmCode *sect = DaoProcess_InitCodeSection( proc, 0 );	DaoRoutine *routine = proc->activeRoutine;	if( sect == NULL ) return;	/* Get the original routine, if this one is a specialized copy: */	while( routine->original ) routine = routine->original;	/*	// Use "routine + sect->c" instead of "sect" as the key for mutex,	// as "sect" may be different for different copy of specialized routine.	// But "sect->c" won't change after being set during compiling.	*/	key = routine + sect->c;	if( cache == NULL ){		cache = DHash_New(0,0); /* Local cache is used to avoid extra locking; */		DaoProcess_SetAuxData( proc, DaoMT_ProcMutexCache, cache );	}	it = DMap_Find( cache, key ); /* Check local cache first; */	if( it == NULL ) it = DMap_Insert( cache, key, DaoMT_GetMutex( routine, key ) );	DMutex_Lock( (DMutex*) it->value.pVoid );	DaoProcess_Execute( proc );	DMutex_Unlock( (DMutex*) it->value.pVoid );	DaoProcess_PopFrame( proc );}

static void SYS_Sleep( DaoProcess *proc, DaoValue *p[], int N ){#ifdef DAO_WITH_THREAD	DMutex    mutex;	DCondVar  condv;#endif	double s = p[0]->xFloat.value;	if( s < 0 ){		DaoProcess_RaiseException( proc, DAO_WARNING_VALUE, "expecting positive value" );		return;	}#ifdef DAO_WITH_THREAD	/* sleep only the current thread: */	DMutex_Init( & mutex );	DCondVar_Init( & condv );	DMutex_Lock( & mutex );	DCondVar_TimedWait( & condv, & mutex, s );	DMutex_Unlock( & mutex );	DMutex_Destroy( & mutex );	DCondVar_Destroy( & condv );#elif UNIX	sleep( (int)s ); /* This may cause the whole process to sleep. */#else	Sleep( s * 1000 );#endif}

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 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 void TEST_Run( DaoProcess *proc, DaoValue* p[], int N ){	DString *prefix = p[0]->xString.value;	DList *funcs = proc->activeNamespace->definedRoutines;	DaoStream *out = DaoVmSpace_StdioStream( proc->vmSpace );	daoint i;	char buf[100];	for ( i = 0; i < funcs->size; ++i ){		DaoRoutine *rout = funcs->items.pRoutine[i];		// find global routines matching prefix		if ( strncmp( rout->routName->chars, prefix->chars, prefix->size ) == 0 ){			// ignore 'main()'			if ( strcmp( rout->routName->chars, "main" ) == 0 )				continue;			// run			if ( !( rout->attribs & DAO_ROUT_DECORATOR ) && rout->parCount == 0 ){				if ( DaoProcess_Call( proc, rout, NULL, NULL, 0 ) == DAO_ERROR ){ // failed					char buf[512];					snprintf( buf, sizeof(buf), "unexpected error running '%s'", rout->routName->chars );					DaoProcess_RaiseError( proc, "Error::Test", buf );					return;				}			}		}	}	DMutex_Lock( &test_mtx );	snprintf( buf, sizeof(buf), "Summary: %i tests, %i passed, %i failed, %i skipped/n", test_count, pass_count, fail_count, skip_count );	DaoStream_WriteChars( out, buf );	test_count = pass_count = fail_count = skip_count = 0;	DMutex_Unlock( &test_mtx );}

void DThread_ResumeVM( DThread *another ){	another->vmpause = 0;	DMutex_Lock( & another->mutex );	DCondVar_Signal( & another->condv );	DMutex_Unlock( & another->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 );}

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 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 );}

/*// 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 );}

int DCondVar_TimedWait( DCondVar *self, DMutex *mtx, double seconds ){	DWORD retc;	DThreadData *p = (DThreadData*)TlsGetValue( thdSpecKey );	DMutex_Lock( & self->thdMutex );	DArray_PushBack( self->thdWaiting, (void*) p->thdObject );	DMutex_Unlock( & self->thdMutex );	if( mtx ) DMutex_Unlock( mtx );	retc = WaitForSingleObject( p->thdObject->condv.myCondVar, (DWORD)( seconds * 1000 ) );	ResetEvent( p->thdObject->condv.myCondVar );	if( mtx ) DMutex_Lock( mtx );	if( p->state & DTHREAD_CANCELED ) DThread_Exit( p->thdObject );	return ( retc == WAIT_TIMEOUT );}

void DThread_PauseVM( DThread *another ){	another->vmpause = 1;	DMutex_Lock( & another->mutex );	if( ! another->vmpaused ){		DCondVar_TimedWait( & another->condv, & another->mutex, 0.01 );	}	DMutex_Unlock( & another->mutex );}

static void DaoMT_Critical( DaoProcess *proc, DaoValue *p[], int n ){	DaoVmCode *sect = DaoGetSectionCode( proc->activeCode );	if( sect == NULL || DaoMT_PushSectionFrame( proc ) == 0 ) return;	if( proc->mutex ) DMutex_Lock( proc->mutex );	DaoProcess_Execute( proc );	if( proc->mutex ) DMutex_Unlock( proc->mutex );	DaoProcess_PopFrame( proc );}

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_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 );}

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 DaoTaskEvent* DaoCallServer_MakeEvent(){	DaoTaskEvent *event;	DaoCallServer *server = daoCallServer;	DMutex_Lock( & server->mutex );	event = (DaoTaskEvent*) DList_PopBack( server->caches );	if( event == NULL ) event = DaoTaskEvent_New();	DMutex_Unlock( & server->mutex );	return event;}

static void DaoCallServer_AddThread( DThreadTask func, void *param ){	DaoCallThread *calth;	DaoCallServer_TryInit( mainVmSpace );	calth = DaoCallThread_New( func, param );	DMutex_Lock( & daoCallServer->mutex );	daoCallServer->total += 1;	DArray_Append( daoCallServer->threads, calth );	DMutex_Unlock( & daoCallServer->mutex );	DThread_Start( & calth->thread, (DThreadTask) DaoCallThread_Run, calth );}

static DaoTaskEvent* DaoCallServer_MakeEvent(){	DaoTaskEvent *event;	DaoCallServer *server;	if( daoCallServer == NULL ) DaoCallServer_Init( mainVmSpace );	server = daoCallServer;	DMutex_Lock( & server->mutex );	event = (DaoTaskEvent*) DArray_PopBack( server->caches );	if( event == NULL ) event = DaoTaskEvent_New();	DMutex_Unlock( & server->mutex );	return event;}

void DaoCallServer_Join(){	DCondVar condv;	if( daoCallServer == NULL ) return;	DCondVar_Init( & condv );	DMutex_Lock( & daoCallServer->mutex );	while( daoCallServer->pending->size || daoCallServer->vacant != daoCallServer->total ){		DCondVar_TimedWait( & condv, & daoCallServer->mutex, 0.01 );	}	DMutex_Unlock( & daoCallServer->mutex );	DCondVar_Destroy( & condv );}

void DCondVar_Signal( DCondVar *self ){	DThread *thread;	DMutex_Lock( & self->thdMutex );	if( self->thdWaiting->size > 0 ){		thread = (DThread*) self->thdWaiting->items.pVoid[0];		SetEvent( thread->condv.myCondVar );		DArray_PopFront( self->thdWaiting );	}	DMutex_Unlock( & self->thdMutex );}

void DCondVar_BroadCast( DCondVar *self ){	DThread *thread;	int i;	DMutex_Lock( & self->thdMutex );	for( i=0; i<self->thdWaiting->size; i++ ){		thread = (DThread*) self->thdWaiting->items.pVoid[i];		SetEvent( thread->condv.myCondVar );	}	DArray_Clear( self->thdWaiting );	DMutex_Unlock( & self->thdMutex );}

void DaoCallServer_AddThread( DThreadTask func, void *param ){	DaoCallThread *calth;	DaoCallServer_TryInit( mainVmSpace );	calth = DaoCallThread_New( func, param );	DMutex_Lock( & daoCallServer->mutex );	daoCallServer->total += 1;	DList_Append( daoCallServer->threads, calth );	DMutex_Unlock( & daoCallServer->mutex );	if( DThread_Start( & calth->thread, (DThreadTask) DaoCallThread_Run, calth ) == 0 ){		if( func != NULL || daoCallServer->total == 0 ){			dao_abort( "failed to create a task thread" );		}	}}