Robust/src/Runtime/DSTM/interface/addPrefetchEnhance.c

   1 #include "addPrefetchEnhance.h"
   2 #include "prelookup.h"
   3
   4 extern int numprefetchsites; // Number of prefetch sites
   5 extern pfcstats_t *evalPrefetch; //Global array that keeps track of operation mode (ON/OFF) for each prefetch site
   6 extern objstr_t *prefetchcache; //Global Prefetch cache
   7 extern pthread_mutex_t prefetchcache_mutex; //Mutex to lock Prefetch Cache
   8 extern unsigned int myIpAddr;
   9
  10 /* This function creates and initializes the
  11  * evalPrefetch global array */
  12 pfcstats_t *initPrefetchStats() {
  13   pfcstats_t *ptr;
  14   if((ptr = calloc(numprefetchsites, sizeof(pfcstats_t))) == NULL) {
  15     printf("%s() Calloc error in %s at line %d\n", __func__, __FILE__, __LINE__);
  16     return NULL;
  17   }
  18   int i;
  19   /* Enable prefetching at the beginning */
  20   for(i=0; i<numprefetchsites; i++) {
  21     ptr[i].operMode = 1;
  22     ptr[i].callcount = 0;
  23     ptr[i].retrycount = RETRYINTERVAL; //N
  24     ptr[i].uselesscount = SHUTDOWNINTERVAL; //M
  25   }
  26   return ptr;
  27 }
  28
  29 int getRetryCount(int siteid) {
  30   return evalPrefetch[siteid].retrycount;
  31 }
  32
  33 int getUselessCount(int siteid) {
  34   return evalPrefetch[siteid].uselesscount;
  35 }
  36
  37 char getOperationMode(int siteid) {
  38   return evalPrefetch[siteid].operMode;
  39 }
  40
  41 /* This function updates counters and mode of operation of a
  42  * prefetch site during runtime. When the prefetch call at a site
  43  * generates oids that are found/not found in the prefetch cache,
  44  * we take action accordingly */
  45 void handleDynPrefetching(int numLocal, int ntuples, int siteid) {
  46   if(numLocal < ntuples) {
  47     /* prefetch not found locally(miss in cache) */
  48     evalPrefetch[siteid].operMode = 1;
  49     evalPrefetch[siteid].uselesscount = SHUTDOWNINTERVAL;
  50   } else {
  51     if(getOperationMode(siteid) != 0) {
  52       evalPrefetch[siteid].uselesscount--;
  53       if(evalPrefetch[siteid].uselesscount <= 0) {
  54         evalPrefetch[siteid].operMode = 0;
  55       }
  56     }
  57   }
  58 }
  59
  60 #if 1
  61 /* This function clears from prefetch cache those
  62  * entries that caused a transaction abort */
  63 void cleanPCache(transrecord_t *record) {
  64   transrecord_t *rec = record;
  65   unsigned int size = rec->lookupTable->size;
  66   chashlistnode_t *ptr = rec->lookupTable->table;
  67   int i;
  68   for(i = 0; i < size; i++) {
  69     chashlistnode_t *curr = &ptr[i]; //for each entry in the cache lookupTable
  70     while(curr != NULL) {
  71       if(curr->key == 0)
  72         break;
  73       objheader_t *header1, *header2;
  74       /* Not found in local machine's object store and found in prefetch cache */
  75       if((header1 = mhashSearch(curr->key)) == NULL && ((header2 = prehashSearch(curr->key)) != NULL)) {
  76         /* Remove from prefetch cache */
  77         prehashRemove(curr->key);
  78       }
  79       curr = curr->next;
  80     }
  81   }
  82 }
  83 #else
  84 /* This function clears from prefetch cache those
  85  * entries that caused a transaction abort */
  86 void cleanPCache(transrecord_t *record) {
  87   transrecord_t *rec = record;
  88   unsigned int size = rec->lookupTable->size;
  89   struct chashentry *ptr = rec->lookupTable->table;
  90   int i;
  91   for(i = 0; i < size; i++) {
  92     struct chashentry *curr = &ptr[i]; //for each entry in the cache lookupTable
  93     if(curr->key == 0)
  94       continue;
  95     objheader_t *header1, *header2;
  96     /* Not found in local machine's object store and found in prefetch cache */
  97     if((header1 = mhashSearch(curr->key)) == NULL && ((header2 = prehashSearch(curr->key)) != NULL)) {
  98       /* Remove from prefetch cache */
  99       prehashRemove(curr->key);
 100     }
 101   }
 102 }
 103 #endif
 104
 105 /* This function updates the prefetch cache with
 106  * entries from the transaction cache when a
 107  * transaction commits
 108  * Return -1 on error else returns 0 */
 109 int updatePrefetchCache(trans_req_data_t *tdata, transrecord_t *rec) {
 110   int retval;
 111   char oidType;
 112   oidType = 'R';
 113   if(tdata->f.numread > 0) {
 114     if((retval = copyToCache(tdata->f.numread, (unsigned int *)(tdata->objread), rec, oidType)) != 0) {
 115       printf("%s(): Error in copying objects read at %s, %d\n", __func__, __FILE__, __LINE__);
 116       return -1;
 117     }
 118   }
 119   if(tdata->f.nummod > 0) {
 120     oidType = 'M';
 121     if((retval = copyToCache(tdata->f.nummod, tdata->oidmod, rec, oidType)) != 0) {
 122       printf("%s(): Error in copying objects read at %s, %d\n", __func__, __FILE__, __LINE__);
 123       return -1;
 124     }
 125   }
 126   return 0;
 127 }
 128
 129 int copyToCache(int numoid, unsigned int *oidarray, transrecord_t *rec, char oidType) {
 130   int i;
 131   for (i = 0; i < numoid; i++) {
 132     unsigned int oid;
 133     if(oidType == 'R') {
 134       char * objread = (char *) oidarray;
 135       oid = *((unsigned int *)(objread+(sizeof(unsigned int)+
 136                                         sizeof(unsigned short))*i));
 137     } else {
 138       oid = oidarray[i];
 139     }
 140     pthread_mutex_lock(&prefetchcache_mutex);
 141     objheader_t * header;
 142     if((header = (objheader_t *) chashSearch(rec->lookupTable, oid)) == NULL) {
 143       printf("%s() obj %x is no longer in transaction cache at %s , %d\n", __func__, oid,__FILE__, __LINE__);
 144       fflush(stdout);
 145       return -1;
 146     }
 147     //copy into prefetch cache
 148     int size;
 149     GETSIZE(size, header);
 150     objheader_t * newAddr;
 151     if((newAddr = prefetchobjstrAlloc(size + sizeof(objheader_t))) == NULL) {
 152       printf("%s(): Error in getting memory from prefetch cache at %s, %d\n", __func__,
 153              __FILE__, __LINE__);
 154       pthread_mutex_unlock(&prefetchcache_mutex);
 155       return -1;
 156     }
 157     pthread_mutex_unlock(&prefetchcache_mutex);
 158     memcpy(newAddr, header, size+sizeof(objheader_t));
 159     //Increment version for every modified object
 160     if(oidType == 'M') {
 161       newAddr->version += 1;
 162       newAddr->notifylist = NULL;
 163     }
 164     //make an entry in prefetch lookup hashtable
 165     void *oldptr;
 166     if((oldptr = prehashSearch(oid)) != NULL) {
 167       prehashRemove(oid);
 168       prehashInsert(oid, newAddr);
 169     } else {
 170       prehashInsert(oid, newAddr);
 171     }
 172   } //end of for
 173   return 0;
 174 }