4 #include "structdefs.h"
6 #include "checkpoint.h"
8 #include "SimpleHash.h"
9 #include "GenericHashtable.h"
10 #include <sys/select.h>
11 #include <sys/types.h>
20 #include <raw_compiler_defs.h>
21 //#include <libints.h>
22 #elif defined THREADSIMULATE
23 // use POSIX message queue
24 // for each core, its message queue named as
30 extern int injectfailures;
31 extern float failurechance;
37 #define TOTALCORE raw_get_num_tiles()
41 #include "instrument.h"
44 struct genhashtable * activetasks;
45 struct genhashtable * failedtasks;
46 struct taskparamdescriptor * currtpd;
48 struct RuntimeHash * forward;
49 struct RuntimeHash * reverse;
52 int corestatus[NUMCORES]; // records status of each core
55 int numsendobjs[NUMCORES]; // records how many objects a core has sent out
56 int numreceiveobjs[NUMCORES]; // records how many objects a core has received
58 struct RuntimeHash locktable;
59 static struct RuntimeHash* locktbl = &locktable;
60 void * curr_heapbase=0;
61 void * curr_heaptop=0;
63 int self_numreceiveobjs;
70 struct Queue objqueue;
75 void calCoords(int core_num, int* coordY, int* coordX);
76 #elif defined THREADSIMULATE
77 static struct RuntimeHash* locktbl;
85 struct thread_data thread_data_array[NUMCORES];
87 static pthread_key_t key;
88 static pthread_rwlock_t rwlock_tbl;
89 static pthread_rwlock_t rwlock_init;
94 bool transStallMsg(int targetcore);
95 void transTerminateMsg(int targetcore);
97 bool getreadlock(void* ptr);
98 void releasereadlock(void* ptr);
100 bool getreadlock_I(void* ptr);
101 void releasereadlock_I(void* ptr);
103 bool getwritelock(void* ptr);
104 void releasewritelock(void* ptr);
109 int main(int argc, char **argv) {
115 bool sendStall = false;
117 bool tocontinue = false;
118 struct QueueItem * objitem = NULL;
119 struct transObjInfo * objInfo = NULL;
121 bool allStall = true;
125 raw_test_pass(0xee01);
127 corenum = raw_get_abs_pos_x() + 4 * raw_get_abs_pos_y();
129 // initialize the arrays
130 if(STARTUPCORE == corenum) {
131 // startup core to initialize corestatus[]
132 for(i = 0; i < NUMCORES; ++i) {
134 numsendobjs[i] = 0; // assume all variables in RAW are local variables! MAY BE WRONG!!!
135 numreceiveobjs[i] = 0;
138 self_numsendobjs = 0;
139 self_numreceiveobjs = 0;
140 for(i = 0; i < 30; ++i) {
148 raw_test_pass(0xee02);
151 // create the lock table, lockresult table and obj queue
153 locktable.bucket = (struct RuntimeNode **) RUNMALLOC_I(sizeof(struct RuntimeNode *)*20);
154 /* Set allocation blocks*/
155 locktable.listhead=NULL;
156 locktable.listtail=NULL;
158 locktable.numelements = 0;
165 objqueue.head = NULL;
166 objqueue.tail = NULL;
168 raw_test_pass(0xee03);
172 if (corenum < NUMCORES) {
175 //setup_interrupts();
176 //start_gdn_avail_ints(recvMsg);
177 raw_user_interrupts_on();
179 raw_test_pass(0xee04);
184 #elif defined THREADSIMULATE
188 pthread_t threads[NUMCORES];
191 // initialize three arrays and msg queue array
192 char * pathhead = "/msgqueue_";
193 int targetlen = strlen(pathhead);
194 for(i = 0; i < NUMCORES; ++i) {
197 numreceiveobjs[i] = 0;
202 corenumstr[0] = i + '0';
203 corenumstr[1] = '\0';
206 corenumstr[1] = i %10 + '0';
207 corenumstr[0] = (i / 10) + '0';
208 corenumstr[2] = '\0';
211 printf("Error: i >= 100\n");
215 char path[targetlen + sourcelen + 1];
216 strcpy(path, pathhead);
217 strncat(path, corenumstr, sourcelen);
218 int oflags = O_RDONLY|O_CREAT|O_NONBLOCK;
219 int omodes = S_IRWXU|S_IRWXG|S_IRWXO;
221 mqd[i]= mq_open(path, oflags, omodes, NULL);
223 printf("[Main] mq_open %s fails: %d, error: %s\n", path, mqd[i], strerror(errno));
226 printf("[Main] mq_open %s returns: %d\n", path, mqd[i]);
231 pthread_key_create(&key, NULL);
233 // create the lock table and initialize its mutex
234 locktbl = allocateRuntimeHash(20);
235 int rc_locktbl = pthread_rwlock_init(&rwlock_tbl, NULL);
236 printf("[Main] initialize the rwlock for lock table: %d error: \n", rc_locktbl, strerror(rc_locktbl));
238 for(i = 0; i < NUMCORES; ++i) {
239 thread_data_array[i].corenum = i;
240 thread_data_array[i].argc = argc;
241 thread_data_array[i].argv = argv;
242 thread_data_array[i].numsendobjs = 0;
243 thread_data_array[i].numreceiveobjs = 0;
244 printf("[main] creating thread %d\n", i);
245 rc[i] = pthread_create(&threads[i], NULL, run, (void *)&thread_data_array[i]);
247 printf("[main] ERROR; return code from pthread_create() is %d\n", rc[i]);
253 //pthread_exit(NULL);
257 void run(void* arg) {
258 struct thread_data * my_tdata = (struct thread_data *)arg;
259 pthread_setspecific(key, (void *)my_tdata->corenum);
260 int argc = my_tdata->argc;
261 char** argv = my_tdata->argv;
262 printf("[run, %d] Thread %d runs: %x\n", my_tdata->corenum, my_tdata->corenum, (int)pthread_self());
268 GC_init(); // Initialize the garbage collector
276 initializeexithandler();
278 raw_test_pass(0xee05);
280 /* Create table for failed tasks */
282 if(corenum > NUMCORES - 1) {
290 raw_test_pass(0xee06);
293 /*failedtasks=genallocatehashtable((unsigned int (*)(void *)) &hashCodetpd,
294 (int (*)(void *,void *)) &comparetpd);*/
297 raw_test_pass(0xee07);
299 /* Create queue of active tasks */
300 activetasks=genallocatehashtable((unsigned int (*)(void *)) &hashCodetpd,
301 (int (*)(void *,void *)) &comparetpd);
303 raw_test_pass(0xee08);
306 /* Process task information */
309 raw_test_pass(0xee09);
312 if(STARTUPCORE == corenum) {
313 /* Create startup object */
314 createstartupobject(argc, argv);
317 raw_test_pass(0xee0a);
322 raw_test_pass(0xee0b);
327 while(receiveObject() != -1) {
331 // check if there are new active tasks can be executed
335 while(receiveObject() != -1) {
340 raw_test_pass(0xee0c);
343 // check if there are some pending objects, if yes, enqueue them and executetasks again
346 raw_test_pass(0xee0d);
348 while(!isEmpty(&objqueue)) {
351 raw_user_interrupts_off();
354 raw_test_pass(0xeee1);
358 objitem = getTail(&objqueue);
359 //obj = objitem->objectptr;
360 objInfo = (struct transObjInfo *)objitem->objectptr;
361 obj = objInfo->objptr;
363 raw_test_pass_reg((int)obj);
365 // grab lock and flush the obj
372 raw_test_pass_reg(grount);
384 raw_invalidate_cache_range(obj, classsize[((struct ___Object___ *)obj)->type]);
386 /*for(k = 0; k < classsize[((struct ___Object___ *)obj)->type]; ++k) {
387 invalidateAddr(obj + k);
389 // enqueue the object
390 for(k = 0; k < objInfo->length; ++k) {
391 int taskindex = objInfo->queues[2 * k];
392 int paramindex = objInfo->queues[2 * k + 1];
393 struct parameterwrapper ** queues = &(paramqueues[corenum][taskindex][paramindex]);
395 raw_test_pass_reg(taskindex);
396 raw_test_pass_reg(paramindex);
398 enqueueObject_I(obj, queues, 1);
400 removeItem(&objqueue, objitem);
401 releasereadlock_I(obj);
402 RUNFREE(objInfo->queues);
404 /*enqueueObject_I(obj, NULL, 0);
405 removeItem(&objqueue, objitem);
406 releasereadlock_I(obj);*/
409 // put it at the end of the queue
410 // and try to execute active tasks already enqueued first
411 removeItem(&objqueue, objitem);
412 addNewItem_I(&objqueue, objInfo);
416 raw_user_interrupts_on();
419 raw_test_pass(0xee0e);
423 raw_test_pass(0xee0f);
428 if(STARTUPCORE == corenum) {
431 raw_test_pass(0xee10);
436 raw_user_interrupts_off();
438 corestatus[corenum] = 0;
439 numsendobjs[corenum] = self_numsendobjs;
440 numreceiveobjs[corenum] = self_numreceiveobjs;
441 // check the status of all cores
444 raw_test_pass_reg(NUMCORES);
446 for(i = 0; i < NUMCORES; ++i) {
448 raw_test_pass(0xe000 + corestatus[i]);
450 if(corestatus[i] != 0) {
456 // check if the sum of send objs and receive obj are the same
458 // no->go on executing
460 for(i = 0; i < NUMCORES; ++i) {
461 sumsendobj += numsendobjs[i];
463 raw_test_pass(0xf000 + numsendobjs[i]);
466 for(i = 0; i < NUMCORES; ++i) {
467 sumsendobj -= numreceiveobjs[i];
469 raw_test_pass(0xf000 + numreceiveobjs[i]);
472 if(0 == sumsendobj) {
475 raw_test_pass(0xee11);
477 raw_test_pass(raw_get_cycle());
478 raw_test_done(1); // All done.
482 raw_user_interrupts_on();
487 raw_test_pass(0xee12);
490 // wait for some time
493 raw_test_pass(0xee13);
498 raw_test_pass(0xee14);
501 // send StallMsg to startup core
503 raw_test_pass(0xee15);
505 sendStall = transStallMsg(STARTUPCORE);
511 raw_test_pass(0xee16);
518 #elif defined THREADSIMULATE
519 /* Start executing the tasks */
523 // check if there are new objects coming
524 bool sendStall = false;
526 int numofcore = pthread_getspecific(key);
528 switch(receiveObject()) {
530 printf("[run, %d] receive an object\n", numofcore);
532 // received an object
533 // check if there are new active tasks can be executed
538 //printf("[run, %d] no msg\n", numofcore);
540 if(STARTUPCORE == numofcore) {
541 corestatus[numofcore] = 0;
542 // check the status of all cores
543 bool allStall = true;
544 for(i = 0; i < NUMCORES; ++i) {
545 if(corestatus[i] != 0) {
551 // check if the sum of send objs and receive obj are the same
553 // no->go on executing
555 for(i = 0; i < NUMCORES; ++i) {
556 sumsendobj += numsendobjs[i];
558 for(i = 0; i < NUMCORES; ++i) {
559 sumsendobj -= numreceiveobjs[i];
561 if(0 == sumsendobj) {
565 int rc_tbl = pthread_rwlock_wrlock(&rwlock_tbl);
566 printf("[run, %d] getting the write lock for locktbl: %d error: \n", numofcore, rc_tbl, strerror(rc_tbl));
567 struct RuntimeIterator* it_lock = RuntimeHashcreateiterator(locktbl);
568 while(0 != RunhasNext(it_lock)) {
569 int key = Runkey(it_lock);
570 pthread_rwlock_t* rwlock_obj = (pthread_rwlock_t*)Runnext(it_lock);
571 int rc_des = pthread_rwlock_destroy(rwlock_obj);
572 printf("[run, %d] destroy the rwlock for object: %d error: \n", numofcore, key, strerror(rc_des));
575 freeRuntimeHash(locktbl);
579 // destroy all message queues
580 char * pathhead = "/msgqueue_";
581 int targetlen = strlen(pathhead);
582 for(i = 0; i < NUMCORES; ++i) {
586 corenumstr[0] = i + '0';
587 corenumstr[1] = '\0';
590 corenumstr[1] = i %10 + '0';
591 corenumstr[0] = (i / 10) + '0';
592 corenumstr[2] = '\0';
595 printf("Error: i >= 100\n");
599 char path[targetlen + sourcelen + 1];
600 strcpy(path, pathhead);
601 strncat(path, corenumstr, sourcelen);
605 printf("[run, %d] terminate!\n", numofcore);
612 // send StallMsg to startup core
613 sendStall = transStallMsg(STARTUPCORE);
619 printf("[run, %d] receive a stall msg\n", numofcore);
620 // receive a Stall Msg, do nothing
621 assert(STARTUPCORE == numofcore); // only startup core can receive such msg
626 printf("[run, %d] receive a terminate msg\n", numofcore);
627 // receive a terminate Msg
628 assert(STARTUPCORE != corenum); // only non-startup core can receive such msg
629 mq_close(mqd[corenum]);
635 printf("[run, %d] Error: invalid message type.\n", numofcore);
645 void createstartupobject(int argc, char ** argv) {
648 /* Allocate startup object */
650 struct ___StartupObject___ *startupobject=(struct ___StartupObject___*) allocate_new(NULL, STARTUPTYPE);
651 struct ArrayObject * stringarray=allocate_newarray(NULL, STRINGARRAYTYPE, argc-1);
653 struct ___StartupObject___ *startupobject=(struct ___StartupObject___*) allocate_new(STARTUPTYPE);
654 struct ArrayObject * stringarray=allocate_newarray(STRINGARRAYTYPE, argc-1);
656 /* Build array of strings */
657 startupobject->___parameters___=stringarray;
658 for(i=1;i<argc;i++) {
659 int length=strlen(argv[i]);
661 struct ___String___ *newstring=NewString(NULL, argv[i],length);
663 struct ___String___ *newstring=NewString(argv[i],length);
665 ((void **)(((char *)& stringarray->___length___)+sizeof(int)))[i-1]=newstring;
668 startupobject->isolate = 1;
669 startupobject->version = 0;
671 /* Set initialized flag for startup object */
672 flagorandinit(startupobject,1,0xFFFFFFFF);
673 enqueueObject(startupobject, NULL, 0);
676 raw_flush_entire_cache();
680 int hashCodetpd(struct taskparamdescriptor *ftd) {
681 int hash=(int)ftd->task;
683 for(i=0;i<ftd->numParameters;i++){
684 hash^=(int)ftd->parameterArray[i];
689 int comparetpd(struct taskparamdescriptor *ftd1, struct taskparamdescriptor *ftd2) {
691 if (ftd1->task!=ftd2->task)
693 for(i=0;i<ftd1->numParameters;i++)
694 if(ftd1->parameterArray[i]!=ftd2->parameterArray[i])
699 /* This function sets a tag. */
701 void tagset(void *ptr, struct ___Object___ * obj, struct ___TagDescriptor___ * tagd) {
703 void tagset(struct ___Object___ * obj, struct ___TagDescriptor___ * tagd) {
705 struct ArrayObject * ao=NULL;
706 struct ___Object___ * tagptr=obj->___tags___;
708 raw_test_pass(0xebb0);
712 raw_test_pass(0xebb1);
714 obj->___tags___=(struct ___Object___ *)tagd;
716 /* Have to check if it is already set */
717 if (tagptr->type==TAGTYPE) {
718 struct ___TagDescriptor___ * td=(struct ___TagDescriptor___ *) tagptr;
720 raw_test_pass(0xebb2);
724 raw_test_pass(0xebb3);
729 int ptrarray[]={2, (int) ptr, (int) obj, (int)tagd};
730 struct ArrayObject * ao=allocate_newarray(&ptrarray,TAGARRAYTYPE,TAGARRAYINTERVAL);
731 obj=(struct ___Object___ *)ptrarray[2];
732 tagd=(struct ___TagDescriptor___ *)ptrarray[3];
733 td=(struct ___TagDescriptor___ *) obj->___tags___;
736 raw_test_pass(0xebb4);
738 ao=allocate_newarray(TAGARRAYTYPE,TAGARRAYINTERVAL);
741 raw_test_pass(0xebb5);
743 ARRAYSET(ao, struct ___TagDescriptor___ *, 0, td);
744 ARRAYSET(ao, struct ___TagDescriptor___ *, 1, tagd);
745 obj->___tags___=(struct ___Object___ *) ao;
746 ao->___cachedCode___=2;
748 raw_test_pass(0xebb6);
753 struct ArrayObject *ao=(struct ArrayObject *) tagptr;
755 raw_test_pass(0xebb7);
757 for(i=0;i<ao->___cachedCode___;i++) {
758 struct ___TagDescriptor___ * td=ARRAYGET(ao, struct ___TagDescriptor___*, i);
760 raw_test_pass(0xebb8);
764 raw_test_pass(0xebb9);
769 if (ao->___cachedCode___<ao->___length___) {
771 raw_test_pass(0xebba);
773 ARRAYSET(ao, struct ___TagDescriptor___ *, ao->___cachedCode___, tagd);
774 ao->___cachedCode___++;
776 raw_test_pass(0xebbb);
780 int ptrarray[]={2,(int) ptr, (int) obj, (int) tagd};
781 struct ArrayObject * aonew=allocate_newarray(&ptrarray,TAGARRAYTYPE,TAGARRAYINTERVAL+ao->___length___);
782 obj=(struct ___Object___ *)ptrarray[2];
783 tagd=(struct ___TagDescriptor___ *) ptrarray[3];
784 ao=(struct ArrayObject *)obj->___tags___;
786 struct ArrayObject * aonew=allocate_newarray(TAGARRAYTYPE,TAGARRAYINTERVAL+ao->___length___);
789 raw_test_pass(0xebbc);
791 aonew->___cachedCode___=ao->___length___+1;
792 for(i=0;i<ao->___length___;i++) {
794 raw_test_pass(0xebbd);
796 ARRAYSET(aonew, struct ___TagDescriptor___*, i, ARRAYGET(ao, struct ___TagDescriptor___*, i));
799 raw_test_pass(0xebbe);
801 ARRAYSET(aonew, struct ___TagDescriptor___ *, ao->___length___, tagd);
803 raw_test_pass(0xebbf);
810 struct ___Object___ * tagset=tagd->flagptr;
812 raw_test_pass(0xb008);
816 raw_test_pass(0xb009);
819 } else if (tagset->type!=OBJECTARRAYTYPE) {
821 int ptrarray[]={2, (int) ptr, (int) obj, (int)tagd};
822 struct ArrayObject * ao=allocate_newarray(&ptrarray,OBJECTARRAYTYPE,OBJECTARRAYINTERVAL);
823 obj=(struct ___Object___ *)ptrarray[2];
824 tagd=(struct ___TagDescriptor___ *)ptrarray[3];
826 struct ArrayObject * ao=allocate_newarray(OBJECTARRAYTYPE,OBJECTARRAYINTERVAL);
828 ARRAYSET(ao, struct ___Object___ *, 0, tagd->flagptr);
829 ARRAYSET(ao, struct ___Object___ *, 1, obj);
830 ao->___cachedCode___=2;
831 tagd->flagptr=(struct ___Object___ *)ao;
833 raw_test_pass(0xb00a);
836 struct ArrayObject *ao=(struct ArrayObject *) tagset;
837 if (ao->___cachedCode___<ao->___length___) {
839 raw_test_pass(0xb00b);
841 ARRAYSET(ao, struct ___Object___*, ao->___cachedCode___++, obj);
845 int ptrarray[]={2, (int) ptr, (int) obj, (int)tagd};
846 struct ArrayObject * aonew=allocate_newarray(&ptrarray,OBJECTARRAYTYPE,OBJECTARRAYINTERVAL+ao->___length___);
847 obj=(struct ___Object___ *)ptrarray[2];
848 tagd=(struct ___TagDescriptor___ *)ptrarray[3];
849 ao=(struct ArrayObject *)tagd->flagptr;
851 struct ArrayObject * aonew=allocate_newarray(OBJECTARRAYTYPE,OBJECTARRAYINTERVAL);
853 aonew->___cachedCode___=ao->___cachedCode___+1;
854 for(i=0;i<ao->___length___;i++) {
855 ARRAYSET(aonew, struct ___Object___*, i, ARRAYGET(ao, struct ___Object___*, i));
857 ARRAYSET(aonew, struct ___Object___ *, ao->___cachedCode___, obj);
858 tagd->flagptr=(struct ___Object___ *) aonew;
860 raw_test_pass(0xb00c);
867 /* This function clears a tag. */
869 void tagclear(void *ptr, struct ___Object___ * obj, struct ___TagDescriptor___ * tagd) {
871 void tagclear(struct ___Object___ * obj, struct ___TagDescriptor___ * tagd) {
873 /* We'll assume that tag is alway there.
874 Need to statically check for this of course. */
875 struct ___Object___ * tagptr=obj->___tags___;
877 if (tagptr->type==TAGTYPE) {
878 if ((struct ___TagDescriptor___ *)tagptr==tagd)
879 obj->___tags___=NULL;
882 printf("ERROR 1 in tagclear\n");
886 struct ArrayObject *ao=(struct ArrayObject *) tagptr;
888 for(i=0;i<ao->___cachedCode___;i++) {
889 struct ___TagDescriptor___ * td=ARRAYGET(ao, struct ___TagDescriptor___ *, i);
891 ao->___cachedCode___--;
892 if (i<ao->___cachedCode___)
893 ARRAYSET(ao, struct ___TagDescriptor___ *, i, ARRAYGET(ao, struct ___TagDescriptor___ *, ao->___cachedCode___));
894 ARRAYSET(ao, struct ___TagDescriptor___ *, ao->___cachedCode___, NULL);
895 if (ao->___cachedCode___==0)
896 obj->___tags___=NULL;
901 printf("ERROR 2 in tagclear\n");
907 struct ___Object___ *tagset=tagd->flagptr;
908 if (tagset->type!=OBJECTARRAYTYPE) {
913 printf("ERROR 3 in tagclear\n");
917 struct ArrayObject *ao=(struct ArrayObject *) tagset;
919 for(i=0;i<ao->___cachedCode___;i++) {
920 struct ___Object___ * tobj=ARRAYGET(ao, struct ___Object___ *, i);
922 ao->___cachedCode___--;
923 if (i<ao->___cachedCode___)
924 ARRAYSET(ao, struct ___Object___ *, i, ARRAYGET(ao, struct ___Object___ *, ao->___cachedCode___));
925 ARRAYSET(ao, struct ___Object___ *, ao->___cachedCode___, NULL);
926 if (ao->___cachedCode___==0)
932 printf("ERROR 4 in tagclear\n");
940 /* This function allocates a new tag. */
942 struct ___TagDescriptor___ * allocate_tag(void *ptr, int index) {
943 struct ___TagDescriptor___ * v=(struct ___TagDescriptor___ *) mygcmalloc((struct garbagelist *) ptr, classsize[TAGTYPE]);
945 struct ___TagDescriptor___ * allocate_tag(int index) {
946 struct ___TagDescriptor___ * v=FREEMALLOC(classsize[TAGTYPE]);
955 /* This function updates the flag for object ptr. It or's the flag
956 with the or mask and and's it with the andmask. */
958 void flagbody(struct ___Object___ *ptr, int flag, struct parameterwrapper ** queues, int length, bool isnew);
960 int flagcomp(const int *val1, const int *val2) {
961 return (*val1)-(*val2);
964 void flagorand(void * ptr, int ormask, int andmask, struct parameterwrapper ** queues, int length) {
966 int oldflag=((int *)ptr)[1];
967 int flag=ormask|oldflag;
970 raw_test_pass_reg((int)ptr);
971 raw_test_pass(0xaa000000 + oldflag);
972 raw_test_pass(0xaa000000 + flag);
974 flagbody(ptr, flag, queues, length, false);
978 bool intflagorand(void * ptr, int ormask, int andmask) {
980 int oldflag=((int *)ptr)[1];
981 int flag=ormask|oldflag;
983 if (flag==oldflag) /* Don't do anything */
986 flagbody(ptr, flag, NULL, 0, false);
992 void flagorandinit(void * ptr, int ormask, int andmask) {
993 int oldflag=((int *)ptr)[1];
994 int flag=ormask|oldflag;
997 raw_test_pass(0xaa100000 + oldflag);
998 raw_test_pass(0xaa100000 + flag);
1000 flagbody(ptr,flag,NULL,0,true);
1003 void flagbody(struct ___Object___ *ptr, int flag, struct parameterwrapper ** vqueues, int vlength, bool isnew) {
1004 struct parameterwrapper * flagptr = NULL;
1006 struct parameterwrapper ** queues = vqueues;
1007 int length = vlength;
1009 int UNUSED, UNUSED2;
1010 int * enterflags = NULL;
1011 if((!isnew) && (queues == NULL)) {
1012 #ifdef THREADSIMULATE
1013 int numofcore = pthread_getspecific(key);
1014 queues = objectqueues[numofcore][ptr->type];
1015 length = numqueues[numofcore][ptr->type];
1018 if(corenum < NUMCORES) {
1020 queues = objectqueues[corenum][ptr->type];
1021 length = numqueues[corenum][ptr->type];
1031 raw_test_pass(0xbb000000 + ptr->flag);
1034 /*Remove object from all queues */
1035 for(i = 0; i < length; ++i) {
1036 flagptr = queues[i];
1037 ObjectHashget(flagptr->objectset, (int) ptr, (int *) &next, (int *) &enterflags, &UNUSED, &UNUSED2);
1038 ObjectHashremove(flagptr->objectset, (int)ptr);
1039 if (enterflags!=NULL)
1040 RUNFREE(enterflags);
1044 void enqueueObject(void * vptr, struct parameterwrapper ** vqueues, int vlength) {
1045 struct ___Object___ *ptr = (struct ___Object___ *)vptr;
1048 struct QueueItem *tmpptr;
1049 struct parameterwrapper * parameter=NULL;
1052 struct parameterwrapper * prevptr=NULL;
1053 struct ___Object___ *tagptr=NULL;
1054 struct parameterwrapper ** queues = vqueues;
1055 int length = vlength;
1057 if(corenum > NUMCORES - 1) {
1061 if(queues == NULL) {
1062 #ifdef THREADSIMULATE
1063 int numofcore = pthread_getspecific(key);
1064 queues = objectqueues[numofcore][ptr->type];
1065 length = numqueues[numofcore][ptr->type];
1067 queues = objectqueues[corenum][ptr->type];
1068 length = numqueues[corenum][ptr->type];
1071 tagptr=ptr->___tags___;
1073 /* Outer loop iterates through all parameter queues an object of
1074 this type could be in. */
1075 for(j = 0; j < length; ++j) {
1076 parameter = queues[j];
1078 if (parameter->numbertags>0) {
1080 goto nextloop;//that means the object has no tag but that param needs tag
1081 else if(tagptr->type==TAGTYPE) {//one tag
1082 struct ___TagDescriptor___ * tag=(struct ___TagDescriptor___*) tagptr;
1083 for(i=0;i<parameter->numbertags;i++) {
1084 //slotid is parameter->tagarray[2*i];
1085 int tagid=parameter->tagarray[2*i+1];
1086 if (tagid!=tagptr->flag)
1087 goto nextloop; /*We don't have this tag */
1089 } else {//multiple tags
1090 struct ArrayObject * ao=(struct ArrayObject *) tagptr;
1091 for(i=0;i<parameter->numbertags;i++) {
1092 //slotid is parameter->tagarray[2*i];
1093 int tagid=parameter->tagarray[2*i+1];
1095 for(j=0;j<ao->___cachedCode___;j++) {
1096 if (tagid==ARRAYGET(ao, struct ___TagDescriptor___*, j)->flag)
1107 for(i=0;i<parameter->numberofterms;i++) {
1108 int andmask=parameter->intarray[i*2];
1109 int checkmask=parameter->intarray[i*2+1];
1110 if ((ptr->flag&andmask)==checkmask) {
1112 raw_test_pass(0xcc000000 + andmask);
1113 raw_test_pass_reg((int)ptr);
1114 raw_test_pass(0xcc000000 + ptr->flag);
1115 raw_test_pass(0xcc000000 + checkmask);
1117 enqueuetasks(parameter, prevptr, ptr, NULL, 0);
1129 void enqueueObject_I(void * vptr, struct parameterwrapper ** vqueues, int vlength) {
1130 struct ___Object___ *ptr = (struct ___Object___ *)vptr;
1133 struct QueueItem *tmpptr;
1134 struct parameterwrapper * parameter=NULL;
1137 struct parameterwrapper * prevptr=NULL;
1138 struct ___Object___ *tagptr=NULL;
1139 struct parameterwrapper ** queues = vqueues;
1140 int length = vlength;
1142 if(corenum > NUMCORES - 1) {
1146 if(queues == NULL) {
1147 #ifdef THREADSIMULATE
1148 int numofcore = pthread_getspecific(key);
1149 queues = objectqueues[numofcore][ptr->type];
1150 length = numqueues[numofcore][ptr->type];
1152 queues = objectqueues[corenum][ptr->type];
1153 length = numqueues[corenum][ptr->type];
1157 raw_test_pass(0xeaa1);
1158 raw_test_pass_reg(queues);
1159 raw_test_pass_reg(length);
1161 tagptr=ptr->___tags___;
1163 /* Outer loop iterates through all parameter queues an object of
1164 this type could be in. */
1165 for(j = 0; j < length; ++j) {
1166 parameter = queues[j];
1168 if (parameter->numbertags>0) {
1170 raw_test_pass(0xeaa2);
1171 raw_test_pass_reg(tagptr);
1174 goto nextloop;//that means the object has no tag but that param needs tag
1175 else if(tagptr->type==TAGTYPE) {//one tag
1176 struct ___TagDescriptor___ * tag=(struct ___TagDescriptor___*) tagptr;
1178 raw_test_pass(0xeaa3);
1180 for(i=0;i<parameter->numbertags;i++) {
1181 //slotid is parameter->tagarray[2*i];
1182 int tagid=parameter->tagarray[2*i+1];
1183 if (tagid!=tagptr->flag) {
1185 raw_test_pass(0xeaa4);
1187 goto nextloop; /*We don't have this tag */
1190 } else {//multiple tags
1191 struct ArrayObject * ao=(struct ArrayObject *) tagptr;
1193 raw_test_pass(0xeaa5);
1195 for(i=0;i<parameter->numbertags;i++) {
1196 //slotid is parameter->tagarray[2*i];
1197 int tagid=parameter->tagarray[2*i+1];
1199 for(j=0;j<ao->___cachedCode___;j++) {
1200 if (tagid==ARRAYGET(ao, struct ___TagDescriptor___*, j)->flag) {
1205 raw_test_pass(0xeaa6);
1215 for(i=0;i<parameter->numberofterms;i++) {
1216 int andmask=parameter->intarray[i*2];
1217 int checkmask=parameter->intarray[i*2+1];
1219 raw_test_pass(0xeaa7);
1220 raw_test_pass(0xcc000000 + andmask);
1221 raw_test_pass_reg(ptr);
1222 raw_test_pass(0xcc000000 + ptr->flag);
1223 raw_test_pass(0xcc000000 + checkmask);
1225 if ((ptr->flag&andmask)==checkmask) {
1227 raw_test_pass(0xeaa8);
1229 enqueuetasks_I(parameter, prevptr, ptr, NULL, 0);
1240 // helper function to compute the coordinates of a core from the core number
1241 void calCoords(int core_num, int* coordY, int* coordX) {
1242 *coordX = core_num % 4;
1243 *coordY = core_num / 4;
1247 /* Message format for RAW version:
1249 * type: 0 -- transfer object
1250 * 1 -- transfer stall msg
1256 * ObjMsg: 0 + size of msg + obj's address + (task index + param index)+
1257 * StallMsg: 1 + corenum + sendobjs + receiveobjs (size is always 4 * sizeof(int))
1258 * LockMsg: 2 + lock type + obj pointer + request core (size is always 4 * sizeof(int))
1259 * 3/4/5 + lock type + obj pointer (size is always 3 * sizeof(int))
1260 * lock type: 0 -- read; 1 -- write
1263 // transfer an object to targetcore
1265 void transferObject(struct transObjInfo * transObj) {
1266 void * obj = transObj->objptr;
1267 int type=((int *)obj)[0];
1268 int size=classsize[type];
1269 int targetcore = transObj->targetcore;
1270 //assert(type < NUMCLASSES); // can only transfer normal object
1274 int self_y, self_x, target_y, target_x;
1276 // for 32 bit machine, the size of fixed part is always 3 words
1277 //int msgsize = sizeof(int) * 2 + sizeof(void *);
1278 int msgsize = 3 + transObj->length * 2;
1281 struct ___Object___ * newobj = (struct ___Object___ *)obj;
1282 /*if(0 == newobj->isolate) {
1286 calCoords(corenum, &self_y, &self_x);
1287 calCoords(targetcore, &target_y, &target_x);
1288 // Build the message header
1289 msgHdr = construct_dyn_hdr(0, msgsize, 0, // msgsize word sent.
1291 target_y, target_x);
1292 gdn_send(msgHdr); // Send the message header to EAST to handle fab(n - 1).
1294 raw_test_pass(0xbbbb);
1295 raw_test_pass(0xb000 + targetcore); // targetcore
1303 raw_test_pass_reg(msgsize);
1307 raw_test_pass_reg(obj);
1309 for(i = 0; i < transObj->length; ++i) {
1310 int taskindex = transObj->queues[2*i];
1311 int paramindex = transObj->queues[2*i+1];
1312 gdn_send(taskindex);
1314 raw_test_pass_reg(taskindex);
1316 gdn_send(paramindex);
1318 raw_test_pass_reg(paramindex);
1322 raw_test_pass(0xffff);
1324 ++(self_numsendobjs);
1325 #elif defined THREADSIMULATE
1326 int numofcore = pthread_getspecific(key);
1328 // use POSIX message queue to transfer objects between cores
1332 if(targetcore < 10) {
1333 corenumstr[0] = targetcore + '0';
1334 corenumstr[1] = '\0';
1336 } else if(targetcore < 100) {
1337 corenumstr[1] = targetcore % 10 + '0';
1338 corenumstr[0] = (targetcore / 10) + '0';
1339 corenumstr[2] = '\0';
1342 printf("Error: targetcore >= 100\n");
1346 char * pathhead = "/msgqueue_";
1347 int targetlen = strlen(pathhead);
1348 char path[targetlen + sourcelen + 1];
1349 strcpy(path, pathhead);
1350 strncat(path, corenumstr, sourcelen);
1351 int oflags = O_WRONLY|O_NONBLOCK;
1352 int omodes = S_IRWXU|S_IRWXG|S_IRWXO;
1353 mqdnum = mq_open(path, oflags, omodes, NULL);
1355 printf("[transferObject, %d] mq_open %s fail: %d, error: %s\n", numofcore, path, mqdnum, strerror(errno));
1359 /*struct ___Object___ * newobj = (struct ___Object___ *)obj;
1360 if(0 == newobj->isolate) {
1361 newobj = RUNMALLOC(size);
1362 memcpy(newobj, obj, size);
1363 newobj->original=obj;
1365 struct transObjInfo * tmptransObj = RUNMALLOC(sizeof(struct transObjInfo));
1366 memcpy(tmptransObj, transObj, sizeof(struct transObjInfo));
1367 int * tmpqueue = RUNMALLOC(sizeof(int)*2*tmptransObj->length);
1368 memcpy(tmpqueue, tmptransObj->queues, sizeof(int)*2*tmptransObj->length);
1369 tmptransObj->queues = tmpqueue;
1370 struct ___Object___ * newobj = RUNMALLOC(sizeof(struct ___Object___));
1371 newobj->type = ((struct ___Object___ *)obj)->type;
1372 newobj->original = (struct ___Object___ *)tmptransObj;
1375 ret=mq_send(mqdnum, (void *)newobj, sizeof(struct ___Object___), 0); // send the object into the queue
1377 printf("[transferObject, %d] mq_send to %s returned: %d, error: %s\n", numofcore, path, ret, strerror(errno));
1381 if(numofcore == STARTUPCORE) {
1382 ++numsendobjs[numofcore];
1384 ++(thread_data_array[numofcore].numsendobjs);
1386 printf("[transferObject, %d] mq_send to %s returned: $%x\n", numofcore, path, ret);
1390 // send terminate message to targetcore
1392 bool transStallMsg(int targetcore) {
1395 int self_y, self_x, target_y, target_x;
1396 // for 32 bit machine, the size is always 4 words
1397 //int msgsize = sizeof(int) * 4;
1400 calCoords(corenum, &self_y, &self_x);
1401 calCoords(targetcore, &target_y, &target_x);
1402 // Build the message header
1403 msgHdr = construct_dyn_hdr(0, msgsize, 0, // msgsize word sent.
1405 target_y, target_x);
1406 gdn_send(msgHdr); // Send the message header to EAST to handle fab(n - 1).
1408 raw_test_pass(0xbbbb);
1409 raw_test_pass(0xb000 + targetcore); // targetcore
1417 raw_test_pass_reg(corenum);
1419 gdn_send(self_numsendobjs);
1421 raw_test_pass_reg(self_numsendobjs);
1423 gdn_send(self_numreceiveobjs);
1425 raw_test_pass_reg(self_numreceiveobjs);
1426 raw_test_pass(0xffff);
1429 #elif defined THREADSIMULATE
1430 struct ___Object___ *newobj = RUNMALLOC(sizeof(struct ___Object___));
1431 // use the first four int field to hold msgtype/corenum/sendobj/receiveobj
1433 int numofcore = pthread_getspecific(key);
1434 newobj->flag = numofcore;
1435 newobj->___cachedHash___ = thread_data_array[numofcore].numsendobjs;
1436 newobj->___cachedCode___ = thread_data_array[numofcore].numreceiveobjs;
1438 // use POSIX message queue to send stall msg to startup core
1439 assert(targetcore == STARTUPCORE);
1443 if(targetcore < 10) {
1444 corenumstr[0] = targetcore + '0';
1445 corenumstr[1] = '\0';
1447 } else if(targetcore < 100) {
1448 corenumstr[1] = targetcore % 10 + '0';
1449 corenumstr[0] = (targetcore / 10) + '0';
1450 corenumstr[2] = '\0';
1453 printf("Error: targetcore >= 100\n");
1457 char * pathhead = "/msgqueue_";
1458 int targetlen = strlen(pathhead);
1459 char path[targetlen + sourcelen + 1];
1460 strcpy(path, pathhead);
1461 strncat(path, corenumstr, sourcelen);
1462 int oflags = O_WRONLY|O_NONBLOCK;
1463 int omodes = S_IRWXU|S_IRWXG|S_IRWXO;
1464 mqdnum = mq_open(path, oflags, omodes, NULL);
1466 printf("[transStallMsg, %d] mq_open %s fail: %d, error: %s\n", numofcore, path, mqdnum, strerror(errno));
1471 ret=mq_send(mqdnum, (void *)newobj, sizeof(struct ___Object___), 0); // send the object into the queue
1473 printf("[transStallMsg, %d] mq_send to %s returned: %d, error: %s\n", numofcore, path, ret, strerror(errno));
1477 printf("[transStallMsg, %d] mq_send to %s returned: $%x\n", numofcore, path, ret);
1478 printf("<transStallMsg> to %s index: %d, sendobjs: %d, receiveobjs: %d\n", path, newobj->flag, newobj->___cachedHash___, newobj->___cachedCode___);
1485 // receive object transferred from other cores
1486 // or the terminate message from other cores
1487 // NOTICE: following format is for threadsimulate version only
1488 // RAW version please see previous description
1489 // format: type + object
1490 // type: -1--stall msg
1492 // return value: 0--received an object
1493 // 1--received nothing
1494 // 2--received a Stall Msg
1495 // 3--received a lock Msg
1496 // RAW version: -1 -- received nothing
1497 // otherwise -- received msg type
1498 int receiveObject() {
1502 int self_y, self_x, target_y, target_x;
1504 if(gdn_input_avail() == 0) {
1506 if(corenum < NUMCORES) {
1507 raw_test_pass(0xd001);
1514 raw_test_pass(0xcccc);
1516 while((gdn_input_avail() != 0) && (msgdataindex < msglength)) {
1517 msgdata[msgdataindex] = gdn_receive();
1518 if(msgdataindex == 0) {
1519 if(msgdata[0] > 2) {
1521 } else if(msgdata[0] > 0) {
1524 } else if((msgdataindex == 1) && (msgdata[0] == 0)) {
1525 msglength = msgdata[msgdataindex];
1528 raw_test_pass_reg(msgdata[msgdataindex]);
1532 /*if(msgdataindex == 0) {
1534 msgtype = gdn_receive();
1541 msgdata = (int *)RUNMALLOC_I(msglength * sizeof(int));
1542 msgdata[msgdataindex] = msgtype;
1545 raw_test_pass_reg(msgtype);
1547 } else if((msgdataindex == 1) && (msgtype == 0)) {
1548 // object transfer msg
1549 msglength = gdn_receive();
1550 msgdata = (int *)RUNMALLOC_I(msglength * sizeof(int));
1551 msgdata[0] = msgtype;
1552 msgdata[msgdataindex] = msglength;
1554 raw_test_pass_reg(msgdata[msgdataindex]);
1557 msgdata[msgdataindex] = gdn_receive();
1559 raw_test_pass_reg(msgdata[msgdataindex]);
1565 raw_test_pass(0xffff);
1567 if(msgdataindex == msglength) {
1568 // received a whole msg
1569 int type, data1, data2; // will receive at least 3 words including type
1575 // receive a object transfer msg
1576 struct transObjInfo * transObj = RUNMALLOC_I(sizeof(struct transObjInfo));
1578 if(corenum > NUMCORES - 1) {
1579 raw_test_done(0xa00a);
1581 // store the object and its corresponding queue info, enqueue it later
1582 transObj->objptr = (void *)data2; // data1 is now size of the msg
1583 transObj->length = (msglength - 3) / 2;
1584 transObj->queues = RUNMALLOC_I(sizeof(int)*(msglength - 3));
1585 for(k = 0; k < transObj->length; ++k) {
1586 transObj->queues[2*k] = msgdata[3+2*k];
1588 raw_test_pass_reg(transObj->queues[2*k]);
1590 transObj->queues[2*k+1] = msgdata[3+2*k+1];
1592 raw_test_pass_reg(transObj->queues[2*k+1]);
1595 //memcpy(transObj->queues, msgdata[3], sizeof(int)*(msglength - 3));
1596 addNewItem_I(&objqueue, (void *)transObj);
1597 ++(self_numreceiveobjs);
1599 raw_test_pass(0xe881);
1602 addNewItem_I(&objqueue, (void *)data2);
1603 ++(self_numreceiveobjs);
1605 raw_test_pass(0xe881);
1611 // receive a stall msg
1612 if(corenum != STARTUPCORE) {
1613 // non startup core can not receive stall msg
1615 raw_test_done(0xa001);
1617 if(data1 < NUMCORES) {
1619 raw_test_pass(0xe882);
1621 corestatus[data1] = 0;
1622 numsendobjs[data1] = data2;
1623 numreceiveobjs[data1] = msgdata[3];
1628 // receive lock request msg
1629 // for 32 bit machine, the size is always 3 words
1630 //int msgsize = sizeof(int) * 3;
1632 // lock request msg, handle it right now
1633 // check to see if there is a lock exist in locktbl for the required obj
1634 int data3 = msgdata[3];
1636 if(!RuntimeHashcontainskey(locktbl, data2)) {
1637 // no locks for this object
1638 // first time to operate on this shared object
1639 // create a lock for it
1640 // the lock is an integer: 0 -- stall, >0 -- read lock, -1 -- write lock
1642 raw_test_pass(0xe883);
1645 RuntimeHashadd_I(locktbl, data2, 1);
1647 RuntimeHashadd_I(locktbl, data2, -1);
1652 raw_test_pass(0xe884);
1654 RuntimeHashget(locktbl, data2, &rwlock_obj);
1656 raw_test_pass_reg(rwlock_obj);
1658 if(0 == rwlock_obj) {
1664 RuntimeHashremovekey(locktbl, data2);
1665 RuntimeHashadd_I(locktbl, data2, rwlock_obj);
1666 } else if((rwlock_obj > 0) && (data1 == 0)) {
1667 // read lock request and there are only read locks
1669 RuntimeHashremovekey(locktbl, data2);
1670 RuntimeHashadd_I(locktbl, data2, rwlock_obj);
1675 raw_test_pass_reg(rwlock_obj);
1679 calCoords(corenum, &self_y, &self_x);
1680 calCoords(targetcore, &target_y, &target_x);
1681 // Build the message header
1682 msgHdr = construct_dyn_hdr(0, msgsize, 0, // msgsize word sent.
1684 target_y, target_x);
1685 gdn_send(msgHdr); // Send the message header to EAST to handle fab(n - 1).
1687 raw_test_pass(0xbbbb);
1688 raw_test_pass(0xb000 + targetcore); // targetcore
1691 // deny the lock request
1692 gdn_send(4); // lock request
1697 // grount the lock request
1698 gdn_send(3); // lock request
1703 gdn_send(data1); // lock type
1705 raw_test_pass_reg(data1);
1707 gdn_send(data2); // lock target
1709 raw_test_pass_reg(data2);
1710 raw_test_pass(0xffff);
1715 // receive lock grount msg
1716 if(corenum > NUMCORES - 1) {
1717 raw_test_done(0xa00b);
1719 if(lockobj == data2) {
1726 // conflicts on lockresults
1727 raw_test_done(0xa002);
1732 // receive lock grount/deny msg
1733 if(corenum > NUMCORES - 1) {
1734 raw_test_done(0xa00c);
1736 if(lockobj == data2) {
1743 // conflicts on lockresults
1744 raw_test_done(0xa003);
1749 // receive lock release msg
1750 if(!RuntimeHashcontainskey(locktbl, data2)) {
1751 // no locks for this object, something is wrong
1752 raw_test_done(0xa004);
1755 RuntimeHashget(locktbl, data2, &rwlock_obj);
1757 raw_test_pass(0xe885);
1758 raw_test_pass_reg(rwlock_obj);
1765 RuntimeHashremovekey(locktbl, data2);
1766 RuntimeHashadd_I(locktbl, data2, rwlock_obj);
1768 raw_test_pass_reg(rwlock_obj);
1778 for(msgdataindex--;msgdataindex > 0; --msgdataindex) {
1779 msgdata[msgdataindex] = -1;
1785 raw_test_pass(0xe886);
1787 if(gdn_input_avail() != 0) {
1794 raw_test_pass(0xe887);
1798 #elif defined THREADSIMULATE
1799 int numofcore = pthread_getspecific(key);
1800 // use POSIX message queue to transfer object
1802 struct mq_attr mqattr;
1803 mq_getattr(mqd[numofcore], &mqattr);
1804 void * msgptr =RUNMALLOC(mqattr.mq_msgsize);
1805 msglen=mq_receive(mqd[numofcore], msgptr, mqattr.mq_msgsize, NULL); // receive the object into the queue
1811 //printf("msg: %s\n",msgptr);
1812 if(((int*)msgptr)[0] == -1) {
1814 struct ___Object___ * tmpptr = (struct ___Object___ *)msgptr;
1815 int index = tmpptr->flag;
1816 corestatus[index] = 0;
1817 numsendobjs[index] = tmpptr->___cachedHash___;
1818 numreceiveobjs[index] = tmpptr->___cachedCode___;
1819 printf("<receiveObject> index: %d, sendobjs: %d, reveiveobjs: %d\n", index, numsendobjs[index], numreceiveobjs[index]);
1822 } /*else if(((int*)msgptr)[0] == -2) {
1827 if(numofcore == STARTUPCORE) {
1828 ++(numreceiveobjs[numofcore]);
1830 ++(thread_data_array[numofcore].numreceiveobjs);
1832 struct ___Object___ * tmpptr = (struct ___Object___ *)msgptr;
1833 struct transObjInfo * transObj = (struct transObjInfo *)tmpptr->original;
1834 tmpptr = (struct ___Object___ *)(transObj->objptr);
1835 int type = tmpptr->type;
1836 int size=classsize[type];
1837 struct ___Object___ * newobj=RUNMALLOC(size);
1838 memcpy(newobj, tmpptr, size);
1839 if(0 == newobj->isolate) {
1840 newobj->original=tmpptr;
1845 for(k = 0; k < transObj->length; ++k) {
1846 int taskindex = transObj->queues[2 * k];
1847 int paramindex = transObj->queues[2 * k + 1];
1848 struct parameterwrapper ** queues = &(paramqueues[numofcore][taskindex][paramindex]);
1849 enqueueObject(newobj, queues, 1);
1851 RUNFREE(transObj->queues);
1858 bool getreadlock(void * ptr) {
1861 int self_y, self_x, target_y, target_x;
1862 int targetcore = ((int)ptr >> 1) % TOTALCORE;
1863 // for 32 bit machine, the size is always 4 words
1864 //int msgsize = sizeof(int) * 4;
1874 if(targetcore == corenum) {
1875 // reside on this core
1878 raw_user_interrupts_off();
1880 if(!RuntimeHashcontainskey(locktbl, (int)ptr)) {
1881 // no locks for this object
1882 // first time to operate on this shared object
1883 // create a lock for it
1884 // the lock is an integer: 0 -- stall, >0 -- read lock, -1 -- write lock
1885 RuntimeHashadd_I(locktbl, (int)ptr, 1);
1888 RuntimeHashget(locktbl, (int)ptr, &rwlock_obj);
1889 if(-1 != rwlock_obj) {
1891 RuntimeHashremovekey(locktbl, (int)ptr);
1892 RuntimeHashadd_I(locktbl, (int)ptr, rwlock_obj);
1898 raw_user_interrupts_on();
1900 if(lockobj == (int)ptr) {
1911 // conflicts on lockresults
1912 raw_test_done(0xa005);
1917 calCoords(corenum, &self_y, &self_x);
1918 calCoords(targetcore, &target_y, &target_x);
1919 // Build the message header
1920 msgHdr = construct_dyn_hdr(0, msgsize, 0, // msgsize word sent.
1922 target_y, target_x);
1923 gdn_send(msgHdr); // Send the message header to EAST to handle fab(n - 1).
1925 raw_test_pass(0xbbbb);
1926 raw_test_pass(0xb000 + targetcore); // targetcore
1928 gdn_send(2); // lock request
1932 gdn_send(0); // read lock
1938 raw_test_pass_reg(ptr);
1942 raw_test_pass_reg(corenum);
1943 raw_test_pass(0xffff);
1946 #elif defined THREADSIMULATE
1947 int numofcore = pthread_getspecific(key);
1949 int rc = pthread_rwlock_tryrdlock(&rwlock_tbl);
1950 printf("[getreadlock, %d] getting the read lock for locktbl: %d error: \n", numofcore, rc, strerror(rc));
1954 if(!RuntimeHashcontainskey(locktbl, (int)ptr)) {
1955 // no locks for this object
1956 // first time to operate on this shared object
1957 // create a lock for it
1958 rc = pthread_rwlock_unlock(&rwlock_tbl);
1959 printf("[getreadlock, %d] release the read lock for locktbl: %d error: \n", numofcore, rc, strerror(rc));
1960 pthread_rwlock_t* rwlock = (pthread_rwlock_t *)RUNMALLOC(sizeof(pthread_rwlock_t));
1961 memcpy(rwlock, &rwlock_init, sizeof(pthread_rwlock_t));
1962 rc = pthread_rwlock_init(rwlock, NULL);
1963 printf("[getreadlock, %d] initialize the rwlock for object %d: %d error: \n", numofcore, (int)ptr, rc, strerror(rc));
1964 rc = pthread_rwlock_trywrlock(&rwlock_tbl);
1965 printf("[getreadlock, %d] getting the write lock for locktbl: %d error: \n", numofcore, rc, strerror(rc));
1970 if(!RuntimeHashcontainskey(locktbl, (int)ptr)) {
1972 RuntimeHashadd(locktbl, (int)ptr, (int)rwlock);
1975 RuntimeHashget(locktbl, (int)ptr, (int*)&rwlock);
1977 rc = pthread_rwlock_unlock(&rwlock_tbl);
1978 printf("[getreadlock, %d] release the write lock for locktbl: %d error: \n", numofcore, rc, strerror(rc));
1980 rc = pthread_rwlock_tryrdlock(rwlock);
1981 printf("[getreadlock, %d] getting read lock for object %d: %d error: \n", numofcore, (int)ptr, rc, strerror(rc));
1988 pthread_rwlock_t* rwlock_obj = NULL;
1989 RuntimeHashget(locktbl, (int)ptr, (int*)&rwlock_obj);
1990 rc = pthread_rwlock_unlock(&rwlock_tbl);
1991 printf("[getreadlock, %d] release the read lock for locktbl: %d error: \n", numofcore, rc, strerror(rc));
1992 int rc_obj = pthread_rwlock_tryrdlock(rwlock_obj);
1993 printf("[getreadlock, %d] getting read lock for object %d: %d error: \n", numofcore, (int)ptr, rc_obj, strerror(rc_obj));
2003 void releasereadlock(void * ptr) {
2006 int self_y, self_x, target_y, target_x;
2007 int targetcore = ((int)ptr >> 1) % TOTALCORE;
2008 // for 32 bit machine, the size is always 3 words
2009 //int msgsize = sizeof(int) * 3;
2012 if(targetcore == corenum) {
2014 raw_user_interrupts_off();
2016 // reside on this core
2017 if(!RuntimeHashcontainskey(locktbl, (int)ptr)) {
2018 // no locks for this object, something is wrong
2019 raw_test_done(0xa006);
2022 RuntimeHashget(locktbl, (int)ptr, &rwlock_obj);
2024 RuntimeHashremovekey(locktbl, (int)ptr);
2025 RuntimeHashadd_I(locktbl, (int)ptr, rwlock_obj);
2028 raw_user_interrupts_on();
2033 calCoords(corenum, &self_y, &self_x);
2034 calCoords(targetcore, &target_y, &target_x);
2035 // Build the message header
2036 msgHdr = construct_dyn_hdr(0, msgsize, 0, // msgsize word sent.
2038 target_y, target_x);
2039 gdn_send(msgHdr); // Send the message header to EAST to handle fab(n - 1).
2041 raw_test_pass(0xbbbb);
2042 raw_test_pass(0xb000 + targetcore); // targetcore
2044 gdn_send(5); // lock release
2048 gdn_send(0); // read lock
2054 raw_test_pass_reg(ptr);
2055 raw_test_pass(0xffff);
2057 #elif defined THREADSIMULATE
2058 int numofcore = pthread_getspecific(key);
2059 int rc = pthread_rwlock_rdlock(&rwlock_tbl);
2060 printf("[releasereadlock, %d] getting the read lock for locktbl: %d error: \n", numofcore, rc, strerror(rc));
2061 if(!RuntimeHashcontainskey(locktbl, (int)ptr)) {
2062 printf("[releasereadlock, %d] Error: try to release a lock without previously grab it\n", numofcore);
2065 pthread_rwlock_t* rwlock_obj = NULL;
2066 RuntimeHashget(locktbl, (int)ptr, (int*)&rwlock_obj);
2067 int rc_obj = pthread_rwlock_unlock(rwlock_obj);
2068 printf("[releasereadlock, %d] unlocked object %d: %d error: \n", numofcore, (int)ptr, rc_obj, strerror(rc_obj));
2069 rc = pthread_rwlock_unlock(&rwlock_tbl);
2070 printf("[releasereadlock, %d] release the read lock for locktbl: %d error: \n", numofcore, rc, strerror(rc));
2075 bool getreadlock_I(void * ptr) {
2077 int self_y, self_x, target_y, target_x;
2078 int targetcore = ((int)ptr >> 1) % TOTALCORE;
2079 // for 32 bit machine, the size is always 4 words
2080 //int msgsize = sizeof(int) * 4;
2090 if(targetcore == corenum) {
2091 // reside on this core
2093 if(!RuntimeHashcontainskey(locktbl, (int)ptr)) {
2094 // no locks for this object
2095 // first time to operate on this shared object
2096 // create a lock for it
2097 // the lock is an integer: 0 -- stall, >0 -- read lock, -1 -- write lock
2098 RuntimeHashadd_I(locktbl, (int)ptr, 1);
2101 RuntimeHashget(locktbl, (int)ptr, &rwlock_obj);
2102 if(-1 != rwlock_obj) {
2104 RuntimeHashremovekey(locktbl, (int)ptr);
2105 RuntimeHashadd_I(locktbl, (int)ptr, rwlock_obj);
2110 if(lockobj == (int)ptr) {
2121 // conflicts on lockresults
2122 raw_test_done(0xa005);
2127 calCoords(corenum, &self_y, &self_x);
2128 calCoords(targetcore, &target_y, &target_x);
2129 // Build the message header
2130 msgHdr = construct_dyn_hdr(0, msgsize, 0, // msgsize word sent.
2132 target_y, target_x);
2133 gdn_send(msgHdr); // Send the message header to EAST to handle fab(n - 1).
2135 raw_test_pass(0xbbbb);
2136 raw_test_pass(0xb000 + targetcore); // targetcore
2138 gdn_send(2); // lock request
2142 gdn_send(0); // read lock
2148 raw_test_pass_reg(ptr);
2152 raw_test_pass_reg(corenum);
2153 raw_test_pass(0xffff);
2158 void releasereadlock_I(void * ptr) {
2160 int self_y, self_x, target_y, target_x;
2161 int targetcore = ((int)ptr >> 1) % TOTALCORE;
2162 // for 32 bit machine, the size is always 3 words
2163 //int msgsize = sizeof(int) * 3;
2166 if(targetcore == corenum) {
2167 // reside on this core
2168 if(!RuntimeHashcontainskey(locktbl, (int)ptr)) {
2169 // no locks for this object, something is wrong
2170 raw_test_done(0xa006);
2173 RuntimeHashget(locktbl, (int)ptr, &rwlock_obj);
2175 RuntimeHashremovekey(locktbl, (int)ptr);
2176 RuntimeHashadd_I(locktbl, (int)ptr, rwlock_obj);
2181 calCoords(corenum, &self_y, &self_x);
2182 calCoords(targetcore, &target_y, &target_x);
2183 // Build the message header
2184 msgHdr = construct_dyn_hdr(0, msgsize, 0, // msgsize word sent.
2186 target_y, target_x);
2187 gdn_send(msgHdr); // Send the message header to EAST to handle fab(n - 1).
2189 raw_test_pass(0xbbbb);
2190 raw_test_pass(0xb000 + targetcore); // targetcore
2192 gdn_send(5); // lock release
2196 gdn_send(0); // read lock
2202 raw_test_pass_reg(ptr);
2203 raw_test_pass(0xffff);
2209 bool getwritelock(void * ptr) {
2212 int self_y, self_x, target_y, target_x;
2213 int targetcore = ((int)ptr >> 1) % TOTALCORE;
2214 // for 32 bit machine, the size is always 4 words
2215 //int msgsize = sizeof(int) * 4;
2219 //raw_user_interrupts_off();
2221 //targetcore = ((int)ptr) % tc;
2223 //raw_user_interrupts_on();
2227 raw_test_pass(0xe551);
2228 raw_test_pass_reg(ptr);
2229 raw_test_pass_reg(targetcore);
2230 raw_test_pass_reg(tc);
2240 if(targetcore == corenum) {
2241 // reside on this core
2244 raw_user_interrupts_off();
2246 if(!RuntimeHashcontainskey(locktbl, (int)ptr)) {
2247 // no locks for this object
2248 // first time to operate on this shared object
2249 // create a lock for it
2250 // the lock is an integer: 0 -- stall, >0 -- read lock, -1 -- write lock
2252 raw_test_pass(0xe552);
2254 RuntimeHashadd_I(locktbl, (int)ptr, -1);
2257 RuntimeHashget(locktbl, (int)ptr, &rwlock_obj);
2259 raw_test_pass(0xe553);
2260 raw_test_pass_reg(rwlock_obj);
2262 if(0 == rwlock_obj) {
2264 RuntimeHashremovekey(locktbl, (int)ptr);
2265 RuntimeHashadd_I(locktbl, (int)ptr, rwlock_obj);
2271 raw_user_interrupts_on();
2274 raw_test_pass(0xe554);
2275 raw_test_pass_reg(lockresult);
2277 if(lockobj == (int)ptr) {
2294 // conflicts on lockresults
2295 raw_test_done(0xa007);
2301 raw_test_pass(0xe555);
2303 calCoords(corenum, &self_y, &self_x);
2304 calCoords(targetcore, &target_y, &target_x);
2305 // Build the message header
2306 msgHdr = construct_dyn_hdr(0, msgsize, 0, // msgsize word sent.
2308 target_y, target_x);
2309 gdn_send(msgHdr); // Send the message header to EAST to handle fab(n - 1).
2311 raw_test_pass(0xbbbb);
2312 raw_test_pass(0xb000 + targetcore); // targetcore
2314 gdn_send(2); // lock request
2318 gdn_send(1); // write lock
2324 raw_test_pass_reg(ptr);
2328 raw_test_pass_reg(corenum);
2329 raw_test_pass(0xffff);
2332 #elif defined THREADSIMULATE
2333 int numofcore = pthread_getspecific(key);
2335 int rc = pthread_rwlock_tryrdlock(&rwlock_tbl);
2336 printf("[getwritelock, %d] getting the read lock for locktbl: %d error: \n", numofcore, rc, strerror(rc));
2340 if(!RuntimeHashcontainskey(locktbl, (int)ptr)) {
2341 // no locks for this object
2342 // first time to operate on this shared object
2343 // create a lock for it
2344 rc = pthread_rwlock_unlock(&rwlock_tbl);
2345 printf("[getwritelock, %d] release the read lock for locktbl: %d error: \n", numofcore, rc, strerror(rc));
2346 pthread_rwlock_t* rwlock = (pthread_rwlock_t *)RUNMALLOC(sizeof(pthread_rwlock_t));
2347 memcpy(rwlock, &rwlock_init, sizeof(pthread_rwlock_t));
2348 rc = pthread_rwlock_init(rwlock, NULL);
2349 printf("[getwritelock, %d] initialize the rwlock for object %d: %d error: \n", numofcore, (int)ptr, rc, strerror(rc));
2350 rc = pthread_rwlock_trywrlock(&rwlock_tbl);
2351 printf("[getwritelock, %d] getting the write lock for locktbl: %d error: \n", numofcore, rc, strerror(rc));
2353 pthread_rwlock_destroy(rwlock);
2357 if(!RuntimeHashcontainskey(locktbl, (int)ptr)) {
2359 RuntimeHashadd(locktbl, (int)ptr, (int)rwlock);
2361 pthread_rwlock_destroy(rwlock);
2363 RuntimeHashget(locktbl, (int)ptr, (int*)&rwlock);
2365 rc = pthread_rwlock_unlock(&rwlock_tbl);
2366 printf("[getwritelock, %d] release the write lock for locktbl: %d error: \n", numofcore, rc, strerror(rc));
2368 rc = pthread_rwlock_trywrlock(rwlock);
2369 printf("[getwritelock, %d] getting write lock for object %d: %d error: \n", numofcore, (int)ptr, rc, strerror(rc));
2376 pthread_rwlock_t* rwlock_obj = NULL;
2377 RuntimeHashget(locktbl, (int)ptr, (int*)&rwlock_obj);
2378 rc = pthread_rwlock_unlock(&rwlock_tbl);
2379 printf("[getwritelock, %d] release the read lock for locktbl: %d error: \n", numofcore, rc, strerror(rc));
2380 int rc_obj = pthread_rwlock_trywrlock(rwlock_obj);
2381 printf("[getwritelock, %d] getting write lock for object %d: %d error: \n", numofcore, (int)ptr, rc_obj, strerror(rc_obj));
2392 void releasewritelock(void * ptr) {
2395 int self_y, self_x, target_y, target_x;
2396 int targetcore = ((int)ptr >> 1) % TOTALCORE;
2397 // for 32 bit machine, the size is always 3 words
2398 //int msgsize = sizeof(int) * 3;
2401 if(targetcore == corenum) {
2403 raw_user_interrupts_off();
2405 // reside on this core
2406 if(!RuntimeHashcontainskey(locktbl, (int)ptr)) {
2407 // no locks for this object, something is wrong
2408 raw_test_done(0xa008);
2412 raw_test_pass(0xe662);
2414 RuntimeHashget(locktbl, (int)ptr, &rwlock_obj);
2416 raw_test_pass_reg(rwlock_obj);
2419 RuntimeHashremovekey(locktbl, (int)ptr);
2420 RuntimeHashadd_I(locktbl, (int)ptr, rwlock_obj);
2422 raw_test_pass_reg(rwlock_obj);
2426 raw_user_interrupts_on();
2432 raw_test_pass(0xe663);
2434 calCoords(corenum, &self_y, &self_x);
2435 calCoords(targetcore, &target_y, &target_x);
2436 // Build the message header
2437 msgHdr = construct_dyn_hdr(0, msgsize, 0, // msgsize word sent.
2439 target_y, target_x);
2440 gdn_send(msgHdr); // Send the message header to EAST to handle fab(n - 1).
2442 raw_test_pass(0xbbbb);
2443 raw_test_pass(0xb000 + targetcore);
2445 gdn_send(5); // lock release
2449 gdn_send(1); // write lock
2455 raw_test_pass_reg(ptr);
2456 raw_test_pass(0xffff);
2458 #elif defined THREADSIMULATE
2459 int numofcore = pthread_getspecific(key);
2460 int rc = pthread_rwlock_rdlock(&rwlock_tbl);
2461 printf("[releasewritelock, %d] getting the read lock for locktbl: %d error: \n", numofcore, rc, strerror(rc));
2462 if(!RuntimeHashcontainskey(locktbl, (int)ptr)) {
2463 printf("[releasewritelock, %d] Error: try to release a lock without previously grab it\n", numofcore);
2466 pthread_rwlock_t* rwlock_obj = NULL;
2467 RuntimeHashget(locktbl, (int)ptr, (int*)&rwlock_obj);
2468 int rc_obj = pthread_rwlock_unlock(rwlock_obj);
2469 printf("[releasewritelock, %d] unlocked object %d: %d error:\n", numofcore, (int)ptr, rc_obj, strerror(rc_obj));
2470 rc = pthread_rwlock_unlock(&rwlock_tbl);
2471 printf("[releasewritelock, %d] release the read lock for locktbl: %d error: \n", numofcore, rc, strerror(rc));
2475 int enqueuetasks(struct parameterwrapper *parameter, struct parameterwrapper *prevptr, struct ___Object___ *ptr, int * enterflags, int numenterflags) {
2476 void * taskpointerarray[MAXTASKPARAMS];
2478 int numparams=parameter->task->numParameters;
2479 int numiterators=parameter->task->numTotal-1;
2484 struct taskdescriptor * task=parameter->task;
2486 ObjectHashadd(parameter->objectset, (int) ptr, 0, (int) enterflags, numenterflags, enterflags==NULL);//this add the object to parameterwrapper
2488 /* Add enqueued object to parameter vector */
2489 taskpointerarray[parameter->slot]=ptr;
2491 /* Reset iterators */
2492 for(j=0;j<numiterators;j++) {
2493 toiReset(¶meter->iterators[j]);
2496 /* Find initial state */
2497 for(j=0;j<numiterators;j++) {
2499 if(toiHasNext(¶meter->iterators[j], taskpointerarray OPTARG(failed)))
2500 toiNext(¶meter->iterators[j], taskpointerarray OPTARG(failed));
2502 /* Need to backtrack */
2503 toiReset(¶meter->iterators[j]);
2507 /* Nothing to enqueue */
2514 /* Enqueue current state */
2516 struct taskparamdescriptor *tpd=RUNMALLOC(sizeof(struct taskparamdescriptor));
2518 tpd->numParameters=numiterators+1;
2519 tpd->parameterArray=RUNMALLOC(sizeof(void *)*(numiterators+1));
2520 for(j=0;j<=numiterators;j++){
2521 tpd->parameterArray[j]=taskpointerarray[j];//store the actual parameters
2524 if ((/*!gencontains(failedtasks, tpd)&&*/!gencontains(activetasks,tpd))) {
2525 genputtable(activetasks, tpd, tpd);
2527 RUNFREE(tpd->parameterArray);
2531 /* This loop iterates to the next parameter combination */
2532 if (numiterators==0)
2535 for(j=numiterators-1; j<numiterators;j++) {
2537 if(toiHasNext(¶meter->iterators[j], taskpointerarray OPTARG(failed)))
2538 toiNext(¶meter->iterators[j], taskpointerarray OPTARG(failed));
2540 /* Need to backtrack */
2541 toiReset(¶meter->iterators[j]);
2545 /* Nothing more to enqueue */
2554 int enqueuetasks_I(struct parameterwrapper *parameter, struct parameterwrapper *prevptr, struct ___Object___ *ptr, int * enterflags, int numenterflags) {
2555 void * taskpointerarray[MAXTASKPARAMS];
2557 int numparams=parameter->task->numParameters;
2558 int numiterators=parameter->task->numTotal-1;
2563 struct taskdescriptor * task=parameter->task;
2565 ObjectHashadd_I(parameter->objectset, (int) ptr, 0, (int) enterflags, numenterflags, enterflags==NULL);//this add the object to parameterwrapper
2567 /* Add enqueued object to parameter vector */
2568 taskpointerarray[parameter->slot]=ptr;
2570 /* Reset iterators */
2571 for(j=0;j<numiterators;j++) {
2572 toiReset(¶meter->iterators[j]);
2575 /* Find initial state */
2576 for(j=0;j<numiterators;j++) {
2578 if(toiHasNext(¶meter->iterators[j], taskpointerarray OPTARG(failed)))
2579 toiNext(¶meter->iterators[j], taskpointerarray OPTARG(failed));
2581 /* Need to backtrack */
2582 toiReset(¶meter->iterators[j]);
2586 /* Nothing to enqueue */
2592 /* Enqueue current state */
2594 struct taskparamdescriptor *tpd=RUNMALLOC_I(sizeof(struct taskparamdescriptor));
2596 tpd->numParameters=numiterators+1;
2597 tpd->parameterArray=RUNMALLOC_I(sizeof(void *)*(numiterators+1));
2598 for(j=0;j<=numiterators;j++){
2599 tpd->parameterArray[j]=taskpointerarray[j];//store the actual parameters
2602 if ((/*!gencontains(failedtasks, tpd)&&*/!gencontains(activetasks,tpd))) {
2603 genputtable_I(activetasks, tpd, tpd);
2605 RUNFREE(tpd->parameterArray);
2609 /* This loop iterates to the next parameter combination */
2610 if (numiterators==0)
2613 for(j=numiterators-1; j<numiterators;j++) {
2615 if(toiHasNext(¶meter->iterators[j], taskpointerarray OPTARG(failed)))
2616 toiNext(¶meter->iterators[j], taskpointerarray OPTARG(failed));
2618 /* Need to backtrack */
2619 toiReset(¶meter->iterators[j]);
2623 /* Nothing more to enqueue */
2632 /* Handler for signals. The signals catch null pointer errors and
2633 arithmatic errors. */
2635 void myhandler(int sig, siginfo_t *info, void *uap) {
2638 printf("sig=%d\n",sig);
2641 sigemptyset(&toclear);
2642 sigaddset(&toclear, sig);
2643 sigprocmask(SIG_UNBLOCK, &toclear,NULL);
2644 longjmp(error_handler,1);
2650 struct RuntimeHash *fdtoobject;
2652 void addreadfd(int fd) {
2655 FD_SET(fd, &readfds);
2658 void removereadfd(int fd) {
2659 FD_CLR(fd, &readfds);
2660 if (maxreadfd==(fd+1)) {
2662 while(maxreadfd>0&&!FD_ISSET(maxreadfd-1, &readfds))
2673 void executetasks() {
2674 void * taskpointerarray[MAXTASKPARAMS+OFFSET];
2677 struct ___Object___ * tmpparam = NULL;
2678 struct parameterdescriptor * pd=NULL;
2679 struct parameterwrapper *pw=NULL;
2689 raw_test_pass(0xe991);
2694 /* Set up signal handlers */
2695 struct sigaction sig;
2696 sig.sa_sigaction=&myhandler;
2697 sig.sa_flags=SA_SIGINFO;
2698 sigemptyset(&sig.sa_mask);
2700 /* Catch bus errors, segmentation faults, and floating point exceptions*/
2701 sigaction(SIGBUS,&sig,0);
2702 sigaction(SIGSEGV,&sig,0);
2703 sigaction(SIGFPE,&sig,0);
2704 sigaction(SIGPIPE,&sig,0);
2713 fdtoobject=allocateRuntimeHash(100);
2717 /* Map first block of memory to protected, anonymous page */
2718 mmap(0, 0x1000, 0, MAP_SHARED|MAP_FIXED|MAP_ANON, -1, 0);
2722 while((hashsize(activetasks)>0)||(maxreadfd>0)) {
2726 raw_test_pass(0xe992);
2729 /* Check if any filedescriptors have IO pending */
2732 struct timeval timeout={0,0};
2736 numselect=select(maxreadfd, &tmpreadfds, NULL, NULL, &timeout);
2738 /* Process ready fd's */
2740 for(fd=0;fd<maxreadfd;fd++) {
2741 if (FD_ISSET(fd, &tmpreadfds)) {
2742 /* Set ready flag on object */
2744 // printf("Setting fd %d\n",fd);
2745 if (RuntimeHashget(fdtoobject, fd,(int *) &objptr)) {
2746 if(intflagorand(objptr,1,0xFFFFFFFF)) { /* Set the first flag to 1 */
2747 enqueueObject(objptr, NULL, 0);
2756 /* See if there are any active tasks */
2757 if (hashsize(activetasks)>0) {
2759 currtpd=(struct taskparamdescriptor *) getfirstkey(activetasks);
2760 genfreekey(activetasks, currtpd);
2762 /* Check if this task has failed, allow a task that contains optional objects to fire */
2763 /*if (gencontains(failedtasks, currtpd)) {
2764 // Free up task parameter descriptor
2765 RUNFREE(currtpd->parameterArray);
2769 numparams=currtpd->task->numParameters;
2770 numtotal=currtpd->task->numTotal;
2772 #ifdef THREADSIMULATE
2773 int isolateflags[numparams];
2775 /* Make sure that the parameters are still in the queues */
2776 for(i=0;i<numparams;i++) {
2777 void * parameter=currtpd->parameterArray[i];
2780 raw_test_pass(0xe993);
2783 if(((struct ___Object___ *)parameter)->type == STARTUPTYPE) {
2785 taskpointerarray[i+OFFSET]=parameter;
2788 // require locks for this parameter if it is not a startup object
2789 getwritelock(parameter);
2793 raw_user_interrupts_off();
2800 while(receiveObject() != -1) {
2804 grount = lockresult;
2813 raw_user_interrupts_on();
2818 raw_test_pass(0xe994);
2820 // can not get the lock, try later
2821 for(j = 0; j < i; ++j) {
2822 releasewritelock(taskpointerarray[j+OFFSET]);
2824 genputtable(activetasks, currtpd, currtpd);
2825 if(hashsize(activetasks) == 1) {
2826 // only one task right now, wait a little while before next try
2834 raw_invalidate_cache_range((int)parameter, classsize[((struct ___Object___ *)parameter)->type]);
2836 for(tmp = 0; tmp < classsize[((struct ___Object___ *)parameter)->type]; ++tmp) {
2837 invalidateAddr(parameter + tmp);
2841 tmpparam = (struct ___Object___ *)parameter;
2842 #ifdef THREADSIMULATE
2843 if(((struct ___Object___ *)parameter)->type == STARTUPTYPE) {
2845 taskpointerarray[i+OFFSET]=parameter;
2848 if(0 == tmpparam->isolate) {
2849 isolateflags[i] = 0;
2850 // shared object, need to flush with current value
2851 //if(!getreadlock(tmpparam->original)) {
2852 // // fail to get read lock of the original object, try this task later
2853 if(!getwritelock(tmpparam->original)) {
2854 // fail to get write lock, release all obtained locks and try this task later
2856 for(j = 0; j < i; ++j) {
2857 if(0 == isolateflags[j]) {
2858 releasewritelock(((struct ___Object___ *)taskpointerarray[j+OFFSET])->original);
2861 genputtable(activetasks, currtpd, currtpd);
2864 if(tmpparam->version != tmpparam->original->version) {
2865 // some task on another core has changed this object
2866 // flush this object
2867 //memcpy(tmpparam, tmpparam->original, classsize[tmpparam->type]);
2868 // release all obtained locks
2870 for(j = 0; j < i; ++j) {
2871 if(0 == isolateflags[j]) {
2872 releasewritelock(((struct ___Object___ *)taskpointerarray[j+OFFSET])->original);
2875 releasewritelock(tmpparam->original);
2877 // dequeue this object
2878 int numofcore = pthread_getspecific(key);
2879 struct parameterwrapper ** queues = objectqueues[numofcore][tmpparam->type];
2880 int length = numqueues[numofcore][tmpparam->type];
2881 for(j = 0; j < length; ++j) {
2882 struct parameterwrapper * pw = queues[j];
2883 if(ObjectHashcontainskey(pw->objectset, (int)tmpparam)) {
2885 int UNUSED, UNUSED2;
2887 ObjectHashget(pw->objectset, (int) tmpparam, (int *) &next, (int *) &enterflags, &UNUSED, &UNUSED2);
2888 ObjectHashremove(pw->objectset, (int)tmpparam);
2889 if (enterflags!=NULL)
2893 // try to enqueue it again to check if it feeds other tasks;
2894 //enqueueObject(tmpparam, NULL, 0);
2895 // Free up task parameter descriptor
2896 RUNFREE(currtpd->parameterArray);
2901 isolateflags[i] = 1;
2904 pd=currtpd->task->descriptorarray[i];
2905 pw=(struct parameterwrapper *) pd->queue;
2906 /* Check that object is still in queue */
2908 if (!ObjectHashcontainskey(pw->objectset, (int) parameter)) {
2910 raw_test_pass(0xe995);
2912 // release grabbed locks
2913 for(j = 0; j < i; ++j) {
2914 releasewritelock(taskpointerarray[j+OFFSET]);
2916 releasewritelock(parameter);
2917 RUNFREE(currtpd->parameterArray);
2923 /* Check if the object's flags still meets requirements */
2927 for(tmpi = 0; tmpi < pw->numberofterms; ++tmpi) {
2928 andmask=pw->intarray[tmpi*2];
2929 checkmask=pw->intarray[tmpi*2+1];
2931 raw_test_pass(0xdd000000 + andmask);
2932 raw_test_pass_reg((int)parameter);
2933 raw_test_pass(0xdd000000 + ((struct ___Object___ *)parameter)->flag);
2934 raw_test_pass(0xdd000000 + checkmask);
2936 if((((struct ___Object___ *)parameter)->flag&andmask)==checkmask) {
2942 // flags are never suitable
2943 // remove this obj from the queue
2945 int UNUSED, UNUSED2;
2948 raw_test_pass(0xe996);
2950 ObjectHashget(pw->objectset, (int) parameter, (int *) &next, (int *) &enterflags, &UNUSED, &UNUSED2);
2951 ObjectHashremove(pw->objectset, (int)parameter);
2952 if (enterflags!=NULL)
2954 // release grabbed locks
2955 for(j = 0; j < i; ++j) {
2956 releasewritelock(taskpointerarray[j+OFFSET]);
2958 releasewritelock(parameter);
2959 RUNFREE(currtpd->parameterArray);
2967 /* Check that object still has necessary tags */
2968 for(j=0;j<pd->numbertags;j++) {
2969 int slotid=pd->tagarray[2*j]+numparams;
2970 struct ___TagDescriptor___ *tagd=currtpd->parameterArray[slotid];
2971 if (!containstag(parameter, tagd)) {
2973 raw_test_pass(0xe997);
2975 RUNFREE(currtpd->parameterArray);
2981 taskpointerarray[i+OFFSET]=parameter;
2984 for(;i<numtotal;i++) {
2985 taskpointerarray[i+OFFSET]=currtpd->parameterArray[i];
2988 #ifdef THREADSIMULATE
2989 for(i = 0; i < numparams; ++i) {
2990 if(0 == isolateflags[i]) {
2991 struct ___Object___ * tmpparam = (struct ___Object___ *)taskpointerarray[i+OFFSET];
2992 if(tmpparam != tmpparam->original) {
2993 taskpointerarray[i+OFFSET] = tmpparam->original;
3002 /* Checkpoint the state */
3003 forward=allocateRuntimeHash(100);
3004 reverse=allocateRuntimeHash(100);
3005 //void ** checkpoint=makecheckpoint(currtpd->task->numParameters, currtpd->parameterArray, forward, reverse);
3008 if (x=setjmp(error_handler)) {
3013 printf("Fatal Error=%d, Recovering!\n",x);
3017 genputtable(failedtasks,currtpd,currtpd);
3018 //restorecheckpoint(currtpd->task->numParameters, currtpd->parameterArray, checkpoint, forward, reverse);
3020 freeRuntimeHash(forward);
3021 freeRuntimeHash(reverse);
3029 raw_test_pass_reg(x);
3031 raw_test_done(0xa009);
3036 /*if (injectfailures) {
3037 if ((((double)random())/RAND_MAX)<failurechance) {
3038 printf("\nINJECTING TASK FAILURE to %s\n", currtpd->task->name);
3039 longjmp(error_handler,10);
3042 /* Actually call task */
3044 ((int *)taskpointerarray)[0]=currtpd->numParameters;
3045 taskpointerarray[1]=NULL;
3050 printf("ENTER %s count=%d\n",currtpd->task->name, (instaccum-instructioncount));
3052 ((void (*) (void **)) currtpd->task->taskptr)(taskpointerarray);
3054 printf("EXIT %s count=%d\n",currtpd->task->name, (instaccum-instructioncount));
3057 ((void (*) (void **)) currtpd->task->taskptr)(taskpointerarray);
3062 for(i = 0; i < numparams; ++i) {
3064 struct ___Object___ * tmpparam = (struct ___Object___ *)taskpointerarray[i+OFFSET];
3066 raw_test_pass(0xe998);
3067 raw_test_pass(0xdd100000 + tmpparam->flag);
3069 releasewritelock(tmpparam);
3071 #elif defined THREADSIMULATE
3072 for(i = 0; i < numparams; ++i) {
3073 if(0 == isolateflags[i]) {
3074 struct ___Object___ * tmpparam = (struct ___Object___ *)taskpointerarray[i+OFFSET];
3075 releasewritelock(tmpparam);
3083 freeRuntimeHash(forward);
3084 freeRuntimeHash(reverse);
3088 // Free up task parameter descriptor
3089 RUNFREE(currtpd->parameterArray);
3102 raw_test_pass(0xe999);
3106 /* This function processes an objects tags */
3107 void processtags(struct parameterdescriptor *pd, int index, struct parameterwrapper *parameter, int * iteratorcount, int *statusarray, int numparams) {
3110 for(i=0;i<pd->numbertags;i++) {
3111 int slotid=pd->tagarray[2*i];
3112 int tagid=pd->tagarray[2*i+1];
3114 if (statusarray[slotid+numparams]==0) {
3115 parameter->iterators[*iteratorcount].istag=1;
3116 parameter->iterators[*iteratorcount].tagid=tagid;
3117 parameter->iterators[*iteratorcount].slot=slotid+numparams;
3118 parameter->iterators[*iteratorcount].tagobjectslot=index;
3119 statusarray[slotid+numparams]=1;
3126 void processobject(struct parameterwrapper *parameter, int index, struct parameterdescriptor *pd, int *iteratorcount, int * statusarray, int numparams) {
3129 struct ObjectHash * objectset=((struct parameterwrapper *)pd->queue)->objectset;
3131 parameter->iterators[*iteratorcount].istag=0;
3132 parameter->iterators[*iteratorcount].slot=index;
3133 parameter->iterators[*iteratorcount].objectset=objectset;
3134 statusarray[index]=1;
3136 for(i=0;i<pd->numbertags;i++) {
3137 int slotid=pd->tagarray[2*i];
3138 int tagid=pd->tagarray[2*i+1];
3139 if (statusarray[slotid+numparams]!=0) {
3140 /* This tag has already been enqueued, use it to narrow search */
3141 parameter->iterators[*iteratorcount].tagbindings[tagcount]=slotid+numparams;
3145 parameter->iterators[*iteratorcount].numtags=tagcount;
3150 /* This function builds the iterators for a task & parameter */
3152 void builditerators(struct taskdescriptor * task, int index, struct parameterwrapper * parameter) {
3153 int statusarray[MAXTASKPARAMS];
3155 int numparams=task->numParameters;
3156 int iteratorcount=0;
3157 for(i=0;i<MAXTASKPARAMS;i++) statusarray[i]=0;
3159 statusarray[index]=1; /* Initial parameter */
3160 /* Process tags for initial iterator */
3162 processtags(task->descriptorarray[index], index, parameter, & iteratorcount, statusarray, numparams);
3166 /* Check for objects with existing tags */
3167 for(i=0;i<numparams;i++) {
3168 if (statusarray[i]==0) {
3169 struct parameterdescriptor *pd=task->descriptorarray[i];
3171 for(j=0;j<pd->numbertags;j++) {
3172 int slotid=pd->tagarray[2*j];
3173 if(statusarray[slotid+numparams]!=0) {
3174 processobject(parameter, i, pd, &iteratorcount, statusarray, numparams);
3175 processtags(pd, i, parameter, &iteratorcount, statusarray, numparams);
3182 /* Next do objects w/ unbound tags*/
3184 for(i=0;i<numparams;i++) {
3185 if (statusarray[i]==0) {
3186 struct parameterdescriptor *pd=task->descriptorarray[i];
3187 if (pd->numbertags>0) {
3188 processobject(parameter, i, pd, &iteratorcount, statusarray, numparams);
3189 processtags(pd, i, parameter, &iteratorcount, statusarray, numparams);
3195 /* Nothing with a tag enqueued */
3197 for(i=0;i<numparams;i++) {
3198 if (statusarray[i]==0) {
3199 struct parameterdescriptor *pd=task->descriptorarray[i];
3200 processobject(parameter, i, pd, &iteratorcount, statusarray, numparams);
3201 processtags(pd, i, parameter, &iteratorcount, statusarray, numparams);
3214 #ifdef THREADSIMULATE
3215 int numofcore = pthread_getspecific(key);
3216 for(i=0;i<numtasks[numofcore];i++) {
3217 struct taskdescriptor * task=taskarray[numofcore][i];
3220 if(corenum > NUMCORES - 1) {
3224 for(i=0;i<numtasks[corenum];i++) {
3225 struct taskdescriptor * task=taskarray[corenum][i];
3228 printf("%s\n", task->name);
3230 for(j=0;j<task->numParameters;j++) {
3231 struct parameterdescriptor *param=task->descriptorarray[j];
3232 struct parameterwrapper *parameter=param->queue;
3233 struct ObjectHash * set=parameter->objectset;
3234 struct ObjectIterator objit;
3236 printf(" Parameter %d\n", j);
3238 ObjectHashiterator(set, &objit);
3239 while(ObjhasNext(&objit)) {
3240 struct ___Object___ * obj=(struct ___Object___ *)Objkey(&objit);
3241 struct ___Object___ * tagptr=obj->___tags___;
3242 int nonfailed=Objdata4(&objit);
3243 int numflags=Objdata3(&objit);
3244 int flags=Objdata2(&objit);
3247 printf(" Contains %lx\n", obj);
3248 printf(" flag=%d\n", obj->flag);
3251 } else if (tagptr->type==TAGTYPE) {
3253 printf(" tag=%lx\n",tagptr);
3258 struct ArrayObject *ao=(struct ArrayObject *)tagptr;
3259 for(;tagindex<ao->___cachedCode___;tagindex++) {
3261 printf(" tag=%lx\n",ARRAYGET(ao, struct ___TagDescriptor___*, tagindex));
3271 /* This function processes the task information to create queues for
3272 each parameter type. */
3274 void processtasks() {
3277 if(corenum > NUMCORES - 1) {
3281 #ifdef THREADSIMULATE
3282 int numofcore = pthread_getspecific(key);
3283 for(i=0;i<numtasks[numofcore];i++) {
3284 struct taskdescriptor *task=taskarray[numofcore][i];
3286 for(i=0;i<numtasks[corenum];i++) {
3287 struct taskdescriptor * task=taskarray[corenum][i];
3291 /* Build objectsets */
3292 for(j=0;j<task->numParameters;j++) {
3293 struct parameterdescriptor *param=task->descriptorarray[j];
3294 struct parameterwrapper *parameter=param->queue;
3295 parameter->objectset=allocateObjectHash(10);
3296 parameter->task=task;
3299 /* Build iterators for parameters */
3300 for(j=0;j<task->numParameters;j++) {
3301 struct parameterdescriptor *param=task->descriptorarray[j];
3302 struct parameterwrapper *parameter=param->queue;
3303 builditerators(task, j, parameter);
3308 void toiReset(struct tagobjectiterator * it) {
3311 } else if (it->numtags>0) {
3314 ObjectHashiterator(it->objectset, &it->it);
3318 int toiHasNext(struct tagobjectiterator *it, void ** objectarray OPTARG(int * failed)) {
3321 /* Get object with tags */
3322 struct ___Object___ *obj=objectarray[it->tagobjectslot];
3323 struct ___Object___ *tagptr=obj->___tags___;
3324 if (tagptr->type==TAGTYPE) {
3325 if ((it->tagobjindex==0)&& /* First object */
3326 (it->tagid==((struct ___TagDescriptor___ *)tagptr)->flag)) /* Right tag type */
3331 struct ArrayObject *ao=(struct ArrayObject *) tagptr;
3332 int tagindex=it->tagobjindex;
3333 for(;tagindex<ao->___cachedCode___;tagindex++) {
3334 struct ___TagDescriptor___ *td=ARRAYGET(ao, struct ___TagDescriptor___ *, tagindex);
3335 if (td->flag==it->tagid) {
3336 it->tagobjindex=tagindex; /* Found right type of tag */
3342 } else if (it->numtags>0) {
3343 /* Use tags to locate appropriate objects */
3344 struct ___TagDescriptor___ *tag=objectarray[it->tagbindings[0]];
3345 struct ___Object___ *objptr=tag->flagptr;
3347 if (objptr->type!=OBJECTARRAYTYPE) {
3348 if (it->tagobjindex>0)
3350 if (!ObjectHashcontainskey(it->objectset, (int) objptr))
3352 for(i=1;i<it->numtags;i++) {
3353 struct ___TagDescriptor___ *tag2=objectarray[it->tagbindings[i]];
3354 if (!containstag(objptr,tag2))
3359 struct ArrayObject *ao=(struct ArrayObject *) objptr;
3362 for(tagindex=it->tagobjindex;tagindex<ao->___cachedCode___;tagindex++) {
3363 struct ___Object___ *objptr=ARRAYGET(ao, struct ___Object___*, tagindex);
3364 if (!ObjectHashcontainskey(it->objectset, (int) objptr))
3366 for(i=1;i<it->numtags;i++) {
3367 struct ___TagDescriptor___ *tag2=objectarray[it->tagbindings[i]];
3368 if (!containstag(objptr,tag2))
3371 it->tagobjindex=tagindex;
3376 it->tagobjindex=tagindex;
3380 return ObjhasNext(&it->it);
3384 int containstag(struct ___Object___ *ptr, struct ___TagDescriptor___ *tag) {
3386 struct ___Object___ * objptr=tag->flagptr;
3387 if (objptr->type==OBJECTARRAYTYPE) {
3388 struct ArrayObject *ao=(struct ArrayObject *)objptr;
3389 for(j=0;j<ao->___cachedCode___;j++) {
3390 if (ptr==ARRAYGET(ao, struct ___Object___*, j))
3398 void toiNext(struct tagobjectiterator *it , void ** objectarray OPTARG(int * failed)) {
3399 /* hasNext has all of the intelligence */
3402 /* Get object with tags */
3403 struct ___Object___ *obj=objectarray[it->tagobjectslot];
3404 struct ___Object___ *tagptr=obj->___tags___;
3405 if (tagptr->type==TAGTYPE) {
3407 objectarray[it->slot]=tagptr;
3409 struct ArrayObject *ao=(struct ArrayObject *) tagptr;
3410 objectarray[it->slot]=ARRAYGET(ao, struct ___TagDescriptor___ *, it->tagobjindex++);
3412 } else if (it->numtags>0) {
3413 /* Use tags to locate appropriate objects */
3414 struct ___TagDescriptor___ *tag=objectarray[it->tagbindings[0]];
3415 struct ___Object___ *objptr=tag->flagptr;
3416 if (objptr->type!=OBJECTARRAYTYPE) {
3418 objectarray[it->slot]=objptr;
3420 struct ArrayObject *ao=(struct ArrayObject *) objptr;
3421 objectarray[it->slot]=ARRAYGET(ao, struct ___Object___ *, it->tagobjindex++);
3424 /* Iterate object */
3425 objectarray[it->slot]=(void *)Objkey(&it->it);