Skip to content

Commit bed41d2

Browse files
ReferenceTypeReferenceType
committed
Resource Recovery Update, Udp Server Optimisation
1 parent 686d0ce commit bed41d2

File tree

20 files changed

+234
-169
lines changed

20 files changed

+234
-169
lines changed

‎Benchmarks/ProtobuffBenchmark/Program.cs‎

Lines changed: 3 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -53,7 +53,7 @@ static void Main(string[] args)
5353
if (runAsClient) Benchmark();
5454

5555
ShowStatus();
56-
Environment.Exit(0);
56+
Console.ReadLine();
5757
}
5858
private static void InitializeServer()
5959
{
@@ -149,6 +149,8 @@ private static void ShowStatus()
149149
{
150150
if (runAsServer)
151151
{
152+
GC.Collect();
153+
152154
server.GetStatistics(out TcpStatistics general, out var _);
153155
Console.WriteLine("-> Server Statistics Snapshot:");
154156
Console.WriteLine(general.ToString());
@@ -177,7 +179,6 @@ private static void ShowStatus()
177179
var elapsedSeconds = (double)lastTimeStamp / 1000;
178180
var messagePerSecond = totMsgClient / elapsedSeconds;
179181

180-
181182
Console.WriteLine("-> Client Statistics Snapshot: ");
182183
Console.WriteLine(TcpStatistics.GetAverageStatistics(stats).ToString());
183184
Console.WriteLine("# Average Request-Response Per second " + (totMsgClient / elapsedSeconds).ToString("N1"));

‎Benchmarks/ProtobuffBenchmark/ProtobuffBenchmark.csproj‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -2,7 +2,7 @@
22

33
<PropertyGroup>
44
<OutputType>Exe</OutputType>
5-
<TargetFramework>net6.0</TargetFramework>
5+
<TargetFramework>net7.0</TargetFramework>
66
<Nullable>enable</Nullable>
77
</PropertyGroup>
88

‎Benchmarks/SecureProtobuffBenchmark/Program.cs‎

Lines changed: 2 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -37,8 +37,7 @@ internal class Program
3737
private static ThreadLocal<long> TotalNumMsgServer = new ThreadLocal<long>(true);
3838
static void Main(string[] args)
3939
{
40-
//TcpTest();
41-
//TcpTest2();
40+
4241
var config = ConsoleInputHandler.ObtainConfig();
4342
runAsClient = config.runAsClient;
4443
runAsServer = config.runAsServer;
@@ -52,7 +51,7 @@ static void Main(string[] args)
5251
if (runAsClient) Benchmark();
5352

5453
ShowStatus();
55-
Environment.Exit(0);
54+
Console.ReadLine();
5655
}
5756
private static void InitializeServer()
5857
{

‎Benchmarks/SecureProtobuffBenchmark/SecureProtobuffBenchmark.csproj‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -2,7 +2,7 @@
22

33
<PropertyGroup>
44
<OutputType>Exe</OutputType>
5-
<TargetFramework>net6.0</TargetFramework>
5+
<TargetFramework>net7.0</TargetFramework>
66
<Nullable>enable</Nullable>
77
</PropertyGroup>
88

‎Benchmarks/TcpBenchmark/TcpMessageBenchmark.csproj‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -2,7 +2,7 @@
22

33
<PropertyGroup>
44
<OutputType>Exe</OutputType>
5-
<TargetFramework>net6.0</TargetFramework>
5+
<TargetFramework>net7.0</TargetFramework>
66
<Nullable>enable</Nullable>
77
<Platforms>AnyCPU;ARM32</Platforms>
88
</PropertyGroup>

‎NetworkLibrary/Components/BufferPool.cs‎

Lines changed: 59 additions & 80 deletions
Original file line numberDiff line numberDiff line change
@@ -2,6 +2,7 @@
22
using System;
33
using System.Collections.Concurrent;
44
using System.Collections.Generic;
5+
using System.Diagnostics;
56
using System.Drawing;
67
using System.Linq;
78
using System.Runtime.CompilerServices;
@@ -11,15 +12,23 @@
1112
namespace NetworkLibrary
1213
{
1314
/*
14-
* This behaves like ArrayPool<>. The butckets are TLS thx to concurrent bag. ( ThreadLocal<ThreadLocalList> m_locals )
15+
* Concurrent bag has Tls list ( ThreadLocal<ThreadLocalList> m_locals )
16+
* each bucket holds a set of weak references to byte arrays
17+
* this arrays are pooled and resuable and we preserve the peak memory usage by this.
18+
* If application calls the GC gen2 collect some of this weak references are cleared,
19+
* this way we trim the pools automatically if they are not referenced by the application.
20+
*
21+
* you can also configure the pool to auto GC collect(also does gen2) if the application is mostly idle and
22+
* we reached to some threshold on workingset memory.
1523
*/
1624
public class BufferPool
1725
{
1826
public static bool ForceGCOnCleanup = true;
27+
public static int MaxMemoryBeforeForceGc = 100000000;
1928
public const int MaxBufferSize = 1073741824;
2029
public const int MinBufferSize = 256;
21-
22-
private static readonly ConcurrentBag<byte[]>[] bufferBuckets = new ConcurrentBag<byte[]>[32];
30+
private static readonly ConcurrentBag<WeakReference<byte[]>> weakReferencePool= new ConcurrentBag<WeakReference<byte[]>>();
31+
private static readonly ConcurrentBag<WeakReference<byte[]>>[] bufferBuckets = new ConcurrentBag<WeakReference<byte[]>>[32];
2332
private static SortedDictionary<int, int> bucketCapacityLimits = new SortedDictionary<int, int>()
2433
{
2534
{ 256,10000 },
@@ -47,6 +56,8 @@ public class BufferPool
4756
{ 1073741824,0 }
4857

4958
};
59+
static readonly Process process = Process.GetCurrentProcess();
60+
static ManualResetEvent autoGcHandle = new ManualResetEvent(false);
5061

5162
static BufferPool()
5263
{
@@ -56,59 +67,48 @@ static BufferPool()
5667
thread.Start();
5768
}
5869

70+
/// <summary>
71+
/// Starts a task where GC.Collect() is called
72+
/// if application consumed less than %1 proccessor time and memory is above threashold
73+
/// </summary>
74+
public static void StartCollectGcOnIdle()
75+
{
76+
autoGcHandle.Set();
77+
}
78+
79+
/// <summary>
80+
/// Stops a task where GC.Collect() is called
81+
/// if application consumed less than %1 proccessor time and memory is above threashold
82+
/// </summary>
83+
public static void StopCollectGcOnIdle()
84+
{
85+
autoGcHandle.Reset();
86+
}
87+
5988
// creates bufferBuckets structure
6089
private static void Init()
6190
{
6291
//bufferBuckets = new ConcurrentDictionary<int, ConcurrentBag<byte[]>>();
6392
for (int i = 8; i < 31; i++)
6493
{
65-
bufferBuckets[i] = new ConcurrentBag<byte[]>();
94+
bufferBuckets[i] = new ConcurrentBag<WeakReference<byte[]>>();
6695
}
6796
}
6897

6998
private static void MaintainMemory()
7099
{
71-
// Check each bucket periodically if the free capacity limit is exceeded
72-
// Dump excess amount
100+
var lastTime = process.TotalProcessorTime;
73101
while (true)
74102
{
103+
autoGcHandle.WaitOne();
75104
Thread.Sleep(10000);
76-
try
77-
{
78-
for (int k = 0; k < bufferBuckets.Length; k++)
79-
{
80-
var size = GetBucketSize(k);
81-
var bag = bufferBuckets[k];
82-
if (bag == null) continue;
83-
84-
if (bucketCapacityLimits[size] < bag.Count)
85-
{
86-
// check 5 times slowly to make sure buffer bucket is not hot.
87-
for (int i = 0; i < 5; i++)
88-
{
89-
Thread.Sleep(100);
90-
if (bucketCapacityLimits[size] >= bag.Count) continue;
91-
}
92-
93-
// trim slowly
94-
while (bag.Count > bucketCapacityLimits[size])
95-
{
96-
bag.TryTake(out var buffer);
97-
buffer = null;
98-
Thread.Sleep(1);
99-
100-
}
101-
Thread.Sleep(100);
102-
if (ForceGCOnCleanup)
103-
GC.Collect();
104-
}
105-
}
106-
}
107-
catch(Exception e)
108-
{
109-
MiniLogger.Log(MiniLogger.LogLevel.Error,"Buffer manager encountered an error: " + e.Message);
110-
}
105+
var currentProcTime = process.TotalProcessorTime;
106+
var deltaT = (lastTime - currentProcTime).TotalMilliseconds;
107+
lastTime = currentProcTime;
111108

109+
if (deltaT <100 && process.WorkingSet64< MaxMemoryBeforeForceGc)
110+
GC.Collect();
111+
process.Refresh();
112112
}
113113

114114
}
@@ -123,18 +123,25 @@ private static void MaintainMemory()
123123
[MethodImpl(MethodImplOptions.AggressiveInlining)]
124124
public static byte[] RentBuffer(int size)
125125
{
126+
byte[] buffer;
126127
if(MaxBufferSize < size)
127128
throw new InvalidOperationException(
128129
string.Format("Unable to rent buffer bigger than max buffer size: {0}",MaxBufferSize));
129130
if (size <= MinBufferSize) return new byte[size];
130131

131132
int idx = GetBucketIndex(size);
132-
if (!bufferBuckets[idx].TryTake(out byte[] buffer))
133+
134+
while(bufferBuckets[idx].TryTake(out WeakReference<byte[]> bufferRef))
133135
{
134-
buffer = new byte[GetBucketSize(idx)];
136+
if (bufferRef.TryGetTarget(out buffer))
137+
{
138+
weakReferencePool.Add(bufferRef);
139+
return buffer;
140+
}
135141
}
142+
buffer = new byte[GetBucketSize(idx)];
143+
return buffer;
136144

137-
return buffer;
138145
}
139146

140147
/// <summary>
@@ -147,45 +154,17 @@ public static void ReturnBuffer(byte[] buffer)
147154
if (buffer.Length <= MinBufferSize) return;
148155

149156
int idx = GetBucketIndex(buffer.Length);
150-
bufferBuckets[idx-1].Add(buffer);
151-
}
152-
153-
/// <summary>
154-
/// Sets Bucket size
155-
/// </summary>
156-
/// <param name="bucketNumber"></param>
157-
/// <param name="amount"></param>
158-
/// <exception cref="InvalidOperationException"></exception>
159-
public static void SetBucketLimit(int bucketNumber, int amount)
160-
{
161-
if (bucketNumber >= 31 || bucketNumber < 8)
162-
throw new InvalidOperationException("Bucket number needs to be between 8 and 31 (inclusive)");
163-
164-
int size = GetBucketSize(bucketNumber);
165-
bucketCapacityLimits[size] = amount;
166-
}
167-
168-
/// <summary>
169-
/// Sets Bucket size
170-
/// </summary>
171-
/// <param name="bucketSize"></param>
172-
/// <param name="amount"></param>
173-
/// <exception cref="InvalidOperationException"></exception>
174-
public static void SetBucketLimitBySize(int bucketSize, int amount)
175-
{
176-
if (bucketSize > 1073741824 || bucketSize < 256)
177-
throw new InvalidOperationException("Bucket number needs to be between 8 and 31 (inclusive)");
178-
179-
if (bucketCapacityLimits.ContainsKey(bucketSize))
180-
bucketCapacityLimits[bucketSize] = amount;
157+
if(weakReferencePool.TryTake(out var wr))
158+
{
159+
wr.SetTarget(buffer);
160+
bufferBuckets[idx - 1].Add(wr);
181161

182-
int idx = GetBucketIndex(bucketSize);
183-
int size = GetBucketSize(idx);
184-
bucketCapacityLimits[size] = amount;
162+
}
163+
else
164+
bufferBuckets[idx-1].Add(new WeakReference<byte[]>(buffer));
165+
buffer = null;
185166
}
186167

187-
188-
189168
[MethodImpl(MethodImplOptions.AggressiveInlining)]
190169
private static int GetBucketSize(int bucketIndex)
191170
{

‎NetworkLibrary/Components/ByteMessageReader.cs‎

Lines changed: 5 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -235,6 +235,11 @@ private void FreeMemory()
235235
BufferPool.ReturnBuffer(internalBufer);
236236
internalBufer = BufferPool.RentBuffer(originalCapacity);
237237
}
238+
239+
public void ReleaseResources()
240+
{
241+
if(internalBufer!= null) { BufferPool.ReturnBuffer(internalBufer); internalBufer = null; }
242+
}
238243
#endregion
239244
}
240245
}

‎NetworkLibrary/Components/MessageBuffer/MessageBuffer.cs‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -28,7 +28,7 @@ public MessageBuffer(int maxIndexedMemory, bool writeLengthPrefix = true)
2828

2929
public bool IsEmpty()
3030
{
31-
return disposedValue || writeStream.Position == 0;
31+
return Volatile.Read(ref disposedValue) || writeStream.Position == 0;
3232
}
3333

3434
public bool TryEnqueueMessage(byte[] bytes)

‎NetworkLibrary/NetworkLibrary.csproj‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -13,7 +13,7 @@
1313
<Authors>RefrenceType</Authors>
1414
<PackageLicenseFile>Licence.txt</PackageLicenseFile>
1515
<Copyright>Apache 2.0</Copyright>
16-
<Version>1.0.2</Version>
16+
<Version>1.0.3</Version>
1717
</PropertyGroup>
1818

1919
<PropertyGroup Condition="'$(Configuration)|$(Platform)'=='Debug|AnyCPU'">

0 commit comments

Comments
 (0)