src/conv-core/conv-conds.c

   1 #include <stdio.h>
   2 #include <stdlib.h>
   3 #include <math.h>
   4
   5 #include "converse.h"
   6
   7 /**
   8  * Structure to hold the requisites for a callback
   9  */
  10 typedef struct _ccd_callback {
  11   CcdVoidFn fn;
  12   void *arg;
  13   int pe;                       /* the pe that sets the callback */
  14 } ccd_callback;
  15
  16
  17
  18 /**
  19  * An element (a single callback) in a list of callbacks
  20  */
  21 typedef struct _ccd_cblist_elem {
  22   ccd_callback cb;
  23   short int next;
  24   short int prev;
  25 } ccd_cblist_elem;
  26
  27
  28
  29 /**
  30  * A list of callbacks stored as an array and handled like a list
  31  */
  32 typedef struct _ccd_cblist {
  33   unsigned short int maxlen;
  34   unsigned short int len;
  35   short int first, last;
  36   short int first_free;
  37   unsigned char flag;
  38   ccd_cblist_elem *elems;
  39 } ccd_cblist;
  40
  41
  42
  43 /** Initialize a list of callbacks. Alloc memory, set counters etc. */
  44 static void init_cblist(ccd_cblist *l, unsigned int ml)
  45 {
  46   int i;
  47   l->elems = (ccd_cblist_elem*) malloc(ml*sizeof(ccd_cblist_elem));
  48   _MEMCHECK(l->elems);
  49   for(i=0;i<ml;i++) {
  50     l->elems[i].next = i+1;
  51     l->elems[i].prev = i-1;
  52   }
  53   l->elems[ml-1].next = -1;
  54   l->len = 0;
  55   l->maxlen = ml;
  56   l->first = l->last = -1;
  57   l->first_free = 0;
  58   l->flag = 0;
  59 }
  60
  61
  62
  63 /** Expand the callback list to a max length of ml */
  64 static void expand_cblist(ccd_cblist *l, unsigned int ml)
  65 {
  66   ccd_cblist_elem *old_elems = l->elems;
  67   int i = 0;
  68   l->elems = (ccd_cblist_elem*) malloc(ml*sizeof(ccd_cblist_elem));
  69   _MEMCHECK(l->elems);
  70   for(i=0;i<(l->len);i++)
  71     l->elems[i] = old_elems[i];
  72   free(old_elems);
  73   for(i=l->len;i<ml;i++) {
  74     l->elems[i].next = i+1;
  75     l->elems[i].prev = i-1;
  76   }
  77   l->elems[ml-1].next = -1;
  78   l->elems[l->len].prev = -1;
  79   l->maxlen = ml;
  80   l->first_free = l->len;
  81 }
  82
  83
  84
  85 /** Remove element referred to by given list index idx. */
  86 static void remove_elem(ccd_cblist *l, int idx)
  87 {
  88   ccd_cblist_elem *e = l->elems;
  89   /* remove lidx from the busy list */
  90   if(e[idx].next != (-1))
  91     e[e[idx].next].prev = e[idx].prev;
  92   if(e[idx].prev != (-1))
  93     e[e[idx].prev].next = e[idx].next;
  94   if(idx==(l->first))
  95     l->first = e[idx].next;
  96   if(idx==(l->last))
  97     l->last = e[idx].prev;
  98   /* put lidx in the free list */
  99   e[idx].prev = -1;
 100   e[idx].next = l->first_free;
 101   if(e[idx].next != (-1))
 102     e[e[idx].next].prev = idx;
 103   l->first_free = idx;
 104   l->len--;
 105 }
 106
 107
 108
 109 /** Remove n elements from the beginning of the list. */
 110 static void remove_n_elems(ccd_cblist *l, int n)
 111 {
 112   int i;
 113   if(n==0 || (l->len < n))
 114     return;
 115   for(i=0;i<n;i++) {
 116     remove_elem(l, l->first);
 117   }
 118 }
 119
 120
 121
 122 /** Append callback to the given cblist, and return the index. */
 123 static int append_elem(ccd_cblist *l, CcdVoidFn fn, void *arg, int pe)
 124 {
 125   int idx;
 126   ccd_cblist_elem *e;
 127   if(l->len == l->maxlen)
 128     expand_cblist(l, l->maxlen*2);
 129   idx = l->first_free;
 130   e = l->elems;
 131   l->first_free = e[idx].next;
 132   e[idx].next = -1;
 133   e[idx].prev = l->last;
 134   if(l->first == (-1))
 135     l->first = idx;
 136   if(l->last != (-1))
 137     e[l->last].next = idx;
 138   l->last = idx;
 139   e[idx].cb.fn = fn;
 140   e[idx].cb.arg = arg;
 141   e[idx].cb.pe = pe;
 142   l->len++;
 143   return idx;
 144 }
 145
 146
 147
 148 /**
 149  * Trigger the callbacks in the provided callback list and *retain* them
 150  * after they are called.
 151  *
 152  * Callbacks that are added after this function is started (e.g. callbacks
 153  * registered from other callbacks) are ignored.
 154  * @note: It is illegal to cancel callbacks from within ccd callbacks.
 155  */
 156 static void call_cblist_keep(ccd_cblist *l,double curWallTime)
 157 {
 158   int i, len = l->len, idx;
 159   for(i=0, idx=l->first;i<len;i++) {
 160     int old = CmiSwitchToPE(l->elems[idx].cb.pe);
 161     (*(l->elems[idx].cb.fn))(l->elems[idx].cb.arg,curWallTime);
 162     int unused = CmiSwitchToPE(old);
 163     idx = l->elems[idx].next;
 164   }
 165 }
 166
 167
 168
 169 /**
 170  * Trigger the callbacks in the provided callback list and *remove* them
 171  * from the list after they are called.
 172  *
 173  * Callbacks that are added after this function is started (e.g. callbacks
 174  * registered from other callbacks) are ignored.
 175  * @note: It is illegal to cancel callbacks from within ccd callbacks.
 176  */
 177 static void call_cblist_remove(ccd_cblist *l,double curWallTime)
 178 {
 179   int i, len = l->len, idx;
 180   /* reentrant */
 181   if (l->flag) return;
 182   l->flag = 1;
 183 #if ! CMK_BIGSIM_CHARM
 184   for(i=0, idx=l->first;i<len;i++) {
 185     int old = CmiSwitchToPE(l->elems[idx].cb.pe);
 186     (*(l->elems[idx].cb.fn))(l->elems[idx].cb.arg,curWallTime);
 187     int unused = CmiSwitchToPE(old);
 188     idx = l->elems[idx].next;
 189   }
 190 #else
 191   for(i=0, idx=l->last;i<len;i++) {
 192     int old = CmiSwitchToPE(l->elems[idx].cb.pe);
 193     (*(l->elems[idx].cb.fn))(l->elems[idx].cb.arg,curWallTime);
 194     int unused = CmiSwitchToPE(old);
 195     idx = l->elems[idx].prev;
 196   }
 197 #endif
 198   remove_n_elems(l,len);
 199   l->flag = 0;
 200 }
 201
 202
 203
 204 #define CBLIST_INIT_LEN   8
 205 #define MAXNUMCONDS       (CcdUSERMAX + 1)
 206
 207 /**
 208  * Lists of conditional callbacks that are maintained by the scheduler
 209  */
 210 typedef struct {
 211   ccd_cblist condcb[MAXNUMCONDS];
 212   ccd_cblist condcb_keep[MAXNUMCONDS];
 213 } ccd_cond_callbacks;
 214
 215 /***/
 216 CpvStaticDeclare(ccd_cond_callbacks, conds);
 217
 218
 219 // Default resolution of .005 seconds aka 5 milliseconds
 220 #define CCD_DEFAULT_RESOLUTION 5.0e-3
 221
 222 /*Make sure this matches the CcdPERIODIC_* list in converse.h*/
 223 #define CCD_PERIODIC_MAX 13
 224 const static double periodicCallInterval[CCD_PERIODIC_MAX]=
 225 {0.001, 0.010, 0.100, 1.0, 5.0, 10.0, 60.0, 2*60.0, 5*60.0, 10*60.0, 3600.0, 12*3600.0, 24*3600.0};
 226
 227 /**
 228  * List of periodic callbacks maintained by the scheduler
 229  */
 230 typedef struct {
 231         int nSkip;/*Number of opportunities to skip*/
 232         double lastCheck;/*Time of last check*/
 233         double resolution;
 234         double nextCall[CCD_PERIODIC_MAX];
 235 } ccd_periodic_callbacks;
 236
 237 /** */
 238 CpvStaticDeclare(ccd_periodic_callbacks, pcb);
 239 CpvDeclare(int, _ccd_numchecks);
 240
 241
 242
 243 #define MAXTIMERHEAPENTRIES       128
 244
 245 /**
 246  * Structure used to manage callbacks in a heap
 247  */
 248 typedef struct {
 249     double time;
 250     ccd_callback cb;
 251 } ccd_heap_elem;
 252
 253
 254 /* Note : The heap is only stored in elements ccd_heap[0] to
 255  * ccd_heap[ccd_heaplen]
 256  */
 257
 258 /** An array of time-scheduled callbacks managed as a heap */
 259 CpvStaticDeclare(ccd_heap_elem*, ccd_heap);
 260 /** The length of the callback heap */
 261 CpvStaticDeclare(int, ccd_heaplen);
 262 /** The max allowed length of the callback heap */
 263 CpvStaticDeclare(int, ccd_heapmaxlen);
 264
 265
 266
 267 /** Swap two elements on the heap */
 268 static void ccd_heap_swap(int index1, int index2)
 269 {
 270   ccd_heap_elem *h = CpvAccess(ccd_heap);
 271   ccd_heap_elem temp;
 272
 273   temp = h[index1];
 274   h[index1] = h[index2];
 275   h[index2] = temp;
 276 }
 277
 278
 279
 280 /**
 281  * Expand the ccd_heap to make more room.
 282  *
 283  * Double the heap size and copy everything over. Initial 128 is reasonably
 284  * big, so expanding won't happen often.
 285  *
 286  * Had a bug previously due to late expansion, should work now - Gengbin 12/4/03
 287 */
 288 static void expand_ccd_heap(void)
 289 {
 290   int i;
 291   int oldlen = CpvAccess(ccd_heapmaxlen);
 292   int newlen = oldlen*2;
 293   ccd_heap_elem *newheap;
 294
 295   CmiPrintf("[%d] Warning: ccd_heap expand from %d to %d\n", CmiMyPe(),oldlen, newlen);
 296
 297   newheap = (ccd_heap_elem*) malloc(sizeof(ccd_heap_elem)*2*(newlen+1));
 298   _MEMCHECK(newheap);
 299   /* need to copy the second half part ??? */
 300   for (i=0; i<=oldlen; i++) {
 301     newheap[i] = CpvAccess(ccd_heap)[i];
 302     newheap[i+newlen] = CpvAccess(ccd_heap)[i+oldlen];
 303   }
 304   free(CpvAccess(ccd_heap));
 305   CpvAccess(ccd_heap) = newheap;
 306   CpvAccess(ccd_heapmaxlen) = newlen;
 307 }
 308
 309
 310
 311 /**
 312  * Insert a new callback into the heap
 313  */
 314 static void ccd_heap_insert(double t, CcdVoidFn fnp, void *arg, int pe)
 315 {
 316   int child, parent;
 317   ccd_heap_elem *h;
 318
 319   if(CpvAccess(ccd_heaplen) >= CpvAccess(ccd_heapmaxlen)) {
 320 /* CmiAbort("Heap overflow (InsertInHeap), exiting...\n"); */
 321     expand_ccd_heap();
 322   }
 323
 324   h = CpvAccess(ccd_heap);
 325
 326   {
 327     ccd_heap_elem *e = &(h[++CpvAccess(ccd_heaplen)]);
 328     e->time = t;
 329     e->cb.fn = fnp;
 330     e->cb.arg = arg;
 331     e->cb.pe = pe;
 332     child  = CpvAccess(ccd_heaplen);
 333     parent = child / 2;
 334     while((parent>0) && (h[child].time<h[parent].time)) {
 335             ccd_heap_swap(child, parent);
 336             child  = parent;
 337             parent = parent / 2;
 338     }
 339   }
 340 }
 341
 342
 343
 344 /**
 345  * Remove the top of the heap
 346  */
 347 static void ccd_heap_remove(void)
 348 {
 349   int parent,child;
 350   ccd_heap_elem *h = CpvAccess(ccd_heap);
 351
 352   parent = 1;
 353   if(CpvAccess(ccd_heaplen)>0) {
 354     /* put deleted value at end of heap */
 355     ccd_heap_swap(1,CpvAccess(ccd_heaplen));
 356     CpvAccess(ccd_heaplen)--;
 357     if(CpvAccess(ccd_heaplen)) {
 358       /* if any left, then bubble up values */
 359             child = 2 * parent;
 360             while(child <= CpvAccess(ccd_heaplen)) {
 361               if(((child + 1) <= CpvAccess(ccd_heaplen))  &&
 362                        (h[child].time > h[child+1].time))
 363                 child++; /* use the smaller of the two */
 364               if(h[parent].time <= h[child].time)
 365                       break;
 366               ccd_heap_swap(parent,child);
 367               parent  = child;      /* go down the tree one more step */
 368               child  = 2 * child;
 369       }
 370     }
 371   }
 372 }
 373
 374
 375
 376 /**
 377  * Identify any (over)due callbacks that were scheduled
 378  * and trigger them.
 379  */
 380 static void ccd_heap_update(double curWallTime)
 381 {
 382   ccd_heap_elem *h = CpvAccess(ccd_heap);
 383   ccd_heap_elem *e = h+CpvAccess(ccd_heapmaxlen);
 384   int i,ne=0;
 385   /* Pull out all expired heap entries */
 386   while ((CpvAccess(ccd_heaplen)>0) && (h[1].time<curWallTime)) {
 387     e[ne++]=h[1];
 388     ccd_heap_remove();
 389   }
 390   /* Now execute those heap entries.  This must be
 391      separated from the removal phase because executing
 392      an entry may change the heap.
 393   */
 394   for (i=0;i<ne;i++) {
 395 /*
 396       ccd_heap_elem *h = CpvAccess(ccd_heap);
 397       ccd_heap_elem *e = h+CpvAccess(ccd_heapmaxlen);
 398 */
 399       int old = CmiSwitchToPE(e[i].cb.pe);
 400       (*(e[i].cb.fn))(e[i].cb.arg,curWallTime);
 401       int unused = CmiSwitchToPE(old);
 402   }
 403 }
 404
 405
 406
 407 void CcdCallBacksReset(void *ignored,double curWallTime);
 408
 409 /**
 410  * Initialize the callback containers
 411  */
 412 void CcdModuleInit(char **ignored)
 413 {
 414    int i;
 415    double curTime;
 416    CpvInitialize(ccd_heap_elem*, ccd_heap);
 417    CpvInitialize(ccd_cond_callbacks, conds);
 418    CpvInitialize(ccd_periodic_callbacks, pcb);
 419    CpvInitialize(int, ccd_heaplen);
 420    CpvInitialize(int, ccd_heapmaxlen);
 421    CpvInitialize(int, _ccd_numchecks);
 422
 423    CpvAccess(ccd_heaplen) = 0;
 424    CpvAccess(ccd_heapmaxlen) = MAXTIMERHEAPENTRIES;
 425    CpvAccess(ccd_heap) =
 426      (ccd_heap_elem*) malloc(sizeof(ccd_heap_elem)*2*(MAXTIMERHEAPENTRIES + 1));
 427    _MEMCHECK(CpvAccess(ccd_heap));
 428    for(i=0;i<MAXNUMCONDS;i++) {
 429      init_cblist(&(CpvAccess(conds).condcb[i]), CBLIST_INIT_LEN);
 430      init_cblist(&(CpvAccess(conds).condcb_keep[i]), CBLIST_INIT_LEN);
 431    }
 432    CpvAccess(_ccd_numchecks) = 1;
 433    CpvAccess(pcb).nSkip = 1;
 434    curTime=CmiWallTimer();
 435    CpvAccess(pcb).lastCheck = curTime;
 436    for (i=0;i<CCD_PERIODIC_MAX;i++)
 437            CpvAccess(pcb).nextCall[i]=curTime+periodicCallInterval[i];
 438    CpvAccess(pcb).resolution = CCD_DEFAULT_RESOLUTION;
 439    CcdCallOnConditionKeep(CcdPROCESSOR_BEGIN_IDLE,CcdCallBacksReset,0);
 440    CcdCallOnConditionKeep(CcdPROCESSOR_END_IDLE,CcdCallBacksReset,0);
 441 }
 442
 443
 444
 445 /**
 446  * Register a callback function that will be triggered when the specified
 447  * condition is raised the next time
 448  */
 449 int CcdCallOnCondition(int condnum, CcdVoidFn fnp, void *arg)
 450 {
 451   CmiAssert(condnum < MAXNUMCONDS);
 452   return append_elem(&(CpvAccess(conds).condcb[condnum]), fnp, arg, CcdIGNOREPE);
 453 }
 454
 455 /**
 456  * Register a callback function that will be triggered on the specified PE
 457  * when the specified condition is raised the next time
 458  */
 459 int CcdCallOnConditionOnPE(int condnum, CcdVoidFn fnp, void *arg, int pe)
 460 {
 461   CmiAssert(condnum < MAXNUMCONDS);
 462   return append_elem(&(CpvAccess(conds).condcb[condnum]), fnp, arg, pe);
 463 }
 464
 465 /**
 466  * Register a callback function that will be triggered *whenever* the specified
 467  * condition is raised
 468  */
 469 int CcdCallOnConditionKeep(int condnum, CcdVoidFn fnp, void *arg)
 470 {
 471   CmiAssert(condnum < MAXNUMCONDS);
 472   return append_elem(&(CpvAccess(conds).condcb_keep[condnum]), fnp, arg, CcdIGNOREPE);
 473 }
 474
 475 /**
 476  * Register a callback function that will be triggered on the specified PE
 477  * *whenever* the specified condition is raised
 478  */
 479 int CcdCallOnConditionKeepOnPE(int condnum, CcdVoidFn fnp, void *arg, int pe)
 480 {
 481   CmiAssert(condnum < MAXNUMCONDS);
 482   return append_elem(&(CpvAccess(conds).condcb_keep[condnum]), fnp, arg, pe);
 483 }
 484
 485
 486 /**
 487  * Cancel a previously registered conditional callback
 488  */
 489 void CcdCancelCallOnCondition(int condnum, int idx)
 490 {
 491   CmiAssert(condnum < MAXNUMCONDS);
 492   remove_elem(&(CpvAccess(conds).condcb[condnum]), idx);
 493 }
 494
 495
 496 /**
 497  * Cancel a previously registered conditional callback
 498  */
 499 void CcdCancelCallOnConditionKeep(int condnum, int idx)
 500 {
 501   CmiAssert(condnum < MAXNUMCONDS);
 502   remove_elem(&(CpvAccess(conds).condcb_keep[condnum]), idx);
 503 }
 504
 505
 506 /**
 507  * Register a callback function that will be triggered on the specified PE
 508  * after a minimum delay of deltaT
 509  */
 510 void CcdCallFnAfterOnPE(CcdVoidFn fnp, void *arg, double deltaT, int pe)
 511 {
 512     double ctime  = CmiWallTimer();
 513     double tcall = ctime + deltaT/1000.0;
 514     ccd_heap_insert(tcall, fnp, arg, pe);
 515 }
 516
 517 /**
 518  * Register a callback function that will be triggered after a minimum
 519  * delay of deltaT
 520  */
 521 void CcdCallFnAfter(CcdVoidFn fnp, void *arg, double deltaT)
 522 {
 523     CcdCallFnAfterOnPE(fnp, arg, deltaT, CcdIGNOREPE);
 524 }
 525
 526
 527 /**
 528  * Raise a condition causing all registered callbacks corresponding to
 529  * that condition to be triggered
 530  */
 531 void CcdRaiseCondition(int condnum)
 532 {
 533   CmiAssert(condnum < MAXNUMCONDS);
 534   double curWallTime=CmiWallTimer();
 535   call_cblist_remove(&(CpvAccess(conds).condcb[condnum]),curWallTime);
 536   call_cblist_keep(&(CpvAccess(conds).condcb_keep[condnum]),curWallTime);
 537 }
 538
 539
 540 /**
 541  * Internal helper function that updates the polling resolution for time
 542  * based callbacks to minimum of two arguments and ensures appropriate
 543  * counters etc are reset
 544  */
 545 double CcdSetMinResolution(double newResolution, double minResolution) {
 546   ccd_periodic_callbacks* o = &CpvAccess(pcb);
 547   double oldResolution = o->resolution;
 548
 549   o->resolution = fmin(newResolution, minResolution);
 550
 551   // Ensure we don't miss the new quantum
 552   if (o->resolution < oldResolution) {
 553     CcdCallBacksReset(NULL, CmiWallTimer());
 554   }
 555
 556   return oldResolution;
 557 }
 558
 559 /**
 560  * Set the polling resolution for time based callbacks
 561  */
 562 double CcdSetResolution(double newResolution) {
 563   return CcdSetMinResolution(newResolution, CCD_DEFAULT_RESOLUTION);
 564 }
 565
 566 /**
 567  * Reset the polling resolution for time based callbacks to its default value
 568  */
 569 double CcdResetResolution() {
 570   ccd_periodic_callbacks* o = &CpvAccess(pcb);
 571   double oldResolution = o->resolution;
 572
 573   o->resolution = CCD_DEFAULT_RESOLUTION;
 574
 575   return oldResolution;
 576 }
 577
 578 /**
 579  * "Safe" operation that only ever increases the polling resolution for time
 580  * based callbacks
 581  */
 582 double CcdIncreaseResolution(double newResolution) {
 583   return CcdSetMinResolution(newResolution, CpvAccess(pcb).resolution);
 584 }
 585
 586 /*
 587  * Trigger callbacks periodically, and also the time-indexed
 588  * functions if their time has arrived
 589  */
 590 void CcdCallBacks(void)
 591 {
 592   int i;
 593   ccd_periodic_callbacks *o=&CpvAccess(pcb);
 594
 595   /* Figure out how many times to skip Ccd processing */
 596   double curWallTime = CmiWallTimer();
 597
 598   unsigned int nSkip=o->nSkip;
 599 #if 1
 600 /* Dynamically adjust the number of messages to skip */
 601   double elapsed = curWallTime - o->lastCheck;
 602   // Adjust the number of skipped messages by a multiple between .5, if we
 603   // skipped too many messages last time, and 2, if we skipped too few.
 604   // Ideally elapsed = resolution and we keep nSkip the same i.e. multiply by 1
 605   if (elapsed > 0.0) {
 606     nSkip = (int)(nSkip * fmax(0.5, fmin(2.0, o->resolution / elapsed)));
 607   }
 608
 609 /* Keep skipping within a sensible range */
 610 #define minSkip 1u
 611 #define maxSkip 20u
 612   if (nSkip<minSkip) nSkip=minSkip;
 613   else if (nSkip>maxSkip) nSkip=maxSkip;
 614 #else
 615 /* Always skip a fixed number of messages */
 616   nSkip=1;
 617 #endif
 618
 619   CpvAccess(_ccd_numchecks)=o->nSkip=nSkip;
 620   o->lastCheck=curWallTime;
 621
 622   ccd_heap_update(curWallTime);
 623
 624   for (i=0;i<CCD_PERIODIC_MAX;i++)
 625     if (o->nextCall[i]<=curWallTime) {
 626       CcdRaiseCondition(CcdPERIODIC+i);
 627       o->nextCall[i]=curWallTime+periodicCallInterval[i];
 628     }
 629     else
 630       break; /*<- because intervals are multiples of one another*/
 631 }
 632
 633
 634
 635 /**
 636  * Called when something drastic changes-- restart ccd_num_checks
 637  */
 638 void CcdCallBacksReset(void *ignored,double curWallTime)
 639 {
 640   ccd_periodic_callbacks *o=&CpvAccess(pcb);
 641   CpvAccess(_ccd_numchecks)=o->nSkip=1;
 642   o->lastCheck=curWallTime;
 643 }
 644
 645