synchronizationtimer.java

一个很好的微工作流内核

    return cloopComboBox;
  }
  
  JComboBox barrierBox() {
    final int[] itersPerBarrierChoices = { 
      1,
      2,
      4,
      8,
      16,
      32,
      64,
      128,
      256,
      512,
      1024,
      2 * 1024,
      4 * 1024,
      8 * 1024,
      16 * 1024,
      32 * 1024,
      64 * 1024, 
      128 * 1024, 
      256 * 1024, 
      512 * 1024, 
      1024 * 1024,
    };
    
    JComboBox barrierComboBox = new JComboBox();
    
    for (int j = 0; j < itersPerBarrierChoices.length; ++j) {
      String lab = p2ToString(itersPerBarrierChoices[j]) + 
        " iterations per barrier";
      barrierComboBox.addItem(lab);
    }
    barrierComboBox.addItemListener(new ItemListener() {
      public void itemStateChanged(ItemEvent evt) {
        JComboBox src = (JComboBox)(evt.getItemSelectable());
        int idx = src.getSelectedIndex();
        RNG.itersPerBarrier.set(itersPerBarrierChoices[idx]);
      }
    });
    
    RNG.itersPerBarrier.set(itersPerBarrierChoices[13]);
    barrierComboBox.setSelectedIndex(13);

    //    RNG.itersPerBarrier.set(itersPerBarrierChoices[15]);
    //    barrierComboBox.setSelectedIndex(15);

    return barrierComboBox;
  }
  
  JComboBox exchangeBox() {
    final int[] exchangerChoices = { 
      1,
      2,
      4,
      8,
      16,
      32,
      64,
      128,
      256,
      512,
      1024,
    };
    
    JComboBox exchComboBox = new JComboBox();
    
    for (int j = 0; j < exchangerChoices.length; ++j) {
      String lab = p2ToString(exchangerChoices[j]) + 
        " max threads per barrier";
      exchComboBox.addItem(lab);
    }
    exchComboBox.addItemListener(new ItemListener() {
      public void itemStateChanged(ItemEvent evt) {
        JComboBox src = (JComboBox)(evt.getItemSelectable());
        int idx = src.getSelectedIndex();
        RNG.exchangeParties.set(exchangerChoices[idx]);
      }
    });
    
    RNG.exchangeParties.set(exchangerChoices[1]);
    exchComboBox.setSelectedIndex(1);
    return exchComboBox;
  }
  
  JComboBox biasBox() {
    final int[] biasChoices = { 
      -1, 
      0, 
      1 
    };
    
    
    JComboBox biasComboBox = new JComboBox();
    
    for (int j = 0; j < biasChoices.length; ++j) {
      String lab = biasToString(biasChoices[j]);
      biasComboBox.addItem(lab);
    }
    biasComboBox.addItemListener(new ItemListener() {
      public void itemStateChanged(ItemEvent evt) {
        JComboBox src = (JComboBox)(evt.getItemSelectable());
        int idx = src.getSelectedIndex();
        RNG.bias.set(biasChoices[idx]);
      }
    });
    
    RNG.bias.set(biasChoices[1]);
    biasComboBox.setSelectedIndex(1);
    return biasComboBox;
  }
  
  JComboBox capacityBox() {
    
    final int[] bufferCapacityChoices = {
      1,
      4,
      64,
      256,
      1024,
      4096,
      16 * 1024,
      64 * 1024,
      256 * 1024,
      1024 * 1024,
    };
    
    JComboBox bcapComboBox = new JComboBox();
    
    for (int j = 0; j < bufferCapacityChoices.length; ++j) {
      String lab = p2ToString(bufferCapacityChoices[j]) + 
        " element bounded buffers";
      bcapComboBox.addItem(lab);
    }
    bcapComboBox.addItemListener(new ItemListener() {
      public void itemStateChanged(ItemEvent evt) {
        JComboBox src = (JComboBox)(evt.getItemSelectable());
        int idx = src.getSelectedIndex();
        DefaultChannelCapacity.set(bufferCapacityChoices[idx]);
      }
    });
    
    
    DefaultChannelCapacity.set(bufferCapacityChoices[3]);
    bcapComboBox.setSelectedIndex(3);
    return bcapComboBox;
  }
  
  JComboBox timeoutBox() {
    
    
    final long[] timeoutChoices = {
      0,
      1,
      10,
      100,
      1000,
      10000,
      100000,
    };
    
    
    JComboBox timeoutComboBox = new JComboBox();
    
    for (int j = 0; j < timeoutChoices.length; ++j) {
      String lab = timeoutChoices[j] + " msec timeouts";
      timeoutComboBox.addItem(lab);
    }
    timeoutComboBox.addItemListener(new ItemListener() {
      public void itemStateChanged(ItemEvent evt) {
        JComboBox src = (JComboBox)(evt.getItemSelectable());
        int idx = src.getSelectedIndex();
        RNG.timeout.set(timeoutChoices[idx]);
      }
    });
    
    RNG.timeout.set(timeoutChoices[3]);
    timeoutComboBox.setSelectedIndex(3);
    return timeoutComboBox;
  }

  ClockDaemon timeDaemon = new ClockDaemon();
  
  void startPoolStatus(final JLabel status) {
    Runnable updater = new Runnable() {
      int lastps = 0;
      public void run() {
        final int ps = Threads.activeThreads.get();
        if (lastps != ps) {
          lastps = ps;
          SwingUtilities.invokeLater(new Runnable() {
            public void run() {
              status.setText("Active threads: " + ps);
            } } );
        }
      }
    };
    timeDaemon.executePeriodically(250, updater, false);
  }

  private final SynchronizedRef contention_ = new SynchronizedRef(null);
  private final SynchronizedInt loopsPerTest_ = new SynchronizedInt(0);

  private final SynchronizedBoolean echoToSystemOut = 
      new SynchronizedBoolean(false);


  private final JButton startstop_ = new JButton("Start");
  
  private WaitableInt testNumber_ = new WaitableInt(1);

  private void runOneTest(Runnable tst) throws InterruptedException { 
    int nt = testNumber_.get(); 
    Threads.pool.execute(tst);
    testNumber_.whenNotEqual(nt, null);
  }

  private void endOneTest() {
    testNumber_.increment();
  }

  private SynchronizedBoolean running_ = new SynchronizedBoolean(false);

  void cancel() { 
    //  not stable enough to cancel during construction
    synchronized (RNG.constructionLock) {
      try {
        Threads.pool.interruptAll();
      }
      catch(Exception ex) {
        System.out.println("\nException during cancel:\n" + ex);
        return;
      }
    }
  }


  void startTestSeries(Runnable tst) throws InterruptedException {
    running_.set(true);
    startstop_.setText("Stop");
    Threads.pool.execute(tst);
  }

  // prevent odd class-gc problems on some VMs?
  class PrintStart implements Runnable {
    public void run() {
      startstop_.setText("Start");
    } 
  } 


  void endTestSeries() {
    running_.set(false);
    SwingUtilities.invokeLater(new PrintStart());
  }

  /*
  void old_endTestSeries() {
    running_.set(false);
    SwingUtilities.invokeLater(new Runnable() {
      public void run() {
        startstop_.setText("Start");
      } } );
  }
  */

  class TestSeries implements Runnable {
    final int firstclass;
    final int firstnthreads;

    TestSeries() { 
      firstclass = 0;
      firstnthreads = 0;
    }

    TestSeries(final int firstc, final int firstnt) { 
      firstclass = firstc;
      firstnthreads = firstnt;
    }

    public void run() {
      Thread.currentThread().setPriority(Thread.NORM_PRIORITY);

      try {
        int t = firstnthreads; 
        int c = firstclass;

        if (t < nthreadsChoices.length &&
            c < TestedClass.classes.length) {

          for (;;) {

            
            // these checks are duplicated in OneTest, but added here
            // to minimize unecessary thread construction, 
            // which can skew results

            if (threadEnabled(t)) {

              TestedClass entry = TestedClass.classes[c];
        
              int nthreads = nthreadsChoices[t];
              int iters = loopsPerTest_.get();
              Fraction pshr = (Fraction)(contention_.get());
        
              if (entry.isEnabled(nthreads, pshr)) {

                runOneTest(new OneTest(c, t));
              }
            }

            if (++c >= TestedClass.classes.length) {
              c = 0;
              if (++t >= nthreadsChoices.length) 
                break;
            }

            nextClassIdx_.set(c);
            nextThreadIdx_.set(t);
            
          }
        }

      }
      catch (InterruptedException ex) { 
        Thread.currentThread().interrupt();
      }
      finally {
        endTestSeries();
      }
    }
  }

  static class BarrierTimer implements Runnable {
    private long startTime_ = 0;
    private long endTime_ = 0;

    public synchronized long getTime() {
      return endTime_ - startTime_;
    }

    public synchronized void run() {
      long now = System.currentTimeMillis();
      if (startTime_ == 0) 
        startTime_ = now;
      else
        endTime_ = now;
    }
  }
      
  class OneTest implements Runnable {
    final int clsIdx; 
    final int nthreadsIdx; 

    OneTest(int idx, int t) {
      clsIdx = idx; 
      nthreadsIdx = t; 
    }

    public void run() {
      Thread.currentThread().setPriority(Thread.NORM_PRIORITY-3);

      boolean wasInterrupted = false;

      final TestedClass entry = TestedClass.classes[clsIdx];

      final JLabel cell = (JLabel)(resultTable_[clsIdx+1][nthreadsIdx+1]);
      final Color oldfg =  cell.getForeground();

      try {


        if (Thread.interrupted()) return;
        if (!threadEnabled(nthreadsIdx)) return;
        
        int nthreads = nthreadsChoices[nthreadsIdx];
        int iters = loopsPerTest_.get();
        Fraction pshr = (Fraction)(contention_.get());
        
        if (!entry.isEnabled(nthreads, pshr))  return;

        BarrierTimer timer = new BarrierTimer();
        CyclicBarrier barrier = new CyclicBarrier(nthreads+1, timer);

        Class cls = entry.cls;
        Class chanCls = entry.buffCls;

        try {
          SwingUtilities.invokeAndWait(new Runnable() {
            public void run() {
              cell.setForeground(Color.blue);
              cell.setText("RUN");
              cell.repaint();
            }
          });
        }
        catch (InvocationTargetException ex) {
          ex.printStackTrace();
          System.exit(-1);
        }
        synchronized (RNG.constructionLock) {
          RNG.reset(nthreads);

          if (chanCls == null) {
            RNG shared = (RNG)(cls.newInstance());
            for (int k = 0; k < nthreads; ++k) {
              RNG pri = (RNG)(cls.newInstance());
              TestLoop l = new TestLoop(shared, pri, pshr, iters, barrier);
              Threads.pool.execute(l.testLoop());
            }
          }
          else {
            Channel shared = (Channel)(chanCls.newInstance());
            if (nthreads == 1) {
              ChanRNG single = (ChanRNG)(cls.newInstance());
              single.setSingle(true);
              PCTestLoop l = new PCTestLoop(single.getDelegate(), single, pshr,
                                            iters, barrier,
                                            shared, shared);
              Threads.pool.execute(l.testLoop(true));
            }
            else if (nthreads % 2 != 0) 
              throw new Error("Must have even number of threads!");
            else {
              int npairs = nthreads / 2;
              
              for (int k = 0; k < npairs; ++k) {
                ChanRNG t = (ChanRNG)(cls.newInstance());
                t.setSingle(false);
                Channel chan = (Channel)(chanCls.newInstance());
                
                PCTestLoop l = new PCTestLoop(t.getDelegate(), t, pshr, 
                                              iters, barrier,
                                              shared, chan);
                
                Threads.pool.execute(l.testLoop(false));
                Threads.pool.execute(l.testLoop(true));
                
              }
            }
          }

          if (echoToSystemOut.get()) {
            System.out.print(
                             entry.name + " " +
                             nthreads + "T " +
                             pshr + "S " +
                             RNG.computeLoops.get() + "I " +
                             RNG.syncMode.get() + "Lm " +
                             RNG.timeout.get() + "TO " +
                             RNG.producerMode.get() + "Pm " +
                             RNG.consumerMode.get() + "Cm " +
                             RNG.bias.get() + "B " +
                             DefaultChannelCapacity.get() + "C " +
                             RNG.exchangeParties.get() + "Xp " +
                             RNG.itersPerBarrier.get() + "Ib : "
                             );
          }