Skip to content

Commit

Permalink
Related to discussion #1793
Browse files Browse the repository at this point in the history
* Expand throttling delay code, add comment.
* Add `ThrowsExceptionRateLimiter` and use it in `MaybeSomethingUpWithRateLimiter_GH1793`, which reproduces the issue.
* Reproduce issue by artifically never acquiring rate limiter leases.
* Fix issue by catching `SemaphoreFullException` and checking that the exception was thrown in a valid failure case.
  • Loading branch information
lukebakken committed Feb 25, 2025
1 parent 13353a2 commit 4271e64
Show file tree
Hide file tree
Showing 4 changed files with 154 additions and 4 deletions.
19 changes: 18 additions & 1 deletion projects/RabbitMQ.Client/Impl/Channel.PublisherConfirms.cs
Original file line number Diff line number Diff line change
Expand Up @@ -344,7 +344,24 @@ private async Task MaybeEndPublisherConfirmationTracking(PublisherConfirmationIn
{
if (_publisherConfirmationsEnabled)
{
_confirmSemaphore.Release();
try
{
_confirmSemaphore.Release();
}
catch (SemaphoreFullException ex)
{
/*
* rabbitmq/rabbitmq-dotnet-client-1793
* If MaybeStartPublisherConfirmationTracking throws an exception *prior* to acquiring
* _confirmSemaphore, the above Release() call will throw SemaphoreFullException.
* In "normal" cases, publisherConfirmationInfo will thus be null, but if not, throw
* a "bug found" exception here.
*/
if (publisherConfirmationInfo is not null)
{
throw new InvalidOperationException(InternalConstants.BugFound, ex);
}
}

if (publisherConfirmationInfo is not null)
{
Expand Down
17 changes: 14 additions & 3 deletions projects/RabbitMQ.Client/ThrottlingRateLimiter.cs
Original file line number Diff line number Diff line change
Expand Up @@ -134,13 +134,24 @@ protected override void Dispose(bool disposing)

private int CalculateDelay()
{
long? availablePermits = _concurrencyLimiter.GetStatistics()?.CurrentAvailablePermits;
if (!(availablePermits < _throttlingThreshold))
RateLimiterStatistics? rateLimiterStatistics = _concurrencyLimiter.GetStatistics();
if (rateLimiterStatistics is null)
{
return 0;
}

return (int)((1.0 - availablePermits / (double)_maxConcurrency) * 1000);
long availablePermits = rateLimiterStatistics.CurrentAvailablePermits;
if (availablePermits >= _throttlingThreshold)
{
/*
* Note: do NOT add a delay because available permits exceeeds the threshold
* below which throttling begins
*/
return 0;
}

double percentageUsed = 1.0 - (availablePermits / (double)_maxConcurrency);
return (int)(percentageUsed * 1000);
}
}
}
68 changes: 68 additions & 0 deletions projects/Test/Integration/GH/NeverAcquiredRateLimiter.cs
Original file line number Diff line number Diff line change
@@ -0,0 +1,68 @@
// This source code is dual-licensed under the Apache License, version
// 2.0, and the Mozilla Public License, version 2.0.
//
// The APL v2.0:
//
//---------------------------------------------------------------------------
// Copyright (c) 2007-2024 Broadcom. All Rights Reserved.
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
// https://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.
//---------------------------------------------------------------------------
//
// The MPL v2.0:
//
//---------------------------------------------------------------------------
// This Source Code Form is subject to the terms of the Mozilla Public
// License, v. 2.0. If a copy of the MPL was not distributed with this
// file, You can obtain one at https://mozilla.org/MPL/2.0/.
//
// Copyright (c) 2007-2024 Broadcom. All Rights Reserved.
//---------------------------------------------------------------------------

using System;
using System.Collections.Generic;
using System.Threading;
using System.Threading.RateLimiting;
using System.Threading.Tasks;

namespace Integration.GH
{
public class NeverAcquiredRateLimiter : RateLimiter
{
public override TimeSpan? IdleDuration => throw new NotImplementedException();
public override RateLimiterStatistics GetStatistics() => throw new NotImplementedException();

protected override ValueTask<RateLimitLease> AcquireAsyncCore(int permitCount, CancellationToken cancellationToken)
{
return new ValueTask<RateLimitLease>(new NotAcquiredRateLimitLease());
}

protected override RateLimitLease AttemptAcquireCore(int permitCount)
{
return new NotAcquiredRateLimitLease();
}
}

public class NotAcquiredRateLimitLease : RateLimitLease
{
public override bool IsAcquired => false;

public override IEnumerable<string> MetadataNames => [];

public override bool TryGetMetadata(string metadataName, out object metadata)
{
metadata = string.Empty;
return true;
}
}
}
54 changes: 54 additions & 0 deletions projects/Test/Integration/GH/TestGitHubIssues.cs
Original file line number Diff line number Diff line change
Expand Up @@ -34,6 +34,7 @@
using System.Text;
using System.Threading;
using System.Threading.Tasks;
using Integration.GH;
using RabbitMQ.Client;
using RabbitMQ.Client.Events;
using RabbitMQ.Client.Exceptions;
Expand Down Expand Up @@ -329,5 +330,58 @@ await ch.BasicPublishAsync(exchange: string.Empty, routingKey: queueName,
_output.WriteLine("saw {0} publishExceptions", publishExceptions.Count);
}
}

[Fact]
public async Task MaybeSomethingUpWithRateLimiter_GH1793()
{
const int messageCount = 16;

_connFactory = new ConnectionFactory
{
AutomaticRecoveryEnabled = true
};

_conn = await _connFactory.CreateConnectionAsync();

var channelOpts = new CreateChannelOptions(
publisherConfirmationsEnabled: true,
publisherConfirmationTrackingEnabled: true,
outstandingPublisherConfirmationsRateLimiter: new NeverAcquiredRateLimiter()
);

_channel = await _conn.CreateChannelAsync(channelOpts);

var properties = new BasicProperties
{
DeliveryMode = DeliveryModes.Persistent
};

for (int i = 0; i < messageCount; i++)
{
int retryCount = 0;
const int maxRetries = 3;
while (retryCount <= maxRetries)
{
try
{
byte[] bytes = Encoding.UTF8.GetBytes("message");
await Assert.ThrowsAnyAsync<InvalidOperationException>(async () =>
{
await _channel.BasicPublishAsync(string.Empty, string.Empty, true, properties, bytes);
});
break;
}
catch (SemaphoreFullException ex0)
{
_output.WriteLine("{0} ex: {1}", _testDisplayName, ex0);
retryCount++;
}
catch (PublishException)
{
retryCount++;
}
}
}
}
}
}

0 comments on commit 4271e64

Please sign in to comment.