programming done :)

Author: wolfram77

CombiningTree.java

@@ -32,6 +32,7 @@ public CombiningTree(int depth) {
   public CombiningTree(int depth, int ary) {
     Node<T>[] parent = createNodes(1, ary);
     root = parent[0];
+    leaf = parent;
     for (int i=1; i<depth; i++) {
       int n = (int) Math.pow(ary, i);
       leaf = createNodes(n, ary);

Main.java

@@ -1,28 +1,39 @@
 import java.util.*;
-import java.util.concurrent.ConcurrentLinkedQueue;
+import java.util.concurrent.*;
+
 
 class Main {
   static Queue<Integer> queue;
   static CombiningTree<Integer> tree;
-  static int TH = 8, NUM = 10;
+  static int TH = 25, ARY = 3, NUM = 100;
+  // queue: used to store old (get) values (unique check)
+  // tree: combining tree where threads do increment ops
+  // TH: number of threads
+  // ARY: arity of combining tree
+  // NUM: number of increment ops each thread performs
+
 
-  // Each safe thread enqs N numbers and deqs N, adding
-  // them to its own deqValues for checking; using
-  // ArrayQueue.
+  // Each thread performs increment op using the combining
+  // tree, and saves old (get) values in the queue, for
+  // uniqueness check later.
   static Thread thread(int id) {
     return new Thread(() -> {
       try {
+      long start = System.currentTimeMillis();
       for (int i=0; i<NUM; i++) {
         Integer r = tree.getAndOp(1,
-          (Integer x, Integer y) -> x);
+          (Integer x, Integer y) -> x + y);
         queue.add(r);
+        Thread.yield();
       }
+      long stop = System.currentTimeMillis();
+      log(id()+": done in "+(stop-start)+"ms");
       } catch (InterruptedException e) {}
     });
   }
 
-  // Checks if each thread dequeued N values, and they are
-  // globally unique.
+  // Check if total sum is as expected.
+  // Check if all old values are unique.
   static boolean wasValid() {
     int a = tree.get().intValue();
     if (a != TH*NUM) return false;
@@ -35,6 +46,20 @@ static boolean wasValid() {
     return true;
   }
 
+  // Setup the combining tree for threads.
+  static void setupTreeAndQueue() {
+    int depth = (int) Math.ceil(Math.log(TH)/Math.log(ARY));
+    tree = new CombiningTree<>(depth, ARY);
+    tree.set(0);
+  }
+
+  // Setup the queue for storing old values.
+  static void setupQueue() {
+    queue = new ConcurrentLinkedQueue<>();
+  }
+
+
+  // Start threads doing increments using tree.
   static Thread[] startOps() {
     Thread[] t = new Thread[TH];
     for (int i=0; i<TH; i++)
@@ -44,24 +69,32 @@ static Thread[] startOps() {
     return t;
   }
 
+  // Wait until all threads done with increments.
   static void awaitOps(Thread[] t) {
     try {
     for (int i=0; i<TH; i++)
       t[i].join();
     } catch (InterruptedException e) {}
   }
 
+
   public static void main(String[] args) {
-    queue = new ConcurrentLinkedQueue<>();
-    tree = new CombiningTree<>(3);
-    tree.set(0);
-    log("Starting "+TH+" threads doing ops ...");
+    setupTree();
+    setupQueue();
+    log(ARY+"-ary "+depth+"-depth Combining tree.");
+    log("Starting "+TH+" threads doing increments ...");
     Thread[] t = startOps();
     awaitOps(t);
+    log("Total: "+tree.get());
     log("\nWas valid? "+wasValid());
   }
 
+
   static void log(String x) {
     System.out.println(x);
   }
+
+  static long id() {
+    return Thread.currentThread().getId();
+  }
 }

Node.java

@@ -24,7 +24,7 @@ class Node<T> {
   // FREE: indicates that values are still being inserted
   // PUSHING: node is full, combined value being pushed
   // PULLED: pulled value from parent being given to threads
-  // TIMEOUT: maximum time active thread waits for node to fill
+  // TIMEOUT: max time active thread waits for node to fill
   // parent: parent node
   // state: either FREE, PUSHING, or PULLING
   // count: number of values in node
@@ -60,16 +60,18 @@ public synchronized void set(T x) {
 
   // Gets current value, and then updates it.
   // x: value to OP (accumulate), op: binary operator
-  // 1. Perform get & op based on 3 possible cases.
-  // 1a. Root node
-  // 1b. Active thread (first to visit node)
-  // 1c. Passive thread (visits later)
+  // 1. Wait until node is free.
+  // 2. Perform get & op based on 3 possible cases.
+  // 2a. Root node
+  // 2b. Active thread (first to visit node)
+  // 2c. Passive thread (visits later)
   public synchronized T getAndOp(T x,
   BinaryOperator<T> op)
   throws InterruptedException {
-    if (parent==null) return getAndOpRoot(x, op); // 1a
-    if (count==0) return getAndOpActive(x, op);   // 1b
-    return getAndOpPassive(x, op); // 1c
+    while (state!=FREE || count==size) wait(); // 1
+    if (parent==null) return getAndOpRoot(x, op); // 2a
+    if (count==0) return getAndOpActive(x, op);   // 2b
+    return getAndOpPassive(x, op); // 2c
   }
 
   // Performs get & op for root node.
@@ -108,7 +110,8 @@ private synchronized T getAndOpActive(T x,
     T r = parent.getAndOp(a, op); // 5
     distribute(r, op); // 6
     state = PULLING;   // 7
-    count--;  // 8
+    notifyAll();       // 7
+    decrementCount();  // 8
     return r; // 9
   }
 
@@ -124,7 +127,7 @@ private synchronized T getAndOpPassive(T x,
   throws InterruptedException {
     int i = insert(x); // 1
     while (state!=PULLING) wait(); // 2
-    if (--count==0) state = FREE;  // 3, 4
+    decrementCount(); // 3, 4
     return value[i]; // 5
   }
 
@@ -133,16 +136,16 @@ private synchronized T getAndOpPassive(T x,
   // x: value to insert
   // 1. Wait unit node is free.
   // 2. Get index to place value in.
-  // 3. Increment number of values in node.
-  // 4. Place the value.
+  // 3. Place the value.
+  // 4. Increment number of values in node.
   // 5. If node is full, notify active thread.
   // 6. Return index where value was placed.
   public synchronized int insert(T x)
   throws InterruptedException {
     while (state!=FREE) wait(); // 1
-    int i = count++; // 2, 3
-    value[i] = x;    // 4
-    if (count==size) notifyAll(); // 5
+    int i = count;    // 2
+    value[i] = x;     // 3
+    incrementCount(); // 4, 5
     return i; // 6
   }
 
@@ -173,6 +176,22 @@ public synchronized void distribute(T r,
     } // 3
   }
 
+  // Increment count once done with insertion.
+  // 1. Increment count.
+  // 2. If node is full, notify active thread.
+  private synchronized void incrementCount() {
+    if (++count<size) return; // 1
+    notifyAll(); // 2
+  }
+
+  // Decrement count once done with distribution.
+  // 1. Decrement count.
+  // 2. If count is zero, node is free.
+  private synchronized void decrementCount() {
+    if (--count>0) return; // 1
+    state = FREE; // 2
+    notifyAll();  // 2
+  }
 
   // Wait until node is full, or timeout.
   // 1. Get start time.

README.md

@@ -31,27 +31,36 @@ tryDeq():
 
 ```bash
 ## OUTPUT
-Starting 10 threads with sequential queue
-2: failed enq
-5: failed deq
-6: failed deq
-7: failed deq
-8: failed deq
-1: failed deq
-4: failed deq
-9: failed deq
-1: dequeued 0/1000 values
-2: dequeued 0/1000 values
-4: dequeued 0/1000 values
-5: dequeued 698/1000 values
-6: dequeued 0/1000 values
-7: dequeued 0/1000 values
-8: dequeued 0/1000 values
-9: dequeued 0/1000 values
-Was LIFO? false
+3-ary 3-depth Combining tree.
+Starting 25 threads doing increments ...
+30: done in 51ms
+14: done in 60ms
+13: done in 68ms
+23: done in 53ms
+12: done in 56ms
+21: done in 54ms
+32: done in 56ms
+31: done in 56ms
+22: done in 54ms
+28: done in 7831ms
+11: done in 7836ms
+10: done in 7838ms
+19: done in 7833ms
+20: done in 7833ms
+29: done in 7831ms
+24: done in 8165ms
+33: done in 8160ms
+25: done in 8165ms
+16: done in 8167ms
+34: done in 8159ms
+15: done in 8167ms
+26: done in 11970ms
+17: done in 11972ms
+27: done in 12239ms
+18: done in 12241ms
+Total: 2500
 
-Starting 10 threads with array queue
-Was LIFO? true
+Was valid? true
 ```
 
 See [ArrayQueue.java] for code, [Main.java] for test, and [repl.it] for output.