Awaiting multiple channels

[withinboredom]

Awaiting multiple channels

[kelunik]

You'll likely want to use Future::iterate instead if you need something like that.

Instead of a new fiber (async) for every receive, you could also put the receive loop into the async call and put all items into a Queue, but whether that makes sense depends on your logic.

[trowski]

Thus the 'controller' MUST get all items off of the channel before the worker halts. If the worker halts before this, then any data still in the channel is lost.

This wasn't the intended behavior, which I fixed several days ago. Update to v2.2.1 and you'll be able to retrieve all data sent on the channel before the child closes.

[withinboredom]

Thanks that was a really good suggestion and the code is quite simpler now :)

I have another related question. Basically, how does one handle the race condition where data is sent in a channel but the worker halts? The code I'm refactoring is pretty straightforward to use this library, but the biggest difference is that this library maintains the channels for you, but closes them once the worker halts. Thus the 'controller' MUST get all items off of the channel before the worker halts. If the worker halts before this, then any data still in the channel is lost.

It would be ideal if the channel were not closed until both sides are __destructed. I think I can write something like this in my code, but I wanted to get your thoughts on it before doing so.

[withinboredom]

That's awesome, I'll give it a go!

[withinboredom]

This is still throwing

while (!$execution->getChannel()->isClosed()) {
				try {
					$prefix[] = $execution->getChannel()->receive(new TimeoutCancellation(1));
				} catch (TimeoutException) {
					Logger::log('Warning: waited for event to prefix!');
				}
			}

Channel source closed unexpectedly

Is there anyway to tell if there is data to be read from the channel?

[trowski]

Channel is a low-level component that makes no assumptions about the data to be transmitted. You'll want to send some message to indicate the other side is closing the channel gracefully. null is a good candidate for this. Instead of looping on checking if the channel is closed, check if you received null.

[withinboredom]

null is a valid value in my case, and I ended up using a different approach. In most cases, the worker is not ending gracefully when hitting this.

The workflow looks something like:

stateDiagram-v2
    [*] --> DequeueBatch
    DequeueBatch --> SendingToWorker
    SendingToWorker --> CrashDetected
    CrashDetected --> ReadingFinalMessages
    ReadingFinalMessages --> DequeueBatch
    SendingToWorker --> Complete
    Complete --> ReadingFinalMessages

I ended up making a change where the worker 'acks' a message once it's been successfully processed (or 'half-ack' if it is taking longer than expected). This way, after a period of time, the 'work' can be resubmitted after the designated worker dies unexpectedly.

It seems to be pretty resilient. At least as resilient as the old parallel implementation (if a bit more complex).

Thank you for you replies, I think this can be closed.