I try to implement chunk (something like lock-free fixedsize queue)
--
import core.atomic;

shared struct Chunk(T, uint N)
{
	shared T[N] data;
	shared uint count;
	shared uint queueCounter;

	@property uint capacity() { return N; }
	@property uint count() { return count; }
	@property bool full() { return count == N; }

	void append(shared T value)
	{
		atomicOp!("+=")(queueCounter, 1);
		while(1)
		{
			uint c = count;
			if(cas(&count, c, c + 1))
			{
				data[c] = value;
				atomicOp!("-=")(queueCounter, 1);
				break;
			}
		}		
	}

	bool waitAll()
	{
		if(!full())
		{
			return false;
		}

		while(0 != queueCounter)
		{
		}

		return true;
	}
}
--

And call it like:

--
import std.parallelism;

struct S
{
    bool dirty;
    int time;
    int[16] data;
}

int main(string[] argv)
{
    const uint N = 14344;

    shared Chunk!(S, N) ch;

    foreach(i; taskPool.parallel(std.range.iota(N), 10))
    {
        shared S item;
        item.time = i;
        ch.append(item);
    }
    while(!ch.waitAll()) {}

    // DONE

    return 0;
}
--

It works fine with N == 14343, but fails without any message with 14344 (value depends on computer).

Why does program fail?

Am I doing correct CAS append?

On Thursday, 31 March 2016 at 18:25:46 UTC, jacob wrote:
> I try to implement chunk (something like lock-free fixedsize queue)
> ...

Check out this implementation https://github.com/MartinNowak/lock-free/blob/master/src/lock_free/rwqueue.d