Skip to content

feat: AEP 7: Queue system interface design #51

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
aneojgurhem wants to merge 4 commits into
base: main
Choose a base branch
from
Open

feat: AEP 7: Queue system interface design #51

Changes from all commits
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
4e17046
feat: AEP 7: Queue system interface design
aneojgurhem Jan 15, 2025
e0e31f4
refactor: rewrite AEP to explain our design choices and the current i…
aneojgurhem Jan 29, 2025
55fd459
docs: add more informations about queue interfaces working
aneojgurhem Feb 3, 2025
ed8dd33
Take comments into account
aneojgurhem Feb 4, 2025
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
228 changes: 228 additions & 0 deletions AEP/aep-00007.md
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -0,0 +1,228 @@
# AEP 7: Queue system interface design

| |ArmoniK Enhancement Proposal|
|---: |:--- |
| **AEP** | 7 |
| **Title** | Queue system interface design |
| **Author** | Jérôme Gurhem <<jgurhem@aneo.fr>> |
| **Status** | Active |
| **Type** | Standard |
| **Creation Date** | 2025-01-15 |

# Abstract

This proposal outlines the rationale and design decisions behind the current message processing and lifecycle management system in ArmoniK. It highlights the primary reasons for adopting a pull-based model, discusses design constraints, and evaluates the trade-offs involved in ensuring reliability and robustness for long-running tasks.

# Motivation

The main motivation for the current architecture is to maintain a historical interface that provides comprehensive traceability and monitoring of tasks and message processing. This interface is central to operational observability and is a key requirement for the system.

# Rationale

Copy link

@wkirschenmann wkirschenmann Feb 5, 2025

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

In ArmoniK, the queue system contains messages representing the tasks ready to be executed (i.e. their input data are all available). In ArmoniK, the messages are processed by several Scheduling Agents.

Simplified message processing Algorithm

  1. Acquire the message
  2. Process the task
    1. Prepare the task
    2. Execute the task
    3. Complete the task
  3. Acknowledge the message

## Functional Requirements

To ensure optimal message handling within ArmoniK, strict control over the number of messages being processed simultaneously is required.
Each message represents a task that can be executed, necessitating fine-grained lifecycle management.
Copy link

@wkirschenmann wkirschenmann Feb 5, 2025

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Why?

Copy link

@wkirschenmann wkirschenmann Feb 5, 2025

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

The explanation is below. The paragraphs structure doesn't show that.


When a message is received, the associated task begins processing immediately.
The message is acknowledged only upon the successful or unsuccessful completion of the task.
Copy link

@wkirschenmann wkirschenmann Feb 5, 2025

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

"acknowledged" should be explained first.
You can also replace with "removed" if you don't want to do that.

It the task completes successfully, the message is acknowledged after tasks results are uploaded to the storage and tasks created by the task in processing are submitted.
If the queue service loses connection with the processing agent, the message is redelivered, ensuring task execution even in the presence of errors.

Additionally, tasks that are pending can be released when a long-running task is in progress.
ArmoniK initiates the processing of a new message only when the current task has been dispatched to a worker, starting the acquisition of a new task during the processing of the previous one.
This design allows to have a task ready for execution as soon as the previous task ends.
Tasks undergoing retry are treated as entirely new tasks, which simplifies tracking and execution, generating new messages accordingly.

Message uniqueness is not required as it is managed elsewhere.
Furthermore, message scheduling must be handled within the queue service to accommodate prioritization mechanisms.
The system should also allow for the seamless integration of new plugins to enhance flexibility and adaptability.
Comment on lines +24 to +39
Copy link

@wkirschenmann wkirschenmann Feb 5, 2025

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Suggested change
To ensure optimal message handling within ArmoniK, strict control over the number of messages being processed simultaneously is required.
Each message represents a task that can be executed, necessitating fine-grained lifecycle management.
When a message is received, the associated task begins processing immediately.
The message is acknowledged only upon the successful or unsuccessful completion of the task.
It the task completes successfully, the message is acknowledged after tasks results are uploaded to the storage and tasks created by the task in processing are submitted.
If the queue service loses connection with the processing agent, the message is redelivered, ensuring task execution even in the presence of errors.
Additionally, tasks that are pending can be released when a long-running task is in progress.
ArmoniK initiates the processing of a new message only when the current task has been dispatched to a worker, starting the acquisition of a new task during the processing of the previous one.
This design allows to have a task ready for execution as soon as the previous task ends.
Tasks undergoing retry are treated as entirely new tasks, which simplifies tracking and execution, generating new messages accordingly.
Message uniqueness is not required as it is managed elsewhere.
Furthermore, message scheduling must be handled within the queue service to accommodate prioritization mechanisms.
The system should also allow for the seamless integration of new plugins to enhance flexibility and adaptability.
1. The Queue System should allow for **multiple Scheduling Agent** to connect and get messages. While it should provide each Scheduling Agent with a different message, it need not to provide a guarantee of uniqueness of distribution.
2. The Queue System should **not lose any message**.
3. As long as there are messages in the Queue System, the Scheduling Agents should receive messages, i.e. they cannot starve if there are messages in the queue.
4. The Queue System should allow the Scheduling Agents to have a **strict control over the number of messages acquired**. This will allow the Scheduling Agent to parallelize message processing, either through pipelining (e.g. prepare a task while executing another task) or by executing several tasks simultaneously.
5. The Queue System should allow to **detect when a Scheduling Agent is not responsive** and provide the messages to another Scheduling Agent. This feature is at the core of the resilience mechanisms of ArmoniK and ensures that all tasks will be processed even if there are errors on the executing nodes.
6. The Queue System should **allow a Scheduling Agent to "liberate" a message**, i.e. the message is signaled for the Queue System to provide the message to another Scheduling Agent. When using a pipelined parallelization, this feature can be used to avoid that a message waits for a too long time when a long task is being executed.
While the properties 1, 2, and 3 are inherent to the solution chosen, the other features should be exposed through the interface.


## Possible Approaches for the Interface
Copy link
Contributor

@lemaitre-aneo lemaitre-aneo Feb 19, 2025

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

You need to talk about long polling


Several approaches can be considered to implement the message processing interface, including event-driven and pull-based mechanisms. Given the need to control pipelining in order to manage the number of concurrently processed messages and determining when processing begins, both approaches are functionally equivalent. In practice, polling would be used on an internal queue to regulate message processing.

Below is an overview of how major market solutions implement message retrieval mechanisms:

| Solution | Approach |
| ------------- | -------- |
| AWS SQS | pull |
| ActiveMQ | both |
| RabbitMQ | both |
| Google PubSub | both |
| Pulsar | both |

## Trade-offs and Constraints

The current design prioritizes simplicity and robustness over a pure event-driven model. Considering the long duration of tasks and the cost of orchestration, some inefficiencies during message reception are acceptable. These inefficiencies are preferable to the development overhead and technical debt involved in transforming the existing pull-based mechanism. Rewriting the codebase to adopt a fully event-driven, push-based pattern is not feasible given the complexity and scale of such a transformation.

The technical limitations of the underlying queueing service also play an important role in the design decisions. SQS, for instance, does not provide an API for push-based message reception, which necessitates to use of a pull-based approach. Additionally, past attempts to use RabbitMQ for push-based message handling revealed significant stability issues, such as connection losses and inconsistent message processing. These challenges have further reinforced the decision to rely on the pull-based model, which offers greater reliability and predictability.

We can consider converting a push-based reception to messages into our pull-based interfaces. Naturally, this implementation requires to be proved as stable and as scalable as the current one. Pull requests are welcome to improve the implementation of our queue system adaptors.
Comment on lines +57 to +61
Copy link

@wkirschenmann wkirschenmann Feb 5, 2025

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Suggested change
The current design prioritizes simplicity and robustness over a pure event-driven model. Considering the long duration of tasks and the cost of orchestration, some inefficiencies during message reception are acceptable. These inefficiencies are preferable to the development overhead and technical debt involved in transforming the existing pull-based mechanism. Rewriting the codebase to adopt a fully event-driven, push-based pattern is not feasible given the complexity and scale of such a transformation.
The technical limitations of the underlying queueing service also play an important role in the design decisions. SQS, for instance, does not provide an API for push-based message reception, which necessitates to use of a pull-based approach. Additionally, past attempts to use RabbitMQ for push-based message handling revealed significant stability issues, such as connection losses and inconsistent message processing. These challenges have further reinforced the decision to rely on the pull-based model, which offers greater reliability and predictability.
We can consider converting a push-based reception to messages into our pull-based interfaces. Naturally, this implementation requires to be proved as stable and as scalable as the current one. Pull requests are welcome to improve the implementation of our queue system adaptors.
### Event-driven approach (push)
Using an event-driven approach allows the Queue System to push the messages to the Scheduling Agent. The control of "when the messages arrives in the Scheduling Agent" is moved to the Queue system. This allows to reduce the latency between the submission of a message in the Queue System and its dispatch in the Scheduling Agent. On the other hand, managing the number of messages received by the Scheduling Agent is more complex, or impossible if the number of messages allowed in each Scheduling Agent can change frequently over time.
### Pull based approach
Using a pull-based approach, the Scheduling Agent has to call the Queue System to try and get a new message. This allows the Scheduling Agent to control when a new message is obtained. However, this increases the latency to get new messages.
Note that in the Scheduling Agent pipelined implementation, polling is used on internal queues to regulate the number of message being in each step of processing.
### Choice
The technical limitations of the underlying queueing service also play an important role in the design decisions. SQS, for instance, does not provide an API for push-based message reception, which necessitates to use of a pull-based approach. A library can expose a push-based interface while using a pull-based Queue System (or vice-versa).
In the end, we believe that the choice made should be the result of two main constraints :
* a pipelined implementation
* a need to minimize the number of messages stuck in the steps prior the task execution
These constraints are more easily taken into account with a pull-based approach as the information to decide whether or not a new message should enter the pipeline are available in the Scheduling Agent rather than the Queue system.
Additionally, past experiments to use RabbitMQ for push-based message handling revealed significant stability issues, such as connection losses and inconsistent message processing. These challenges have further reinforced the decision to rely on the pull-based model, which offers greater reliability and predictability.

Copy link
Contributor

@lemaitre-aneo lemaitre-aneo Feb 19, 2025

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I think it is really crucial to explain long polling as it is a good tradeoff between the two, and it is what is actually used.


## Current Interface

Queue interfaces can be found in our dotnet package [ArmoniK.Core](https://www.nuget.org/packages/ArmoniK.Core.Base).
The package provides the elements necessary to implement queue plugins that can be dynamically loaded by ArmoniK.Core allowing users to implement plugins that match their requirements.

An agent from a partition calls the `PullMessagesAsync` method to get messages representing tasks from the associated partition.
The partition is given outside of the `IPullQueueStorage` interface and should be passed to the implementation through other means.
It is usually done with `.Net` options system and environment variables.
This also means there is no uniformization to set up the partition and it will depend on the implementation of the interface.

```csharp
/// <summary>
/// Interface to retrieve messages from the queue
/// </summary>
public interface IPullQueueStorage : IQueueStorage
{
/// <summary>
/// Gets messages from the queue
/// </summary>
/// <param name="nbMessages">Number of messages to retrieve</param>
/// <param name="cancellationToken">Token used to cancel the execution of the method</param>
/// <returns>
/// Enumerator allowing async iteration over the message queue
/// </returns>
IAsyncEnumerable<IQueueMessageHandler> PullMessagesAsync(int nbMessages,
CancellationToken cancellationToken = default);
}
```

The `PullMessagesAsync` method returns an `IAsyncEnumerable<IQueueMessageHandler>` where `IQueueMessageHandler` is an interface representing the lifecycle of a message.

```csharp
/// <summary>
/// Interface to handle queue messages lifecycle.
/// </summary>
public interface IQueueMessageHandler : IAsyncDisposable
{
/// <summary>
/// Used to signal that the message ownership has been lost
/// </summary>
[Obsolete("ArmoniK now manages loss of link with the queue")]
CancellationToken CancellationToken { get; set; }

/// <summary>
/// Id of the message
/// </summary>
string MessageId { get; }

/// <summary>
/// Task Id contained in the message
/// </summary>
string TaskId { get; }

/// <summary>
/// Status of the message. Used when the handler is disposed to notify the queue.
/// </summary>
QueueMessageStatus Status { get; set; }

/// <summary>
/// Date of reception of the message
/// </summary>
DateTime ReceptionDateTime { get; init; }
}
```

The `QueueMessageStatus` is set by ArmoniK during the execution of the task.
Then, the `DisposeAsync` method inherited from `IAsyncDisposable` is used to process the message from the queue by acknowledging it, not acknowledging it, or requeuing it depending on the status.
The statuses are defined as follow:

```csharp
/// <summary>
/// Represents the status of a queue message
/// </summary>
public enum QueueMessageStatus
{
/// <summary>
/// Message is waiting for being processed.
/// </summary>
Waiting,

/// <summary>
/// Message processing has failed. The message should be put back at the begin of the queue.
Copy link
Contributor

@lemaitre-aneo lemaitre-aneo Feb 19, 2025

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I am not sure about this one.

Copy link
Contributor Author

@aneojgurhem aneojgurhem Feb 19, 2025

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

That's how it is currently written in our code.

/// </summary>
Failed,

/// <summary>
/// The message is being processed.
/// </summary>
Running,

/// <summary>
/// Task is not ready to be processed. The message should be put at the end of the queue.
/// </summary>
Postponed,

/// <summary>
/// The message has been processed. It can safely be removed from the queue.
/// </summary>
Processed,

/// <summary>
/// The message processing has been cancelled. the message can safely be removed from the queue.
/// </summary>
Cancelled,

/// <summary>
/// Message has been retried too many times and is considered as poisonous for the queue
/// </summary>
Poisonous,
}
```

Interface to insert tasks into queue is simpler.
For a given partition, ArmoniK gives an `IEnumerable<MessageData>` where each `MessageData` represents a task.
It contains the task identifier, the session identifier and the options of the task.
The options has a field for the priority of the task.
The queue system and the implementation of the interfaces are then responsible to distribute tasks.

```csharp
/// <summary>
/// Interface to insert messages into the queue
/// </summary>
public interface IPushQueueStorage : IQueueStorage
{
/// <summary>
/// Puts messages into the queue, handles priorities of messages
/// </summary>
/// <param name="messages">Collection of messages</param>
/// <param name="partitionId">Id of the partition</param>
/// <param name="cancellationToken">Token used to cancel the execution of the method</param>
/// <returns>
/// Task representing the asynchronous execution of the method
/// </returns>
public Task PushMessagesAsync(IEnumerable<MessageData> messages,
string partitionId,
CancellationToken cancellationToken = default);
}
```

```csharp
/// <summary>
/// Data structure to hold message data
/// </summary>
/// <param name="TaskId">Unique identifier of the task</param>
/// <param name="SessionId">Unique name of the session to which this message belongs</param>
/// <param name="Options">Task options</param>
public record MessageData(string TaskId,
string SessionId,
TaskOptions Options);
```

## Issues with the current interface

There are a few issues from the current interfaces:
- The split of the interface into two due to the configuration. The partition should be given in the pull method directly and uniformize partition selection.
- Clarify message processing instead of relying on the `DisposeAsync` method from the `IQueueMessageHandler`.

These issues will be addressed in a new version of these interfaces.

# Conclusion

The architecture reflects a deliberate balance between design complexity, operational requirements, and system constraints. By combining a historical interface with strict control over message processing and lifecycle management, ArmoniK achieves a reliable and scalable system. Although the system does not adopt a fully event-driven approach, it remains robust and resilient, meeting the demands of long-running tasks and orchestration challenges. The trade-offs in this design are justified by the system’s operational stability and its ability to handle large-scale workloads efficiently.
Copy link

@wkirschenmann wkirschenmann Feb 5, 2025

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Suggested change
The architecture reflects a deliberate balance between design complexity, operational requirements, and system constraints. By combining a historical interface with strict control over message processing and lifecycle management, ArmoniK achieves a reliable and scalable system. Although the system does not adopt a fully event-driven approach, it remains robust and resilient, meeting the demands of long-running tasks and orchestration challenges. The trade-offs in this design are justified by the system’s operational stability and its ability to handle large-scale workloads efficiently.
The architecture reflects a deliberate balance between design complexity, operational requirements, and system constraints. The interface reflects an architecture that allows a strict and precise control over message processing and lifecycle management. This allows ArmoniK to achieve a reliable and scalable system. The trade-offs in this design are justified by the system’s operational stability and its ability to handle large-scale workloads efficiently.

Although the system does not adopt a fully event-driven approach, it remains robust and resilient, meeting the demands of long-running tasks and orchestration challenges. has been removed as it could be interpreted that a full event-driven approach should be a target. This is not the case.


# Copyright

This document is placed in the public domain or under the CC0-1.0-Universal license, whichever is more permissive