The classic producer–consumer problem (aka the bounded-buffer problem) in D.

prodcom.d

import std.stdio;
import core.thread;
import core.sync.condition;
import core.time;
import std.random;

class Buffer(T) {

	private {
		T[] array;
		size_t count;
		Mutex lock;
		Condition notEmpty, notFull;
	}

	this(size_t size) {
		array.length = size;
		lock = new Mutex;
		notEmpty = new Condition(lock);
		notFull = new Condition(lock);
	}

	auto add(T item) {
		synchronized(lock) {
			while (count == array.length) {
				debug writeln("queue full, producer waiting");
				notFull.wait();
			}
			array[count++] = item;
			debug writeln(array[0..count]);
			notEmpty.notify();
		}
	}

	auto remove() {
		synchronized(lock) {
			T result;
			while (count < 1) {
				notEmpty.wait();
			}
			--count;
			result = array[count];
			debug writeln(array[0..count]);
			notFull.notify();
			return result;
		}
	}

}

class Producer(T) : Thread {
	private {
		Buffer!T buffer;
	}
	this (Buffer!T buffer) {
		super(&run);
		this.buffer = buffer;
	}

	void run() {
		foreach (i; 0 .. 32) {
			Thread.sleep(1.seconds);
			writeln("producing ", i);
			buffer.add(i);
		}
	}
}

class Consumer(T) : Thread {
	private {
		Buffer!T buffer;
	}
	this (Buffer!T buffer, string str) {
		super(&run);
		this.buffer = buffer;
		this.name = str;
	}

	void run() {
		int c = 0;
		while (c < 16) {
			Thread.sleep(uniform(0,5).seconds);
			auto item = buffer.remove();
			writeln("consumed ", item);
			++c;
		}
	}
}

void main() {
	alias int T;
	auto buffer = new Buffer!T(10);

	auto producer = new Producer!T(buffer);
	auto consumer1 = new Consumer!T(buffer, "consumer1");
	auto consumer2 = new Consumer!T(buffer, "consumer2");

	producer.start();
	consumer1.start();
	consumer2.start();

	consumer1.join();
	consumer2.join();
	producer.join();
}