feat(client): add parallel execution control for receiver callbacks

[flc1125]

Add an attribute to support customizing the number of asynchronous parallel executions.

Currently, creating coroutines without limits significantly impacts server performance. Since we currently only have serial and arbitrary options, this attribute has been added to facilitate customizing the number of asynchronous coroutines.

Certainly, the actual total number of goroutines is determined by WithPollGoroutines and WithParallelGoroutines, and it involves a multiplication operation.

These two may seem redundant, but their purposes are actually different.

• WithPollGoroutines defines the concurrency level for reading events from the protocol;

• WithParallelGoroutines, on the other hand, specifies the parallel processing count after the events are read.

[duglin]

Am i reading this right... we don't support blocking any more?

[flc1125]

I changed the value of WithParallelGoroutines to 1, which can achieve the same effect. Moreover, it will be more versatile.

[flc1125]

see https://github.com/flc1125/cloudevents-sdk-go/blob/ce30137fe5dbc884c4860f2114622957b739c996/v2/client/options.go#L122

[duglin]

it may not matter but you're blocking at a different place now... if I'm reading this right.
Before it was: read -> block while processing
Now it's: read -> func(process)() -> read -> wait for a free spot to open in the channel
Right? This means that we're getting the next event BEFORE we're ready, where before we could only get one AFTER we were ready. Not sure it matters, but something to think about.

Also, we need testcases.

[flc1125]

Before it was: read -> block while processing
Now it's: read -> func(process)() -> read -> wait for a free spot to open in the channel

You're right, indeed. Given this situation, should we maintain the original approach or treat it as a breaking change?
I'd like to confirm our final approach so I can make some adjustments accordingly.

PS: I'll supplement the unit tests later.

[duglin]

I think it depends on whether calling:

• c.responder.Respond(ctx)
• c.receiver.Receive(ctx)

before we're ready is risky or not. For example, if we get the next "msg" and then pause, and this entire for-loop stops. If another process tries to continue with the same responser/receiver, that one "msg" will be lost.

Abstractly it sounds like a bug. But in practice I don't know if the idea of continuing to get messages outside of the instance of this for-loop makes any sense. E.g. if we're pulling messages from a queue during the "receiver.Receive()" call, then I think this situation might occur.

[flc1125]

We can make assumptions based on the logic of WithPollGoroutines. When there are multiple asynchronous fetches occurring simultaneously, it essentially means the scenario you just described is happening.

Can I understand it this way: event messages, when distributed across different processes (goroutines or others), inherently have the potential to encounter such situations? At the very least, the message currently being read would appear as "lost" or "non-existent" to other processes (goroutines or others).

Therefore, I believe this scenario might not be a cause for concern.

Of course, there is one extreme case where the user specifically desires to read messages "sequentially," meaning no other processes (goroutines or others) are allowed to handle the "next" message additionally.

However, I think this situation is similar to using blockingCallback, where the user intentionally makes such a choice. When this happens, it implies there is no scenario where another process (goroutine or other) is missing a message still waiting for the channel to become available.

[duglin]

Couple of things come to mind:
1 - I make a distinction between "lost a message because the processor of that message crashed", and "lost a message because it was pulled it from a queue and purposely never even tried to be processed". The former is kind of expected. The latter is a bug IMO. While the latter isn't 100% "purposely dropped it"... the logic comes close since we don't take into account the situation where this entire go process just dies while that message is "on hold". The queue it's being pulled from should be where it's "on hold".

2 - if we're pulling messages from a queue then this "on hold" message could be processed before the next message if there's another processor pulling messages from that queue. Now, I'm not suggesting that it's our job to try to synchronous things in such a scenario, however, I think someone would find it really odd that this "on hold" message might be delayed significantly before it's even started to be processed while the next message might be processed immediately.

3 - "this situation is similar to using blockingCallback"... exactly. We don't support "blocking" any more. At least not with the same "when are things pulled off of the original queue" semantics.

I'm curious to know @embano1's thoughts, but I'm leaning towards a "only pull the next message when we're ready" model, otherwise it kind of feels like a potential bug or we're introducing another queue into the flow... granted a queue of size one, but still a queue, that has zero persistence/resiliency to it.

[flc1125]

I think you're right. We need to maintain predictability for end users regarding events. Hiding certain event-related operations might make users feel things are beyond their control.

Perhaps we could slightly adjust the channel position—that might preserve the original logic. Of course, this is just my initial thought—I'll look into it more carefully when I have time to confirm.

[flc1125]

@duglin Involves some methods, functions, and variable naming aspects. If there are better suggestions, they can also be provided to me for adjustment.