基于三元组合树的并发计数器实现

骑猪看日落 2022-08-19 08:29 129阅读 0赞

课本上给了二元的组合树计数器，但是三元情况有很大不同，经过一番摸索，写出了可以运行的三元组合树，代码可能不是那么优雅，欢迎交流讨论。

完整代码如下：

package Combine;
    
    
    /**
     * ThrilNode declaration for software thril-combining tree.
     * @author juneshen
     * based on the work of Maurice Herlihy
     */
    public class ThrilNode {
      enum CStatus{IDLE, FIRST, SECOND,THIRD,RESULT_SECOND, RESULT_THIRD,ROOT};
      boolean locked;   // is ThrilNode locked?
      boolean locked1;
      CStatus cStatus;  // combining status
      int firstValue, secondValue,thirdValue; // values to be combined
      int result;       // result of combining
      ThrilNode parent;      // reference to parent
      
      /** Creates a root ThrilNode */
      public ThrilNode() {
        cStatus = CStatus.ROOT;
        locked = false;
        locked1=false;
      }
      /** Create a non-root ThrilNode */
      public ThrilNode(ThrilNode _parent) {
        parent = _parent;
        cStatus = CStatus.IDLE;
        locked = false;
        locked1=false;
      }
      
      synchronized boolean precombine() throws InterruptedException {
        while (locked||locked1) wait();
    
        switch (cStatus) {
          case IDLE:
            cStatus = CStatus.FIRST;
            return true;
          case FIRST://前面已经有一个进程冲上去了，自己只能等待，目前有两个线程访问过这个节点
            locked = true;
            cStatus = CStatus.SECOND;
            return false;
    	  case SECOND:  //前面已经有两个进程冲上去了，自己只能等待，目前有三个线程访问过这个节点
            locked = true;
            cStatus = CStatus.THIRD;
            return false;
          case ROOT://冲到头了，停止往上冲
            return false;
          default:
            throw new PanicException("in precombine unexpected ThrilNode state " + cStatus);
        }
      }
      
      synchronized int combine(int combined) throws InterruptedException {
        while (locked) wait();
        locked = true;
        locked1=true;
        firstValue = combined;
        switch (cStatus) {
          case FIRST:
            return firstValue;//就只有一个线程访问这个节点，直接返回自己的请求值
          case SECOND:
            return firstValue+secondValue;//把自己的请求和前面的一个线程的请求合并
    	  case THIRD:
            return firstValue+secondValue+thirdValue;//把自己的请求和前2个的线程的请求合并
          default:
            throw new PanicException("in combine unexpected ThrilNode state " + cStatus);
        }
      }
      
      synchronized int op(int combined) throws InterruptedException {
        switch (cStatus) {//操作阶段，线程2和线程3都变成被动线程，提交自己的请求，等待获胜的主动线程把组合的结果带下来
          case ROOT:
            int oldValue = result;
            result += combined;
            return oldValue;
          case SECOND:
              secondValue = combined;
              locked = false;
              locked1=true;
              notifyAll();
              while (cStatus != CStatus.RESULT_SECOND) wait();
              cStatus = CStatus.IDLE;
              
              locked1 = false;
              locked = false;
              notifyAll();
              
          //    System.out.println("second in op lock is free");
              return result+firstValue;
          case THIRD:
            thirdValue = combined;
            locked = false;
            locked1=true;
            notifyAll();
            while (cStatus != CStatus.RESULT_THIRD) wait();
            locked = true;
            notifyAll();
            cStatus = CStatus.RESULT_SECOND;
      //      System.out.println("third in op lock is free");
            return result+firstValue+secondValue;
          default:
            throw new PanicException("in op unexpected ThrilNode state "+cStatus);
        }
      }
      synchronized void distribute(int prior) throws InterruptedException {
        switch (cStatus) {
          case FIRST:
            cStatus = CStatus.IDLE;
            locked1 = false;
            locked=false;
            break;
          case SECOND:
            result = prior ;
            cStatus = CStatus.RESULT_SECOND;
            locked=true;//6666
          //  System.out.println("in dis second"+result);
            break;
    	  case THIRD:
    		locked=true;
            result = prior;
            cStatus = CStatus.RESULT_THIRD;
            
            break;
          default:
            throw new PanicException("in distribute unexpected ThrilNode state");
        }
        notifyAll();
      }
      
    }

package Combine;
    /*
     * Combine.java
     *
     * 
     */
    import java.util.Stack;
    import java.math.*;
    /**
     *
     * @author juneshen
     * based on the work of mph.
     */
    public class ThrilTree  {
    
        
      ThrilNode[] leaf;
      
      /** Creates a new instance of Combine */     
      public ThrilTree(int size) {
    	 
        ThrilNode[] nodes = new ThrilNode[(size - 1)/2];
        nodes[0] = new ThrilNode();
        for (int i = 1; i < nodes.length; i++) {
          nodes[i] = new ThrilNode(nodes[(i-1)/3]);//三叉完全树的性质，i节点的父节点为[(i-1)/3]
        }
        leaf = new ThrilNode[(size)/3];//叶子节点的数目
        for (int i = 0; i < leaf.length; i++) {
          leaf[i] = nodes[nodes.length - i - 1];
        }
      }
      
      public int getAndIncrement() throws InterruptedException {
        Stack<ThrilNode> stack = new Stack<ThrilNode>();
        ThrilNode myLeaf = leaf[ThreadID.get() / 3];
        ThrilNode node = myLeaf;
        // phase one
        while (node.precombine()) {
      //    System.out.println("t id="+ThreadID.get());
          node = node.parent;
        }
        ThrilNode stop = node;
        // phase two
        node = myLeaf;
        int combined = 1;
        while (node != stop) {
          combined = node.combine(combined);
          stack.push(node);
          node = node.parent;
        }
        // phase 3
        int prior = stop.op(combined);
        // phase 4
        while (!stack.empty()) {
          node = stack.pop();
          node.distribute(prior);
        }
        return prior;
      }
      
    }

/*
     * TreeTest.java
     * JUnit based test
     *
     * Created on June 16, 2006, 9:32 PM
     */
    
    package Combine;
    
    import junit.framework.*;
    import java.util.Stack;
    
    /**
     *
     * @author mph
     */
    public class TreeTest extends TestCase {
      
      final static int THREADS = 27;
      final static int TRIES = 1024*1024;      
      static boolean[] test = new boolean[THREADS * TRIES];
      ThrilTree instance = new ThrilTree(THREADS);
      Thread[] thread = new Thread[THREADS];
      
      public TreeTest(String testName) {
        super(testName);
      }
    
      public static Test suite() {
        TestSuite suite = new TestSuite(TreeTest.class);
        
        return suite;
      }
    
      public void testGetAndIncrement() throws Exception {
        System.out.printf("Parallel, %d threads, %d tries\n", THREADS, TRIES);
        
        long startTime=System.currentTimeMillis();   //获取开始时间
        for (int i = 0; i < THREADS; i++) {
          thread[i] = new MyThread();
        }
        for (int i = 0; i < THREADS; i ++) {
          thread[i].start();
        }
        for (int i = 0; i < THREADS; i ++) {
          thread[i].join();
        }
        check(test);
        
        long endTime=System.currentTimeMillis(); //获取结束时间  
        
        
        double elapsedTime = ((double) (endTime - startTime)) / 1000.0;
        double throughput = (TRIES*THREADS / elapsedTime);
        System.out.printf(" 吞吐量: %f, 运行时间: %f \n",throughput,elapsedTime);
      }
    
      
       class MyThread extends Thread {
        public void run() {
          try {
            for (int j = 0; j < TRIES; j++) {
              int i = instance.getAndIncrement();
              if (test[i]) {
                System.out.printf("ERROR duplicate value %d\n", i);
              } else {
                test[i] = true;
              }
            }
          } catch (InterruptedException e) {}
        }
      }
       
       void check(boolean[] test) throws Exception {
         for (int i = 0; i < test.length; i++) {
           if (!test[i]) {
             System.out.println("missing value at " + i);
             throw new Exception();
           }
         }
       }
      
    }

package Combine;
    /**
     * Thrown by an attempt to open a <code>TMObject</code> to indicate
     * that the current transaction cannot commit.
     **/
    public class PanicException extends java.lang.RuntimeException {
      /**
       * Creates a new <code>PanicException</code> instance with no detail message.
       */
      public PanicException() {
        super();
      }  
      /**
       * Creates a new <code>Panic</code> instance with the specified detail message.
       * @param msg the detail message.
       */
      public PanicException(String msg) {
        super(msg);
      }
    }

package Combine;
    
    
    public class ThreadID {
      // The next thread ID to be assigned.
      private static volatile int nextID = 0;
      // my thread-local ID
      private static LocalID threadID = new LocalID();
      public static int get() {
        return threadID.get();
      }
      static class LocalID extends ThreadLocal<Integer> {    
        protected synchronized Integer initialValue() {
          return nextID++;
        }
      }
    }