Add all the A tasks

Task/Atomic-updates/0DESCRIPTION

@@ -0,0 +1,16 @@
+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 [[wp:Clamping (graphics)|clamping]] the transferred amount to ensure the values remain nonnegative
+
+----
+
+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''.

Task/Atomic-updates/1META.yaml

@@ -0,0 +1,4 @@
+---
+note: Concurrency
+requires:
+- Concurrency

Task/Atomic-updates/Ada/atomic-updates.ada

@@ -0,0 +1,85 @@
+with Ada.Text_IO;  use Ada.Text_IO;
+with Ada.Numerics.Discrete_Random;
+
+procedure Test_Updates is
+
+   type Bucket_Index is range 1..13;
+   package Random_Index is new Ada.Numerics.Discrete_Random (Bucket_Index);
+   use Random_Index;
+   type Buckets is array (Bucket_Index) of Natural;
+
+   protected type Safe_Buckets is
+      procedure Initialize (Value : Buckets);
+      function Get (I : Bucket_Index) return Natural;
+      procedure Transfer (I, J : Bucket_Index; Amount : Integer);
+      function Snapshot return Buckets;
+   private
+      Data : Buckets := (others => 0);
+   end Safe_Buckets;
+
+   protected body Safe_Buckets is
+      procedure Initialize (Value : Buckets) is
+      begin
+         Data := Value;
+      end Initialize;
+
+      function Get (I : Bucket_Index) return Natural is
+      begin
+         return Data (I);
+      end Get;
+
+      procedure Transfer (I, J : Bucket_Index; Amount : Integer) is
+         Increment : constant Integer :=
+            Integer'Max (-Data (J), Integer'Min (Data (I), Amount));
+      begin
+         Data (I) := Data (I) - Increment;
+         Data (J) := Data (J) + Increment;
+      end Transfer;
+
+      function Snapshot return Buckets is
+      begin
+         return Data;
+      end Snapshot;
+   end Safe_Buckets;
+
+   Data : Safe_Buckets;
+
+   task Equalize;
+   task Mess_Up;
+
+   task body Equalize is
+      Dice : Generator;
+      I, J : Bucket_Index;
+   begin
+      loop
+         I := Random (Dice);
+         J := Random (Dice);
+         Data.Transfer (I, J, (Data.Get (I) - Data.Get (J)) / 2);
+      end loop;
+   end Equalize;
+
+   task body Mess_Up is
+      Dice : Generator;
+   begin
+      loop
+         Data.Transfer (Random (Dice), Random (Dice), 100);
+      end loop;
+   end Mess_Up;
+
+begin
+   Data.Initialize ((1,2,3,4,5,6,7,8,9,10,11,12,13));
+   loop
+      delay 1.0;
+      declare
+         State : Buckets := Data.Snapshot;
+         Sum   : Natural := 0;
+      begin
+         for Index in State'Range loop
+            Sum := Sum + State (Index);
+            Put (Integer'Image (State (Index)));
+         end loop;
+         Put (" =" & Integer'Image (Sum));
+         New_Line;
+      end;
+   end loop;
+end Test_Updates;

Task/Atomic-updates/C/atomic-updates-1.c

@@ -0,0 +1,111 @@
+#include <stdio.h>
+#include <stdlib.h>
+#include <stdbool.h>
+#include <unistd.h>
+#include <time.h>
+#include <pthread.h>
+
+#define N_BUCKETS 15
+
+pthread_mutex_t bucket_mutex[N_BUCKETS];
+int buckets[N_BUCKETS];
+
+pthread_t equalizer;
+pthread_t randomizer;
+
+void transfer_value(int from, int to, int howmuch)
+{
+  bool swapped = false;
+
+  if ( (from == to) || ( howmuch < 0 ) ||
+       (from < 0 ) || (to < 0) || (from >= N_BUCKETS) || (to >= N_BUCKETS) ) return;
+
+  if ( from > to ) {
+    int temp1 = from;
+    from = to;
+    to = temp1;
+    swapped = true;
+    howmuch = -howmuch;
+  }
+
+  pthread_mutex_lock(&bucket_mutex[from]);
+  pthread_mutex_lock(&bucket_mutex[to]);
+
+  if ( howmuch > buckets[from] && !swapped )
+    howmuch = buckets[from];
+  if ( -howmuch > buckets[to] && swapped )
+    howmuch = -buckets[to];
+
+  buckets[from] -= howmuch;
+  buckets[to] += howmuch;
+
+  pthread_mutex_unlock(&bucket_mutex[from]);
+  pthread_mutex_unlock(&bucket_mutex[to]);
+}
+
+void print_buckets()
+{
+  int i;
+  int sum=0;
+
+  for(i=0; i < N_BUCKETS; i++) pthread_mutex_lock(&bucket_mutex[i]);
+  for(i=0; i < N_BUCKETS; i++) {
+    printf("%3d ", buckets[i]);
+    sum += buckets[i];
+  }
+  printf("= %d\n", sum);
+  for(i=0; i < N_BUCKETS; i++) pthread_mutex_unlock(&bucket_mutex[i]);
+}
+
+void *equalizer_start(void *t)
+{
+  for(;;) {
+    int b1 = rand()%N_BUCKETS;
+    int b2 = rand()%N_BUCKETS;
+    int diff = buckets[b1] - buckets[b2];
+    if ( diff < 0 )
+      transfer_value(b2, b1, -diff/2);
+    else
+      transfer_value(b1, b2, diff/2);
+  }
+  return NULL;
+}
+
+void *randomizer_start(void *t)
+{
+  for(;;) {
+    int b1 = rand()%N_BUCKETS;
+    int b2 = rand()%N_BUCKETS;
+    int diff = rand()%(buckets[b1]+1);
+    transfer_value(b1, b2, diff);
+  }
+  return NULL;
+}
+
+int main()
+{
+  int i, total=0;
+
+  for(i=0; i < N_BUCKETS; i++) pthread_mutex_init(&bucket_mutex[i], NULL);
+
+  for(i=0; i < N_BUCKETS; i++) {
+    buckets[i] = rand() % 100;
+    total += buckets[i];
+    printf("%3d ", buckets[i]);
+  }
+  printf("= %d\n", total);
+
+  // we should check if these succeeded
+  pthread_create(&equalizer, NULL, equalizer_start, NULL);
+  pthread_create(&randomizer, NULL, randomizer_start, NULL);
+
+  for(;;) {
+    sleep(1);
+    print_buckets();
+  }
+
+  // we do not provide a "good" way to stop this run, so the following
+  // is never reached indeed...
+  for(i=0; i < N_BUCKETS; i++) pthread_mutex_destroy(bucket_mutex+i);
+  return EXIT_SUCCESS;
+}

Task/Atomic-updates/C/atomic-updates-2.c

@@ -0,0 +1,51 @@
+#include <stdio.h>
+#include <stdlib.h>
+#include <omp.h>
+
+#define irand(n) (n * (double)rand()/(RAND_MAX + 1.0))
+
+int bucket[10];
+int main()
+{
+	int i;
+	for (i = 0; i < 10; i++) bucket[i] = 1000;
+	omp_set_num_threads(3);
+
+	#pragma omp parallel private(i)
+	for (i = 0; i < 10000; i++) {
+		int from, to, mode, diff = 0, sum;
+
+		from = irand(10);
+		do { to = irand(10); } while (from == to);
+		mode = irand(10);
+
+		switch (mode) {
+		case 0:
+		case 1:
+		case 2:	/* equalize */
+			diff = (bucket[from] - bucket[to]) / 2;
+			break;
+
+		case 3: /* report */
+			sum = 0;
+			for (int j = 0; j < 10; j++) {
+				printf("%d ", bucket[j]);
+				sum += bucket[j];
+			}
+			printf(" Sum: %d\n", sum);
+			continue;
+
+		default: /* random transfer */
+			diff = irand(bucket[from]);
+			break;
+		}
+
+		#pragma omp critical
+		{
+			bucket[from] -= diff;
+			bucket[to]   += diff;
+		}
+	}
+
+	return 0;
+}

Task/Atomic-updates/C/atomic-updates-3.c

@@ -0,0 +1,8 @@
+1000 1000 1000 1798 1000 1000 1000 1000 202 1000  Sum: 10000
+595 800 2508 2750 470 1209 283 314 601 470  Sum: 10000
+5 521 3339 1656 351 1038 1656 54 508 872  Sum: 10000
+.
+.
+.
+752 490 385 2118 1503 508 384 509 1110 2241  Sum: 10000
+752 823 385 2118 1544 508 10 509 1110 2241  Sum: 10000

Task/Atomic-updates/Clojure/atomic-updates-1.clj

@@ -0,0 +1,7 @@
+(defn xfer [m from to amt]
+  (let [{f-bal from t-bal to} m
+        f-bal (- f-bal amt)
+        t-bal (+ t-bal amt)]
+    (if (or (neg? f-bal) (neg? t-bal))
+      (throw (IllegalArgumentException. "Call results in negative balance."))
+      (assoc m from f-bal to t-bal))))

Task/Atomic-updates/Clojure/atomic-updates-2.clj

@@ -0,0 +1,2 @@
+(def *data* (atom {:a 100 :b 100})) ;; *data* is an atom holding a map
+(swap! *data* xfer :a :b 50) ;; atomically results in *data* holding {:a 50 :b 150}

Task/Atomic-updates/Clojure/atomic-updates-3.clj

@@ -0,0 +1,22 @@
+(defn equalize [m a b]
+  (let [{a-val a b-val b} m
+        diff (- a-val b-val)
+        amt (/ diff 2)]
+    (xfer m a b amt)))
+
+(defn randomize [m a b]
+  (let [{a-val a b-val b} m
+        min-val (min a-val b-val)
+        amt (rand-int (- min-val) min-val)]
+    (xfer m a b amt)))
+
+(defn test-conc [f data a b n name]
+  (dotimes [i n]
+    (swap! data f a b)
+    (println (str "total is " (reduce + (vals @data)) " after " name " iteration " i))))
+
+(def thread-eq (Thread. #(test-conc equalize *data* :a :b 1000 "equalize")))
+(def thread-rand (Thread. #(test-conc randomize *data* :a :b 1000 "randomize")))
+
+(.start thread-eq)
+(.start thread-rand)

Task/Atomic-updates/Go/atomic-updates-1.go

@@ -0,0 +1,106 @@
+package main
+
+import (
+    "fmt"
+    "math/rand"
+    "time"
+)
+
+// Data type required by task.
+type bucketList interface {
+    // Two operations required by task.  Updater parameter not specified
+    // by task, but useful for displaying update counts as an indication
+    // that transfer operations are happening "as often as possible."
+    bucketValue(bucket int) int
+    transfer(b1, b2, ammount, updater int)
+
+    // Operation not specified by task, but needed for synchronization.
+    snapshot(bucketValues []int, transferCounts []int)
+
+    // Operation not specified by task, but useful.
+    buckets() int // number of buckets
+}
+
+// Total of all bucket values, declared as a const to demonstrate that
+// it doesn't change.
+const originalTotal = 1000
+
+// Updater ids, used for maintaining transfer counts.
+const (
+    idOrder = iota
+    idChaos
+    nUpdaters
+)
+
+func main() {
+    // Create a concrete object implementing the bucketList interface.
+    bl := newChList(10, originalTotal, nUpdaters)
+
+    // Three concurrent tasks.
+    go order(bl)
+    go chaos(bl)
+    buddha(bl)
+}
+
+// The concurrent tasks exercise the data operations by going through
+// the bucketList interface.  They do no explicit synchronization and
+// are not responsible for maintaining invariants.
+
+// Exercise (1.) required by task: make values more equal.
+func order(bl bucketList) {
+    r := rand.New(rand.NewSource(time.Now().UnixNano()))
+    nBuckets := bl.buckets()
+    for {
+        b1 := r.Intn(nBuckets)
+        b2 := r.Intn(nBuckets)
+        v1 := bl.bucketValue(b1)
+        v2 := bl.bucketValue(b2)
+        if v1 > v2 {
+            bl.transfer(b1, b2, (v1-v2)/2, idOrder)
+        } else {
+            bl.transfer(b2, b1, (v2-v1)/2, idOrder)
+        }
+    }
+}
+
+// Exercise (2.) required by task: redistribute values.
+func chaos(bl bucketList) {
+    r := rand.New(rand.NewSource(time.Now().UnixNano()))
+    nBuckets := bl.buckets()
+    for {
+        b1 := r.Intn(nBuckets)
+        b2 := r.Intn(nBuckets)
+        bl.transfer(b1, b2, r.Intn(bl.bucketValue(b1)+1), idChaos)
+    }
+}
+
+// Exercise (3.) requred by task: display total.
+func buddha(bl bucketList) {
+    nBuckets := bl.buckets()
+    s := make([]int, nBuckets)
+    tc := make([]int, nUpdaters)
+    var total, nTicks int
+
+    fmt.Println("sum  ---updates---    mean  buckets")
+    tr := time.Tick(time.Second / 10)
+    for {
+        var sum int
+        <-tr
+        bl.snapshot(s, tc)
+        for _, l := range s {
+            if l < 0 {
+                panic("sob") // invariant not preserved
+            }
+            sum += l
+        }
+        // Output number of updates per tick and cummulative mean
+        // updates per tick to demonstrate "as often as possible"
+        // of task exercises 1 and 2.
+        total += tc[0] + tc[1]
+        nTicks++
+        fmt.Printf("%d %6d %6d %7d  %v\n", sum, tc[0], tc[1], total/nTicks, s)
+        if sum != originalTotal {
+            panic("weep") // invariant not preserved
+        }
+    }
+}

Task/Atomic-updates/Go/atomic-updates-2.go

@@ -0,0 +1,78 @@
+// chList (ch for channel-synchronized) is a concrete type implementing
+// the bucketList interface.  The bucketList interface declared methods,
+// the struct type here declares data.  chList methods are repsonsible
+// for synchronization so they are goroutine-safe.  They are also
+// responsible for maintaining the invariants that the sum of all buckets
+// stays constant and that no bucket value goes negative.
+type chList struct {
+    b  []int     // bucket data specified by task
+    s  chan bool // syncronization object
+    tc []int     // a transfer count for each updater
+}
+
+// Constructor.
+func newChList(nBuckets, initialSum, nUpdaters int) *chList {
+    bl := &chList{
+        b:  make([]int, nBuckets),
+        s:  make(chan bool, 1),
+        tc: make([]int, nUpdaters),
+    }
+    // Distribute initialSum across buckets.
+    for i, dist := nBuckets, initialSum; i > 0; {
+        v := dist / i
+        i--
+        bl.b[i] = v
+        dist -= v
+    }
+    // Synchronization is needed to maintain the invariant that the total
+    // of all bucket values stays the same.  This is an implementation of
+    // the straightforward solution mentioned in the task description,
+    // ensuring that only one transfer happens at a time.  Channel s
+    // holds a token.  All methods must take the token from the channel
+    // before accessing data and then return the token when they are done.
+    // it is equivalent to a mutex.  The constructor makes data available
+    // by initially dropping the token in the channel after all data is
+    // initialized.
+    bl.s <- true
+    return bl
+}
+
+// Four methods implementing the bucketList interface.
+func (bl *chList) bucketValue(b int) int {
+    <-bl.s // get token before accessing data
+    r := bl.b[b]
+    bl.s <- true // return token
+    return r
+}
+
+func (bl *chList) transfer(b1, b2, a int, ux int) {
+    if b1 == b2 { // null operation
+        return
+    }
+    // Get access.
+    <-bl.s
+    // Clamping maintains invariant that bucket values remain nonnegative.
+    if a > bl.b[b1] {
+        a = bl.b[b1]
+    }
+    // Transfer.
+    bl.b[b1] -= a
+    bl.b[b2] += a
+    bl.tc[ux]++ // increment transfer count
+    // Release "lock".
+    bl.s <- true
+}
+
+func (bl *chList) snapshot(s []int, tc []int) {
+    <-bl.s
+    copy(s, bl.b)
+    copy(tc, bl.tc)
+    for i := range bl.tc {
+        bl.tc[i] = 0
+    }
+    bl.s <- true
+}
+
+func (bl *chList) buckets() int {
+    return len(bl.b)
+}

Task/Atomic-updates/Go/atomic-updates-3.go

@@ -0,0 +1,83 @@
+// rwList, rw is for RWMutex-synchronized.
+type rwList struct {
+    b []int // bucket data specified by task
+
+    // Syncronization objects.
+    m   []sync.Mutex // mutex for each bucket
+    all sync.RWMutex // mutex for entire list, for snapshot operation
+
+    tc []int // a transfer count for each updater
+}
+
+// Constructor.
+func newRwList(nBuckets, initialSum, nUpdaters int) *rwList {
+    bl := &rwList{
+        b:  make([]int, nBuckets),
+        m:  make([]sync.Mutex, nBuckets),
+        tc: make([]int, nUpdaters),
+    }
+    for i, dist := nBuckets, initialSum; i > 0; {
+        v := dist / i
+        i--
+        bl.b[i] = v
+        dist -= v
+    }
+    return bl
+}
+
+// Four methods implementing the bucketList interface.
+func (bl *rwList) bucketValue(b int) int {
+    bl.m[b].Lock() // lock on bucket ensures read is atomic
+    r := bl.b[b]
+    bl.m[b].Unlock()
+    return r
+}
+
+func (bl *rwList) transfer(b1, b2, a int, ux int) {
+    if b1 == b2 { // null operation
+        return
+    }
+    // RLock on list allows other simultaneous transfers.
+    bl.all.RLock()
+    // Locking lowest bucket first prevents deadlock
+    // with multiple tasks working at the same time.
+    if b1 < b2 {
+        bl.m[b1].Lock()
+        bl.m[b2].Lock()
+    } else {
+        bl.m[b2].Lock()
+        bl.m[b1].Lock()
+    }
+    // clamp
+    if a > bl.b[b1] {
+        a = bl.b[b1]
+    }
+    // transfer
+    bl.b[b1] -= a
+    bl.b[b2] += a
+    bl.tc[ux]++ // increment transfer count
+    // release
+    bl.m[b1].Unlock()
+    bl.m[b2].Unlock()
+    bl.all.RUnlock()
+    // With current Go, the program can hang without a call to gosched here.
+    // It seems that functions in the sync package don't touch the scheduler,
+    // (which is good) but we need to touch it here to give the channel
+    // operations in buddha a chance to run.  (The current Go scheduler
+    // is basically cooperative rather than preemptive.)
+    runtime.Gosched()
+}
+
+func (bl *rwList) snapshot(s []int, tc []int) {
+    bl.all.Lock() // RW lock on list prevents transfers during snap.
+    copy(s, bl.b)
+    copy(tc, bl.tc)
+    for i := range bl.tc {
+        bl.tc[i] = 0
+    }
+    bl.all.Unlock()
+}
+
+func (bl *rwList) buckets() int {
+    return len(bl.b)
+}

Task/Atomic-updates/Go/atomic-updates-4.go

@@ -0,0 +1,63 @@
+// lf for lock-free
+type lfList struct {
+    b []int32
+    sync.RWMutex
+    tc []int
+}
+
+// Constructor.
+func newLfList(nBuckets, initialSum, nUpdaters int) *lfList {
+    bl := &lfList{
+        b:  make([]int32, nBuckets),
+        tc: make([]int, nUpdaters),
+    }
+    for i, dist := int32(nBuckets), int32(initialSum); i > 0; {
+        v := dist / i
+        i--
+        bl.b[i] = v
+        dist -= v
+    }
+    return bl
+}
+
+// Four methods implementing the bucketList interface.
+func (bl *lfList) bucketValue(b int) int {
+    return int(atomic.LoadInt32(&bl.b[b]))
+}
+
+func (bl *lfList) transfer(b1, b2, a int, ux int) {
+    if b1 == b2 {
+        return
+    }
+    bl.RLock()
+    for {
+        t := int32(a)
+        v1 := atomic.LoadInt32(&bl.b[b1])
+        if t > v1 {
+            t = v1
+        }
+        if atomic.CompareAndSwapInt32(&bl.b[b1], v1, v1-t) {
+            atomic.AddInt32(&bl.b[b2], t)
+            break
+        }
+        // else retry
+    }
+    bl.tc[ux]++
+    bl.RUnlock()
+    runtime.Gosched()
+}
+
+func (bl *lfList) snapshot(s []int, tc []int) {
+    bl.Lock()
+    for i, bv := range bl.b {
+        s[i] = int(bv)
+    }
+    for i := range bl.tc {
+        tc[i], bl.tc[i] = bl.tc[i], 0
+    }
+    bl.Unlock()
+}
+
+func (bl *lfList) buckets() int {
+    return len(bl.b)
+}

Task/Atomic-updates/Go/atomic-updates-5.go

@@ -0,0 +1,109 @@
+// mnList (mn for monitor-synchronized) is a concrete type implementing
+// the bucketList interface.  The monitor is a goroutine, all communication
+// with it is done through channels, which are the members of mnList.
+// All data implementing the buckets is encapsulated in the monitor.
+type mnList struct {
+    vrCh chan *valueReq
+    trCh chan *transferReq
+    srCh chan *snapReq
+    nbCh chan chan int
+}
+
+// Constructor makes channels and starts monitor.
+func newMnList(nBuckets, initialSum, nUpdaters int) *mnList {
+    mn := &mnList{
+        make(chan *valueReq),
+        make(chan *transferReq),
+        make(chan *snapReq),
+        make(chan chan int),
+    }
+    go monitor(mn, nBuckets, initialSum, nUpdaters)
+    return mn
+}
+
+// Monitor goroutine ecapsulates data and enters a loop to handle requests.
+// The loop handles one request at a time, thus serializing all access.
+func monitor(mn *mnList, nBuckets, initialSum, nUpdaters int) {
+    // bucket representation
+    b := make([]int, nBuckets)
+    for i, dist := nBuckets, initialSum; i > 0; {
+        v := dist / i
+        i--
+        b[i] = v
+        dist -= v
+    }
+    // transfer count representation
+    count := make([]int, nUpdaters)
+
+    // monitor loop
+    for {
+        select {
+        // value request operation
+        case vr := <-mn.vrCh:
+            vr.resp <- b[vr.bucket]
+
+        // transfer operation
+        case tr := <-mn.trCh:
+            // clamp
+            if tr.amount > b[tr.from] {
+                tr.amount = b[tr.from]
+            }
+            // transfer
+            b[tr.from] -= tr.amount
+            b[tr.to] += tr.amount
+            count[tr.updaterId]++
+
+        // snap operation
+        case sr := <-mn.srCh:
+            copy(sr.bucketSnap, b)
+            copy(sr.countSnap, count)
+            for i := range count {
+                count[i] = 0
+            }
+            sr.resp <- true
+
+        // number of buckets
+        case nb := <-mn.nbCh:
+            nb <- nBuckets
+        }
+    }
+}
+
+type valueReq struct {
+    bucket int
+    resp   chan int
+}
+
+func (mn *mnList) bucketValue(b int) int {
+    resp := make(chan int)
+    mn.vrCh <- &valueReq{b, resp}
+    return <-resp
+}
+
+type transferReq struct {
+    from, to  int
+    amount    int
+    updaterId int
+}
+
+func (mn *mnList) transfer(b1, b2, a, ux int) {
+    mn.trCh <- &transferReq{b1, b2, a, ux}
+}
+
+type snapReq struct {
+    bucketSnap []int
+    countSnap  []int
+    resp       chan bool
+}
+
+func (mn *mnList) snapshot(s []int, tc []int) {
+    resp := make(chan bool)
+    mn.srCh <- &snapReq{s, tc, resp}
+    <-resp
+}
+
+func (mn *mnList) buckets() int {
+    resp := make(chan int)
+    mn.nbCh <- resp
+    return <-resp
+}

Task/Atomic-updates/Haskell/atomic-updates.hs

@@ -0,0 +1,69 @@
+module AtomicUpdates (main) where
+
+import Control.Concurrent (forkIO, threadDelay)
+import Control.Concurrent.MVar (MVar, newMVar, readMVar, modifyMVar_)
+import Control.Monad (forever, forM_)
+import Data.IntMap (IntMap, (!), toAscList, fromList, adjust)
+import System.Random (randomRIO)
+import Text.Printf (printf)
+
+-------------------------------------------------------------------------------
+
+type Index = Int
+type Value = Integer
+data Buckets = Buckets Index (MVar (IntMap Value))
+
+makeBuckets   :: Int -> IO Buckets
+size          :: Buckets -> Index
+currentValue  :: Buckets -> Index -> IO Value
+currentValues :: Buckets          -> IO (IntMap Value)
+transfer      :: Buckets -> Index -> Index -> Value -> IO ()
+
+-------------------------------------------------------------------------------
+
+makeBuckets n = do v <- newMVar (fromList [(i, 100) | i <- [1..n]])
+                   return (Buckets n v)
+
+size (Buckets n _) = n
+
+currentValue (Buckets _ v) i = fmap (! i) (readMVar v)
+currentValues (Buckets _ v) = readMVar v
+
+transfer b@(Buckets n v) i j amt | amt < 0        = transfer b j i (-amt)
+                                 | otherwise      = do
+  modifyMVar_ v $ \map -> let amt' = min amt (map ! i)
+                            in return $ adjust (subtract amt') i
+                                      $ adjust (+        amt') j
+                                      $ map
+
+-------------------------------------------------------------------------------
+
+roughen, smooth, display :: Buckets -> IO ()
+
+pick buckets = randomRIO (1, size buckets)
+
+roughen buckets = forever loop where
+  loop = do i <- pick buckets
+            j <- pick buckets
+            iv <- currentValue buckets i
+            transfer buckets i j (iv `div` 3)
+
+smooth buckets = forever loop where
+  loop = do i <- pick buckets
+            j <- pick buckets
+            iv <- currentValue buckets i
+            jv <- currentValue buckets j
+            transfer buckets i j ((iv - jv) `div` 4)
+
+display buckets = forever loop where
+  loop = do threadDelay 1000000
+            bmap <- currentValues buckets
+            putStrLn (report $ map snd $ toAscList bmap)
+  report list = "\nTotal: " ++ show (sum list) ++ "\n" ++ bars
+    where bars = concatMap row $ map (*40) $ reverse [1..5]
+          row lim = printf "%3d " lim ++ [if x >= lim then '*' else ' ' | x <- list] ++ "\n"
+
+main = do buckets <- makeBuckets 100
+          forkIO (roughen buckets)
+          forkIO (smooth buckets)
+          display buckets

Task/Atomic-updates/Java/atomic-updates.java

@@ -0,0 +1,119 @@
+import java.util.Arrays;
+import java.util.Random;
+
+public class AtomicUpdates
+{
+  public static class Buckets
+  {
+    private final int[] data;
+
+    public Buckets(int[] data)
+    {
+      this.data = data.clone();
+    }
+
+    public int getBucket(int index)
+    {
+      synchronized (data)
+      {  return data[index];  }
+    }
+
+    public int transfer(int srcBucketIndex, int destBucketIndex, int amount)
+    {
+      if (amount == 0)
+        return 0;
+      // Negative transfers will happen in the opposite direction
+      if (amount < 0)
+      {
+        int tempIndex = srcBucketIndex;
+        srcBucketIndex = destBucketIndex;
+        destBucketIndex = tempIndex;
+        amount = -amount;
+      }
+      synchronized (data)
+      {
+        if (amount > data[srcBucketIndex])
+          amount = data[srcBucketIndex];
+        if (amount <= 0)
+          return 0;
+        data[srcBucketIndex] -= amount;
+        data[destBucketIndex] += amount;
+        return amount;
+      }
+    }
+
+    public int[] getBuckets()
+    {
+      synchronized (data)
+      {  return data.clone();  }
+    }
+  }
+
+  public static int getTotal(int[] values)
+  {
+    int totalValue = 0;
+    for (int i = values.length - 1; i >= 0; i--)
+      totalValue += values[i];
+    return totalValue;
+  }
+
+  public static void main(String[] args)
+  {
+    final int NUM_BUCKETS = 10;
+    Random rnd = new Random();
+    final int[] values = new int[NUM_BUCKETS];
+    for (int i = 0; i < values.length; i++)
+      values[i] = rnd.nextInt(10);
+    System.out.println("Initial Array: " + getTotal(values) + " " + Arrays.toString(values));
+    final Buckets buckets = new Buckets(values);
+
+    new Thread(new Runnable() {
+        public void run()
+        {
+          Random r = new Random();
+          while (true)
+          {
+            int srcBucketIndex = r.nextInt(NUM_BUCKETS);
+            int destBucketIndex = r.nextInt(NUM_BUCKETS);
+            int amount = (buckets.getBucket(srcBucketIndex) - buckets.getBucket(destBucketIndex)) >> 1;
+            if (amount != 0)
+              buckets.transfer(srcBucketIndex, destBucketIndex, amount);
+          }
+        }
+      }
+    ).start();
+
+    new Thread(new Runnable() {
+        public void run()
+        {
+          Random r = new Random();
+          while (true)
+          {
+            int srcBucketIndex = r.nextInt(NUM_BUCKETS);
+            int destBucketIndex = r.nextInt(NUM_BUCKETS);
+            int srcBucketAmount = buckets.getBucket(srcBucketIndex);
+            int destBucketAmount = buckets.getBucket(destBucketIndex);
+            int amount = r.nextInt(srcBucketAmount + destBucketAmount + 1) - destBucketAmount;
+            if (amount != 0)
+              buckets.transfer(srcBucketIndex, destBucketIndex, amount);
+          }
+        }
+      }
+    ).start();
+
+    while (true)
+    {
+      long nextPrintTime = System.currentTimeMillis() + 3000;
+      long curTime;
+      while ((curTime = System.currentTimeMillis()) < nextPrintTime)
+      {
+        try
+        {  Thread.sleep(nextPrintTime - curTime);  }
+        catch (InterruptedException e)
+        {  }
+      }
+      int[] bucketValues = buckets.getBuckets();
+      System.out.println("Current values: " + getTotal(bucketValues) + " " + Arrays.toString(bucketValues));
+    }
+  }
+}

Task/Atomic-updates/Perl/atomic-updates.pl

@@ -0,0 +1,52 @@
+use strict;
+use 5.10.0;
+
+use threads 'yield';
+use threads::shared;
+
+my @a :shared = (100) x 10;
+my $stop :shared = 0;
+
+sub pick2 {
+	my $i = int(rand(10));
+	my $j;
+	$j = int(rand(10)) until $j != $i;
+	($i, $j)
+}
+
+sub even {
+	lock @a;
+	my ($i, $j) = pick2;
+	my $sum = $a[$i] + $a[$j];
+	$a[$i] = int($sum / 2);
+	$a[$j] = $sum - $a[$i];
+}
+
+sub rand_move {
+	lock @a;
+	my ($i, $j) = pick2;
+
+	my $x = int(rand $a[$i]);
+	$a[$i] -= $x;
+	$a[$j] += $x;
+}
+
+sub show {
+	lock @a;
+	my $sum = 0;
+	$sum += $_ for (@a);
+	printf "%4d", $_ for @a;
+	print " total $sum\n";
+}
+
+my $t1 = async { even		until $stop }
+my $t2 = async { rand_move	until $stop }
+my $t3 = async {
+	for (1 .. 10) {
+		show;
+		sleep(1);
+	}
+	$stop = 1;
+};
+
+$t1->join; $t2->join; $t3->join;

Task/Atomic-updates/PicoLisp/atomic-updates.l

@@ -0,0 +1,71 @@
+(de *Buckets . 15)  # Number of buckets
+
+# E/R model
+(class +Bucket +Entity)
+(rel key (+Key +Number))  # Key  1 .. *Buckets
+(rel val (+Number))       # Value 1 .. 999
+
+
+# Start with an empty DB
+(call 'rm "-f" "buckets.db")  # Remove old DB (if any)
+(pool "buckets.db")           # Create new DB file
+
+
+# Create *Buckets buckets with values between 1 and 999
+(for K *Buckets
+   (new T '(+Bucket)  'key K  'val (rand 1 999)) )
+(commit)
+
+
+# Pick a random bucket
+(de pickBucket ()
+   (db 'key '+Bucket (rand 1 *Buckets)) )
+
+
+# First process
+(unless (fork)
+   (seed *Pid)  # Ensure local random sequence
+   (loop
+      (let (B1 (pickBucket)  B2 (pickBucket))  # Pick two buckets 'B1' and 'B2'
+         (dbSync)                              # Atomic DB operation
+         (let (V1 (; B1 val)  V2 (; B2 val))   # Get current values
+            (cond
+               ((> V1 V2)
+                  (dec> B1 'val)               # Make them closer to equal
+                  (inc> B2 'val) )
+               ((> V2 V1)
+                  (dec> B2 'val)
+                  (inc> B1 'val) ) ) )
+         (commit 'upd) ) ) )                   # Close transaction
+
+# Second process
+(unless (fork)
+   (seed *Pid)  # Ensure local random sequence
+   (loop
+      (let (B1 (pickBucket)  B2 (pickBucket))  # Pick two buckets 'B1' and 'B2'
+         (unless (== B1 B2)                    # Found two different ones?
+            (dbSync)                              # Atomic DB operation
+            (let (V1 (; B1 val)  V2 (; B2 val))   # Get current values
+               (cond
+                  ((> V1 V2 0)
+                     (inc> B1 'val)               # Redistribute them
+                     (dec> B2 'val) )
+                  ((> V2 V1 0)
+                     (inc> B2 'val)
+                     (dec> B1 'val) ) ) )
+            (commit 'upd) ) ) ) )                 # Close transaction
+
+# Third process
+(unless (fork)
+   (loop
+      (dbSync)                         # Atomic DB operation
+      (let Lst (collect 'key '+Bucket) # Get all buckets
+         (for This Lst                 # Print current values
+            (printsp (: val)) )
+         (prinl                        # and total sum
+            "-- Total: "
+            (sum '((This) (: val)) Lst) ) )
+      (rollback)
+      (wait 2000) ) )                  # Sleep two seconds
+
+(wait)

Task/Atomic-updates/Python/atomic-updates.py

@@ -0,0 +1,74 @@
+from __future__ import with_statement # required for Python 2.5
+import threading
+import random
+import time
+
+terminate = threading.Event()
+
+class Buckets:
+    def __init__(self, nbuckets):
+        self.nbuckets = nbuckets
+        self.values = [random.randrange(10) for i in range(nbuckets)]
+        self.lock = threading.Lock()
+
+    def __getitem__(self, i):
+        return self.values[i]
+
+    def transfer(self, src, dst, amount):
+        with self.lock:
+            amount = min(amount, self.values[src])
+            self.values[src] -= amount
+            self.values[dst] += amount
+
+    def snapshot(self):
+        # copy of the current state (synchronized)
+        with self.lock:
+            return self.values[:]
+
+def randomize(buckets):
+    nbuckets = buckets.nbuckets
+    while not terminate.isSet():
+        src = random.randrange(nbuckets)
+        dst = random.randrange(nbuckets)
+        if dst!=src:
+            amount = random.randrange(20)
+            buckets.transfer(src, dst, amount)
+
+def equalize(buckets):
+    nbuckets = buckets.nbuckets
+    while not terminate.isSet():
+        src = random.randrange(nbuckets)
+        dst = random.randrange(nbuckets)
+        if dst!=src:
+            amount = (buckets[src] - buckets[dst]) // 2
+            if amount>=0: buckets.transfer(src, dst, amount)
+            else: buckets.transfer(dst, src, -amount)
+
+def print_state(buckets):
+    snapshot = buckets.snapshot()
+    for value in snapshot:
+        print '%2d' % value,
+    print '=', sum(snapshot)
+
+# create 15 buckets
+buckets = Buckets(15)
+
+# the randomize thread
+t1 = threading.Thread(target=randomize, args=[buckets])
+t1.start()
+
+# the equalize thread
+t2 = threading.Thread(target=equalize, args=[buckets])
+t2.start()
+
+# main thread, display
+try:
+    while True:
+        print_state(buckets)
+        time.sleep(1)
+except KeyboardInterrupt: # ^C to finish
+    terminate.set()
+
+# wait until all worker threads finish
+t1.join()
+t2.join()

Task/Atomic-updates/Ruby/atomic-updates.rb

@@ -0,0 +1,94 @@
+require 'thread'
+
+# A collection of buckets, filled with random non-negative integers.
+# There are atomic operations to look at the bucket contents, and
+# to move amounts between buckets.
+class BucketStore
+
+  # Creates a BucketStore with +nbuckets+ buckets. Fills each bucket
+  # with a random non-negative integer.
+  def initialize nbuckets
+    # Create an array for the buckets
+    @buckets = (0...nbuckets).map { rand(1024) }
+
+    # Mutex used to make operations atomic
+    @mutex = Mutex.new
+  end
+
+  # Returns an array with the contents of all buckets.
+  def buckets
+    @mutex.synchronize { Array.new(@buckets) }
+  end
+
+  # Transfers _amount_ to bucket at array index _destination_,
+  # from bucket at array index _source_.
+  def transfer destination, source, amount
+    # Do nothing if both buckets are same
+    return nil if destination == source
+
+    @mutex.synchronize do
+      # Clamp amount to prevent negative value in bucket
+      amount = [amount, @buckets[source]].min
+
+      @buckets[source] -= amount
+      @buckets[destination] += amount
+    end
+    nil
+  end
+end
+
+# Create bucket store
+bucket_store = BucketStore.new 8
+
+# Get total amount in the store
+TOTAL = bucket_store.buckets.inject { |a, b| a += b }
+
+# Start a thread to equalize buckets
+Thread.new do
+  loop do
+    # Pick 2 buckets
+    buckets = bucket_store.buckets
+    first = rand buckets.length
+    second = rand buckets.length
+
+    # Swap buckets so that _first_ has not more than _second_
+    first, second = second, first if buckets[first] > buckets[second]
+
+    # Transfer half of the difference, rounded down
+    bucket_store.transfer first, second, (buckets[second] - buckets[first]) / 2
+  end
+end
+
+# Start a thread to distribute values among buckets
+Thread.new do
+  loop do
+    # Pick 2 buckets
+    buckets = bucket_store.buckets
+    first = rand buckets.length
+    second = rand buckets.length
+
+    # Transfer random amount to _first_ from _second_
+    bucket_store.transfer first, second, rand(buckets[second])
+  end
+end
+
+# Loop to display buckets
+loop do
+  sleep 1
+
+  buckets = bucket_store.buckets
+
+  # Compute the total value in all buckets.
+  # We calculate this outside BucketStore so BucketStore can't cheat by
+  # always reporting the same value.
+  n = buckets.inject { |a, b| a += b }
+
+  # Display buckets and total
+  printf "%s, total %d\n", (buckets.map { |v| sprintf "%4d", v }.join " "), n
+
+  if n != TOTAL
+    # This should never happen
+    $stderr.puts "ERROR: Total changed from #{TOTAL} to #{n}"
+    exit 1
+  end
+end

Task/Atomic-updates/Scala/atomic-updates.scala

@@ -0,0 +1,75 @@
+object AtomicUpdates {
+
+  class Buckets(ns: Int*) {
+
+    import scala.actors.Actor._
+
+    val buckets = ns.toArray
+
+    case class Get(index: Int)
+    case class Transfer(fromIndex: Int, toIndex: Int, amount: Int)
+    case object GetAll
+
+    val handler = actor {
+      loop {
+        react {
+          case Get(index) => reply(buckets(index))
+          case Transfer(fromIndex, toIndex, amount) =>
+            assert(amount >= 0)
+            val actualAmount = Math.min(amount, buckets(fromIndex))
+            buckets(fromIndex) -= actualAmount
+            buckets(toIndex) += actualAmount
+          case GetAll => reply(buckets.toList)
+        }
+      }
+    }
+
+    def get(index: Int): Int = (handler !? Get(index)).asInstanceOf[Int]
+    def transfer(fromIndex: Int, toIndex: Int, amount: Int) = handler ! Transfer(fromIndex, toIndex, amount)
+    def getAll: List[Int] = (handler !? GetAll).asInstanceOf[List[Int]]
+  }
+
+  def randomPair(n: Int): (Int, Int) = {
+    import scala.util.Random._
+    val pair = (nextInt(n), nextInt(n))
+    if (pair._1 == pair._2) randomPair(n) else pair
+  }
+
+  def main(args: Array[String]) {
+    import scala.actors.Scheduler._
+    val buckets = new Buckets(List.range(1, 11): _*)
+    val stop = new java.util.concurrent.atomic.AtomicBoolean(false)
+    val latch = new java.util.concurrent.CountDownLatch(3)
+    execute {
+      while (!stop.get) {
+        val (i1, i2) = randomPair(10)
+        val (n1, n2) = (buckets.get(i1), buckets.get(i2))
+        val m = (n1 + n2) / 2
+        if (n1 < n2)
+          buckets.transfer(i2, i1, n2 - m)
+        else
+          buckets.transfer(i1, i2, n1 - m)
+      }
+      latch.countDown
+    }
+    execute {
+      while (!stop.get) {
+        val (i1, i2) = randomPair(10)
+        val n = buckets.get(i1)
+        buckets.transfer(i1, i2, if (n == 0) 0 else scala.util.Random.nextInt(n))
+      }
+      latch.countDown
+    }
+    execute {
+      for (i <- 1 to 20) {
+        val all = buckets.getAll
+        println(all.sum + ":" + all)
+        Thread.sleep(500)
+      }
+      stop.set(true)
+      latch.countDown
+    }
+    latch.await
+    shutdown
+  }
+}

Task/Atomic-updates/Tcl/atomic-updates.tcl

@@ -0,0 +1,102 @@
+package require Thread
+package require Tk
+
+# Make the shared state
+canvas .c		;# So we can allocate the display lines in one loop
+set m [thread::mutex create]
+for {set i 0} {$i<100} {incr i} {
+    set bucket b$i	;# A handle for every bucket...
+    tsv::set buckets $bucket 50
+    lappend buckets $bucket
+    lappend lines [.c create line 0 0 0 0]
+}
+tsv::set still going 1
+
+# Make the "make more equal" task
+lappend tasks [thread::create {
+    # Perform an atomic update of two cells
+    proc transfer {b1 b2 val} {
+	variable m
+	thread::mutex lock $m
+	set v [tsv::get buckets $b1]
+	if {$val > $v} {
+	    set val $v
+	}
+	tsv::incr buckets $b1 [expr {-$val}]
+	tsv::incr buckets $b2 $val
+	thread::mutex unlock $m
+    }
+
+    # The task itself; we loop this round frequently
+    proc task {mutex buckets} {
+	variable m $mutex b $buckets i 0
+	while {[tsv::get still going]} {
+	    set b1 [lindex $b $i]
+	    if {[incr i] == [llength $b]} {set i 0}
+	    set b2 [lindex $b $i]
+
+	    if {[tsv::get buckets $b1] > [tsv::get buckets $b2]} {
+		transfer $b1 $b2 1
+	    } else {
+		transfer $b1 $b2 -1
+	    }
+	}
+    }
+    thread::wait
+}]
+
+# Make the "mess things up" task
+lappend tasks [thread::create {
+    # Utility to pick a random item from a list
+    proc pick list {
+	lindex $list [expr {int(rand() * [llength $list])}]
+    }
+    proc transfer {b1 b2 val} {
+	variable m
+	thread::mutex lock $m
+	set v [tsv::get buckets $b1]
+	if {$val > $v} {
+	    set val $v
+	}
+	tsv::incr buckets $b1 [expr {-$val}]
+	tsv::incr buckets $b2 $val
+	thread::mutex unlock $m
+    }
+
+    # The task to move a large amount between two random buckets
+    proc task {mutex buckets} {
+	variable m $mutex b $buckets
+	while {[tsv::get still going]} {
+	    set b1 [pick $b]
+	    set b2 [pick $b]
+	    transfer $b1 $b2 [expr {[tsv::get buckets $b1] / 3}]
+	}
+    }
+    thread::wait
+}]
+
+# The "main" task; we keep GUI operations in the main thread
+proc redisplay {} {
+    global m buckets lines
+    thread::mutex lock $m
+    set i 1
+    foreach b $buckets l $lines {
+	.c coords $l $i 0 $i [tsv::get buckets $b]
+	incr i 2
+    }
+    thread::mutex unlock $m
+    after 100 redisplay
+}
+
+# Start tasks and display
+.c configure -width 201 -height 120
+pack .c
+redisplay
+foreach t $tasks {
+    thread::send -async $t [list task $m $buckets]
+}
+
+# Wait for user to close window, then tidy up
+tkwait window .
+tsv::set still going 0
+thread::broadcast thread::exit