Skip to content

Commit dff2afb

Browse files
FRACerqueiraFRACerqueira
committed
Renamed for readability syntax : LockAcquireWhenAutoScale -> LockWhenScaling
1 parent f2a8df1 commit dff2afb

11 files changed

Lines changed: 115 additions & 227 deletions

File tree

‎README.md‎

Lines changed: 10 additions & 107 deletions
Original file line numberDiff line numberDiff line change
@@ -60,7 +60,7 @@ A ring buffer makes a bounded queue when separate indices are used for inserting
6060
- Bug fixed when used with Rabbitmq.
6161
- Removed need to set Automatic Recovery to false for use with Rabbitmq
6262
- Removed Master/Slave, ReportScale, BufferHealth, ScaleWhen..., RollbackWhen... and TriggerByAccqWhen... concept (Break changes)
63-
- Added command LockAcquireWhenAutoScale
63+
- Added command LockWhenScaling
6464
- Added command AutoScaleAcquireFault
6565
- Added command HeartBeat
6666
- Added command BackgroundLogger
@@ -116,9 +116,9 @@ var rb = await RingBuffer<int>.New("MyBuffer")
116116

117117
Console.WriteLine($"Ring Buffer name({rb.Name}) created.");
118118
Console.WriteLine($"Ring Buffer Current capacity is : {rb.CurrentCapacity}");
119-
Console.WriteLine($"Ring Buffer name({rb.Name}) IsInitCapacity = {rb.IsInitCapacity}.");
120-
Console.WriteLine($"Ring Buffer name({rb.Name}) IsMaxCapacity = {rb.IsMaxCapacity}.");
121-
Console.WriteLine($"Ring Buffer name({rb.Name}) IsMinCapacity = {rb.IsMinCapacity}.");
119+
Console.WriteLine($"Ring Buffer name({rb.Name}) IsInitCapacity({rb.Capacity}) = {rb.IsInitCapacity}.");
120+
Console.WriteLine($"Ring Buffer name({rb.Name}) IsMaxCapacity({rb.MaxCapacity}) = {rb.IsMaxCapacity}.");
121+
Console.WriteLine($"Ring Buffer name({rb.Name}) IsMinCapacity({rb.MinCapacity}) = {rb.IsMinCapacity}.");
122122

123123
using (var buffer = await rb.AcquireAsync(token))
124124
{
@@ -152,29 +152,11 @@ var rb = await RingBuffer<int>.New("MyBuffer")
152152
.MaxCapacity(9)
153153
.BuildWarmupAsync(token);
154154

155-
Console.WriteLine($"Ring Buffer name({rb.Name}) created.");
156-
Console.WriteLine($"Ring Buffer Current capacity is : {rb.CurrentCapacity}");
157-
Console.WriteLine($"Ring Buffer name({rb.Name}) IsInitCapacity = {rb.IsInitCapacity}.");
158-
Console.WriteLine($"Ring Buffer name({rb.Name}) IsMaxCapacity = {rb.IsMaxCapacity}.");
159-
Console.WriteLine($"Ring Buffer name({rb.Name}) IsMinCapacity = {rb.IsMinCapacity}.");
160-
161155
if (!await rb.SwitchToAsync(ScaleSwitch.MaxCapacity))
162156
{
163157
//manual scale was not scheduled
164158
//do something
165-
}
166-
167-
using (var buffer = await rb.AcquireAsync(token))
168-
{
169-
if (buffer.Successful)
170-
{
171-
Console.WriteLine($"Buffer is ok({buffer.Successful}:{buffer.ElapsedTime}) value: {buffer.Current}");
172-
}
173-
else
174-
{
175-
//do something
176-
}
177-
}
159+
}
178160
```
179161

180162
### Trigger Scale Usage
@@ -198,27 +180,9 @@ var rb = await RingBuffer<int>.New("MyBuffer")
198180
.MinCapacity(3)
199181
.MaxCapacity(9)
200182
.BuildWarmupAsync(token);
201-
202-
Console.WriteLine($"Ring Buffer name({rb.Name}) created.");
203-
Console.WriteLine($"Ring Buffer Current capacity is : {rb.CurrentCapacity}");
204-
Console.WriteLine($"Ring Buffer name({rb.Name}) IsInitCapacity = {rb.IsInitCapacity}.");
205-
Console.WriteLine($"Ring Buffer name({rb.Name}) IsMaxCapacity = {rb.IsMaxCapacity}.");
206-
Console.WriteLine($"Ring Buffer name({rb.Name}) IsMinCapacity = {rb.IsMinCapacity}.");
207-
208-
using (var buffer = await rb.AcquireAsync(token))
209-
{
210-
if (buffer.Successful)
211-
{
212-
Console.WriteLine($"Buffer is ok({buffer.Successful}:{buffer.ElapsedTime}) value: {buffer.Current}");
213-
}
214-
else
215-
{
216-
//do something
217-
}
218-
}
219183
```
220184

221-
### Lock/Unlock de Acquire/Switch Usage
185+
### Lock Acquire/Switch Usage
222186
[**Top**](#table-of-contents)
223187

224188
When the scaling up or down process is executed, acquisition or scale switching is not blocked.
@@ -233,29 +197,10 @@ var rb = await RingBuffer<int>.New("MyBuffer")
233197
.Logger(HostApp.Services.GetService<ILogger<Program>>())
234198
.Factory((_) => { return Task.FromResult(rnd.Next(1, 10)); })
235199
.ScaleTimer(50, TimeSpan.FromSeconds(5))
236-
.LockAcquireWhenAutoScale()
237-
.AutoScaleAcquireFault()
200+
.LockWhenScaling()
238201
.MinCapacity(3)
239202
.MaxCapacity(9)
240203
.BuildWarmupAsync(token);
241-
242-
Console.WriteLine($"Ring Buffer name({rb.Name}) created.");
243-
Console.WriteLine($"Ring Buffer Current capacity is : {rb.CurrentCapacity}");
244-
Console.WriteLine($"Ring Buffer name({rb.Name}) IsInitCapacity = {rb.IsInitCapacity}.");
245-
Console.WriteLine($"Ring Buffer name({rb.Name}) IsMaxCapacity = {rb.IsMaxCapacity}.");
246-
Console.WriteLine($"Ring Buffer name({rb.Name}) IsMinCapacity = {rb.IsMinCapacity}.");
247-
248-
using (var buffer = await rb.AcquireAsync(token))
249-
{
250-
if (buffer.Successful)
251-
{
252-
Console.WriteLine($"Buffer is ok({buffer.Successful}:{buffer.ElapsedTime}) value: {buffer.Current}");
253-
}
254-
else
255-
{
256-
//do something
257-
}
258-
}
259204
```
260205

261206
### HeartBeat Usage
@@ -275,24 +220,6 @@ var rb = await RingBuffer<int>.New("MyBuffer")
275220
.Factory((_) => { return Task.FromResult(rnd.Next(1, 10)); })
276221
.BuildWarmupAsync(token);
277222

278-
Console.WriteLine($"Ring Buffer name({rb.Name}) created.");
279-
Console.WriteLine($"Ring Buffer Current capacity is : {rb.CurrentCapacity}");
280-
Console.WriteLine($"Ring Buffer name({rb.Name}) IsInitCapacity = {rb.IsInitCapacity}.");
281-
Console.WriteLine($"Ring Buffer name({rb.Name}) IsMaxCapacity = {rb.IsMaxCapacity}.");
282-
Console.WriteLine($"Ring Buffer name({rb.Name}) IsMinCapacity = {rb.IsMinCapacity}.");
283-
284-
using (var buffer = await rb.AcquireAsync(token))
285-
{
286-
if (buffer.Successful)
287-
{
288-
Console.WriteLine($"Buffer is ok({buffer.Successful}:{buffer.ElapsedTime}) value: {buffer.Current}");
289-
}
290-
else
291-
{
292-
//do something
293-
}
294-
}
295-
296223
private void MyHeartBeat(RingBufferValue<int> item)
297224
{
298225
//do anything ex: health check
@@ -302,7 +229,7 @@ private void MyHeartBeat(RingBufferValue<int> item)
302229
### Background Logger Usage
303230
[**Top**](#table-of-contents)
304231

305-
Log execution is done automatically by the component (Level debug and Error) in the same execution thread. This process can burden execution if the log recording process takes a long time.
232+
Log execution is done automatically by the component (Level Debug, Warning and Error) in the same execution thread. This process can burden execution if the log recording process takes a long time.
306233

307234
For this scenario, you can use the log execution in the background in an asynchronous process done by the component.
308235

@@ -315,24 +242,6 @@ var rb = await RingBuffer<int>.New("MyBuffer")
315242
.BackgroundLogger()
316243
.Factory((_) => { return Task.FromResult(rnd.Next(1, 10)); })
317244
.BuildWarmupAsync(token);
318-
319-
Console.WriteLine($"Ring Buffer name({rb.Name}) created.");
320-
Console.WriteLine($"Ring Buffer Current capacity is : {rb.CurrentCapacity}");
321-
Console.WriteLine($"Ring Buffer name({rb.Name}) IsInitCapacity = {rb.IsInitCapacity}.");
322-
Console.WriteLine($"Ring Buffer name({rb.Name}) IsMaxCapacity = {rb.IsMaxCapacity}.");
323-
Console.WriteLine($"Ring Buffer name({rb.Name}) IsMinCapacity = {rb.IsMinCapacity}.");
324-
325-
using (var buffer = await rb.AcquireAsync(token))
326-
{
327-
if (buffer.Successful)
328-
{
329-
Console.WriteLine($"Buffer is ok({buffer.Successful}:{buffer.ElapsedTime}) value: {buffer.Current}");
330-
}
331-
else
332-
{
333-
//do something
334-
}
335-
}
336245
```
337246

338247
## RabbitMQ Usage
@@ -363,19 +272,13 @@ var rb = await RingBuffer<IChannel>.New("RabbitChanels")
363272
.Capacity(10)
364273
.Logger(applogger!)
365274
.BackgroundLogger()
366-
.Factory((cts) => ModelFactory(cts)!)
275+
.Factory((cts) => ChannelFactory(cts))
367276
.ScaleTimer(100, TimeSpan.FromSeconds(10))
368277
.MaxCapacity(20)
369278
.MinCapacity(5)
370279
.AutoScaleAcquireFault()
371280
.BuildWarmupAsync(token);
372281

373-
Console.WriteLine($"Ring Buffer name({rb.Name}) created.");
374-
Console.WriteLine($"Ring Buffer Current capacity is : {rb.CurrentCapacity}");
375-
Console.WriteLine($"Ring Buffer name({rb.Name}) IsInitCapacity = {rb.IsInitCapacity}.");
376-
Console.WriteLine($"Ring Buffer name({rb.Name}) IsMaxCapacity = {rb.IsMaxCapacity}.");
377-
Console.WriteLine($"Ring Buffer name({rb.Name}) IsMinCapacity = {rb.IsMinCapacity}.");
378-
379282
using (var buffer = await rb.AcquireAsync(token))
380283
{
381284
if (buffer.Successful)
@@ -389,7 +292,7 @@ using (var buffer = await rb.AcquireAsync(token))
389292
}
390293
}
391294

392-
private async Task<IChannel> ModelFactory(CancellationToken cancellation)
295+
private async Task<IChannel> ChannelFactory(CancellationToken cancellation)
393296
{
394297
return await connectionRabbit!.CreateChannelAsync(cancellationToken: cancellation);
395298
}

‎samples/RingBufferPlusRabbitSample/Program.cs‎

Lines changed: 83 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -89,7 +89,7 @@ static async Task<IChannel> ChannelFactory(CancellationToken cancellation)
8989
Console.WriteLine($"Ring Buffer name({rb.Name}) IsMaxCapacity = {rb.IsMaxCapacity}.");
9090
Console.WriteLine($"Ring Buffer name({rb.Name}) IsMinCapacity = {rb.IsMinCapacity}.");
9191

92-
Console.WriteLine($"Wait... 20 sec. to start {threadCount} thread");
92+
Console.WriteLine($"Wait... 20 sec. to start {threadCount} thread using Non lock Acquire");
9393
Thread.Sleep(TimeSpan.FromSeconds(20));
9494

9595
Console.WriteLine($"Running 60 seconds..");
@@ -148,6 +148,88 @@ static async Task<IChannel> ChannelFactory(CancellationToken cancellation)
148148
}
149149
sw.Reset();
150150

151+
threads.Clear();
152+
153+
cts.Dispose();
154+
cts = CancellationTokenSource.CreateLinkedTokenSource(tokenapplifetime);
155+
156+
Console.WriteLine($"Wait... 20 sec. to start {threadCount} thread using lock Acquire");
157+
158+
rb = await RingBuffer<IChannel>.New("RabbitChanels")
159+
.Capacity(10)
160+
.Logger(hostApp.Services.GetService<ILogger<Program>>())
161+
.BackgroundLogger()
162+
.Factory((cts) => ChannelFactory(cts)!)
163+
.ScaleTimer(50, TimeSpan.FromSeconds(5))
164+
.MaxCapacity(20)
165+
.MinCapacity(5)
166+
.LockWhenScaling()
167+
.AutoScaleAcquireFault()
168+
.BuildWarmupAsync(cts.Token);
169+
170+
Console.WriteLine($"Ring Buffer name({rb.Name}) created.");
171+
Console.WriteLine($"Ring Buffer Current capacity = {rb.CurrentCapacity}");
172+
Console.WriteLine($"Ring Buffer name({rb.Name}) IsInitCapacity = {rb.IsInitCapacity}.");
173+
Console.WriteLine($"Ring Buffer name({rb.Name}) IsMaxCapacity = {rb.IsMaxCapacity}.");
174+
Console.WriteLine($"Ring Buffer name({rb.Name}) IsMinCapacity = {rb.IsMinCapacity}.");
175+
176+
Thread.Sleep(TimeSpan.FromSeconds(20));
177+
178+
Console.WriteLine($"Running 60 seconds..");
179+
Thread.Sleep(TimeSpan.FromSeconds(1));
180+
181+
dtref = DateTime.Now.AddSeconds(60);
182+
qtdstart = 0;
183+
for (int i = 0; i < threadCount; i++)
184+
{
185+
Thread thread = new(async () =>
186+
{
187+
var id = Interlocked.Increment(ref qtdstart);
188+
Console.WriteLine($"Thread {qtdstart} started ");
189+
while (true)
190+
{
191+
if (DateTime.Now >= dtref)
192+
{
193+
Console.WriteLine($"wait({id}) 60 seconds (idle)");
194+
Thread.Sleep(TimeSpan.FromSeconds(60));
195+
break;
196+
}
197+
using var bufferedItem = await rb!.AcquireAsync();
198+
if (bufferedItem.Successful)
199+
{
200+
var body = new ReadOnlyMemory<byte>(messageBodyBytes);
201+
await bufferedItem.Current!.BasicPublishAsync("", "log", body);
202+
}
203+
else
204+
{
205+
if (!cts.IsCancellationRequested)
206+
{
207+
Console.WriteLine($"RingBuffer-{id}({bufferedItem.Successful}:{bufferedItem.ElapsedTime}) Channel Capacity({rb!.CurrentCapacity})");
208+
}
209+
}
210+
}
211+
Console.WriteLine($"Thread {id} ended");
212+
Interlocked.Decrement(ref qtdstart);
213+
});
214+
thread.Start();
215+
threads.Add(thread);
216+
}
217+
218+
Console.WriteLine($"Waiting for {threadCount} threads to finish...");
219+
while (qtdstart > 0)
220+
{
221+
Thread.Sleep(10);
222+
}
223+
224+
Console.WriteLine("Dispose ring buffer");
225+
cts.Cancel();
226+
sw = Stopwatch.StartNew();
227+
while (sw.ElapsedMilliseconds < 10000)
228+
{
229+
Thread.Sleep(1000);
230+
Console.WriteLine($"Ring Buffer {rb!.Name} current capacity : {rb!.CurrentCapacity}");
231+
}
232+
sw.Reset();
151233
}
152234

153235
public static string RandomString(int length)

‎src/RingBufferPlus.Tests/RingBufferManager.cs‎

Lines changed: 3 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -83,7 +83,7 @@ public async Task SwitchToAsync_ShouldScaleToMinCapacity()
8383
// Arrange
8484
var builder = new RingBufferBuilder<int>("TestBuffer",null);
8585
builder.Capacity(5);
86-
builder.LockAcquireWhenAutoScale();
86+
builder.LockWhenScaling();
8787
builder.MaxCapacity(10);
8888
builder.MinCapacity(2);
8989
builder.Factory((_) => Task.FromResult(0));
@@ -104,7 +104,7 @@ public async Task SwitchToAsync_ShouldScaleToMaxCapacity()
104104
// Arrange
105105
var builder = new RingBufferBuilder<int>("TestBuffer", null);
106106
builder.Capacity(5);
107-
builder.LockAcquireWhenAutoScale();
107+
builder.LockWhenScaling();
108108
builder.MaxCapacity(10);
109109
builder.MinCapacity(2);
110110
builder.Factory((_) => Task.FromResult(0));
@@ -124,7 +124,7 @@ public async Task SwitchToAsync_ShouldScaleToInitCapacity()
124124
// Arrange
125125
var builder = new RingBufferBuilder<int>("TestBuffer", null);
126126
builder.Capacity(5);
127-
builder.LockAcquireWhenAutoScale();
127+
builder.LockWhenScaling();
128128
builder.MaxCapacity(10);
129129
builder.MinCapacity(2);
130130
builder.Factory((_) => Task.FromResult(0));

‎src/RingBufferPlus/Commands/IRingBufferScaleCapacity.cs‎

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -32,11 +32,11 @@ public interface IRingBufferScaleCapacity<T> : IRingBufferBuild<T>
3232
IRingBufferScaleCapacity<T> MaxCapacity(int value);
3333

3434
/// <summary>
35-
/// Sets acquisition lock when running auto scale.
35+
/// Sets acquisition/Switch lock when running scaleUp/ScaleDown.
3636
/// </summary>
3737
/// <param name="value">True to acquisition lock.Default true</param>
3838
/// <returns></returns>
39-
IRingBufferScaleCapacity<T> LockAcquireWhenAutoScale(bool value = true);
39+
IRingBufferScaleCapacity<T> LockWhenScaling(bool value = true);
4040

4141
/// <summary>
4242
/// Sets the condition to autoscale (scale up) capacity when an acquire fault occurs. The Manually change scale will always return false if autoscale is enabled.

‎src/RingBufferPlus/Core/RingBufferBuilder.cs‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -120,7 +120,7 @@ public IRingBuffer<T> BackgroundLogger(bool value = true)
120120
return this;
121121
}
122122

123-
public IRingBufferScaleCapacity<T> LockAcquireWhenAutoScale(bool value = true)
123+
public IRingBufferScaleCapacity<T> LockWhenScaling(bool value = true)
124124
{
125125
_lockAcquire = value;
126126
return this;

0 commit comments

Comments
 (0)