Data update

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

@@ -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/Crystal/atomic-updates.cr

@@ -0,0 +1,103 @@
+require "wait_group"
+
+class Buckets
+  @buckets : Array(Int32)
+  @locks : Array(Mutex)
+
+  def initialize (n)
+    @buckets = Array.new(n, 100)
+    @locks = Array.new(n) { Mutex.new }
+  end
+
+  def size
+    @buckets.size
+  end
+
+  def [] (idx)
+    @locks[idx].synchronize do
+      @buckets[idx]
+    end
+  end
+
+  def transfer (from, to, amount)
+    return if from == to
+    from, to, amount = to, from, amount.abs if amount < 0
+    l1, l2 = from, to
+    l1, l2 = l2, l1 if l1 > l2
+    @locks[l1].synchronize do
+      @locks[l2].synchronize do
+        amount = Math.min(@buckets[from], amount)
+        @buckets[from] -= amount
+        @buckets[to]   += amount
+      end
+    end
+  end
+
+  def snapshot
+    (0...@buckets.size).each do |i| @locks[i].lock end
+    result = @buckets.dup
+    (0...@buckets.size).reverse_each do |i| @locks[i].unlock end
+    result
+  end
+end
+
+# pick two buckets and make their values closer to equal
+def equalize (buckets, channel)
+  loop do
+    b1 = rand(buckets.size)
+    b2 = rand(buckets.size)
+    diff = buckets[b1] - (buckets[b1] + buckets[b2]) // 2
+    buckets.transfer(b1, b2, diff)
+    # check termination
+    select
+    when stop = channel.receive
+      return
+    else # continue
+    end
+  end
+end
+
+# pick two buckets and arbitrarily redistribute their values
+def distribute (buckets, channel)
+  loop do
+    b1 = rand(buckets.size)
+    b2 = rand(buckets.size)
+    buckets.transfer(b1, b2, rand(0..buckets[b1]))
+    #check termination
+    select
+    when stop = channel.receive
+      return
+    else # continue
+    end
+  end
+end
+
+# display contents of buckets and sum
+def display (buckets, channel)
+  loop do
+    bs = buckets.snapshot
+    puts "Total: %4d, buckets: %s" % { bs.sum, bs }
+    sleep 1.second
+    #check termination
+    select
+    when stop = channel.receive
+      return
+    else # continue
+    end
+  end
+end
+
+wg = WaitGroup.new
+stop_channels = Array.new(3) { Channel(Bool).new }
+
+buckets = Buckets.new(10)
+
+wg.spawn { display(buckets, stop_channels[0]) }
+wg.spawn { equalize(buckets, stop_channels[1]) }
+wg.spawn { distribute(buckets, stop_channels[2]) }
+
+sleep 10.seconds
+
+stop_channels.each do |ch| ch.send(true) end
+
+wg.wait

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

@@ -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