• jacob (67/67) Mar 31 2016 I try to implement chunk (something like lock-free fixedsize
• MrSmith (3/6) Mar 31 2016 Check out this implementation

jacob <jj75607 gmail.com> writes:

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?

MrSmith <mrsmith33 yandex.ru> writes:

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