synchronizationtimer.java

一个很好的微工作流内核

        }
        
        // Uncomment if AWT doesn't update right
        //        Thread.sleep(100);

        barrier.barrier(); // start

        barrier.barrier(); // stop

        long tm = timer.getTime();
        long totalIters = nthreads * iters;
        double dns = tm * 1000.0 * PRECISION / totalIters;
        long ns = Math.round(dns);

        setTime(ns, clsIdx, nthreadsIdx);

        if (echoToSystemOut.get()) {
          System.out.println(formatTime(ns, true));
        }

      }
      catch (BrokenBarrierException ex) { 
        wasInterrupted = true;
      }
      catch (InterruptedException ex) {
        wasInterrupted = true;
        Thread.currentThread().interrupt();
      }
      catch (Exception ex) { 
        ex.printStackTrace();
        System.out.println("Construction Exception?");
        System.exit(-1);
      }
      finally {
        final boolean clear = wasInterrupted;
        SwingUtilities.invokeLater(new Runnable() {
          public void run() {
            if (clear) cell.setText("");
            cell.setForeground(oldfg);
            cell.repaint();
          }
        });

        Thread.currentThread().setPriority(Thread.NORM_PRIORITY);
        endOneTest();
      }
    }
  }

}

class Threads implements ThreadFactory {

  static final SynchronizedInt activeThreads = new SynchronizedInt(0);

  static final Threads factory = new Threads();

  static final PooledExecutor pool = new PooledExecutor();

  static { 
    pool.setKeepAliveTime(10000); 
    pool.setThreadFactory(factory);
  }

  static class MyThread extends Thread {
    public MyThread(Runnable cmd) { 
      super(cmd); 
    }

    public void run() {
      activeThreads.increment();

      try {
        super.run();
      }
      finally {
        activeThreads.decrement();
      }
    }
  }

  public Thread newThread(Runnable cmd) {
    return new MyThread(cmd);
  }
}



class TestLoop {

  final RNG shared;
  final RNG primary;
  final int iters;
  final Fraction pshared;
  final CyclicBarrier barrier;
  final boolean[] useShared;
  final int firstidx;

  public TestLoop(RNG sh, RNG pri, Fraction pshr, int it, CyclicBarrier br) {
    shared = sh; 
    primary = pri; 
    pshared = pshr; 
    iters = it; 
    barrier = br; 

    firstidx = (int)(primary.get());

    int num = (int)(pshared.numerator());
    int denom = (int)(pshared.denominator());

    if (num == 0 || primary == shared) {
      useShared = new boolean[1];
      useShared[0] = false;
    }
    else if (num >= denom) {
      useShared = new boolean[1];
      useShared[0] = true;
    }
    else {
      // create bool array and randomize it.
      // This ensures that always same number of shared calls.

      // denom slots is too few. iters is too many. an arbitrary compromise is:
      int xfactor = 1024 / denom;
      if (xfactor < 1) xfactor = 1;
      useShared = new boolean[denom * xfactor];
      for (int i = 0; i < num * xfactor; ++i) 
        useShared[i] = true;
      for (int i = num * xfactor; i < denom  * xfactor; ++i) 
        useShared[i] = false;

      for (int i = 1; i < useShared.length; ++i) {
        int j = ((int) (shared.next() & 0x7FFFFFFF)) % (i + 1);
        boolean tmp = useShared[i];
        useShared[i] = useShared[j];
        useShared[j] = tmp;
      }
    }
  }

  public Runnable testLoop() {
    return new Runnable() {
      public void run() {
        int itersPerBarrier = RNG.itersPerBarrier.get();
        try {
          int delta = -1;
          if (primary.getClass().equals(PrioritySemRNG.class)) {
            delta = 2 - (int)((primary.get() % 5));
          }
          Thread.currentThread().setPriority(Thread.NORM_PRIORITY+delta);
          
          int nshared = (int)(iters * pshared.asDouble());
          int nprimary = iters - nshared;
          int idx = firstidx;
          
          barrier.barrier();
          
          for (int i = iters; i > 0; --i) {
            ++idx;
            if (i % itersPerBarrier == 0)
              primary.exchange();
            else {
              
              RNG r;
              
              if (nshared > 0 && useShared[idx % useShared.length]) {
                --nshared;
                r = shared;
              }
              else {
                --nprimary;
                r = primary;
              }
              long rnd = r.next();
              if (rnd % 2 == 0 && Thread.currentThread().isInterrupted()) 
                break;
            }
          }
        }
        catch (BrokenBarrierException ex) {
        }
        catch (InterruptedException ex) {
          Thread.currentThread().interrupt();
        }
        finally {
          try {
            barrier.barrier();
          }
          catch (BrokenBarrierException ex) { 
          }
          catch (InterruptedException ex) {
            Thread.currentThread().interrupt();
          }
          finally {
            Thread.currentThread().setPriority(Thread.NORM_PRIORITY);
          }

        }
      }
    };
  }
}

class PCTestLoop extends TestLoop {
  final Channel primaryChannel;
  final Channel sharedChannel;

  public PCTestLoop(RNG sh, RNG pri, Fraction pshr, int it, 
    CyclicBarrier br, Channel shChan, Channel priChan) {
    super(sh, pri, pshr, it, br);
    sharedChannel = shChan;
    primaryChannel = priChan;
  }

  public Runnable testLoop(final boolean isProducer) {
    return new Runnable() {
      public void run() {
        int delta = -1;
        Thread.currentThread().setPriority(Thread.NORM_PRIORITY+delta);
        int itersPerBarrier = RNG.itersPerBarrier.get();
        try { 
          
          int nshared = (int)(iters * pshared.asDouble());
          int nprimary = iters - nshared;
          int idx = firstidx;
          
          barrier.barrier(); 
          
          ChanRNG target = (ChanRNG)(primary);
          
          for (int i = iters; i > 0; --i) {
            ++idx;
            if (i % itersPerBarrier == 0)
              primary.exchange();
            else {
              Channel c;
            
              if (nshared > 0 && useShared[idx % useShared.length]) {
                --nshared;
                c = sharedChannel;
              }
              else {
                --nprimary;
                c = primaryChannel;
              }
              
              long rnd;
              if (isProducer) 
                rnd = target.producerNext(c);
              else 
                rnd = target.consumerNext(c);
              
              if (rnd % 2 == 0 && Thread.currentThread().isInterrupted()) 
                break;
            }
          }
        }
        catch (BrokenBarrierException ex) {
        }
        catch (InterruptedException ex) {
          Thread.currentThread().interrupt();
        }
        finally {
          try {
            barrier.barrier();
          }
          catch (InterruptedException ex) {
            Thread.currentThread().interrupt();
          }
          catch (BrokenBarrierException ex) { 
          }
          finally {
            Thread.currentThread().setPriority(Thread.NORM_PRIORITY);
          }
        }
      }
    };
  }
}

// -------------------------------------------------------------


abstract class RNG implements Serializable, Comparable {
  static final int firstSeed = 4321;
  static final int rmod = 2147483647;
  static final int rmul = 16807;

  static int lastSeed = firstSeed;
  static final int smod = 32749;
  static final int smul = 3125;

  static final Object constructionLock = RNG.class;

  // Use construction lock for all params to disable
  // changes in midst of construction of test objects.

  static final SynchronizedInt computeLoops = 
    new SynchronizedInt(16, constructionLock);
  static final SynchronizedInt syncMode = 
    new SynchronizedInt(0, constructionLock);
  static final SynchronizedInt producerMode = 
    new SynchronizedInt(0, constructionLock);
  static final SynchronizedInt consumerMode = 
    new SynchronizedInt(0, constructionLock);
  static final SynchronizedInt bias = 
    new SynchronizedInt(0, constructionLock);
  static final SynchronizedLong timeout = 
    new SynchronizedLong(100, constructionLock);
  static final SynchronizedInt exchangeParties = 
    new SynchronizedInt(1, constructionLock);
  static final SynchronizedInt sequenceNumber = 
    new SynchronizedInt(0, constructionLock);
  static final SynchronizedInt itersPerBarrier = 
    new SynchronizedInt(0, constructionLock);

  static Rendezvous[] exchangers_;

  static void reset(int nthreads) {
    synchronized(constructionLock) {
      sequenceNumber.set(-1);
      int parties = exchangeParties.get();
      if (nthreads < parties) parties = nthreads;
      if (nthreads % parties != 0) 
        throw new Error("need even multiple of parties");
      exchangers_ = new Rendezvous[nthreads / parties];
      for (int i = 0; i < exchangers_.length; ++i) {
        exchangers_[i] = new Rendezvous(parties);
      }
    }
  }

  static long nextSeed() {
    synchronized(constructionLock) {
      long s = lastSeed;
      lastSeed = (lastSeed * smul) % smod;
      if (lastSeed == 0) 
        lastSeed = (int)(System.currentTimeMillis());
      return s;
    }
  }

  final int cloops = computeLoops.get();
  final int pcBias = bias.get();
  final int smode = syncMode.get();
  final int pmode = producerMode.get();
  final int cmode = consumerMode.get();
  final long waitTime = timeout.get();
  Rendezvous exchanger_ = null;

  synchronized Rendezvous getExchanger() {
    if (exchanger_ == null) {
      synchronized (constructionLock) {
        int idx = sequenceNumber.increment();
        exchanger_ = exchangers_[idx % exchangers_.length];
      }
    }
    return exchanger_;
  }

  public void exchange() throws InterruptedException {
    Rendezvous ex = getExchanger(); 
    Runnable r = (Runnable)(ex.rendezvous(new UpdateCommand(this)));
    if (r != null) r.run();
  }

  public int compareTo(Object other) {
    int h1 = hashCode();
    int h2 = other.hashCode();
    if (h1 < h2) return -1;
    else if (h1 > h2) return 1;
    else return 0;
  }

  protected final long compute(long l) { 
    int loops = (int)((l & 0x7FFFFFFF) % (cloops * 2)) + 1;
    for (int i = 0; i < loops; ++i) l = (l * rmul) % rmod;
    return (l == 0)? firstSeed : l; 
  }

  abstract protected void set(long l);
  abstract protected long internalGet();
  abstract protected void internalUpdate();

  public long get()    { return internalGet(); }
  public void update() { internalUpdate();  }
  public long next()   { internalUpdate(); return internalGet(); }
}


class UpdateCommand implements Runnable, Serializable, Comparable {
  private final RNG obj_;
  final long cmpVal;
  public UpdateCommand(RNG o) { 
    obj_ = o; 
    cmpVal = o.get();
  }

  public void run() { obj_.update(); } 

  public int compareTo(Object x) {
    UpdateCommand u = (UpdateCommand)x;
    if (cmpVal < u.cmpVal) return -1;
    else if (cmpVal > u.cmpVal) return 1;
    else return 0;
  }
}


class GetFunction implements Callable {
  private final RNG obj_;
  public GetFunction(RNG o) { obj_ = o;  }
  public Object call() { return new Long(obj_.get()); } 
}

class NextFunction implements Callable {
  private final RNG obj_;
  public NextFunction(RNG o) { obj_ = o;  }
  public Object call() { return new Long(obj_.next()); } 
}


class NoSynchRNG extends RNG {
  protected long current_ = nextSeed();

  protected void set(long l) { current_ = l; }
  protected long internalGet() { return current_; }  
  protected void internalUpdate() { set(compute(internalGet())); }
}

class PublicSynchRNG extends NoSynchRNG {
  public synchronized long get() { return internalGet(); }  
  public synchronized void update() { internalUpdate();  }
  public synchronized long next() { internalUpdate(); return internalGet(); }
}

class AllSynchRNG extends PublicSynchRNG {
  protected synchronized void set(long l) { current_ = l; }
  protected synchronized long internalGet() { return current_; }
  protected synchronized void internalUpdate() { set(compute(internalGet())); }
}


class AClongRNG extends RNG {
  protected final SynchronizedLong acurrent_ = 
    new SynchronizedLong(nextSeed());

  protected void set(long l) { throw new Error("No set allowed"); }
  protected long internalGet() { return acurrent_.get(); }

  protected void internalUpdate() { 
    int retriesBeforeSleep = 100;
    int maxSleepTime = 100;
    int retries = 0;
    for (;;) {
      long v = internalGet();
      long n = compute(v);
      if (acurrent_.commit(v, n))
        return;
      else if (++retries >= retriesBeforeSleep) {
        try {