

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 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: seconds;
 
__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) {
    auto rng = Xorshift(1);
 
    while (data.running) {
        immutable i = uniform(0, data.length, rng);
        immutable j = uniform(0, data.length, rng);
        immutable amount = uniform!"[]"(0, 20, rng);
        data.transfer(i, j, amount);
    }
}
 
void equalize(size_t N)(Buckets!N data) {
    auto rng = Xorshift(1);
 
    while (data.running) {
        immutable i = uniform(0, data.length, rng);
        immutable j = uniform(0, data.length, 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(1.seconds);
    }
    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();
}