synchronizationtimer.java

一个很好的微工作流内核

          Thread.sleep(n % maxSleepTime);
        }
        catch (InterruptedException ex) {
          Thread.currentThread().interrupt();
        }
        retries = 0;
      }
    }        
  }
  
}

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

  protected void set(long l) { acurrent_.set(l); }
  protected long internalGet() { return acurrent_.get(); }
  protected void internalUpdate() { set(compute(internalGet())); }
  
}

abstract class DelegatedRNG extends RNG  {
  protected RNG delegate_ = null;
  public synchronized void setDelegate(RNG d) { delegate_ = d; }
  protected synchronized RNG getDelegate() { return delegate_; }

  public long get() { return getDelegate().get(); }
  public void update() { getDelegate().update(); }
  public long next() { return getDelegate().next(); }

  protected void set(long l) { throw new Error(); }
  protected long internalGet() { throw new Error(); }
  protected void internalUpdate() { throw new Error(); }

}

class SDelegatedRNG extends DelegatedRNG {
  public SDelegatedRNG() { setDelegate(new NoSynchRNG()); }
  public synchronized long get() { return getDelegate().get(); }
  public synchronized void update() { getDelegate().update(); }
  public synchronized long next() { return getDelegate().next(); }
}


class SyncDelegatedRNG extends DelegatedRNG {
  protected final Sync cond_;
  public SyncDelegatedRNG(Sync c) { 
    cond_ = c; 
    setDelegate(new NoSynchRNG());
  }


  protected final void acquire() throws InterruptedException {
    if (smode == 0) {
      cond_.acquire();
    }
    else {
      while (!cond_.attempt(waitTime)) {}
    }
  }
      
  public long next() { 
    try {
      acquire();

      getDelegate().update();
      long l = getDelegate().get();
      cond_.release(); 
      return l;
    }
    catch(InterruptedException x) { 
      Thread.currentThread().interrupt(); 
      return 0;
    }
  }

  public long get()  { 
    try {
      acquire();
      long l = getDelegate().get();
      cond_.release(); 
      return l;
    }
    catch(InterruptedException x) { 
      Thread.currentThread().interrupt(); 
      return 0;
    }
  }

  public void update()  { 
    try {
      acquire();
      getDelegate().update();
      cond_.release(); 
    }
    catch(InterruptedException x) { 
      Thread.currentThread().interrupt(); 
    }
  }


}

class MutexRNG extends SyncDelegatedRNG {
  public MutexRNG() { super(new Mutex()); }
}


class SemRNG extends SyncDelegatedRNG {
  public SemRNG() { super(new Semaphore(1)); }
}

class WpSemRNG extends SyncDelegatedRNG {
  public WpSemRNG() { super(new WaiterPreferenceSemaphore(1)); }
}

class FifoRNG extends SyncDelegatedRNG {
  public FifoRNG() { super(new FIFOSemaphore(1)); }
}

class PrioritySemRNG extends SyncDelegatedRNG {
  public PrioritySemRNG() { super(new PrioritySemaphore(1)); }
}

class RlockRNG extends SyncDelegatedRNG {
  public RlockRNG() { super(new ReentrantLock()); }
}


class RWLockRNG extends NoSynchRNG {
  protected final ReadWriteLock lock_;
  public RWLockRNG(ReadWriteLock l) { 
    lock_ = l; 
  }
      
  protected final void acquireR() throws InterruptedException {
    if (smode == 0) {
      lock_.readLock().acquire();
    }
    else {
      while (!lock_.readLock().attempt(waitTime)) {}
    }
  }

  protected final void acquireW() throws InterruptedException {
    if (smode == 0) {
      lock_.writeLock().acquire();
    }
    else {
      while (!lock_.writeLock().attempt(waitTime)) {}
    }
  }


  public long next() { 
    long l = 0;
    try {
      acquireR();
      l = current_;
      lock_.readLock().release(); 
    }
    catch(InterruptedException x) { 
      Thread.currentThread().interrupt(); 
      return 0;
    }

    l = compute(l);

    try {
      acquireW();
      set(l);
      lock_.writeLock().release(); 
      return l;
    }
    catch(InterruptedException x) { 
      Thread.currentThread().interrupt(); 
      return 0;
    }
  }


  public long get()  { 
    try {
      acquireR();
      long l = current_;
      lock_.readLock().release(); 
      return l;
    }
    catch(InterruptedException x) { 
      Thread.currentThread().interrupt(); 
      return 0;
    }
  }

  public void update()  { 
    long l = 0;

    try {
      acquireR();
      l = current_;
      lock_.readLock().release(); 
    }
    catch(InterruptedException x) { 
      Thread.currentThread().interrupt(); 
      return;
    }

    l = compute(l);

    try {
      acquireW();
      set(l);
      lock_.writeLock().release(); 
    }
    catch(InterruptedException x) { 
      Thread.currentThread().interrupt(); 
    }
  }

}

class WpRWlockRNG extends RWLockRNG {
  public WpRWlockRNG() { super(new WriterPreferenceReadWriteLock()); }
}

class ReaderPrefRWlockRNG extends RWLockRNG {
  public ReaderPrefRWlockRNG() { 
    super(new ReaderPreferenceReadWriteLock()); 
  }


}

class FIFORWlockRNG extends RWLockRNG {
  public FIFORWlockRNG() { super(new FIFOReadWriteLock()); }
}


class ReentrantRWlockRNG extends RWLockRNG {
  public ReentrantRWlockRNG() { 
    super(new ReentrantWriterPreferenceReadWriteLock()); 
  }

  public void update()  {  // use embedded acquires
    long l = 0;

    try {
      acquireW();

      try {
        acquireR();
        l = current_;
        lock_.readLock().release(); 
      }
      catch(InterruptedException x) { 
        Thread.currentThread().interrupt(); 
        return;
      }

      l = compute(l);

      set(l);
      lock_.writeLock().release(); 
    }
    catch(InterruptedException x) { 
      Thread.currentThread().interrupt(); 
    }
  }

}


abstract class ExecutorRNG extends DelegatedRNG {
  Executor executor_;


  synchronized void setExecutor(Executor e) { executor_ = e; }
  synchronized Executor getExecutor() { return executor_; }

  Runnable delegatedUpdate_ = null;
  Callable delegatedNext_ = null;

  synchronized Runnable delegatedUpdateCommand() {
    if (delegatedUpdate_ == null)
      delegatedUpdate_ = new UpdateCommand(getDelegate());
    return delegatedUpdate_;
  }

  synchronized Callable delegatedNextFunction() {
    if (delegatedNext_ == null)
      delegatedNext_ = new NextFunction(getDelegate());
    return delegatedNext_;
  }

  public void update() { 
    try {
      getExecutor().execute(delegatedUpdateCommand()); 
    }
    catch (InterruptedException ex) {
      Thread.currentThread().interrupt();
    }
  }

  // Each call to next gets result of previous future 
  FutureResult nextResult_ = null;

  public synchronized long next() { 
    long res = 0;
    try {
      if (nextResult_ == null) { // direct call first time through
        nextResult_ = new FutureResult();
        nextResult_.set(new Long(getDelegate().next()));
      }
      FutureResult currentResult = nextResult_;

      nextResult_ = new FutureResult();
      Runnable r = nextResult_.setter(delegatedNextFunction());
      getExecutor().execute(r); 

      res =  ((Long)(currentResult.get())).longValue();

    }
    catch (InterruptedException ex) {
      Thread.currentThread().interrupt();
    }
    catch (InvocationTargetException ex) {
      ex.printStackTrace();
      throw new Error("Bad Callable?");
    }
    return res;
  }
}

class DirectExecutorRNG extends ExecutorRNG {
  public DirectExecutorRNG() { 
    setDelegate(new PublicSynchRNG()); 
    setExecutor(new DirectExecutor()); 
  }
}

class LockedSemRNG extends ExecutorRNG {
  public LockedSemRNG() { 
    setDelegate(new NoSynchRNG()); 
    setExecutor(new LockedExecutor(new Semaphore(1))); 
  }
}

class QueuedExecutorRNG extends ExecutorRNG {
  static final QueuedExecutor exec = new QueuedExecutor();
  static { exec.setThreadFactory(Threads.factory); }
  public QueuedExecutorRNG() { 
    setDelegate(new PublicSynchRNG()); 
    setExecutor(exec); 
  }
}

class ForcedStartRunnable implements Runnable {
  protected final Latch latch_ = new Latch();
  protected final Runnable command_;

  ForcedStartRunnable(Runnable command) { command_ = command; }

  public Latch started() { return latch_; }

  public void run() {
    latch_.release();
    command_.run();
  }
}


class ForcedStartThreadedExecutor extends ThreadedExecutor {
  public void execute(Runnable command) throws InterruptedException {
    ForcedStartRunnable wrapped = new ForcedStartRunnable(command);
    super.execute(wrapped);
    wrapped.started().acquire();
  }
}

class ThreadedExecutorRNG extends ExecutorRNG {
  static final ThreadedExecutor exec = new ThreadedExecutor();
  static { exec.setThreadFactory(Threads.factory); }

  public ThreadedExecutorRNG() { 
    setDelegate(new PublicSynchRNG()); 
    setExecutor(exec); 
  }
}


class PooledExecutorRNG extends ExecutorRNG {
  static final PooledExecutor exec = Threads.pool;

  public PooledExecutorRNG() { 
    setDelegate(new PublicSynchRNG()); 
    setExecutor(exec); 
  }
}


class ChanRNG extends DelegatedRNG {

  boolean single_;

  ChanRNG() {
    setDelegate(new PublicSynchRNG());
  }

  public synchronized void setSingle(boolean s) { single_ = s; }
  public synchronized boolean isSingle() { return single_; }

  public long producerNext(Channel c) throws InterruptedException {
    RNG r = getDelegate();
    if (isSingle()) {
      c.put(r);
      r = (RNG)(c.take());
      r.update();
    }
    else {
      if (pcBias < 0) {
        r.update();
        r.update(); // update consumer side too
      }
      else if (pcBias == 0) {
        r.update();
      }
      
      if (pmode == 0) {
        c.put(r);
      }
      else {
        while (!(c.offer(r, waitTime))) {}
      }
    }
    return r.get();
  }

  public long consumerNext(Channel c) throws InterruptedException {
    RNG r = null;
    if (cmode == 0) {
      r =  (RNG)(c.take());
    }
    else {
      while (r == null) r = (RNG)(c.poll(waitTime));
    }
    
    if (pcBias == 0) {
      r.update();
    }
    else if (pcBias > 0) {
      r.update();
      r.update();
    }
    return r.get();
  }
}