Advanced tutorial "Encode Concurrency Blocks in RhoLang"

[golovach-ivan]

Create tutorial specialized in concurrency.
Lets encode something "classic", widespread, popular, known to people.

Lets encode java.util.concurrent concurrency building blocks.

Plan A (good level)

{ Runnable, Callable, Future, Executor, AtomicInteger, Lock, Semaphore, LinkedBlockingQueue, CountDownLatch, Exchanger }.

Plan B (high level)

Plan A + { CompletableFuture, ReentrantLock, ReadWriteLock, ConcurrentMap, CyclicBarrier, Phaser }.

Benefit to RChain

1. Community needs more RhoLang tutorials.
2. Tech info for meetups and social media.

Budget and Objective

Estimated Budget of Task: 3000$ for Plan A, 5000$ for Plan B.
Estimated Timeline Required to Complete the Task: 20-30 days
How will we measure completion? Plan A: 10 pages with described implementations + intro on GitHub (11 pages), Plan B: 15 pages with described implementations + intro on GitHub (16 pages).

[JoshOrndorff]

@golovach-ivan This is an interesting idea. I don't fully understand it, but I would like to. The same is true about some of your other projects. Can we get together for a call next week? I'm @JoshyOrndorff on discord, or you can answer here.

[JoshOrndorff]

@golovach-ivan contacted me via discord DM saying

Hi.
Sorry, was too concentrated on tutorial writting (one more RhoLang tutorial:) ).

I decided write something non-trivial

1. to learn RhoLang
2. to collect code for formal verification

And i rewrite java.util.concurrent in RhoLang
I developed simple pattern (Atomic State) and Wait Set impl is easy in RhoLang, so
https://github.com/golovach-ivan/Correct-by-Construction/tree/master/rho-j.u.c was born

For example - Semaphore (with explicit waitSet) = https://github.com/golovach-ivan/Correct-by-Construction/blob/master/rho-j.u.c/Semaphore.md

contract Semaphore(@initPermits, acquire, release) = {
new stateRef in {

  stateRef!(initPermits, []) |

  contract acquire(ack) = {
    for (@permits, @waitSet <- stateRef) { 
      if (permits > 0) {
        stateRef!(initPermits - 1, waitSet) |
        ack!(Nil)
      } else {
        stateRef!(initPermits, waitSet ++ [*ack])
      }
    }
  } |

  contract release(_) = {
    for (@permits, @waitSet <- stateRef) {
      match waitSet {
        [ack...waitSetTail] => { 
          stateRef!(permits, waitSetTail) |
          @ack!(Nil) }
        [] => 
          stateRef!(permits + 1, waitSet)
      }
    }
  }
}

}
This project is part of #991

If this is interesting - i need sponsorship for this issue for Sep
In my opinion i did 70%-80% of Plan A(3k$), so my bounty maybe 2k$-2.5k$

Ivan, I am interested in learning more about this. But I won't sponsor it until I clearly understand it. When can you offer a walkthrough for me and others who may be interested like @David405 @dckc

[golovach-ivan]

Issue basics: any language with implicit concurrency (RhoLang, Go, Java, etc) needs

1. Concurrency Primitives library to add some ordering between actions.
2. One or more concurrency theories for program structuring.

Example: Ordered Message Queue

In Java for Ordered Message Queue between concurrent activities (threads) someone can

• chose some impl of java.util.concurrent.BlockingQueue (ArrayBlockingQueue, DelayQueue, LinkedBlockingDeque, LinkedBlockingQueue, LinkedTransferQueue, PriorityBlockingQueue, SynchronousQueue)
• implement hand-made based on implicit or explicit Monitors

BlockingQueue backed by array based on explicit Monitor (Lock/Condition)

class BlockingQueue<T> {
    private final Lock lock = new ReentrantLock();
    private final Condition notFull = lock.newCondition();
    private final Condition notEmpty = lock.newCondition();

    private final int maxSize;
    private final Queue<T> queue = new LinkedList<>();

    public BlockingQueue(int maxSize) {
        this.maxSize = maxSize;
    }

    public void put(T item) throws InterruptedException {
        lock.lock();
        try {
            while (queue.size() == maxSize) {
                notFull.await();
            }
            notEmpty.signal();
            queue.add(item);
        } finally {
            lock.unlock();
        }
    }

    public T take() throws InterruptedException {
        lock.lock();
        try {
            while (queue.size() == 0) {
                notEmpty.await();
            }
            notFull.signal();
            return queue.remove();
        } finally {
            lock.unlock();
        }
    }
}

Question: what should do RhoLang developer in this situation?

I developer the aproach to systematically implement mutex as Cell pattern and conditional variables as blocked channel read and translate 10+ classes from java.util.concurrent to Rholang.

In my opinion correct translation is

new BlockingQueue in {
  contract BlockingQueue(@maxSize, putOp, takeOp) = {
    new stateRef in {
  
      stateRef!([], true) |
    
      contract putOp(@newElem, ack) = {
        for (@arr, @true <- stateRef) {
          stateRef!(arr ++ [newElem], arr.length() + 1 < maxSize) | 
          ack!(Nil) } } |
     
      contract takeOp(ret) = {
        for (@[head...tail], @notFull <- stateRef) {
          stateRef!(tail, true) | 
          ret!(head) } }
    }    
  }
}

Issue home page = java.util.concurrent in Rholang

[golovach-ivan]

One more interesting example: CountDownLatch

In Java (based on implicit Monitor)

class CountDownLatch {
    private int count;

    public CountDownLatch(int count) { 
        this.count = count; 
    }

    public synchronized void await() throws InterruptedException {
        while (count != 0) { this.wait(); }
    }

    public synchronized void countDown() {
        if (count == 1) { this.notifyAll(); }
        if (count > 0) { count--; }
    }
}

In RhoLang

new CountDownLatch in {
  contract CountDownLatch(@initCount, awaitOp, countDownOp) = {  
    new stateRef in {    
  
      stateRef!(initCount) |

      contract awaitOp(ack) = {
        for (@0 <- stateRef) {          
          stateRef!(0) | ack!(Nil) } } |  

      contract countDownOp(_) = {
        for (@{count /\ ~0} <- stateRef) {          
          stateRef!(count - 1) } } 
    }    
  }
}  

Complete source code for CountDownLatch (with demo)

new CountDownLatch in {
  contract CountDownLatch(@initCount, awaitOp, countDownOp) = {  
    new stateRef in {    
    
      stateRef!(initCount) |
  
      contract awaitOp(ack) = {
        for (@0 <- stateRef) {          
          stateRef!(0) | 
          ack!(Nil) } } |  

      contract countDownOp(_) = {
        for (@{count /\ ~0} <- stateRef) {          
          stateRef!(count - 1) } } 
    }    
  } |
  
  new countDown, await in {
    // ========== CLOSE GATE WITH 3
    CountDownLatch!(3, *await, *countDown) |
    
    // ========== START 5 WAITERS
    new n in {
      n!(0) | n!(1) | n!(2) | n!(3) | n!(4) | for (@i <= n) { 
        new ack in { 
          await!(*ack) | for (_ <- ack) { stdout!([i, "I woke up!"]) } } } } |     
    
    // ========== 3 TIMES DECREASE GATE COUNTER
    new ack in { 
      stdoutAck!("knock-knock", *ack) | for (_ <- ack) {
        countDown!(Nil) |
        stdoutAck!("KNOCK-KNOCK", *ack) | for (_ <- ack) {
          countDown!(Nil) |
          stdoutAck!("WAKE UP !!!", *ack) | for (_ <- ack) { 
            countDown!(Nil) } } } }    
  }
}

>> "knock-knock"
>> "KNOCK-KNOCK"
>> "WAKE UP !!!"
>> [4, "I woke up!"]
>> [1, "I woke up!"]
>> [0, "I woke up!"]
>> [3, "I woke up!"]
>> [2, "I woke up!"]