Design for deduplicating concurrent tasks in flight simultaneously

Question

I have 1-20 agents that issue requests over time with roughly ~50 in total in flight at any given time. Many of these requests are identical. The tasks are idempotent, but if two identical requests were to evaluate in parallel, the evaluation is still costly even though the final results could be stored in a thread safe manner. So ideally duplicate requests are coalesced, evaluated only once, and the result broadcast to all requesting agents.

My current design is the following:

• Create N awaitable concurrent queues, each with:
  • dedicated worker thread without any sort of work stealing.
  • dedicated results cache only accessed by its worker thread.
  • multiple producers, single consumer WRT the queue.
• After a request comes in, use consistent ring hashing to push the task into one of the work queues.
• Each worker thread pops from its queue.
• Peek at the next item in the queue to see if that request is identical to the just popped one so requests are coalesced.
• Once no longer able to coalesce:
  • Checks the results cache if an identical request has already been processed.
  • If not found, process the task and store the result in the results cache.
• Return the result.

Pros:

• Guarantees a task is performed only once (so long as the results are cached) since identical tasks are serialized.
• Minimizes locking and sync primitives - effectively the concurrent queue is the only thing dealing with sync primitives.
• Implementation is pretty simple.

Cons:

• Reliant on consistent ring hashing to distribute work evenly.
  • But does not appear to be an issue. Distribution is good and enough work in flight simultaneously that nothing is starved or overloaded.
• Unable to batch complete all identical requests in flight simultaneously. Only able to coalesce adjacent identical requests.
• Results cache memory usage.

Is there an alternative design that I should consider that could minimize the existing cons? Any other feedback on this approach? Implementation details I should watch out for?

Answer 1

This looks like a question about:

• Load-balancing,
• Partitioning, and
• Non-blocking design (or, preventing unnecessary blocking by making trivial tweaks to the design)

---

Facts gathered from the question:

• There is already a results cache.
• Under the current design, each results cache is local to a worker.

---

Inputs, outputs, and mappings:

• Input --> HashedInput (fast)
• Input --> Output (slow)
• Output --> HashedOutput (fast)
• InputOutputTable (fast; constant time; atomic)
  • Key: HashedInput
  • Value: either Output or HashedOutput

---

Processes (actions):

• Request: Input, looking for Output

  • If InputOutputTable contains Input as Key, use corresponding Value as output
  • Else
    • Insert entry, mark corresponding Value as Pending
    • Send Input to Load Balancer (see below)
    • Wait until value is computed and populated
    • Return populated Value

• WorkerProcess:

  • Sleep until queue is not empty
  • Pop from queue
  • Optionally, check whether the request has been processed before, either at the same worker or elsewhere; See (lengthy) discussion below.
  • Process
  • Populate InputOutputTable

---

Question: should I partition the InputOutputTable?

Answer: If it fits in the memory of one computer, and if all processing (i.e. all workers) take place on a single computer, there is no benefit in partitioning it.

However, avoiding all-at-once resizing (rehashing) is a worthy goal (avoids the occasional worst-case timing); for this reason alone, Linear Hashing and Distributed Hash Table techniques are worth consideration, even if partitioning is not needed.

Link to Wikipedia: Hash Table - Alternatives to all-at-once rehashing

The matter of whether to do so comes down to: whether the benefit justifies the integration effort and maintenance costs; whether a performant and reliable (i.e. bug-free and having few/no downsides) implementation without restrictive licensing is available for use; etc.

---

Question: Should I put the InputOutputTable in front of or behind the workers?

Answer: In front of.

If InputOutputTable is put behind (i.e. guarded by) either the Load Balancer or the group/individual worker, incoming requests will not be able to get a timely response, until all previous requests on the queue have finished processing.

In other words, it is highly desirable for incoming requests to quickly check whether a given input has been processed and populated in the table, without waiting on any queueing or processing that happens on the Load Balancer or the group/individual worker.

---

Question: Should each worker maintain its own table of recently processed requests?

Presupposition 1: This question pre-supposes that the input space has been partitioned and assigned to workers.

If the input space is not partitioned, in which every worker can potentially receive requests over the entire input space, then every worker will need to know the entire table of recently processed requests over the entire input space. This is not much different from just having one global table of recently processed requests - a role that has already been fulfilled by InputOutputTable.

Presupposition 2: This question pre-supposes that a worker can potentially see a request that it has just processed.

However, if the InputOutputTable has been used correctly, such requests should not have reached the worker at all.

Answer / conclusion: the question seems moot.

---

Question: should I partition the input space in order to achieve load-balancing for uniquely new work (never processed before)?

Answer: Doesn't seem to be necessary, and if one chooses to do so, it is merely a choice between load-balancing strategies.

Depends on: Whether there is computation efficiency gains for requests that are adjacent (from partial reuse of intermediates or results). Presumably no, because it it were the case, OP would have highlighted this fact in the question upfront.

Depends on: Whether strict fairness (strict first-come first-serve / FIFO behavior) is deemed necessary across the entire input space. OP has implied that fairness is not required; an approximation to that fairness (i.e. being FIFO after partitioning into individual workers) would be good enough.

Detail:

By the time the request is sent to the Load Balancer, it is already given that a request for that entry has been received before, because an entry has already been created on the InputOutputTable.

(Due to asynchronicity, it will not be immediately known whether processing has started or finished. It is only given that, at the time of checking, the result hasn't been populated on the InputOutputTable yet.)

---

Question: What blocking / non-blocking behaviors should be expected and/or achievable?

Answer:

Expectations

• If the result is immediately available (cached), the requester expects to be given this result, without minimal waiting (as low as reasonably achievable).

Options for what can be achievable

• An external requester (agent) can request blockingly.
  • If the result is not immediately available, it can ask to be put to sleep, and only to be waken up when the result becomes available.
• An external requester (agent) can ask non-blockingly.
  • If the result is not immediately available,
    • It expects to be told so, immediately;
    • Optionally, it can be told whether it's the first requester to ask for this input.

---

Question: To achieve the expectations, what else would need to be implemented?

Answer: The list of requesters waiting on unfinished requests need to be maintained; requesters waiting on a single item should be notified (waken up) when the item is populated.

Implementation caveat:

It is inadvisable to put the synchronization primitives into the individual (item-level) entries on the InputOutputTable.

Reason: A hash table (in particular, a DHT etc) is supposed to manage data. While a synchronization primitive may look small in terms of userland memory size, there are hidden resource allocations if that synchronization primitive is ever activated (when contention has happened, i.e. two different requesters made the same request). Once activated, extra steps are needed to reclaim those hidden resource allocations. Not doing so may result in a slow but deadly resource leak.

Lesson: The synchronization primitives that are used to notify (wake up) waiting requesters need to be properly cleaned up once they are past their time of usefulness. Furthermore, this cleanup needs to happen timely; it should not be contingent upon the eviction of the computed result from the InputOutputTable, for that would be an unacceptably long time.

Suggested design tweak:

In addition to InputOutputTable, have another WaitTable that maps HashedInput to a list of Requesters. Once an item has been processed and the list of requesters have been notified, the corresponding entry should be removed from WaitTable.