

Task

Define a data type consisting of a fixed number of 'buckets', each containing a nonnegative integer value, which supports operations to:

 get the current value of any bucket
 remove a specified amount from one specified bucket and add it to another, preserving the total of all bucket values, and clamping the transferred amount to ensure the values remain non-negative

In order to exercise this data type, create one set of buckets, and start three concurrent tasks:

 As often as possible, pick two buckets and make their values closer to equal.
 As often as possible, pick two buckets and arbitrarily redistribute their values.
 At whatever rate is convenient, display (by any means) the total value and, optionally, the individual values of each bucket.


The display task need not be explicit; use of e.g. a debugger or trace tool is acceptable provided it is simple to set up to provide the display.


This task is intended as an exercise in atomic operations.   The sum of the bucket values must be preserved even if the two tasks attempt to perform transfers simultaneously, and a straightforward solution is to ensure that at any time, only one transfer is actually occurring — that the transfer operation is atomic.




import locks
import math
import os
import random
 
const N = 10          # Number of buckets.
const MaxInit = 99    # Maximum initial value for buckets.
 
var buckets: array[1..N, Natural]       # Array of buckets.
var bucketLocks: array[1..N, Lock]      # Array of bucket locks.
var randomLock: Lock                    # Lock to protect the random number generator.
var terminate: array[3, Channel[bool]]  # Used to ask threads to terminate.
 
#---------------------------------------------------------------------------------------------------
 
proc getTwoIndexes(): tuple[a, b: int] =
  ## Get two indexes from the random number generator.
 
  result.a = rand(1..N)
  result.b = rand(2..N)
  if result.b == result.a: result.b = 1
 
#---------------------------------------------------------------------------------------------------
 
proc equalize(num: int) {.thread.} =
  ## Try to equalize two buckets.
 
  var b1, b2: int         # Bucket indexes.
 
  while true:
 
    # Select the two buckets to "equalize".
    withLock randomLock:
     (b1, b2) = getTwoIndexes()
    if b1 > b2: swap b1, b2       # We want "b1 < b2" to avoid deadlocks.
 
    # Perform equalization.
    withLock bucketLocks[b1]:
      withLock bucketLocks[b2]:
        let target = (buckets[b1] + buckets[b2]) div 2
        let delta = target - buckets[b1]
        inc buckets[b1], delta
        dec buckets[b2], delta
 
    # Check termination.
    let (available, stop) = tryRecv terminate[num]
    if available and stop: break
 
#---------------------------------------------------------------------------------------------------
 
proc distribute(num: int) {.thread.} =
  ## Redistribute contents of two buckets.
 
  var b1, b2: int         # Bucket indexes.
  var factor: float       # Ratio used to compute the new value for "b1".
 
  while true:
 
    # Select the two buckets for redistribution and the redistribution factor.
    withLock randomLock:
      (b1, b2) = getTwoIndexes()
      factor = rand(0.0..1.0)
    if b1 > b2: swap b1, b2       # We want "b1 < b2" to avoid deadlocks..
 
    # Perform redistribution.
    withLock bucketLocks[b1]:
      withLock bucketLocks[b2]:
        let sum = buckets[b1] + buckets[b2]
        let value = (sum.toFloat * factor).toInt
        buckets[b1] = value
        buckets[b2] = sum - value
 
    # Check termination.
    let (available, stop) = tryRecv terminate[num]
    if available and stop: break
 
#---------------------------------------------------------------------------------------------------
 
proc display(num: int) {.thread.} =
  ## Display the content of buckets and the sum (which should be constant).
 
  while true:
    for i in 1..N: acquire bucketLocks[i]
    echo buckets, "   Total = ", sum(buckets)
    for i in countdown(N, 1): release bucketLocks[i]
    os.sleep(1000)
 
    # Check termination.
    let (available, stop) = tryRecv terminate[num]
    if available and stop: break
 
#———————————————————————————————————————————————————————————————————————————————————————————————————
 
randomize()
 
# Initialize the buckets with a random value.
for bucket in buckets.mitems:
  bucket = rand(1..MaxInit)
 
# Initialize the locks.
randomLock.initLock()
for lock in bucketLocks.mitems:
  lock.initLock()
 
# Open the channels.
for c in terminate.mitems:
  c.open()
 
# Create and launch the threads.
var tequal, tdist, tdisp: Thread[int]
tequal.createThread(equalize, 0)
tdist.createThread(distribute, 1)
tdisp.createThread(display, 2)
 
sleep(10000)
 
# Ask the threads to stop.
for c in terminate.mitems:
  c.send(true)
 
joinThreads([tequal, tdist, tdisp])
 
# Free resources.
randomLock.deinitLock()
for lock in bucketLocks.mitems:
  lock.deinitLock()
for c in terminate.mitems:
  c.close()