CDE

Task/Atomic-updates/C-sharp/atomic-updates.cs

@@ -0,0 +1,125 @@
+using System; //Rand class
+using System.Threading; //Thread, Mutex classes
+public class ThreadSafeBuckets
+{
+    //This class is thread safe, and ensures that all operations on it are atomic.
+    //Calling threads do not need to ensure safety.
+    Random rand = new Random();
+    int[] Buckets;
+    object[] locks; //Mutexes for each bucket so they can lock individually
+    public int BucketCount { get; private set; }
+    public ThreadSafeBuckets(int bucketcount)
+    {
+        //Create buckets+mutexes and fill them with a random amount
+        BucketCount = bucketcount;
+        Buckets = new int[bucketcount];
+        locks = new object[bucketcount];
+        int startingtotal = 0;
+        for (int i = 0; i < BucketCount; i++)
+        {
+            locks[i] = new object();
+            Buckets[i] = rand.Next(30);
+            startingtotal += Buckets[i];
+        }
+        //Print the starting total
+        Console.WriteLine("Starting total: " + startingtotal);
+    }
+    public int GetBucketValue(int i)
+    {
+        return Buckets[i];
+    }
+    public void Transfer(int i, int j, int amount)
+    {
+        //Transfer amount from bucket i to bucket j
+        if (i > BucketCount || j > BucketCount || i < 0 || j < 0 ||
+            i == j || amount < 0)
+            return;
+
+        //To prevent deadlock, always lock the lower bucket first
+        lock (locks[Math.Min(i, j)])
+            lock (locks[Math.Max(i, j)])
+            {
+                //Make sure don't transfer out more than is in the bucket
+                amount = Math.Min(amount, Buckets[i]);
+
+                //Do the transfer
+                Buckets[i] -= amount;
+                Buckets[j] += amount;
+            }
+    }
+
+    public void PrintBuckets()
+    {
+        int counter = 0;
+        //Lock all the buckets in sequential order and print their contents
+        for (int i = 0; i < BucketCount; i++)
+        {
+            Monitor.Enter(locks[i]);
+            Console.Write(Buckets[i] + " ");
+            counter += Buckets[i];
+        }
+        //Print the bucket total, then unlock all the mutexes
+        Console.Write("= " + counter);
+        Console.WriteLine();
+
+        foreach (var l in locks)
+            Monitor.Exit(l);
+    }
+}
+
+class Program
+{
+    static ThreadSafeBuckets TSBs;
+
+    public static void Main(){
+        //Create the thread-safe bucket list
+        TSBs = new ThreadSafeBuckets(10);
+        TSBs.PrintBuckets();
+        //Create and start the Equalizing Thread
+        new Thread(new ThreadStart(EqualizerThread)).Start();
+        Thread.Sleep(1);
+        //Create and start the Randamizing Thread
+        new Thread(new ThreadStart(RandomizerThread)).Start();
+        //Use this thread to do the printing
+        PrinterThread();
+    }
+    //EqualizerThread runs on it's own thread and randomly averages two buckets
+    static void EqualizerThread()
+    {
+        Random rand = new Random();
+        while (true)
+        {
+            //Pick two buckets
+            int b1 = rand.Next(TSBs.BucketCount);
+            int b2 = rand.Next(TSBs.BucketCount);
+            //Get the difference
+            int diff = TSBs.GetBucketValue(b1) - TSBs.GetBucketValue(b2);
+            //Transfer to equalize
+            if (diff < 0)
+                TSBs.Transfer(b2, b1, -diff / 2);
+            else
+                TSBs.Transfer(b1, b2, diff/2);
+        }
+    }
+    //RandomizerThread redistributes the values between two buckets
+    static void RandomizerThread()
+    {
+        Random rand = new Random();
+        while (true)
+        {
+            int b1 = rand.Next(TSBs.BucketCount);
+            int b2 = rand.Next(TSBs.BucketCount);
+            int diff = rand.Next(TSBs.GetBucketValue(b1));
+            TSBs.Transfer(b1, b2, diff);
+        }
+    }
+    //PrinterThread prints the current bucket contents
+    static void PrinterThread()
+    {
+        while (true)
+        {
+            Thread.Sleep(50); //Only print every few milliseconds to let the other threads work
+            TSBs.PrintBuckets();
+        }
+    }
+}

Task/Atomic-updates/D/atomic-updates.d

@@ -0,0 +1,118 @@
+import std.stdio: writeln;
+import std.conv: text;
+import std.random: uniform, Xorshift;
+import std.algorithm: min, max;
+import std.parallelism: task;
+import core.thread: Thread;
+import core.sync.mutex: Mutex;
+import core.time: dur;
+
+__gshared uint transfersCount;
+
+final class Buckets(size_t nBuckets) if (nBuckets > 0) {
+    alias TBucketValue = uint;
+
+    // The trailing padding avoids cache line contention
+    // when run with two or more cores.
+    align(128) private static struct Bucket {
+        TBucketValue value;
+        Mutex mtx;
+        alias value this;
+    }
+
+    private Bucket[nBuckets] buckets;
+    private bool running;
+
+    public this() {
+        this.running = true;
+        foreach (ref b; buckets)
+            b = Bucket(uniform(0, 100), new Mutex);
+    }
+
+    public TBucketValue opIndex(in size_t index) const pure nothrow {
+        return buckets[index];
+    }
+
+    public void transfer(in size_t from, in size_t to,
+                         in TBucketValue amount) {
+        immutable low  = min(from, to);
+        immutable high = max(from, to);
+        buckets[low].mtx.lock();
+        buckets[high].mtx.lock();
+
+        scope(exit) {
+            buckets[low].mtx.unlock();
+            buckets[high].mtx.unlock();
+        }
+
+        immutable realAmount = min(buckets[from].value, amount);
+        buckets[from] -= realAmount;
+        buckets[to  ] += realAmount;
+        transfersCount++;
+    }
+
+    @property size_t length() const pure nothrow {
+        return this.buckets.length;
+    }
+
+    void toString(in void delegate(const(char)[]) sink) {
+        TBucketValue total = 0;
+        foreach (ref b; buckets) {
+            b.mtx.lock();
+            total += b;
+        }
+
+        scope(exit)
+            foreach (ref b; buckets)
+                b.mtx.unlock();
+
+        sink(text(buckets));
+        sink(" ");
+        sink(text(total));
+    }
+}
+
+void randomize(size_t N)(Buckets!N data) {
+    immutable maxi = data.length - 1;
+    auto rng = Xorshift(1);
+
+    while (data.running) {
+        immutable i = uniform(0, maxi, rng);
+        immutable j = uniform(0, maxi, rng);
+        immutable amount = uniform(0, 20, rng);
+        data.transfer(i, j, amount);
+    }
+}
+
+void equalize(size_t N)(Buckets!N data) {
+    immutable maxi = data.length - 1;
+    auto rng = Xorshift(1);
+
+    while (data.running) {
+        immutable i = uniform(0, maxi, rng);
+        immutable j = uniform(0, maxi, rng);
+        immutable a = data[i];
+        immutable b = data[j];
+        if (a > b)
+            data.transfer(i, j, (a - b) / 2);
+        else
+            data.transfer(j, i, (b - a) / 2);
+    }
+}
+
+void display(size_t N)(Buckets!N data) {
+    foreach (immutable _; 0 .. 10) {
+        writeln(transfersCount, " ", data);
+        transfersCount = 0;
+        Thread.sleep(dur!"msecs"(1000));
+    }
+    data.running = false;
+}
+
+void main() {
+    writeln("N. transfers, buckets, buckets sum:");
+    auto data = new Buckets!20();
+    task!randomize(data).executeInNewThread();
+    task!equalize(data).executeInNewThread();
+    task!display(data).executeInNewThread();
+}

Task/Atomic-updates/E/atomic-updates.e

@@ -0,0 +1,123 @@
+#!/usr/bin/env rune
+pragma.syntax("0.9")
+
+def pi := (-1.0).acos()
+def makeEPainter := <unsafe:com.zooko.tray.makeEPainter>
+def colors := <awt:makeColor>
+
+# --------------------------------------------------------------
+# --- Definitions
+
+/** Execute 'task' repeatedly as long 'indicator' is unresolved. */
+def doWhileUnresolved(indicator, task) {
+  def loop() {
+    if (!Ref.isResolved(indicator)) {
+      task()
+      loop <- ()
+    }
+  }
+  loop <- ()
+}
+
+/** The data structure specified for the task. */
+def makeBuckets(size) {
+    def values := ([100] * size).diverge() # storage
+    def buckets {
+        to size() :int { return size }
+        /** get current quantity in bucket 'i' */
+        to get(i :int) { return values[i] }
+        /** transfer 'amount' units, as much as possible, from bucket 'i' to bucket 'j'
+            or vice versa if 'amount' is negative */
+        to transfer(i :int, j :int, amount :int) {
+            def amountLim := amount.min(values[i]).max(-(values[j]))
+            values[i] -= amountLim
+            values[j] += amountLim
+        }
+    }
+    return buckets
+}
+
+/** A view of the current state of the buckets. */
+def makeDisplayComponent(buckets) {
+  def c := makeEPainter(def paintCallback {
+    to paintComponent(g) {
+      def pixelsW := c.getWidth()
+      def pixelsH := c.getHeight()
+      def bucketsW := buckets.size()
+
+      g.setColor(colors.getWhite())
+      g.fillRect(0, 0, pixelsW, pixelsH)
+
+      g.setColor(colors.getDarkGray())
+      var sum := 0
+      for i in 0..!bucketsW {
+        sum += def value := buckets[i]
+        def x0 := (i       * pixelsW / bucketsW).floor()
+        def x1 := ((i + 1) * pixelsW / bucketsW).floor()
+        g.fillRect(x0 + 1, pixelsH - value,
+                   x1 - x0 - 1, value)
+      }
+
+      g.setColor(colors.getBlack())
+      g."drawString(String, int, int)"(`Total: $sum`, 2, 20)
+    }
+  })
+  c.setPreferredSize(<awt:makeDimension>(500, 300))
+  return c
+}
+
+# --------------------------------------------------------------
+# --- Application setup
+
+def buckets := makeBuckets(100)
+def done # Promise indicating when the window is closed
+
+# Create the window
+def frame := <unsafe:javax.swing.makeJFrame>("Atomic transfers")
+frame.setContentPane(def display := makeDisplayComponent(buckets))
+frame.addWindowListener(def mainWindowListener {
+  to windowClosing(event) :void {
+    bind done := null
+  }
+  match _ {}
+})
+frame.setLocation(50, 50)
+frame.pack()
+
+# --------------------------------------------------------------
+# --- Tasks
+
+# Neatens up buckets
+var ni := 0
+doWhileUnresolved(done, fn {
+  def i := ni
+  def j := (ni + 1) %% buckets.size()
+  buckets.transfer(i, j, (buckets[i] - buckets[j]) // 4)
+  ni := j
+})
+
+# Messes up buckets
+var mi := 0
+doWhileUnresolved(done, fn {
+    def i := (mi + entropy.nextInt(3)) %% buckets.size()
+    def j := (i + entropy.nextInt(3)) %% buckets.size() #entropy.nextInt(buckets.size())
+    buckets.transfer(i, j, (buckets[i] / pi).floor())
+    mi := j
+})
+
+# Updates display at fixed 10 Hz
+# (Note: tries to catch up; on slow systems slow this down or it will starve the other tasks)
+def clock := timer.every(100, def _(_) {
+  if (Ref.isResolved(done)) {
+    clock.stop()
+  } else {
+    display.repaint()
+  }
+})
+clock.start()
+
+# --------------------------------------------------------------
+# --- All ready, go visible and wait
+
+frame.show()
+interp.waitAtTop(done)

Task/Atomic-updates/Euphoria/atomic-updates.euphoria

@@ -0,0 +1,69 @@
+function move(sequence s, integer amount, integer src, integer dest)
+    if src < 1 or src > length(s) or dest < 1 or dest > length(s) or amount < 0 then
+        return -1
+    else
+        if src != dest and amount then
+            if amount > s[src] then
+                amount = s[src]
+            end if
+            s[src] -= amount
+            s[dest] += amount
+        end if
+        return s
+    end if
+end function
+
+sequence buckets
+buckets = repeat(100,10)
+
+procedure equalize()
+    integer i, j, diff
+    while 1 do
+        i = rand(length(buckets))
+        j = rand(length(buckets))
+        diff = buckets[i] - buckets[j]
+        if  diff >= 2 then
+            buckets = move(buckets, floor(diff / 2), i, j)
+        elsif diff <= -2 then
+            buckets = move(buckets, -floor(diff / 2), j, i)
+        end if
+        task_yield()
+    end while
+end procedure
+
+procedure redistribute()
+    integer i, j
+    while 1 do
+        i = rand(length(buckets))
+        j = rand(length(buckets))
+        if buckets[i] then
+            buckets = move(buckets, rand(buckets[i]), i, j)
+        end if
+        task_yield()
+    end while
+end procedure
+
+function sum(sequence s)
+    integer sum
+    sum = 0
+    for i = 1 to length(s) do
+        sum += s[i]
+    end for
+    return sum
+end function
+
+atom task
+
+task = task_create(routine_id("equalize"), {})
+task_schedule(task, 1)
+
+task = task_create(routine_id("redistribute"), {})
+task_schedule(task, 1)
+
+task_schedule(0, {0.5, 0.5})
+
+for i = 1 to 24 do
+    print(1,buckets)
+    printf(1," sum: %d\n", {sum(buckets)})
+    task_yield()
+end for