Robust/TransSim/FlexScheduler.java

   1 import java.util.*;
   2
   3 public class FlexScheduler extends Thread {
   4   Executor e;
   5   int abortThreshold;
   6   int abortRatio;
   7   int deadlockcount;
   8   int checkdepth;
   9   int barriercount;
  10
  11   public FlexScheduler(Executor e, int policy, int abortThreshold, int abortRatio, int checkdepth, Plot p) {
  12     this(e, policy, p);
  13     this.abortThreshold=abortThreshold;
  14     this.abortRatio=abortRatio;
  15     this.checkdepth=checkdepth;
  16   }
  17
  18   public void run() {
  19     dosim();
  20   }
  21
  22   public FlexScheduler(Executor e, int policy, Plot p) {
  23     this.e=e;
  24     barriercount=e.numThreads();
  25     aborted=new boolean[e.numThreads()];
  26     currentevents=new Event[e.numThreads()];
  27     rdobjmap=new Hashtable();
  28     wrobjmap=new Hashtable();
  29     this.policy=policy;
  30     r=new Random(100);
  31     eq=new PriorityQueue();
  32     backoff=new int[e.numThreads()];
  33     retrycount=new int[e.numThreads()];
  34     transferred=new int[e.numThreads()];
  35     objtoinfo=new Hashtable();
  36     threadinfo=new ThreadInfo[e.numThreads()];
  37     blocked=new boolean[e.numThreads()];
  38
  39     for(int i=0;i<e.numThreads();i++) {
  40       backoff[i]=BACKOFFSTART;
  41       threadinfo[i]=new ThreadInfo(this);
  42     }
  43     this.p=p;
  44     if (p!=null) {
  45       serCommit=p.getSeries("COMMIT");
  46       serStart=p.getSeries("START");
  47       serAbort=p.getSeries("ABORT");
  48       serStall=p.getSeries("STALL");
  49       serWake=p.getSeries("WAKE");
  50       serAvoid=p.getSeries("AVOIDDEADLOCK");
  51     }
  52   }
  53
  54   Plot p;
  55   Series serCommit;
  56   Series serStart;
  57   Series serAbort;
  58   Series serStall;
  59   Series serAvoid;
  60   Series serWake;
  61
  62   public int getDeadLockCount() {
  63     return deadlockcount;
  64   }
  65
  66   //Where to start the backoff delay at
  67   public static final int BACKOFFSTART=1;
  68
  69   //Commit options
  70   public static final int LAZY=0;
  71   public static final int COMMIT=1;
  72   public static final int ATTACK=2;
  73   public static final int SUICIDE=3;
  74   public static final int TIMESTAMP=4;
  75   public static final int LOCK=5;
  76   public static final int LOCKCOMMIT=6;
  77   public static final int RANDOM=7;
  78   public static final int KARMA=8;
  79   public static final int POLITE=9;
  80   public static final int ERUPTION=10;
  81   public static final int THREAD=11;
  82   public static final int ATTACKTIME=12;
  83   public static final int ATTACKTHREAD=13;
  84
  85   PriorityQueue eq;
  86   int policy;
  87   boolean[] aborted;
  88   long shorttesttime;
  89   long starttime=-1;
  90   Hashtable rdobjmap;
  91   Hashtable wrobjmap;
  92   int abortcount;
  93   int commitcount;
  94   long backoffcycles;
  95   long stallcycles;
  96   long abortedcycles;
  97   Event[] currentevents;
  98   Random r;
  99   int[] backoff;
 100   int[] retrycount;
 101   int[] transferred;
 102   Hashtable objtoinfo;
 103   ThreadInfo[] threadinfo;
 104
 105   boolean[] blocked;
 106
 107   public boolean isEager() {
 108     return policy==ATTACK||policy==SUICIDE||policy==TIMESTAMP||policy==RANDOM||policy==KARMA||policy==POLITE||policy==ERUPTION||policy==THREAD||policy==ATTACKTIME||policy==ATTACKTHREAD;
 109   }
 110
 111   public boolean countObjects() {
 112     return policy==KARMA||policy==ERUPTION;
 113   }
 114
 115   public boolean isLock() {
 116     return policy==LOCK||policy==LOCKCOMMIT;
 117   }
 118
 119   public int getAborts() {
 120     return abortcount;
 121   }
 122
 123   public int getCommits() {
 124     return commitcount;
 125   }
 126
 127   public long getTime() {
 128     return shorttesttime-starttime;
 129   }
 130
 131   public long getStallTime() {
 132     return stallcycles;
 133   }
 134
 135   public long getBackoffTime() {
 136     return backoffcycles;
 137   }
 138
 139   public long getAbortedTime() {
 140     return abortedcycles;
 141   }
 142
 143   //Computes wasted time
 144   public void timewasted(int currthread, long currtime) {
 145     Event e=currentevents[currthread];
 146     Transaction trans=e.getTransaction();
 147     int eIndex=e.getEvent();
 148     long eTime=e.getTime();
 149     long timeleft=eTime-currtime;
 150     long totaltime=0;
 151     for(int i=0;i<=eIndex;i++)
 152       totaltime+=trans.getTime(i);
 153     totaltime-=timeleft;//subtract off time to the next event
 154     abortedcycles+=totaltime;
 155   }
 156
 157   //Aborts another thread...
 158   public void reschedule(int currthread, long currtime, long backofftime) {
 159     long time=currtime+backofftime;
 160     backoffcycles+=backofftime;
 161     currentevents[currthread].makeInvalid();
 162     if (threadinfo[currthread].isStalled()) {
 163       //remove from waiter list
 164       threadinfo[currthread].setStall(false);
 165       getmapping(threadinfo[currthread].getObjIndex()).getWaiters().remove(currentevents[currthread]);
 166     }
 167     if (serAbort!=null) {
 168       serAbort.addPoint(time, currthread);
 169     }
 170     Transaction trans=currentevents[currthread].getTransaction();
 171
 172     releaseObjects(trans, currthread, time);
 173     Event nev=new Event(time+trans.getTime(0), trans, 0, currthread, currentevents[currthread].getTransNum());
 174     currentevents[currthread]=nev;
 175     eq.add(nev);
 176   }
 177
 178   //Aborts another thread...
 179   public void stall(Event ev, long time, long delay) {
 180     stallcycles+=delay;
 181     ev.setTime(time+delay);
 182     eq.add(ev);
 183   }
 184
 185   private void releaseObjects(Transaction trans, int currthread, long time) {
 186     //remove all events
 187     for(int i=0;i<trans.numEvents();i++) {
 188       ObjIndex object=trans.getObjIndex(i);
 189
 190       if (object!=null&&rdobjmap.containsKey(object)) {
 191         ((Set)rdobjmap.get(object)).remove(new Integer(currthread));
 192       }
 193       if (object!=null&&wrobjmap.containsKey(object)) {
 194         ((Set)wrobjmap.get(object)).remove(new Integer(currthread));
 195       }
 196       if (object!=null&&objtoinfo.containsKey(object)) {
 197         ObjectInfo oi=(ObjectInfo)objtoinfo.get(object);
 198         if (oi.getOwner()==currentevents[currthread].getThread()) {
 199           oi.releaseOwner();
 200
 201           //wake up one waiter
 202           for(Iterator waitit=oi.getWaiters().iterator();waitit.hasNext();) {
 203             //requeue everyone who was waiting on us and start them back up
 204             Event waiter=(Event)waitit.next();
 205             waitit.remove();
 206             waiter.setTime(time);
 207             threadinfo[waiter.getThread()].setStall(false);
 208             if (serWake!=null)
 209               serWake.addPoint(time,waiter.getThread());
 210             oi.setOwner(waiter.getThread());
 211             eq.add(waiter);
 212             break;
 213           }
 214         }
 215       }
 216     }
 217   }
 218
 219   /* Initializes things and returns number of transactions */
 220   public int startinitial() {
 221     int tcount=0;
 222     for(int i=0;i<e.numThreads();i++) {
 223       Transaction trans=e.getThread(i).getTransaction(0);
 224       long time=trans.getTime(0);
 225       Event ev=new Event(time, trans, 0, i, 0);
 226       currentevents[i]=ev;
 227       eq.add(ev);
 228       tcount+=e.getThread(i).numTransactions();
 229     }
 230     return tcount;
 231   }
 232
 233   public void dosim() {
 234     long lasttime=0;
 235     //start first transactions
 236     int numtrans=startinitial();
 237     System.out.println("Number of transactions="+numtrans);
 238     int tcount=0;
 239     while(!eq.isEmpty()) {
 240       Event ev=(Event)eq.poll();
 241       if (!ev.isValid()) {
 242         continue;
 243       }
 244
 245       Transaction trans=ev.getTransaction();
 246
 247       int event=ev.getEvent();
 248       long currtime=ev.getTime();
 249       lasttime=currtime;
 250       if (trans.started&&starttime==-1)
 251         starttime=currtime;
 252
 253       if (trans.numEvents()==(event+1)) {
 254         tryCommit(ev, trans);
 255         tcount++;
 256         if ((tcount%100000)==0)
 257           System.out.println("Attempted "+tcount+"transactions "+policy);
 258       } else {
 259         enqueueEvent(ev, trans);
 260       }
 261     }
 262     shorttesttime=lasttime;
 263     if (p!=null)
 264       p.close();
 265   }
 266
 267   private ObjectInfo getmapping(ObjIndex obj) {
 268     if (!objtoinfo.containsKey(obj))
 269       objtoinfo.put(obj, new ObjectInfo(this));
 270     return (ObjectInfo)objtoinfo.get(obj);
 271   }
 272
 273   public void tryCommit(Event ev, Transaction trans) {
 274     //ready to commit this one
 275     long currtime=ev.getTime();
 276     releaseObjects(trans, ev.getThread(), currtime);
 277
 278     //See if we have been flagged as aborted for the lazy case
 279     boolean abort=aborted[ev.getThread()];
 280     aborted[ev.getThread()]=false;
 281     if (!abort) {
 282       //if it is a transaction, increment commit count
 283       if (trans.numEvents()>1||trans.getEvent(0)!=Transaction.DELAY) {
 284         commitcount++;
 285         if (serCommit!=null) {
 286           serCommit.addPoint(ev.getTime(),ev.getThread());
 287         }
 288       }
 289       //Reset our backoff counter
 290       threadinfo[ev.getThread()].priority=0;
 291       threadinfo[ev.getThread()].aborted=false;
 292       backoff[ev.getThread()]=BACKOFFSTART;
 293       retrycount[ev.getThread()]=0;
 294       transferred[ev.getThread()]=0;
 295
 296       //abort the other threads
 297       for(int i=0;i<trans.numEvents();i++) {
 298         ObjIndex object=trans.getObjIndex(i);
 299         int op=trans.getEvent(i);
 300         //Mark commits to objects
 301         if (isLock()&&(op==Transaction.WRITE||op==Transaction.READ)) {
 302           if (object==null) {
 303             System.out.println(op);
 304           }
 305           getmapping(object).recordCommit();
 306         }
 307         //Check for threads we might cause to abort
 308         if (op==Transaction.WRITE) {
 309           HashSet abortset=new HashSet();
 310           if (rdobjmap.containsKey(object)) {
 311             for(Iterator it=((Set)rdobjmap.get(object)).iterator();it.hasNext();) {
 312               Integer threadid=(Integer)it.next();
 313               abortset.add(threadid);
 314               if (isLock()) {
 315                 ObjectInfo oi=getmapping(object);
 316                 oi.recordAbort();
 317               }
 318             }
 319           }
 320           if (wrobjmap.containsKey(object)) {
 321             for(Iterator it=((Set)wrobjmap.get(object)).iterator();it.hasNext();) {
 322               Integer threadid=(Integer)it.next();
 323               abortset.add(threadid);
 324               if (isLock()&&(!rdobjmap.containsKey(object)||!((Set)rdobjmap.get(object)).contains(threadid))) {
 325                 //if this object hasn't already cause this thread to
 326                 //abort, then flag it as an abort cause
 327                 ObjectInfo oi=getmapping(object);
 328                 oi.recordAbort();
 329               }
 330             }
 331           }
 332           for(Iterator abit=abortset.iterator();abit.hasNext();) {
 333             Integer threadid=(Integer)abit.next();
 334             if (policy==LAZY||policy==LOCK) {
 335               //just flag to abort when it trie to commit
 336               aborted[threadid]=true;
 337               if (serAbort!=null)
 338                 serAbort.addPoint(currtime, threadid);
 339             } else if (policy==COMMIT||policy==LOCKCOMMIT) {
 340               //abort it immediately
 341               timewasted(threadid, currtime);
 342               reschedule(threadid, currtime, 0);
 343               abortcount++;
 344             }
 345           }
 346         }
 347       }
 348     } else {
 349       abortcount++;
 350       timewasted(ev.getThread(), currtime);
 351     }
 352
 353     //add next transaction event...could be us if we aborted
 354     int nexttransnum=abort?ev.getTransNum():ev.getTransNum()+1;
 355     if (nexttransnum<e.getThread(ev.getThread()).numTransactions()) {
 356       Transaction nexttrans=e.getThread(ev.getThread()).getTransaction(nexttransnum);
 357       if (serStart!=null) {
 358         if (nexttrans.numEvents()>1||nexttrans.getEvent(0)!=Transaction.DELAY) {
 359           serStart.addPoint(ev.getTime(),ev.getThread());
 360         }
 361       }
 362
 363       Event nev=new Event(currtime+nexttrans.getTime(0), nexttrans, 0, ev.getThread(), nexttransnum);
 364       currentevents[ev.getThread()]=nev;
 365       eq.add(nev);
 366     }
 367   }
 368
 369   public Set rdConflictSet(int thread, ObjIndex obj) {
 370     if (!wrobjmap.containsKey(obj))
 371       return null;
 372     HashSet conflictset=new HashSet();
 373     for(Iterator it=((Set)wrobjmap.get(obj)).iterator();it.hasNext();) {
 374       Integer threadid=(Integer)it.next();
 375       if (threadid.intValue()!=thread)
 376         conflictset.add(threadid);
 377     }
 378     if (conflictset.isEmpty())
 379       return null;
 380     else
 381       return conflictset;
 382   }
 383
 384   public Set wrConflictSet(int thread, ObjIndex obj) {
 385     HashSet conflictset=new HashSet();
 386     if (rdobjmap.containsKey(obj)) {
 387       for(Iterator it=((Set)rdobjmap.get(obj)).iterator();it.hasNext();) {
 388         Integer threadid=(Integer)it.next();
 389         if (threadid.intValue()!=thread)
 390           conflictset.add(threadid);
 391       }
 392     }
 393     for(Iterator it=((Set)wrobjmap.get(obj)).iterator();it.hasNext();) {
 394       Integer threadid=(Integer)it.next();
 395       if (threadid.intValue()!=thread)
 396         conflictset.add(threadid);
 397     }
 398     if (conflictset.isEmpty())
 399       return null;
 400     else
 401       return conflictset;
 402   }
 403
 404   //Takes as parameter -- current transaction read event ev, conflicting
 405   //set of threads, and the current time
 406   //Returning false causes current transaction not continue to be scheduled
 407
 408
 409   public boolean handleConflicts(Event ev, Set threadstokill, long time) {
 410     if (policy==RANDOM) {
 411       boolean b=r.nextBoolean();
 412       if (b) {
 413         //delay
 414         int thread=ev.getThread();
 415         int dback=backoff[thread]*2;
 416         if (dback>0)
 417           backoff[thread]=dback;
 418         stall(ev, time, r.nextInt(backoff[thread]));
 419         return false;
 420       } else {
 421         //abort other transactions
 422         for(Iterator thit=threadstokill.iterator();thit.hasNext();) {
 423           Integer thread=(Integer)thit.next();
 424           timewasted(thread, time);
 425           reschedule(thread, time, 0);
 426           abortcount++;
 427         }
 428         return true;
 429       }
 430     } else if (policy==KARMA) {
 431       int maxpriority=0;
 432       for(Iterator thit=threadstokill.iterator();thit.hasNext();) {
 433         Integer thread=(Integer)thit.next();
 434         if (threadinfo[thread].priority>maxpriority)
 435           maxpriority=threadinfo[thread].priority;
 436       }
 437       if (maxpriority>(threadinfo[ev.getThread()].priority+retrycount[ev.getThread()])) {
 438         //stall for a little while
 439         threadinfo[ev.getThread()].priority--;
 440         retrycount[ev.getThread()]++;
 441         int rtime=r.nextInt(3000);
 442         stall(ev, time, rtime);
 443         return false;
 444       } else {
 445         //we win
 446         for(Iterator thit=threadstokill.iterator();thit.hasNext();) {
 447           Integer thread=(Integer)thit.next();
 448           int dback=backoff[thread]*2;
 449           if (dback>0)
 450             backoff[thread]=dback;
 451           int atime=r.nextInt(backoff[thread]);
 452           timewasted(thread, time);
 453           reschedule(thread, time, atime);
 454           abortcount++;
 455         }
 456         return true;
 457       }
 458     } else if (policy==ERUPTION) {
 459       int maxpriority=0;
 460       //abort other transactions
 461       for(Iterator thit=threadstokill.iterator();thit.hasNext();) {
 462         Integer thread=(Integer)thit.next();
 463         if (threadinfo[thread].priority>maxpriority)
 464           maxpriority=threadinfo[thread].priority;
 465       }
 466       if (maxpriority>(threadinfo[ev.getThread()].priority+retrycount[ev.getThread()])) {
 467         //we lose
 468         threadinfo[ev.getThread()].priority--;
 469         //stall for a little while
 470         int rtime=r.nextInt(3000);
 471         stall(ev, time, rtime);
 472         int ourpriority=threadinfo[ev.getThread()].priority;
 473         ourpriority-=transferred[ev.getThread()];
 474         for(Iterator thit=threadstokill.iterator();thit.hasNext();) {
 475           Integer thread=(Integer)thit.next();
 476           threadinfo[thread].priority+=ourpriority;
 477         }
 478         transferred[ev.getThread()]=threadinfo[ev.getThread()].priority;
 479         retrycount[ev.getThread()]++;
 480
 481         return false;
 482       } else {
 483         //we win
 484         for(Iterator thit=threadstokill.iterator();thit.hasNext();) {
 485           Integer thread=(Integer)thit.next();
 486           int dback=backoff[thread]*2;
 487           if (dback>0)
 488             backoff[thread]=dback;
 489           int atime=r.nextInt(backoff[thread]);
 490           timewasted(thread, time);
 491           reschedule(thread, time, atime);
 492           abortcount++;
 493         }
 494         return true;
 495       }
 496     } else if (policy==POLITE) {
 497       int retry=(++retrycount[ev.getThread()]);
 498       if (retry>=22) {
 499         retrycount[ev.getThread()]=0;
 500         for(Iterator thit=threadstokill.iterator();thit.hasNext();) {
 501           Integer thread=(Integer)thit.next();
 502           timewasted(thread, time);
 503           reschedule(thread, time, 0);
 504           abortcount++;
 505         }
 506         return true;
 507       } else {
 508         //otherwise stall
 509         int stalltime=(1<<(retry-1))*12;
 510         if (stalltime<0)
 511           stalltime=1<<30;
 512         stall(ev, time, r.nextInt(stalltime));
 513         return false;
 514       }
 515     } else if (policy==ATTACK) {
 516       for(Iterator thit=threadstokill.iterator();thit.hasNext();) {
 517         Integer thread=(Integer)thit.next();
 518         timewasted(thread, time);
 519         reschedule(thread, time, r.nextInt(backoff[thread.intValue()]));
 520         int dback=backoff[thread.intValue()]*2;
 521         if (dback>0)
 522           backoff[thread.intValue()]=dback;
 523         abortcount++;
 524       }
 525       return true;
 526     } else if (policy==SUICIDE) {
 527       timewasted(ev.getThread(), time);
 528       reschedule(ev.getThread(), time, r.nextInt(backoff[ev.getThread()]));
 529       int dback=backoff[ev.getThread()]*2;
 530       if (dback>0)
 531         backoff[ev.getThread()]=dback;
 532       abortcount++;
 533       return false;
 534     } else if (policy==TIMESTAMP) {
 535       long opponenttime=0;
 536
 537       for(Iterator thit=threadstokill.iterator();thit.hasNext();) {
 538         Integer thread=(Integer)thit.next();
 539         Event other=currentevents[thread.intValue()];
 540         int eventnum=other.getEvent();
 541         long otime=other.getTransaction().getTime(other.getEvent());
 542         if (otime>opponenttime)
 543           opponenttime=otime;
 544       }
 545       if (opponenttime>ev.getTransaction().getTime(ev.getEvent())) {
 546         //kill ourself
 547         timewasted(ev.getThread(), time);
 548         reschedule(ev.getThread(), time, 0);
 549         abortcount++;
 550         return false;
 551       } else {
 552         //kill the opponents
 553         for(Iterator thit=threadstokill.iterator();thit.hasNext();) {
 554           Integer thread=(Integer)thit.next();
 555           timewasted(thread, time);
 556           reschedule(thread, time, 0);
 557           abortcount++;
 558         }
 559         return true;
 560       }
 561     } else if (policy==THREAD) {
 562       long tid=1000;
 563
 564       for(Iterator thit=threadstokill.iterator();thit.hasNext();) {
 565         Integer thread=(Integer)thit.next();
 566         Event other=currentevents[thread.intValue()];
 567         int eventnum=other.getEvent();
 568         long otid=thread.intValue();
 569         if (tid>otid)
 570           tid=otid;
 571       }
 572       if (ev.getThread()>tid) {
 573         //kill ourself
 574         timewasted(ev.getThread(), time);
 575         reschedule(ev.getThread(), time, 0);
 576         abortcount++;
 577         return false;
 578       } else {
 579         //kill the opponents
 580         for(Iterator thit=threadstokill.iterator();thit.hasNext();) {
 581           Integer thread=(Integer)thit.next();
 582           timewasted(thread, time);
 583           reschedule(thread, time, 0);
 584           abortcount++;
 585         }
 586         return true;
 587       }
 588     } else if (policy==ATTACKTIME) {
 589       boolean timebased=false;
 590       int tev=ev.getThread();
 591       timebased|=threadinfo[tev].aborted;
 592       for(Iterator thit=threadstokill.iterator();thit.hasNext();) {
 593         Integer thread=(Integer)thit.next();
 594         timebased|=threadinfo[thread.intValue()].aborted;
 595       }
 596       if (timebased) {
 597         long opponenttime=0;
 598
 599         for(Iterator thit=threadstokill.iterator();thit.hasNext();) {
 600           Integer thread=(Integer)thit.next();
 601           Event other=currentevents[thread.intValue()];
 602           int eventnum=other.getEvent();
 603           long otime=other.getTransaction().getTime(other.getEvent());
 604           if (otime>opponenttime)
 605             opponenttime=otime;
 606         }
 607         if (opponenttime>ev.getTransaction().getTime(ev.getEvent())) {
 608           //kill ourself
 609           timewasted(ev.getThread(), time);
 610           reschedule(ev.getThread(), time, 0);
 611           threadinfo[ev.getThread()].aborted=true;
 612           abortcount++;
 613           return false;
 614         } else {
 615           //kill the opponents
 616           for(Iterator thit=threadstokill.iterator();thit.hasNext();) {
 617             Integer thread=(Integer)thit.next();
 618             timewasted(thread, time);
 619             reschedule(thread, time, 0);
 620             threadinfo[thread.intValue()].aborted=true;
 621             abortcount++;
 622           }
 623           return true;
 624         }
 625       } else {
 626         for(Iterator thit=threadstokill.iterator();thit.hasNext();) {
 627           Integer thread=(Integer)thit.next();
 628           timewasted(thread, time);
 629           reschedule(thread, time, 0);
 630           threadinfo[thread.intValue()].aborted=true;
 631           abortcount++;
 632         }
 633         return true;
 634       }
 635     } else if (policy==ATTACKTHREAD) {
 636       boolean threadbased=false;
 637       int tev=ev.getThread();
 638       threadbased|=threadinfo[tev].aborted;
 639       for(Iterator thit=threadstokill.iterator();thit.hasNext();) {
 640         Integer thread=(Integer)thit.next();
 641         threadbased|=threadinfo[thread.intValue()].aborted;
 642       }
 643       if (threadbased) {
 644         long opponentthr=1000;
 645
 646         for(Iterator thit=threadstokill.iterator();thit.hasNext();) {
 647           Integer thread=(Integer)thit.next();
 648           Event other=currentevents[thread.intValue()];
 649           int eventnum=other.getEvent();
 650           long othr=thread.intValue();
 651           if (othr<opponentthr)
 652             opponentthr=othr;
 653         }
 654         if (opponentthr<tev) {
 655           //kill ourself
 656           timewasted(ev.getThread(), time);
 657           reschedule(ev.getThread(), time, 0);
 658           threadinfo[ev.getThread()].aborted=true;
 659           abortcount++;
 660           return false;
 661         } else {
 662           //kill the opponents
 663           for(Iterator thit=threadstokill.iterator();thit.hasNext();) {
 664             Integer thread=(Integer)thit.next();
 665             timewasted(thread, time);
 666             reschedule(thread, time, 0);
 667             threadinfo[thread.intValue()].aborted=true;
 668             abortcount++;
 669           }
 670           return true;
 671         }
 672       } else {
 673         for(Iterator thit=threadstokill.iterator();thit.hasNext();) {
 674           Integer thread=(Integer)thit.next();
 675           timewasted(thread, time);
 676           reschedule(thread, time, 0);
 677           threadinfo[thread.intValue()].aborted=true;
 678           abortcount++;
 679         }
 680         return true;
 681       }
 682     }
 683
 684     //Not eager
 685     return true;
 686   }
 687
 688   //Handle current event (read, write, delay) in a transaction and
 689   //enqueue the next one
 690
 691   public void enqueueEvent(Event ev, Transaction trans) {
 692     //just enqueue next event
 693     int event=ev.getEvent();
 694     long currtime=ev.getTime();
 695     ObjIndex object=trans.getObjIndex(event);
 696     int operation=trans.getEvent(event);
 697
 698     if ((operation==Transaction.READ||operation==Transaction.WRITE)&&isLock()) {
 699       ObjectInfo oi=getmapping(object);
 700
 701       if (oi.isRisky()) {
 702         if (oi.isOwned()&&oi.getOwner()!=ev.getThread()) {
 703           //we're going to wait
 704           boolean deadlocked=true;
 705           ObjectInfo toi=oi;
 706           for(int i=0;i<checkdepth;i++) {
 707             //check if stalling would close the loop
 708             if (toi.getOwner()==ev.getThread())
 709               break;
 710             //see if cycle is broken
 711             if (!threadinfo[toi.getOwner()].isStalled()) {
 712               deadlocked=false;
 713               break;
 714             }
 715             //follow one more in depth
 716             toi=getmapping(threadinfo[toi.getOwner()].getObjIndex());
 717           }
 718
 719           if (!deadlocked) {
 720             //don't wait on stalled threads, we could deadlock
 721             threadinfo[ev.getThread()].setStall(true);
 722             threadinfo[ev.getThread()].setObjIndex(object);
 723             if (serStall!=null)
 724               serStall.addPoint(ev.getTime(),ev.getThread());
 725             oi.addWaiter(ev);
 726             return;
 727           } else {
 728             if (serAvoid!=null)
 729               serAvoid.addPoint(ev.getTime(),ev.getThread());
 730             deadlockcount++;
 731           }
 732         } else {
 733           //we have object
 734           oi.setOwner(ev.getThread());
 735         }
 736       }
 737     }
 738
 739     //process the current event
 740     if (operation==Transaction.READ) {
 741       //record read event
 742       if (!rdobjmap.containsKey(object))
 743         rdobjmap.put(object,new HashSet());
 744       if (((Set)rdobjmap.get(object)).add(new Integer(ev.getThread()))) {
 745         //added new object
 746         if (countObjects()) {
 747           threadinfo[ev.getThread()].priority++;
 748         }
 749       }
 750       if (isEager()) {
 751         //do eager contention management
 752         Set conflicts=rdConflictSet(ev.getThread(), object);
 753         if (conflicts!=null) {
 754           if (!handleConflicts(ev, conflicts, currtime)) {
 755             ((Set)rdobjmap.get(object)).remove(new Integer(ev.getThread()));
 756             return;
 757           }
 758         }
 759       }
 760     } else if (operation==Transaction.WRITE) {
 761       //record write event
 762       if (!wrobjmap.containsKey(object))
 763         wrobjmap.put(object,new HashSet());
 764       if (((Set)wrobjmap.get(object)).add(new Integer(ev.getThread()))) {
 765         if (countObjects()) {
 766           threadinfo[ev.getThread()].priority++;
 767         }
 768       }
 769       if (isEager()) {
 770         Set conflicts=wrConflictSet(ev.getThread(), object);
 771         if (conflicts!=null) {
 772           if (!handleConflicts(ev, conflicts, currtime)) {
 773             ((Set)wrobjmap.get(object)).remove(new Integer(ev.getThread()));
 774             return;
 775           }
 776         }
 777       }
 778     } else if (operation==Transaction.BARRIER) {
 779       barriercount--;
 780       if (barriercount==0) {
 781         for(int i=0;i<e.numThreads();i++) {
 782           //enqueue the next event
 783           Event bev=currentevents[i];
 784           int bevent=bev.getEvent();
 785           long bcurrtime=bev.getTime();
 786           Transaction btrans=bev.getTransaction();
 787           long deltatime=btrans.getTime(bevent+1)-btrans.getTime(bevent);
 788           Event nev=new Event(deltatime+currtime, btrans, bevent+1, bev.getThread(), bev.getTransNum());
 789           currentevents[bev.getThread()]=nev;
 790           eq.add(nev);
 791         }
 792         barriercount=e.numThreads();
 793       } else {
 794         //Do nothing
 795         //wait until all threads in barrier
 796       }
 797       return;
 798     }
 799     retrycount[ev.getThread()]=0;
 800     transferred[ev.getThread()]=0;
 801     //enqueue the next event
 802     long deltatime=trans.getTime(event+1)-trans.getTime(event);
 803     Event nev=new Event(deltatime+currtime, trans, event+1, ev.getThread(), ev.getTransNum());
 804     currentevents[ev.getThread()]=nev;
 805     eq.add(nev);
 806   }
 807
 808
 809   class Event implements Comparable {
 810     boolean valid;
 811     long time;
 812     int num;
 813     Transaction t;
 814     int threadid;
 815     int transnum;
 816
 817     public void makeInvalid() {
 818       valid=false;
 819     }
 820
 821     public boolean isValid() {
 822       return valid;
 823     }
 824
 825     public int getTransNum() {
 826       return transnum;
 827     }
 828
 829     public Transaction getTransaction() {
 830       return t;
 831     }
 832
 833     public int getEvent() {
 834       return num;
 835     }
 836
 837     public long getTime() {
 838       return time;
 839     }
 840
 841     public void setTime(long time) {
 842       this.time=time;
 843     }
 844
 845     public int getThread() {
 846       return threadid;
 847     }
 848
 849     public Event(long time, Transaction t, int num, int threadid, int transnum) {
 850       this.time=time;
 851       this.t=t;
 852       this.num=num;
 853       this.threadid=threadid;
 854       this.transnum=transnum;
 855       valid=true;
 856     }
 857
 858     //break ties to allow commits to occur earliest
 859     public int compareTo(Object o) {
 860       Event e=(Event)o;
 861       long delta=time-e.time;
 862       if (delta!=0) {
 863         if (delta>0)
 864           return 1;
 865         else
 866           return -1;
 867       }
 868       if (((getEvent()+1)==getTransaction().numEvents())&&
 869           (e.getEvent()+1)!=e.getTransaction().numEvents())
 870         return -1;
 871       if (((getEvent()+1)!=getTransaction().numEvents())&&
 872           (e.getEvent()+1)==e.getTransaction().numEvents())
 873         return 1;
 874       return 0;
 875     }
 876   }
 877 }