# HG changeset patch # User Me # Date 1288742265 25200 # Node ID bf57b83019e597b0fc8ac3fdacd3bb454b62f73b # Parent 6c6d7fbd7e2559c981c5ef1015427ec5e395c477 Added atomic fn exec, singleton, trans start and end and fixed bus (contd) Fixed double-free of waiting requests bug Fixed place didn't clone in request handling Changed malloc and free handler prototypes diff -r 6c6d7fbd7e25 -r bf57b83019e5 SSR.h --- a/SSR.h Thu Oct 14 17:06:28 2010 -0700 +++ b/SSR.h Tue Nov 02 16:57:45 2010 -0700 @@ -13,10 +13,17 @@ #include "VMS/Hash_impl/PrivateHash.h" #include "VMS/VMS.h" + +//=========================================================================== + +#define NUM_STRUCS_IN_SEM_ENV 1000 + +//=========================================================================== /*This header defines everything specific to the SSR semantic plug-in */ typedef struct _SSRSemReq SSRSemReq; - +typedef void (*PtrToAtomicFn ) ( void * ); //executed atomically in master +//=========================================================================== /*Semantic layer-specific data sent inside a request from lib called in app * to request handler called in MasterLoop @@ -29,19 +36,48 @@ receive_type, // and receive_any first of the receives -- Handlers receive_from_to,// rely upon this ordering of enum transfer_to, - transfer_out + transfer_out, + malloc_req, + free_req, + singleton, + atomic, + trans_start, + trans_end }; struct _SSRSemReq { enum SSRReqType reqType; - VirtProcr *sendPr; - VirtProcr *receivePr; - int32 msgType; - void *msg; + VirtProcr *sendPr; + VirtProcr *receivePr; + int32 msgType; + void *msg; SSRSemReq *nextReqInHashEntry; + + void *initData; + VirtProcrFnPtr fnPtr; + int32 coreToScheduleOnto; + + int32 sizeToMalloc; + void *ptrToFree; + + int32 singletonID; + void *endJumpPt; + + PtrToAtomicFn fnToExecInMaster; + void *dataForFn; + + int32 transID; } /* SSRSemReq */; + +typedef struct + { + VirtProcr *VPCurrentlyExecuting; + PrivQueueStruc *waitingVPQ; + } +SSRTrans; + typedef struct { PrivQueueStruc **readyVPQs; @@ -49,10 +85,28 @@ int32 numVirtPr; int32 nextCoreToGetNewPr; int32 primitiveStartTime; + + //fix limit on num with dynArray + int32 singletonHasBeenExecutedFlags[NUM_STRUCS_IN_SEM_ENV]; + SSRTrans transactionStrucs[NUM_STRUCS_IN_SEM_ENV]; } SSRSemEnv; +typedef struct _TransListElem TransListElem; +struct _TransListElem + { + int32 transID; + TransListElem *nextTrans; + }; + +typedef struct + { + int32 highestTransEntered; + TransListElem *lastTransEntered; + } +SSRSemData; + //=========================================================================== void @@ -70,6 +124,9 @@ int32 SSR__giveIdealNumWorkUnits(); +int32 +SSR__give_number_of_cores_to_schedule_onto(); + //======================= void @@ -84,12 +141,19 @@ SSR__create_procr_with( VirtProcrFnPtr fnPtr, void *initData, VirtProcr *creatingPr ); +inline VirtProcr * +SSR__create_procr_with_affinity( VirtProcrFnPtr fnPtr, void *initData, + VirtProcr *creatingPr, int32 coreToScheduleOnto); + void SSR__dissipate_procr( VirtProcr *procrToDissipate ); //======================= void * -SSR__malloc_size_to( int numBytes, VirtProcr *ownerPr ); +SSR__malloc_to( int numBytes, VirtProcr *ownerPr ); + +void +SSR__free( void *ptrToFree, VirtProcr *owningPr ); void SSR__transfer_ownership_of_from_to( void *data, VirtProcr *oldOwnerPr, @@ -121,10 +185,20 @@ SSR__receive_from_to( VirtProcr *sendPr, VirtProcr *receivePr ); -//======================= +//======================= Concurrency Stuff ====================== +void +SSR__start_singleton( int32 singletonID, void *endSingletonLabelAddr, + VirtProcr *animPr ); void -SSR__free_semantic_request( SSRSemReq *semReq ); +SSR__animate_short_fn_in_isolation( PtrToAtomicFn ptrToFnToExecInMaster, + void *data, VirtProcr *animPr ); + +void +SSR__start_transaction( int32 transactionID, VirtProcr *animPr ); + +void +SSR__end_transaction( int32 transactionID, VirtProcr *animPr ); //========================= Internal use only ============================= diff -r 6c6d7fbd7e25 -r bf57b83019e5 SSR_PluginFns.c --- a/SSR_PluginFns.c Thu Oct 14 17:06:28 2010 -0700 +++ b/SSR_PluginFns.c Tue Nov 02 16:57:45 2010 -0700 @@ -6,13 +6,49 @@ #include #include -#include #include "VMS/Queue_impl/PrivateQueue.h" #include "SSR.h" #include "SSR_Request_Handlers.h" +//=========================== Local Fn Prototypes =========================== +void inline +resumePr( VirtProcr *procr, SSRSemEnv *semEnv ); +void inline +handleSemReq( VMSReqst *req, VirtProcr *requestingPr, SSRSemEnv *semEnv ); + +void +handleDissipate( VirtProcr *requestingPr, SSRSemEnv *semEnv ); + +void +handleCreate( VMSReqst *req, VirtProcr *requestingPr, SSRSemEnv *semEnv ); + + +//============================== Scheduler ================================== +// +/*For SSR, scheduling a slave simply takes the next work-unit off the + * ready-to-go work-unit queue and assigns it to the slaveToSched. + *If the ready-to-go work-unit queue is empty, then nothing to schedule + * to the slave -- return FALSE to let Master loop know scheduling that + * slave failed. + */ +VirtProcr * +SSR__schedule_virt_procr( void *_semEnv, int coreNum ) + { VirtProcr *schedPr; + SSRSemEnv *semEnv; + + semEnv = (SSRSemEnv *)_semEnv; + + schedPr = readPrivQ( semEnv->readyVPQs[coreNum] ); + //Note, using a non-blocking queue -- it returns NULL if queue empty + + return( schedPr ); + } + + +//=========================== Request Handler ============================= +// /*Will get requests to send, to receive, and to create new processors. * Upon send, check the hash to see if a receive is waiting. * Upon receive, check hash to see if a send has already happened. @@ -27,99 +63,132 @@ SSR__Request_Handler( VirtProcr *requestingPr, void *_semEnv ) { SSRSemEnv *semEnv; VMSReqst *req; - SSRSemReq *semReq; semEnv = (SSRSemEnv *)_semEnv; - req = VMS__take_top_request_from( requestingPr ); - + req = VMS__take_next_request_out_of( requestingPr ); + while( req != NULL ) { - if( VMS__isSemanticReqst( req ) ) - { - semReq = VMS__take_sem_reqst_from( req ); - if( semReq == NULL ) goto DoneHandlingReqst; - switch( semReq->reqType ) - { - case send_type: handleSendType( semReq, semEnv); - break; - case send_from_to: handleSendFromTo( semReq, semEnv); - break; - case receive_type: handleReceiveType( semReq, semEnv); - break; - case receive_from_to: handleReceiveFromTo(semReq, semEnv); - break; - case transfer_to: handleTransferTo( semReq, semEnv); - SSR__free_semantic_request( semReq ); - break; - case transfer_out: handleTransferOut( semReq, semEnv); - SSR__free_semantic_request( semReq ); - break; - } - //NOTE: freeing semantic request data strucs handled inside these + switch( req->reqType ) + { case semantic: handleSemReq( req, requestingPr, semEnv); + break; + case createReq: handleCreate( req, requestingPr, semEnv); + break; + case dissipate: handleDissipate( requestingPr, semEnv); + break; + case VMSSemantic: VMS__handle_VMSSemReq(req, requestingPr, semEnv, + &resumePr); + break; + default: + break; } - else if( VMS__isCreateReqst( req ) ) //only plugin can add to ready Q - { VirtProcr * - newPr = (VirtProcr *)req->semReqData; - semEnv->numVirtPr += 1; + + req = VMS__take_next_request_out_of( requestingPr ); + } //while( req != NULL ) - //Assign new processor to next core in line & queue it up - #ifdef DEBUG - newPr->coreAnimatedBy = 0; - #else - newPr->coreAnimatedBy = semEnv->nextCoreToGetNewPr; - if( semEnv->nextCoreToGetNewPr >= NUM_CORES - 1 ) - semEnv->nextCoreToGetNewPr = 0; - else - semEnv->nextCoreToGetNewPr += 1; - #endif - writePrivQ( newPr, semEnv->readyVPQs[requestingPr->coreAnimatedBy]); - - //resume procr that asked for registration - writePrivQ( requestingPr, - semEnv->readyVPQs[requestingPr->coreAnimatedBy]); - } - else if( VMS__isDissipateReqst( req ) ) - { - //free any semantic data allocated to the virt procr - - //Now, call VMS to free_all AppVP state -- stack and so on - VMS__handle_dissipate_reqst( requestingPr ); - - semEnv->numVirtPr -= 1; - if( semEnv->numVirtPr == 0 ) - { //no more work, so shutdown - VMS__handle_shutdown_reqst( requestingPr ); - } - } - - DoneHandlingReqst: - //Free VMS portion of request, no matter what -- sem request data - // struc instances may still be around.. VMS__free_request doesn't - // affect the semantic request that was carried by it - req = VMS__free_top_and_give_next_request_from( requestingPr ); - } //while( req != NULL ) } -//=========================================================================== +void inline +handleSemReq( VMSReqst *req, VirtProcr *reqPr, SSRSemEnv *semEnv ) + { SSRSemReq *semReq; -/*For SSR, scheduling a slave simply takes the next work-unit off the - * ready-to-go work-unit queue and assigns it to the slaveToSched. - *If the ready-to-go work-unit queue is empty, then nothing to schedule - * to the slave -- return FALSE to let Master loop know scheduling that - * slave failed. - */ -VirtProcr * -SSR__schedule_virt_procr( void *_semEnv, int coreNum ) - { VirtProcr *schedPr; - SSRSemEnv *semEnv; - - semEnv = (SSRSemEnv *)_semEnv; - - schedPr = readPrivQ( semEnv->readyVPQs[coreNum] ); - //Note, using a non-blocking queue -- it returns NULL if queue empty - - return( schedPr ); + semReq = VMS__take_sem_reqst_from(req); + if( semReq == NULL ) return; + switch( semReq->reqType ) //sem handlers are all in other file + { + case send_type: handleSendType( semReq, semEnv); + break; + case send_from_to: handleSendFromTo( semReq, semEnv); + break; + case receive_type: handleReceiveType( semReq, semEnv); + break; + case receive_from_to: handleReceiveFromTo(semReq, semEnv); + break; + case transfer_to: handleTransferTo( semReq, semEnv); + break; + case transfer_out: handleTransferOut( semReq, semEnv); + break; + case malloc_req: handleMalloc( semReq, reqPr, semEnv); + break; + case free_req: handleFree( semReq, reqPr, semEnv); + break; + case singleton: handleSingleton( semReq, reqPr, semEnv); + break; + case atomic: handleAtomic( semReq, reqPr, semEnv); + break; + case trans_start: handleTransStart( semReq, reqPr, semEnv); + break; + case trans_end: handleTransEnd( semReq, reqPr, semEnv); + break; + } } + + +//=========================== VMS Request Handlers ============================== +// +void +handleDissipate( VirtProcr *requestingPr, SSRSemEnv *semEnv ) + { + //free any semantic data allocated to the virt procr + + //Now, call VMS to free_all AppVP state -- stack and so on + VMS__dissipate_procr( requestingPr ); + + semEnv->numVirtPr -= 1; + if( semEnv->numVirtPr == 0 ) + { //no more work, so shutdown + VMS__shutdown(); + } + } + +void +handleCreate( VMSReqst *req, VirtProcr *requestingPr, SSRSemEnv *semEnv ) + { SSRSemReq *semReq; + VirtProcr *newPr; + + semReq = VMS__take_sem_reqst_from( req ); + + //This is running in master, so use internal version + newPr = VMS__create_procr( semReq->fnPtr, semReq->initData ); + + semEnv->numVirtPr += 1; + + //Assign new processor to a core & transition it to ready + #ifdef SEQUENTIAL + newPr->coreAnimatedBy = 0; + + #else + int32 + coreToScheduleOnto = semReq->coreToScheduleOnto; + + if(coreToScheduleOnto < 0 || coreToScheduleOnto >= NUM_CORES ) + { //out-of-range, so round-robin assignment + newPr->coreAnimatedBy = semEnv->nextCoreToGetNewPr; + + if( semEnv->nextCoreToGetNewPr >= NUM_CORES - 1 ) + semEnv->nextCoreToGetNewPr = 0; + else + semEnv->nextCoreToGetNewPr += 1; + } + else //core num in-range, so use it + { newPr->coreAnimatedBy = coreToScheduleOnto; + } + #endif + + //For SSR, caller needs ptr to created processor returned to it + requestingPr->dataRetFromReq = newPr; + + resumePr( newPr, semEnv ); + resumePr( requestingPr, semEnv ); + } + + +//=========================== Helper ============================== +void inline +resumePr( VirtProcr *procr, SSRSemEnv *semEnv ) + { + writePrivQ( procr, semEnv->readyVPQs[ procr->coreAnimatedBy] ); + } diff -r 6c6d7fbd7e25 -r bf57b83019e5 SSR_Request_Handlers.c --- a/SSR_Request_Handlers.c Thu Oct 14 17:06:28 2010 -0700 +++ b/SSR_Request_Handlers.c Tue Nov 02 16:57:45 2010 -0700 @@ -6,7 +6,6 @@ #include #include -#include #include "VMS/VMS.h" #include "VMS/Queue_impl/PrivateQueue.h" @@ -15,19 +14,41 @@ +//=========================== Local Fn Prototypes =========================== +void inline +resumePr( VirtProcr *procr, SSRSemEnv *semEnv ); + + + //=========================================================================== // Helpers +/*Only clone the elements of req used in these reqst handlers + */ +inline SSRSemReq * +cloneReq( SSRSemReq *semReq ) + { SSRSemReq *clonedReq; + + clonedReq = VMS__malloc( sizeof(SSRSemReq) ); + clonedReq->reqType = semReq->reqType; + clonedReq->sendPr = semReq->sendPr; + clonedReq->msg = semReq->msg; + clonedReq->nextReqInHashEntry = NULL; + + return clonedReq; + } + HashEntry * giveEntryElseInsertReqst( char *key, SSRSemReq *semReq, HashTable *commHashTbl ) { HashEntry *entry; - SSRSemReq *waitingReq; + SSRSemReq *waitingReq; entry = getEntryFromTable( (char *)key, commHashTbl ); if( entry == NULL ) { //no waiting sends or receives, so add this request and exit - addValueIntoTable( key, semReq, commHashTbl ); + // note: have to clone the request because it's on stack of sender + addValueIntoTable( key, cloneReq( semReq ), commHashTbl ); return NULL; } waitingReq = (SSRSemReq *)entry->content; @@ -60,7 +81,7 @@ * separate processors can send to the same receiver, and hashing on the * receive processor, so they will stack up. */ -void +void inline handleSendType( SSRSemReq *semReq, SSRSemEnv *semEnv ) { VirtProcr *sendPr, *receivePr; int key[] = {0,0,0}; @@ -82,7 +103,7 @@ key[1] = (int)(semReq->msgType); //key[2] acts as the 0 that terminates the string - entry = giveEntryElseInsertReqst( key, semReq, commHashTbl); + entry = giveEntryElseInsertReqst( (char *)key, semReq, commHashTbl); if( entry == NULL ) return; //was just inserted waitingReq = (SSRSemReq *)entry->content; @@ -90,36 +111,36 @@ //At this point, know have waiting request(s) -- either sends or recv //Note, can only have max of one receive waiting, and cannot have both // sends and receives waiting (they would have paired off) - // but can have multiple send_type requests waiting (from diff senders) + // but can have multiple sends from diff sending VPs, all same msg-type if( waitingReq->reqType == send_type ) { //waiting request is another send, so stack this up on list - semReq-> nextReqInHashEntry = waitingReq->nextReqInHashEntry; - waitingReq->nextReqInHashEntry = semReq; - PRINT2_DEBUG("linked requests: %d, %d | ", semReq, waitingReq ) - PRINT2_DEBUG("type: %d, %d\n", semReq->reqType, waitingReq->reqType) + // but first clone the sending request so it persists. + SSRSemReq *clonedReq = cloneReq( semReq ); + clonedReq-> nextReqInHashEntry = waitingReq->nextReqInHashEntry; + waitingReq->nextReqInHashEntry = clonedReq; + PRINT2_DEBUG("linked requests: %d, %d | ", clonedReq, waitingReq ) return; } else - { //waiting request is a receive, so pair it to this send - //first, remove the waiting receive request from the list in entry + { //waiting request is a receive, so it pairs to this send + //First, remove the waiting receive request from the entry entry->content = waitingReq->nextReqInHashEntry; + VMS__free( waitingReq ); //Don't use contents -- so free it + if( entry->content == NULL ) { //TODO: mod hash table to double-link, so can delete entry from // table without hashing the key and looking it up again - deleteEntryFromTable( entry->key, commHashTbl ); //frees entry too + deleteEntryFromTable( entry->key, commHashTbl ); //frees hashEntry } //attach msg that's in this send request to receiving procr - // when comes back from suspend, will have msg in semanticData - receivePr->semanticData = semReq->msg; + // when comes back from suspend will have msg in dataRetFromReq + receivePr->dataRetFromReq = semReq->msg; //bring both processors back from suspend - writePrivQ( sendPr, semEnv->readyVPQs[sendPr->coreAnimatedBy] ); - writePrivQ( receivePr, semEnv->readyVPQs[receivePr->coreAnimatedBy] ); + resumePr( sendPr, semEnv ); + resumePr( receivePr, semEnv ); - //don't need semReq anymore -- free it - SSR__free_semantic_request( waitingReq ); - SSR__free_semantic_request( semReq ); return; } } @@ -128,7 +149,7 @@ /*Looks like can make single handler for both sends.. */ //TODO: combine both send handlers into single handler -void +void inline handleSendFromTo( SSRSemReq *semReq, SSRSemEnv *semEnv) { VirtProcr *sendPr, *receivePr; int key[] = {0,0,0}; @@ -143,47 +164,44 @@ key[1] = (int)sendPr; //key[2] acts at the 0 that terminates the string - entry = giveEntryElseInsertReqst( key, semReq, commHashTbl); + entry = giveEntryElseInsertReqst( (char *)key, semReq, commHashTbl); if( entry == NULL ) return; //was just inserted waitingReq = (SSRSemReq *)entry->content; //At this point, know have waiting request(s) -- either sends or recv if( waitingReq->reqType == send_from_to ) - { printf("\n ERROR: shouldn't be two send from-tos waiting \n"); + { printf("\n ERROR: shouldn't be two send-from-tos waiting \n"); } else { //waiting request is a receive, so it completes pair with this send - - //remove the waiting receive request from the entry + //First, remove the waiting receive request from the entry entry->content = waitingReq->nextReqInHashEntry; + VMS__free( waitingReq ); //Don't use contents -- so free it + //can only be one waiting req for "from-to" semantics if( entry->content != NULL ) { printf("\nERROR in handleSendFromTo\n"); - printf("waitReq: %d | next req: %d\n", waitingReq, entry->content); } - deleteEntryFromTable( entry->key, commHashTbl ); //frees entry too + deleteEntryFromTable( entry->key, commHashTbl ); //frees HashEntry //attach msg that's in this send request to receiving procr - // when comes back from suspend, will have msg in semanticData - receivePr->semanticData = semReq->msg; + // when comes back from suspend, will have msg in dataRetFromReq + receivePr->dataRetFromReq = semReq->msg; //bring both processors back from suspend writePrivQ( sendPr, semEnv->readyVPQs[sendPr->coreAnimatedBy] ); writePrivQ( receivePr, semEnv->readyVPQs[receivePr->coreAnimatedBy] ); - //done with requests, so free them - SSR__free_semantic_request( waitingReq ); - SSR__free_semantic_request( semReq ); return; } } -//======================================================= - +//============================== Receives =========================== +// /*Removed this one for now, because forces either a search or going to a * two-level hash table, where one level the key is the receivePr, in the * other level, the key is the type. @@ -210,14 +228,14 @@ * If ever add receive_any, looking like this second option easier and even * less costly. */ -void +void inline handleReceiveAny( SSRSemReq *semReq, SSRSemEnv *semEnv) { } -void +void inline handleReceiveType( SSRSemReq *semReq, SSRSemEnv *semEnv) { VirtProcr *sendPr, *receivePr; int key[] = {0,0,0}; @@ -229,13 +247,13 @@ key[0] = (int)receivePr; key[1] = (int)(semReq->msgType); - //key[2] acts at the 0 that terminates the string + //key[2] acts as the 0 that terminates the string - entry = giveEntryElseInsertReqst( key, semReq, commHashTbl); + entry = giveEntryElseInsertReqst((char*)key, semReq, commHashTbl);//clones if( entry == NULL ) return; //was just inserted - waitingReq = (SSRSemReq *)entry->content; + waitingReq = (SSRSemReq *)entry->content; //previously cloned by insert //At this point, know have waiting request(s) -- should be send(s) if( waitingReq->reqType == send_type ) @@ -243,21 +261,20 @@ //first, remove the waiting send request from the list in entry entry->content = waitingReq->nextReqInHashEntry; if( entry->content == NULL ) - { deleteEntryFromTable( entry->key, commHashTbl ); //frees entry too + { deleteEntryFromTable( entry->key, commHashTbl ); //frees HashEntry } //attach msg that's in the send request to receiving procr - // when comes back from suspend, will have msg in semanticData - receivePr->semanticData = waitingReq->msg; + // when comes back from suspend, will have msg in dataReturnedFromReq + receivePr->dataRetFromReq = waitingReq->msg; //bring both processors back from suspend sendPr = waitingReq->sendPr; + VMS__free( waitingReq ); + writePrivQ( sendPr, semEnv->readyVPQs[sendPr->coreAnimatedBy] ); writePrivQ( receivePr, semEnv->readyVPQs[receivePr->coreAnimatedBy] ); - //done with requests, so free them - SSR__free_semantic_request( waitingReq ); - SSR__free_semantic_request( semReq ); return; } printf("\nLang Impl Error: Should never be two waiting receives!\n"); @@ -266,7 +283,7 @@ /* */ -void +void inline handleReceiveFromTo( SSRSemReq *semReq, SSRSemEnv *semEnv) { VirtProcr *sendPr, *receivePr; int key[] = {0,0,0}; @@ -281,7 +298,7 @@ key[1] = (int)sendPr; //key[2] acts at the 0 that terminates the string - entry = giveEntryElseInsertReqst( key, semReq, commHashTbl); + entry = giveEntryElseInsertReqst( (char *)key, semReq, commHashTbl); if( entry == NULL ) return; //was just inserted waitingReq = (SSRSemReq *)entry->content; @@ -296,17 +313,16 @@ deleteEntryFromTable( entry->key, commHashTbl ); //frees entry too //attach msg that's in the send request to receiving procr - // when comes back from suspend, will have msg in semanticData - receivePr->semanticData = waitingReq->msg; + // when comes back from suspend, will have msg in dataRetFromReq + receivePr->dataRetFromReq = waitingReq->msg; //bring both processors back from suspend sendPr = waitingReq->sendPr; + VMS__free( waitingReq ); + writePrivQ( sendPr, semEnv->readyVPQs[sendPr->coreAnimatedBy] ); writePrivQ( receivePr, semEnv->readyVPQs[receivePr->coreAnimatedBy] ); - //done with requests, so free them - SSR__free_semantic_request( waitingReq ); - SSR__free_semantic_request( semReq ); return; } printf("\nLang Impl Error: Should never be two waiting receives!\n"); @@ -315,16 +331,164 @@ //=============================================== -void +void inline handleTransferTo( SSRSemReq *semReq, SSRSemEnv *semEnv) { } -void +void inline handleTransferOut( SSRSemReq *semReq, SSRSemEnv *semEnv) { } +/* + */ +void inline +handleMalloc( SSRSemReq *semReq, VirtProcr *requestingPr, SSRSemEnv *semEnv ) + { void *ptr; + + ptr = VMS__malloc( semReq->sizeToMalloc ); + requestingPr->dataRetFromReq = ptr; + resumePr( requestingPr, semEnv ); + } + +/* + */ +void inline +handleFree( SSRSemReq *semReq, VirtProcr *requestingPr, SSRSemEnv *semEnv ) + { + VMS__free( semReq->ptrToFree ); + resumePr( requestingPr, semEnv ); + } + + +//=========================================================================== +// +/*Uses ID as index into array of flags. If flag already set, resumes from + * end-label. Else, sets flag and resumes normally. + */ +void inline +handleSingleton( SSRSemReq *semReq, VirtProcr *requestingPr, + SSRSemEnv *semEnv ) + { + if( semEnv->singletonHasBeenExecutedFlags[ semReq->singletonID ] ) + requestingPr->nextInstrPt = semReq->endJumpPt; + else + semEnv->singletonHasBeenExecutedFlags[ semReq->singletonID ] = TRUE; + + resumePr( requestingPr, semEnv ); + } + + +/*This executes the function in the masterVP, take the function + * pointer out of the request and call it, then resume the VP. + */ +void inline +handleAtomic( SSRSemReq *semReq, VirtProcr *requestingPr, SSRSemEnv *semEnv ) + { + semReq->fnToExecInMaster( semReq->dataForFn ); + resumePr( requestingPr, semEnv ); + } + +/*First, it looks at the VP's semantic data, to see the highest transactionID + * that VP + * already has entered. If the current ID is not larger, it throws an + * exception stating a bug in the code. + *Otherwise it puts the current ID + * there, and adds the ID to a linked list of IDs entered -- the list is + * used to check that exits are properly ordered. + *Next it is uses transactionID as index into an array of transaction + * structures. + *If the "VP_currently_executing" field is non-null, then put requesting VP + * into queue in the struct. (At some point a holder will request + * end-transaction, which will take this VP from the queue and resume it.) + *If NULL, then write requesting into the field and resume. + */ +void inline +handleTransStart( SSRSemReq *semReq, VirtProcr *requestingPr, + SSRSemEnv *semEnv ) + { SSRSemData *semData; + TransListElem *nextTransElem; + + //check ordering of entering transactions is correct + semData = requestingPr->semanticData; + if( semData->highestTransEntered > semReq->transID ) + { //throw VMS exception, which shuts down VMS. + VMS__throw_exception( "transID smaller than prev", requestingPr, NULL); + } + //add this trans ID to the list of transactions entered -- check when + // end a transaction + semData->highestTransEntered = semReq->transID; + nextTransElem = VMS__malloc( sizeof(TransListElem) ); + nextTransElem->transID = semReq->transID; + nextTransElem->nextTrans = semData->lastTransEntered; + semData->lastTransEntered = nextTransElem; + + //get the structure for this transaction ID + SSRTrans * + transStruc = &(semEnv->transactionStrucs[ semReq->transID ]); + + if( transStruc->VPCurrentlyExecuting == NULL ) + { + transStruc->VPCurrentlyExecuting = requestingPr; + resumePr( requestingPr, semEnv ); + } + else + { //note, might make future things cleaner if save request with VP and + // add this trans ID to the linked list when gets out of queue. + // but don't need for now, and lazy.. + writePrivQ( requestingPr, transStruc->waitingVPQ ); + } + } + + +/*Use the trans ID to get the transaction structure from the array. + *Look at VP_currently_executing to be sure it's same as requesting VP. + * If different, throw an exception, stating there's a bug in the code. + *Next, take the first element off the list of entered transactions. + * Check to be sure the ending transaction is the same ID as the next on + * the list. If not, incorrectly nested so throw an exception. + * + *Next, get from the queue in the structure. + *If it's empty, set VP_currently_executing field to NULL and resume + * requesting VP. + *If get somethine, set VP_currently_executing to the VP from the queue, then + * resume both. + */ +void inline +handleTransEnd(SSRSemReq *semReq, VirtProcr *requestingPr, SSRSemEnv *semEnv) + { SSRSemData *semData; + VirtProcr *waitingPr; + SSRTrans *transStruc; + TransListElem *lastTrans; + + transStruc = &(semEnv->transactionStrucs[ semReq->transID ]); + + //make sure transaction ended in same VP as started it. + if( transStruc->VPCurrentlyExecuting != requestingPr ) + { + VMS__throw_exception( "trans ended in diff VP", requestingPr, NULL ); + } + + //make sure nesting is correct -- last ID entered should == this ID + semData = requestingPr->semanticData; + lastTrans = semData->lastTransEntered; + if( lastTrans->transID != semReq->transID ) + { + VMS__throw_exception( "trans incorrectly nested", requestingPr, NULL ); + } + + semData->lastTransEntered = semData->lastTransEntered->nextTrans; + + + waitingPr = readPrivQ( transStruc->waitingVPQ ); + transStruc->VPCurrentlyExecuting = waitingPr; + + if( waitingPr != NULL ) + resumePr( waitingPr, semEnv ); + + resumePr( requestingPr, semEnv ); + } diff -r 6c6d7fbd7e25 -r bf57b83019e5 SSR_Request_Handlers.h --- a/SSR_Request_Handlers.h Thu Oct 14 17:06:28 2010 -0700 +++ b/SSR_Request_Handlers.h Tue Nov 02 16:57:45 2010 -0700 @@ -14,20 +14,34 @@ /*This header defines everything specific to the SSR semantic plug-in */ -void +void inline handleSendType( SSRSemReq *semReq, SSRSemEnv *semEnv); -void +void inline handleSendFromTo( SSRSemReq *semReq, SSRSemEnv *semEnv); -void +void inline handleReceiveAny( SSRSemReq *semReq, SSRSemEnv *semEnv); -void +void inline handleReceiveType( SSRSemReq *semReq, SSRSemEnv *semEnv); -void +void inline handleReceiveFromTo( SSRSemReq *semReq, SSRSemEnv *semEnv); -void +void inline handleTransferTo( SSRSemReq *semReq, SSRSemEnv *semEnv); -void +void inline handleTransferOut( SSRSemReq *semReq, SSRSemEnv *semEnv); +void inline +handleMalloc( SSRSemReq *semReq, VirtProcr *requestingPr, SSRSemEnv *semEnv); +void inline +handleFree( SSRSemReq *semReq, VirtProcr *requestingPr, SSRSemEnv *semEnv ); +void inline +handleTransEnd(SSRSemReq *semReq, VirtProcr *requestingPr, SSRSemEnv*semEnv); +void inline +handleTransStart( SSRSemReq *semReq, VirtProcr *requestingPr, + SSRSemEnv *semEnv ); +void inline +handleAtomic( SSRSemReq *semReq, VirtProcr *requestingPr, SSRSemEnv *semEnv); +void inline +handleSingleton( SSRSemReq *semReq, VirtProcr *requestingPr, + SSRSemEnv *semEnv ); #endif /* _SSR_REQ_H */ diff -r 6c6d7fbd7e25 -r bf57b83019e5 SSR_lib.c --- a/SSR_lib.c Thu Oct 14 17:06:28 2010 -0700 +++ b/SSR_lib.c Tue Nov 02 16:57:45 2010 -0700 @@ -94,7 +94,7 @@ { SSRSemEnv *semEnv; VirtProcr *seedPr; - #ifdef DEBUG + #ifdef SEQUENTIAL SSR__init_Seq(); //debug sequential exe #else SSR__init(); //normal multi-thd @@ -110,7 +110,7 @@ writePrivQ( seedPr, semEnv->readyVPQs[seedPr->coreAnimatedBy] ); semEnv->numVirtPr = 1; - #ifdef DEBUG + #ifdef SEQUENTIAL VMS__start_the_work_then_wait_until_done_Seq(); //debug sequential exe #else VMS__start_the_work_then_wait_until_done(); //normal multi-thd @@ -132,14 +132,19 @@ return NUM_SCHED_SLOTS * NUM_CORES; } +int32 +SSR__give_number_of_cores_to_schedule_onto() + { + return NUM_CORES; + } + /*For now, use TSC -- later, make these two macros with assembly that first * saves jump point, and second jumps back several times to get reliable time */ void inline SSR__start_primitive() - { int32 *saveAddr; - saveAddr = &(((SSRSemEnv *)(_VMSMasterEnv->semanticEnv))->primitiveStartTime); - saveLowTimeStampCountInto( (((SSRSemEnv *)(_VMSMasterEnv->semanticEnv))->primitiveStartTime) ); + { saveLowTimeStampCountInto( ((SSRSemEnv *)(_VMSMasterEnv->semanticEnv))-> + primitiveStartTime ); } /*Just quick and dirty for now -- make reliable later @@ -152,7 +157,7 @@ { int32 endTime, startTime; //TODO: fix by repeating time-measurement saveLowTimeStampCountInto( endTime ); - startTime = ((SSRSemEnv *)(_VMSMasterEnv->semanticEnv))->primitiveStartTime; + startTime =((SSRSemEnv*)(_VMSMasterEnv->semanticEnv))->primitiveStartTime; return (endTime - startTime); } @@ -161,6 +166,7 @@ /*Initializes all the data-structures for a SSR system -- but doesn't * start it running yet! * + *This runs in the main thread -- before VMS starts up * *This sets up the semantic layer over the VMS system * @@ -172,6 +178,7 @@ { VMS__init(); //masterEnv, a global var, now is partially set up by init_VMS + // after this, have VMS__malloc and VMS__free available SSR__init_Helper(); } @@ -189,7 +196,7 @@ SSR__init_Helper() { SSRSemEnv *semanticEnv; PrivQueueStruc **readyVPQs; - int coreIdx; + int coreIdx, i; //Hook up the semantic layer's plug-ins to the Master virt procr _VMSMasterEnv->requestHandler = &SSR__Request_Handler; @@ -197,14 +204,14 @@ //create the semantic layer's environment (all its data) and add to // the master environment - semanticEnv = malloc( sizeof( SSRSemEnv ) ); + semanticEnv = VMS__malloc( sizeof( SSRSemEnv ) ); _VMSMasterEnv->semanticEnv = semanticEnv; //create the ready queue, hash tables used for pairing send to receive // and so forth //TODO: add hash tables for pairing sends with receives, and // initialize the data ownership system - readyVPQs = malloc( NUM_CORES * sizeof(PrivQueueStruc *) ); + readyVPQs = VMS__malloc( NUM_CORES * sizeof(PrivQueueStruc *) ); for( coreIdx = 0; coreIdx < NUM_CORES; coreIdx++ ) { @@ -215,7 +222,19 @@ semanticEnv->nextCoreToGetNewPr = 0; - semanticEnv->commHashTbl = makeHashTable( 1<<16, NULL ); //start big + semanticEnv->commHashTbl = makeHashTable( 1<<16, &VMS__free );//start big + + //TODO: bug -- turn these arrays into dyn arrays to eliminate limit + //semanticEnv->singletonHasBeenExecutedFlags = makeDynArrayInfo( ); + //semanticEnv->transactionStrucs = makeDynArrayInfo( ); + for( i = 0; i < NUM_STRUCS_IN_SEM_ENV; i++ ) + { + semanticEnv->singletonHasBeenExecutedFlags[i] = FALSE; + semanticEnv->transactionStrucs[i].waitingVPQ = makePrivQ(); + } + + + } @@ -225,21 +244,22 @@ SSR__cleanup_after_shutdown() { SSRSemEnv *semanticEnv; int coreIdx; - + +/* It's all allocated inside VMS's big chunk -- that's about to be freed, so + * nothing to do here semanticEnv = _VMSMasterEnv->semanticEnv; -//TODO: double check all sem env locations freed - for( coreIdx = 0; coreIdx < NUM_CORES; coreIdx++ ) { - free( semanticEnv->readyVPQs[coreIdx]->startOfData ); - free( semanticEnv->readyVPQs[coreIdx] ); + VMS__free( semanticEnv->readyVPQs[coreIdx]->startOfData ); + VMS__free( semanticEnv->readyVPQs[coreIdx] ); } - free( semanticEnv->readyVPQs ); + VMS__free( semanticEnv->readyVPQs ); freeHashTable( semanticEnv->commHashTbl ); - free( _VMSMasterEnv->semanticEnv ); - VMS__cleanup_after_shutdown(); + VMS__free( _VMSMasterEnv->semanticEnv ); + */ + VMS__cleanup_at_end_of_shutdown(); } @@ -248,38 +268,76 @@ /* */ inline VirtProcr * -SSR__create_procr_with( VirtProcrFnPtr fnPtr, void *initData, - VirtProcr *creatingPr ) - { VirtProcr *newPr; +SSR__create_procr_with( VirtProcrFnPtr fnPtr, void *initData, + VirtProcr *creatingPr ) + { SSRSemReq reqData; - newPr = VMS__create_procr( fnPtr, initData ); + //the semantic request data is on the stack and disappears when this + // call returns -- it's guaranteed to remain in the VP's stack for as + // long as the VP is suspended. + reqData.reqType = 0; //know type because in a VMS create req + reqData.coreToScheduleOnto = -1; //means round-robin schedule + reqData.fnPtr = fnPtr; + reqData.initData = initData; + reqData.sendPr = creatingPr; - //After create, have to send request to plugin for any sem env - // modifications -- such as putting the new procr into the ready Q - //Need a processor to "animate" the creation -- it's one the register - // request is attached to, and one suspended in order to send req - // to plugin - VMS__send_req_to_register_new_procr( newPr, creatingPr ); + VMS__send_create_procr_req( &reqData, creatingPr ); - return newPr; + return creatingPr->dataRetFromReq; + } + +inline VirtProcr * +SSR__create_procr_with_affinity( VirtProcrFnPtr fnPtr, void *initData, + VirtProcr *creatingPr, int32 coreToScheduleOnto ) + { SSRSemReq reqData; + + //the semantic request data is on the stack and disappears when this + // call returns -- it's guaranteed to remain in the VP's stack for as + // long as the VP is suspended. + reqData.reqType = 0; //know type because in a VMS create req + reqData.coreToScheduleOnto = coreToScheduleOnto; + reqData.fnPtr = fnPtr; + reqData.initData = initData; + reqData.sendPr = creatingPr; + + VMS__send_create_procr_req( &reqData, creatingPr ); } inline void SSR__dissipate_procr( VirtProcr *procrToDissipate ) { - VMS__dissipate_procr( procrToDissipate ); + VMS__send_dissipate_req( procrToDissipate ); } //=========================================================================== void * -SSR__malloc_size_to( int numBytes, VirtProcr *ownerPr ) - { -//TODO: Put in the ownership system from DKU -- have it working, just adapt -// it to here - return malloc( numBytes ); +SSR__malloc_to( int32 sizeToMalloc, VirtProcr *owningPr ) + { SSRSemReq reqData; + + reqData.reqType = malloc_req; + reqData.sendPr = owningPr; + reqData.sizeToMalloc = sizeToMalloc; + + VMS__send_sem_request( &reqData, owningPr ); + + return owningPr->dataRetFromReq; + } + + +/*Sends request to Master, which does the work of freeing + */ +void +SSR__free( void *ptrToFree, VirtProcr *owningPr ) + { SSRSemReq reqData; + + reqData.reqType = free_req; + reqData.sendPr = owningPr; + reqData.ptrToFree = ptrToFree; + + VMS__send_sem_request( &reqData, owningPr ); } @@ -287,7 +345,9 @@ SSR__transfer_ownership_of_from_to( void *data, VirtProcr *oldOwnerPr, VirtProcr *newOwnerPr ) { - + //TODO: put in the ownership system that automatically frees when no + // owners of data left -- will need keeper for keeping data around when + // future created processors might need it but don't exist yet } @@ -333,22 +393,20 @@ void SSR__send_of_type_to( VirtProcr *sendPr, void *msg, const int type, VirtProcr *receivePr) - { SSRSemReq *reqData; + { SSRSemReq reqData; - reqData = malloc( sizeof(SSRSemReq) ); - reqData->receivePr = receivePr; - reqData->sendPr = sendPr; - reqData->reqType = send_type; - reqData->msgType = type; - reqData->msg = msg; - reqData->nextReqInHashEntry = NULL; + reqData.receivePr = receivePr; + reqData.sendPr = sendPr; + reqData.reqType = send_type; + reqData.msgType = type; + reqData.msg = msg; + reqData.nextReqInHashEntry = NULL; //On ownership -- remove inside the send and let ownership sit in limbo // as a potential in an entry in the hash table, when this receive msg // gets paired to a send, the ownership gets added to the receivePr -- // the next work-unit in the receivePr's trace will have ownership. - VMS__add_sem_request( reqData, sendPr ); - VMS__suspend_procr( sendPr ); //will suspend then resume and continue + VMS__send_sem_request( &reqData, sendPr ); //When come back from suspend, no longer own data reachable from msg //TODO: release ownership here @@ -356,27 +414,18 @@ void SSR__send_from_to( void *msg, VirtProcr *sendPr, VirtProcr *receivePr ) - { SSRSemReq *reqData; + { SSRSemReq reqData; //hash on the receiver, 'cause always know it, but sometimes want to // receive from anonymous sender - reqData = malloc( sizeof(SSRSemReq) ); - reqData->receivePr = receivePr; - reqData->sendPr = sendPr; - reqData->reqType = send_from_to; - reqData->msg = msg; - reqData->nextReqInHashEntry = NULL; + reqData.receivePr = receivePr; + reqData.sendPr = sendPr; + reqData.reqType = send_from_to; + reqData.msg = msg; + reqData.nextReqInHashEntry = NULL; - //On ownership -- remove inside the send and let ownership sit in limbo - // as a potential in an entry in the hash table, when this receive msg - // gets paired to a send, the ownership gets added to the receivePr -- - // the next work-unit in the receivePr's trace will have ownership. - VMS__add_sem_request( reqData, sendPr ); - VMS__suspend_procr( sendPr ); //will suspend then resume and continue - - //When come back from suspend, no longer own data reachable from msg - //TODO: release ownership here + VMS__send_sem_request( &reqData, sendPr ); } @@ -390,19 +439,17 @@ void * SSR__receive_type_to( const int type, VirtProcr *receivePr ) - { void *msg; - SSRSemReq *reqData; + { + SSRSemReq reqData; - reqData = malloc( sizeof(SSRSemReq) ); - reqData->receivePr = receivePr; - reqData->reqType = receive_type; - reqData->msgType = type; - reqData->nextReqInHashEntry = NULL; + reqData.receivePr = receivePr; + reqData.reqType = receive_type; + reqData.msgType = type; + reqData.nextReqInHashEntry = NULL; - VMS__add_sem_request( reqData, receivePr ); - VMS__suspend_procr( receivePr ); - msg = receivePr->semanticData; - return msg; + VMS__send_sem_request( &reqData, receivePr ); + + return receivePr->dataRetFromReq; } @@ -415,39 +462,110 @@ */ void * SSR__receive_from_to( VirtProcr *sendPr, VirtProcr *receivePr ) - { SSRSemReq *reqData; + { SSRSemReq reqData; //hash on the receiver, 'cause always know it, but sometimes want to // receive from anonymous sender - reqData = malloc( sizeof(SSRSemReq) ); - reqData->receivePr = receivePr; - reqData->sendPr = sendPr; - reqData->reqType = receive_from_to; - reqData->nextReqInHashEntry = NULL; + reqData.receivePr = receivePr; + reqData.sendPr = sendPr; + reqData.reqType = receive_from_to; + reqData.nextReqInHashEntry = NULL; - //On ownership -- remove inside the send after receive successful. - // Below, add ownership when come back from suspend - //Reason: Thinking of impl ownership mech such that it automatically - // frees any data has no owners -- so have to add receiver before - // remove sender - VMS__add_sem_request( reqData, receivePr ); - //TODO: add ownership of locs reachable from msg inside reqst handler - VMS__suspend_procr( receivePr ); //will suspend then resume and continue + VMS__send_sem_request( &reqData, receivePr ); - //When come back from suspend, the msg data is in receivePr->semData - return receivePr->semanticData; + return receivePr->dataRetFromReq; } //=========================================================================== -/*Just thin wrapper for now -- semantic request is still a simple thing - * (July 3, 2010) +/*Uses ID as index into array of flags. If flag already set, resumes from + * end-label. Else, sets flag and resumes normally. */ -inline void -SSR__free_semantic_request( SSRSemReq *semReq ) +void +SSR__start_singleton( int32 singletonID, void *endSingletonLabelAddr, + VirtProcr *animPr ) { - free( semReq ); + SSRSemReq reqData; + + // + reqData.reqType = singleton; + reqData.singletonID = singletonID; + reqData.endJumpPt = endSingletonLabelAddr; + + VMS__send_sem_request( &reqData, animPr ); } +/*This executes the function in the masterVP, so it executes in isolation + * from any other copies -- only one copy of the function can ever execute + * at a time. + * + *It suspends to the master, and the request handler takes the function + * pointer out of the request and calls it, then resumes the VP. + *Only very short functions should be called this way -- for longer-running + * isolation, use transaction-start and transaction-end, which run the code + * between as work-code. + */ +void +SSR__animate_short_fn_in_isolation( PtrToAtomicFn ptrToFnToExecInMaster, + void *data, VirtProcr *animPr ) + { + SSRSemReq reqData; + + // + reqData.reqType = atomic; + reqData.fnToExecInMaster = ptrToFnToExecInMaster; + reqData.dataForFn = data; + + VMS__send_sem_request( &reqData, animPr ); + } + + +/*This suspends to the master. + *First, it looks at the VP's data, to see the highest transactionID that VP + * already has entered. If the current ID is not larger, it throws an + * exception stating a bug in the code. Otherwise it puts the current ID + * there, and adds the ID to a linked list of IDs entered -- the list is + * used to check that exits are properly ordered. + *Next it is uses transactionID as index into an array of transaction + * structures. + *If the "VP_currently_executing" field is non-null, then put requesting VP + * into queue in the struct. (At some point a holder will request + * end-transaction, which will take this VP from the queue and resume it.) + *If NULL, then write requesting into the field and resume. + */ +void +SSR__start_transaction( int32 transactionID, VirtProcr *animPr ) + { + SSRSemReq reqData; + + // + reqData.sendPr = animPr; + reqData.reqType = trans_start; + reqData.transID = transactionID; + + VMS__send_sem_request( &reqData, animPr ); + } + +/*This suspends to the master, then uses transactionID as index into an + * array of transaction structures. + *It looks at VP_currently_executing to be sure it's same as requesting VP. + * If different, throws an exception, stating there's a bug in the code. + *Next it looks at the queue in the structure. + *If it's empty, it sets VP_currently_executing field to NULL and resumes. + *If something in, gets it, sets VP_currently_executing to that VP, then + * resumes both. + */ +void +SSR__end_transaction( int32 transactionID, VirtProcr *animPr ) + { + SSRSemReq reqData; + + // + reqData.sendPr = animPr; + reqData.reqType = trans_end; + reqData.transID = transactionID; + + VMS__send_sem_request( &reqData, animPr ); + } \ No newline at end of file