dmd-concurrency

[Shammah Chancellor]

On 2013-11-20 07:34:36 +0000, Chris Williams said:

> On Wednesday, 20 November 2013 at 04:24:14 UTC, Daniel Murphy wrote:
>> This is the correct forum to post phobos proposals on.
> 
> Well then, here's what I had written:
> 
> A few applications I've considered implementing seem like they would be 
> easier if there was a channel-based messaging system in 
> std.concurrency. I'm happy to do this implementation, but I thought I 
> would try to get some sort of sign-off before doing so. Following, I 
> will lay out my argument for the addition, and then the API that I am 
> considering.
> 
> ---
> 
> One fairly common task is thread-pooling. With the standard 
> send/receive model currently implemented, you have to choose a specific 
> thread to target when you send a task. While it's true that you can 
> simply iterate through your list of threads over and over, to spread 
> the load evenly over them, that presumes that all tasks take even 
> processing time. It makes more sense to be able to push data into a 
> shared channel (secretly a work queue), and the first thread that 
> finishes its previous task will be able to immediately pull the task 
> before everyone else. This also means that the necessity of passing 
> around references to your threads so that they can be looped over goes 
> away.
> 
> I haven't tested it, but it looks like this sort of thing might be 
> quasi-possible using the register/unregister/locate methods. As each 
> thread starts, it can register itself with a named group (i.e. 
> channel), and then anyone who wants to send an item to an arbitrary 
> thread in that group can call locate() to retrieve one thread and call 
> send() against the Tid. The target thread would then need to unregister 
> itself while it is doing work, then re-register itself. My complaint 
> against this is the need to unregister and re-register. If the thread 
> issuing commands sends a large number of tasks all at once, they will 
> all go to the same thread (if coded poorly) or the caller will need to 
> use yield() or sleep() to allow the target thread to receive the task 
> and unregister, so that locate() can find a different thread. That's 
> not terribly efficient. I am also concerned that there's the chance 
> that all threads will be unregistered when we call locate(), whereas a 
> channeling system would be able to expand the mailbox during the times 
> that all threads are busy.
> 
> The actual implementation within concurrency.d also concerns me as (if 
> I read it correctly), the most recent item to register() will be the 
> one which locate() finds, rather than the thread which has been 
> registered the longest. While I suppose it's probably not too large of 
> an issue if the same two threads keep taking all the tasks - that means 
> that your load can't exceed two threads worth of processing power - it 
> still seems like a LIFO system would be better. The registry is also 
> based on an array rather than a set, which can make removal an O(n) 
> operation, if the contents of the registry have to be shifted left, to 
> fill an empty spot.
> 
> Overall, I think that adding a shared message box system would be a 
> straightforward way to improve the handling of thread pooling via the 
> actor model.
> 
> ---
> 
> A less common use-case but I was also considering some world-simulators 
> (e.g. for studying economics or building a game map) and here the 
> ability to broadcast messages to a large set of other actors, based on 
> location, interest, etc. seems useful. In this case, messages would 
> need to be copied out to each subscriber in the channel rather than 
> having an existence as a point to point connection. For a networked 
> game, most likely you would want to break each channel into two, where 
> locally all senders on a channel push to a single listener that pipes 
> the messages over the network, and then remotely the messages would be 
> broadcast to many listeners again, but that's a reasonably 
> straightforward task for someone to implement on top of the channel 
> functionality. I don't think that such functionality is needed in 
> Phobos itself. Mostly, the presence of the broadcasting functionality 
> in the standard library allows them to use the easy and safe actor 
> model for more creative uses than a straight one-to-one pipe.
> 
> ---
> 
> Overall, my hope would be to develop something that is conceptually no 
> more difficult to deal with than the current send()/receive() model, 
> but also able to be used in a wide variety of ways. The API that I 
> would propose to develop is:
> 
> interface Channel {
> 	void send(T...)(T vals);
> 	void prioritySend(T...)(T vals);
> 	void receive(T...)(out Tid sender, T ops);
> 	receiveOnlyRet!(T) receiveOnly(T...)();
> 	bool receiveTimeout(T...)(Duration d, T ops);
> 
> 	void setMaxMailboxSize(Tid tid, size_t messages, OnCrowding doThis);
> 	void setMaxMailboxSize(Tid tid, size_t messages, bool function(Tid) 
> doThisFunc);
> }
> 
> class SingleChannel : Channel {}	// Send inserts a message into a 
> shared message box. Receive removes message
> 
> class DuplicateChannel(bool echo = true) : Channel {}	// Send inserts 
> the message into a message box per-recipient. Receive removes message 
> in the calling thread's channel message box. If echo is false, messages 
> will not be sent back to the sender, even if they are a registered 
> listener
> 
> void registerSend(Channel c, Tid tid = thisTid);	// used by function 
> sendAll(). Channel can be of either type
> void unregisterSend(Channel c, Tid tid = thisTid);
> void registerReceive(Channel c, Tid tid = thisTid);	// used by function 
> receiveAll(). Channel can be of either type
> void unregisterReceive(Channel c, Tid tid = thisTid);
> 
> void sendAll(T...)(T ops); // Sends a copy of message to all channels 
> this thread has registered for.
> void receiveAll(T...)(out Channel c, out Tid sender, T ops); // 
> Receives a message of type T from any channel that we are registered 
> for. Returns channel and sender
> 
> I believe that the look and feel stays fairly consistent with the 
> current set of functions in std.concurrency. I've added the ability for 
> the recipient to infer information about the sender since, in the 
> duplication model, I believe there are quite a few cases where this 
> would be important information. And of course, I've added the option to 
> register/unregister threads other than ourselves to allow a greater 
> range of code layouts, though it's possible that the lack of this sort 
> of thing in the original code is due to some sort of safety concern?
> 
> The most straightforward way to implement the DuplicateChannel would be 
> to use the individual threads' message boxes, but this would mean that 
> data put into a channel could be pulled out via the traditional 
> receive() method. Currently, my intention would be to partition these 
> two systems (the direct send()/receive() model and the channel model), 
> unless anyone has any reason to think they should be merged into a 
> single whole?
> 
> Those are my thoughts, anyways. Comments? Complaints?

How does one receive from multiple channels out-of-order?   I would 
rather this sent it to the subscribed Tid via send, rather than having 
an additional queue.   It could possible send a ChannelMessage which 
has a reference to the sending channel and the message.   I understand 
this is a different model than what Go and whatnot use, but I think 
it's more pratical in some circumstances.   Maybe both ways would be 
good?  I personally use this method in my vibe-d server.