except Main, wrote comments

Author: wolfram77

CombiningTree.java

@@ -1,8 +1,28 @@
 import java.util.function.*;
 
 
+// A Combining Tree is an N-ary tree of nodes, that follows
+// software combining to reduce memory contention while
+// updating a shared value.
+// 
+// The shared value is placed at the root of the tree, and
+// threads perfrom getAndOp() at the leaf nodes. Each leaf
+// node handles N threads. The combined value is then
+// propagated up the tree by an active thread. There can be
+// several active threads, but eventually one active thread
+// updates the root node. It then shares this message to all
+// other threads behind it.
+// 
+// This was contention at a single memory location is avoided.
+// However, i guess that such a design is normally useful
+// only in hardware, as it becomes too slow in software.
+// Useful for educational purposes.
+
 class CombiningTree<T> {
+  Node<T> root;
   Node<T>[] leaf;
+  // root: root node
+  // leaf: leaf nodes
 
 
   public CombiningTree(int depth) {
@@ -11,11 +31,13 @@ public CombiningTree(int depth) {
 
   public CombiningTree(int depth, int ary) {
     Node<T>[] parent = createNodes(1, ary);
+    root = parent[0];
     for (int i=1; i<depth; i++) {
       int n = (int) Math.pow(ary, i);
       leaf = createNodes(n, ary);
       for (int j=0; j<n; j++)
         leaf[j].parent = parent[j/ary];
+      parent = leaf;
     }
   }
 
@@ -28,14 +50,32 @@ private Node<T>[] createNodes(int n, int ary) {
   }
 
 
+  // Get value of tree (at root).
+  public T get() {
+    return root.get();
+  }
+
+  // Set value of tree (at root).
+  public void set(T x) {
+    root.set(x);
+  }
+
+
+  // Gets current value, and then updates it.
+  // x: value to OP with, op: binary op
+  // 1. Select leaf index using thread id.
+  // 2. Perform get & op at desired leaf.
   public T getAndOp(T x, BinaryOperator<T> op)
   throws InterruptedException {
-    int i = (int) Thread.currentThread().getId();
-    return getAndOp(x, op, i);
+    int i = (int) Thread.currentThread().getId(); // 1
+    return getAndOp(x, op, i); // 2
   }
 
+  // Gets current value, and then updates it.
+  // x: value to OP with, op: binary op, i: leaf index
+  // 1. Perform get & op at desired leaf (ensure in limit).
   public T getAndOp(T x, BinaryOperator<T> op, int i)
   throws InterruptedException {
-    return leaf[i % leaf.length].getAndOp(x, op);
+    return leaf[i % leaf.length].getAndOp(x, op); // 1
   }
 }

Main.java

@@ -1,99 +1,64 @@
 import java.util.*;
+import java.util.concurrent.ConcurrentLinkedQueue;
 
 class Main {
-  static Deque<Integer> queue;
-  static CombiningTree<Integer> concurrentQueue;
-  static List<Integer>[] deqValues;
-  static int TH = 10, NUM = 1000;
-
-  // Each unsafe thread enqs N numbers and deqs N, adding
-  // them to its own deqValues for checking; using Java's
-  // sequential queue implementation, ArrayDeque.
-  static Thread unsafe(int id, int x, int N) {
-    return new Thread(() -> {
-      String action = "enq";
-      try {
-      for (int i=0, y=x; i<N; i++)
-        queue.addLast(y++);
-      Thread.sleep(1000);
-      action = "deq";
-      for (int i=0; i<N; i++)
-        deqValues[id].add(queue.removeFirst());
-      }
-      catch (Exception e) { log(id+": failed "+action); }
-    });
-  }
+  static Queue<Integer> queue;
+  static CombiningTree<Integer> tree;
+  static int TH = 8, NUM = 10;
 
   // Each safe thread enqs N numbers and deqs N, adding
   // them to its own deqValues for checking; using
   // ArrayQueue.
-  static Thread safe(int id, int x, int N) {
+  static Thread thread(int id) {
     return new Thread(() -> {
-      String action = "enq";
       try {
-      for (int i=0, y=x; i<N; i++)
-        concurrentQueue.enq(y++);
-      Thread.sleep(1000);
-      action = "deq";
-      for (int i=0; i<N; i++)
-        deqValues[id].add(concurrentQueue.deq());
+      for (int i=0; i<NUM; i++) {
+        Integer r = tree.getAndOp(1,
+          (Integer x, Integer y) -> x);
+        queue.add(r);
       }
-      catch (Exception e) { log(id+": failed "+action);
-      e.printStackTrace(); }
+      } catch (InterruptedException e) {}
     });
   }
 
   // Checks if each thread dequeued N values, and they are
   // globally unique.
-  static boolean wasLIFO(int N) {
-    Set<Integer> set = new HashSet<>();
-    boolean passed = true;
-    for (int i=0; i<TH; i++) {
-      int n = deqValues[i].size();
-      if (n != N) {
-        log(i+": dequeued "+n+"/"+N+" values");
-        passed = false;
-      }
-      for (Integer x : deqValues[i])
-        if (set.contains(x)) {
-          log(i+": has duplicate value "+x);
-          passed = false;
-        }
-      set.addAll(deqValues[i]);
+  static boolean wasValid() {
+    int a = tree.get().intValue();
+    if (a != TH*NUM) return false;
+    Set<Integer> s = new HashSet<>();
+    while (queue.size()>0) {
+      Integer n = queue.remove();
+      if (s.contains(n)) return false;
+      s.add(n);
     }
-    return passed;
+    return true;
   }
 
-  @SuppressWarnings("unchecked")
-  static void testThreads(boolean safe) {
-    queue = new ArrayDeque<>();
-    concurrentQueue = new CombiningTree<>(TH*NUM);
-    deqValues = new List[TH];
+  static Thread[] startOps() {
+    Thread[] t = new Thread[TH];
     for (int i=0; i<TH; i++)
-      deqValues[i] = new ArrayList<>();
-    Thread[] threads = new Thread[TH];
-    for (int i=0; i<TH; i++) {
-      threads[i] = safe?
-        safe(i, i*NUM, NUM) :
-        unsafe(i, i*NUM, NUM);
-      threads[i].start();      
-    }
+      t[i] = thread(i);
+    for (int i=0; i<TH; i++)
+      t[i].start();
+    return t;
+  }
+
+  static void awaitOps(Thread[] t) {
     try {
     for (int i=0; i<TH; i++)
-      threads[i].join();
-    }
-    catch (Exception e) {}
+      t[i].join();
+    } catch (InterruptedException e) {}
   }
 
   public static void main(String[] args) {
-    log("Starting "+TH+" threads with sequential queue");
-    testThreads(false);
-    log("Was LIFO? "+wasLIFO(NUM));
-    log("");
-    log("Starting "+TH+" threads with array queue");
-    testThreads(true);
-    log("Was LIFO? "+wasLIFO(NUM));
-    log("");
+    queue = new ConcurrentLinkedQueue<>();
+    tree = new CombiningTree<>(3);
+    tree.set(0);
+    log("Starting "+TH+" threads doing ops ...");
+    Thread[] t = startOps();
+    awaitOps(t);
+    log("\nWas valid? "+wasValid());
   }
 
   static void log(String x) {