Drafty Synced!T example with std.concurrancy

sealed.d

import std.concurrency, std.stdio, std.container, core.sync.mutex, core.thread;


struct Synced(T) if(!(is(T == class)))
{
private:
    shared T* payload;
    __gshared Mutex mut; //mutex should have worked as shared

    public this(Args)(Args args...){
        payload = cast(shared)new T(args);
        mut = new Mutex();  
    }

    public void apply(scope void delegate(Proxy arg) action){
        mut.lock();
        scope(exit) mut.unlock();
        action(Proxy(payload));
    }

    struct Proxy{
    private:
        shared T* payload;
    public:
        alias _getPayload this;
        @property ref T _getPayload(){
            return *cast(T*)payload;
        }
    }
}

void process(int n){
    Synced!(Array!int) myStock;
    receive(
        (Synced!(Array!int) r){
            myStock = r;            
        }
    );
    for(int i=0; i<count; i++){
        myStock.apply((x){
            x.insertBack(i);
            pragma(msg, typeof(&x));//prints Proxy*, somewhat foolproof
        });
    }
}
enum count = 12;

void main(){
    //Gotcha:  Synced!(Array!int)() == T.init 
    //so create an Array with 1 element
    auto stock = Synced!(Array!int)(0);
    int nThreads = 10;
    Tid[] tids;
    foreach(i; 0..nThreads){
        tids ~= spawn(&process, i); 
    }
    foreach(i; 0..nThreads){
        send(tids[i], stock);
    }
    bool notFull = true;
    //spin until stock is full
    while(notFull){
        stock.apply((x){
            if(x.length == nThreads*count + 1){
                writeln(x[]); //to witness interleaving 
                notFull = false;
            }
        });
        Thread.sleep(dur!"msecs"(1));
    }
    //count up elements
    int[] counts = new int[count];
    stock.apply((x){
        foreach(v; x[])
            counts[v]++;
    });

    assert(counts[0] == nThreads+1);
    foreach(v; 1..count)
        assert(counts[v] == nThreads);
}