Skip to content

Adjust credits manually before each pull #1088

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

Draft
lemaitre-aneo wants to merge 1 commit into
base: main
Choose a base branch
from
Draft

Adjust credits manually before each pull #1088

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
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
34 changes: 28 additions & 6 deletions Adaptors/Amqp/src/PullQueueStorage.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -97,6 +97,21 @@ public async IAsyncEnumerable<IQueueMessageHandler> PullMessagesAsync(string
$"{partitionId}###SenderLink{i}",
$"{partitionId}###q{i}")))
.ToArray());
var receiversSet = new bool[receiversArray.Length];
await using var rollbackCredits = new Deferrer(async () =>
{
// Rollback credits for all receivers whose credits have been set during the pull
for (var i = 0; i < receiversArray.Length; i++)
{
if (receiversSet[i])
{
var receiver = await receiversArray[i];
receiver.SetCredit(Options.LinkCredit,
false);
}
}
});

while (nbPulledMessage < nbMessages)
{
var currentNbMessages = nbPulledMessage;
Expand All @@ -111,6 +126,16 @@ public async IAsyncEnumerable<IQueueMessageHandler> PullMessagesAsync(string
try
{
receiver = await receiversArray[i];

// Do not set credits if they have been set for this receiver in the same pull
if (!receiversSet[i])
{
// Set credit to number of messages to fetch + prefetch
receiver.SetCredit(nbMessages - nbPulledMessage + Options.LinkCredit,
false);
receiversSet[i] = true;
}

message = await receiver.ReceiveAsync(TimeSpan.FromMilliseconds(100))
.ConfigureAwait(false);
break;
Expand Down Expand Up @@ -177,12 +202,9 @@ private AsyncLazy<IReceiverLink> CreateReceiver(string partitionId,
$"{partitionId}###ReceiverLink{link}",
$"{partitionId}###q{link}");

/* linkCredit_: the maximum number of messages the
* remote peer can send to the receiver.
* With the goal of minimizing/deactivating
* prefetching, a value of 1 gave us the desired
* behavior. We pick a default value of 2 to have "some cache". */
rl.SetCredit(Options.LinkCredit);
// Prefetch
rl.SetCredit(Options.LinkCredit,
false);
return rl;
});
}
16 changes: 11 additions & 5 deletions Adaptors/QueueCommon/src/Amqp.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -78,12 +78,18 @@ public class Amqp
public int MaxRetries { get; set; }

/// <summary>
/// Link credit for flow control in the AMQP connection. The minimum valued supported is 1
/// For more details see:
/// <a
/// href="https: //www.rabbitmq.com/blog/2024/09/02/amqp-flow-control">
/// </a>
/// Link credit for flow control in the AMQP connection. The minimum valued supported is 0
/// </summary>
/// <remarks>
/// <p>
/// Credits are set to LinkCredit + number of messages to pull each time a pull occur.
/// Consequently, the option LinkCredit represent the number of messages to prefetch by the AMQP driver.
/// </p>
/// <p>
/// For more details see:
/// <a href="https: //www.rabbitmq.com/blog/2024/09/02/amqp-flow-control" />
/// </p>
/// </remarks>
public int LinkCredit { get; set; }

/// <summary>
Expand Down
Loading