Fix UDP memory leak #5404

Arkatufus · 2021-12-01T17:38:57Z

Fix UDP memory leak problem because the rented byte buffer memory never got released when SocketAsyncEventArgs got lost.

Closes #5325

Moved ByteBuffer from Udp and UdpConnected into PreallocatedSocketEventAgrsPool. The socket event pool is resposible for renting and releasing buffers now, so we only need to care about renting and releasing socket events.
SocketAsyncEventArgs never leaves Udp, UdpListener, UdpConnect, and UdpConnection class. Required data are copied into the response messages instead of embedding SocketAsyncEventArgs inside the Received message. SocketAsyncEventArgs will either be released immediately if Socket.ReceiveAsync returned false or released when the OnComplete event fires.
Removed SocketAsyncEventArgs pooling functionality, caching SocketAsyncEventArgs is causing a lot more grief than its worth, we would have to catch all the edge cases where it can fail. SocketAsyncEventArgs is not a simple struct that can be cleaned and reused, it actually have internal states that can cause a lot of problems when reused.

src/core/Akka/IO/Buffers/DirectBufferPool.cs

Arkatufus · 2021-12-01T18:24:21Z

src/core/Akka/IO/Buffers/DirectBufferPool.cs

-            if (buf.Count == _bufferSize && _segments.Contains(buf.Array))
-                _buffers.Push(buf);
+            if (buf.Count != _bufferSize || !_segments.Contains(buf.Array))
+                throw new Exception("Wrong ArraySegment<byte> was released to DirectBufferPool");


Sanity check to make sure that any ArraySegment<byte> returned to the buffer actually belongs to the buffer, should only happen during development.

Why would it only happen during development?

Arkatufus · 2021-12-01T18:25:48Z

src/core/Akka/IO/SocketEventArgsPool.cs

    }

    internal class PreallocatedSocketEventAgrsPool : ISocketEventArgsPool
    {
+        private readonly IBufferPool _bufferPool;


Buffer pool belongs to the socket event pool now, no one should touch this. ever.

Add a comment saying this and explaining why

Arkatufus · 2021-12-01T18:29:09Z

src/core/Akka/IO/SocketEventArgsPool.cs

        private readonly EventHandler<SocketAsyncEventArgs> _onComplete;
-        private readonly ConcurrentStack<SocketAsyncEventArgs> _pool = new ConcurrentStack<SocketAsyncEventArgs>();
+        private readonly ConcurrentQueue<SocketAsyncEventArgs> _pool = new ConcurrentQueue<SocketAsyncEventArgs>();


Changed to make sure that the content of the pool are rotated evenly, instead of only using the top ones. Minimizes the problem where SocketAsyncEventArgs could not be reused because it is locked by the previous socket.
To be honest, this whole class is a bad idea because SocketAsyncEventArgs are only supposed to be recycled if they're used by a single socket.

Arkatufus · 2021-12-01T18:30:49Z

src/core/Akka/IO/SocketEventArgsPool.cs

            }
        }

        public SocketAsyncEventArgs Acquire(IActorRef actor)
        {
-            if (!_pool.TryPop(out var e))
-                e = CreateSocketAsyncEventArgs();
+            var buffer = _bufferPool.Rent();


Byte buffer pool rent can only happen here and no where else.

Arkatufus · 2021-12-01T18:32:06Z

src/core/Akka/IO/SocketEventArgsPool.cs

+                }
+                catch (InvalidOperationException)
+                {
+                    // it can be that for some reason socket is in use and haven't closed yet. Dispose anyway to avoid leaks.


This will always be the case if this pool are being used by multiple sockets...

Arkatufus · 2021-12-01T18:33:11Z

src/core/Akka/IO/SocketEventArgsPool.cs


-            e.UserToken = actor;
+                    e.SetBuffer(buffer.Array, buffer.Offset, buffer.Count);


This is fine because we only rent once and if we fail to acquire an instance, we immediately dispose it.

Arkatufus · 2021-12-01T18:35:42Z

src/core/Akka/IO/Udp.cs

+
+            protected SocketCompleted(SocketAsyncEventArgs eventArgs)
+            {
+                Data = ByteString.CopyFrom(eventArgs.Buffer, eventArgs.Offset, eventArgs.BytesTransferred);


Instead of embedding the SocketAsyncEventArgs, we extract the data and store that instead.

And we make a copy of the data, so there's no problems if the SocketAsyncEventArgs get processed and the buffer pool is released.

Arkatufus · 2021-12-01T18:40:59Z

src/core/Akka/IO/Udp.cs

        {
            switch (e.LastOperation)
            {
                case SocketAsyncOperation.Receive:
                case SocketAsyncOperation.ReceiveFrom:
                    return new Udp.SocketReceived(e);
-                case SocketAsyncOperation.Send:


These are never used. Removed to remove red herrings and possible problem in the future.

src/core/Akka/IO/UdpConnected.cs

src/core/Akka/IO/UdpConnectedManager.cs

Arkatufus · 2021-12-01T18:52:23Z

src/core/Akka/IO/UdpConnection.cs

                default: return false;
            }
        }

        private void DoRead(SocketReceived received, IActorRef handler)
        {
-            var e = received.EventArgs;
-            var buffer = new ByteBuffer(e.Buffer, e.Offset, e.BytesTransferred);


This should never happen anymore, all handled in the SocketAsyncEventArgs pool

Arkatufus · 2021-12-01T18:52:50Z

src/core/Akka/IO/UdpConnection.cs

            var e = Udp.SocketEventArgsPool.Acquire(Self);
-            var buffer = Udp.BufferPool.Rent();


Same here, we don't have to worry about this anymore.

Aaronontheweb · 2021-12-01T19:24:26Z

@Arkatufus

Have a number of UDP test failures
API approval failures

…t into IO_Fix_UDP_memory_leak

Arkatufus · 2021-12-01T20:48:04Z

src/core/Akka/IO/SocketEventArgsPool.cs

            _onComplete = onComplete;
-            for (var i = 0; i < initSize; i++)


Removed pool functionality

Aaronontheweb

Left some questions, nitpicks, and minor suggestions but this looks like a great structural improvement for Akka.IO.Udp. Nice work.

Aaronontheweb · 2021-12-01T21:09:27Z

src/core/Akka.Tests/IO/UdpConnectedIntegrationSpec.cs

+            poolInfo = udpConnection.SocketEventArgsPool.BufferPoolInfo;
+            poolInfo.Type.Should().Be(typeof(DirectBufferPool));
+            poolInfo.Free.Should().Be(poolInfo.TotalSize);
+            poolInfo.Used.Should().Be(0);


Aaronontheweb · 2021-12-01T21:13:13Z

src/core/Akka.Tests/IO/UdpIntegrationSpec.cs

+            poolInfo = udp.SocketEventArgsPool.BufferPoolInfo;
+            poolInfo.Type.Should().Be(typeof(DirectBufferPool));
+            poolInfo.Free.Should().Be(poolInfo.TotalSize);
+            poolInfo.Used.Should().Be(0);


Aaronontheweb · 2021-12-01T21:13:48Z

src/core/Akka/IO/Buffers/DirectBufferPool.cs

@@ -131,6 +149,17 @@ public DirectBufferPool(int bufferSize, int buffersPerSegment, int initialSegmen
            }
        }

+        public BufferPoolInfo Diagnostics()


Aaronontheweb · 2021-12-01T21:14:34Z

src/core/Akka/IO/Buffers/DirectBufferPool.cs

-            if (buf.Count == _bufferSize && _segments.Contains(buf.Array))
-                _buffers.Push(buf);
+            if (buf.Count != _bufferSize || !_segments.Contains(buf.Array))
+                throw new Exception("Wrong ArraySegment<byte> was released to DirectBufferPool");


Why would it only happen during development?

Aaronontheweb · 2021-12-01T21:15:12Z

src/core/Akka/IO/SocketEventArgsPool.cs

    }

    internal class PreallocatedSocketEventAgrsPool : ISocketEventArgsPool
    {
+        private readonly IBufferPool _bufferPool;


Add a comment saying this and explaining why

Aaronontheweb · 2021-12-01T21:17:10Z

src/core/Akka/IO/Udp.cs

+
+            protected SocketCompleted(SocketAsyncEventArgs eventArgs)
+            {
+                Data = ByteString.CopyFrom(eventArgs.Buffer, eventArgs.Offset, eventArgs.BytesTransferred);


And we make a copy of the data, so there's no problems if the SocketAsyncEventArgs get processed and the buffer pool is released.

src/core/Akka/IO/UdpConnected.cs

Arkatufus · 2021-12-01T21:29:04Z

I mean its there so we can catch it while we're developing Akka, I think its better if I change that to an error log instead.

Aaronontheweb · 2021-12-01T21:52:21Z

You can use 'Debug.Assert' for that

…use Assert.Debug instead of throwing.

…t into IO_Fix_UDP_memory_leak

Arkatufus · 2021-12-01T23:43:50Z

Done with the changes

Arkatufus added 2 commits December 2, 2021 00:34

Fix UDP memory leak

27ccaed

Revert changes that are not connected to UDP

c459e51

Aaronontheweb self-requested a review December 1, 2021 18:04

Aaronontheweb added the akka-io label Dec 1, 2021

Code comment cleanup

9c72778

Arkatufus commented Dec 1, 2021

View reviewed changes

src/core/Akka/IO/Buffers/DirectBufferPool.cs Show resolved Hide resolved

Arkatufus commented Dec 1, 2021

View reviewed changes

src/core/Akka/IO/UdpConnected.cs Show resolved Hide resolved

Arkatufus commented Dec 1, 2021

View reviewed changes

src/core/Akka/IO/UdpConnectedManager.cs Show resolved Hide resolved

Arkatufus commented Dec 1, 2021

View reviewed changes

Merge branch 'dev' into IO_Fix_UDP_memory_leak

b99cca2

Arkatufus added 5 commits December 2, 2021 02:45

Update API Approval list

74862f7

Merge branch 'IO_Fix_UDP_memory_leak' of github.com:Arkatufus/akka.ne…

bb2ec2d

…t into IO_Fix_UDP_memory_leak

Remove SocketEventArgsPool functionality

8768a78

Fix specs

5a4251e

Merge branch 'dev' into IO_Fix_UDP_memory_leak

80c8d00

Arkatufus commented Dec 1, 2021

View reviewed changes

Aaronontheweb approved these changes Dec 1, 2021

View reviewed changes

Merge branch 'dev' into IO_Fix_UDP_memory_leak

94ef7a7

Aaronontheweb enabled auto-merge (squash) December 1, 2021 23:01

Arkatufus added 2 commits December 2, 2021 06:41

Add documentation in comments to describe the architectural changes, …

541b776

…use Assert.Debug instead of throwing.

Merge branch 'IO_Fix_UDP_memory_leak' of github.com:Arkatufus/akka.ne…

838faa0

…t into IO_Fix_UDP_memory_leak

Arkatufus disabled auto-merge December 2, 2021 14:38

Arkatufus added 2 commits December 2, 2021 21:39

Remove the logger, we're using Debug.Assert instead

926b220

Clean up assert code

5faf777

Arkatufus mentioned this pull request Dec 2, 2021

UDP Connected: Unhandled message SocketReceived #5318

Closed

Aaronontheweb enabled auto-merge (squash) December 2, 2021 14:46

Aaronontheweb disabled auto-merge December 2, 2021 16:23

Aaronontheweb merged commit a920b07 into akkadotnet:dev Dec 2, 2021

Aaronontheweb mentioned this pull request Dec 2, 2021

Add SimpleBufferPool implementation #5326

Closed

Aaronontheweb mentioned this pull request Dec 13, 2021

v1.4.29 Release #5426

Merged

Arkatufus deleted the IO_Fix_UDP_memory_leak branch December 13, 2021 21:10

Arkatufus restored the IO_Fix_UDP_memory_leak branch April 22, 2022 15:02

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Fix UDP memory leak #5404

Fix UDP memory leak #5404

Arkatufus commented Dec 1, 2021 •

edited

Loading

Arkatufus Dec 1, 2021

Aaronontheweb Dec 1, 2021

Arkatufus Dec 1, 2021

Aaronontheweb Dec 1, 2021

Arkatufus Dec 1, 2021

Arkatufus Dec 1, 2021

Arkatufus Dec 1, 2021

Arkatufus Dec 1, 2021

Arkatufus Dec 1, 2021

Aaronontheweb Dec 1, 2021

Arkatufus Dec 1, 2021

Arkatufus Dec 1, 2021

Arkatufus Dec 1, 2021

Aaronontheweb commented Dec 1, 2021

Arkatufus Dec 1, 2021

Aaronontheweb left a comment

Aaronontheweb Dec 1, 2021

Aaronontheweb Dec 1, 2021

Aaronontheweb Dec 1, 2021

Aaronontheweb Dec 1, 2021

Aaronontheweb Dec 1, 2021

Aaronontheweb Dec 1, 2021

Arkatufus commented Dec 1, 2021

Aaronontheweb commented Dec 1, 2021

Arkatufus commented Dec 1, 2021


		e.UserToken = actor;
		e.SetBuffer(buffer.Array, buffer.Offset, buffer.Count);

		var e = Udp.SocketEventArgsPool.Acquire(Self);
		var buffer = Udp.BufferPool.Rent();

Fix UDP memory leak #5404

Fix UDP memory leak #5404

Conversation

Arkatufus commented Dec 1, 2021 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Aaronontheweb commented Dec 1, 2021

Choose a reason for hiding this comment

Aaronontheweb left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Arkatufus commented Dec 1, 2021

Aaronontheweb commented Dec 1, 2021

Arkatufus commented Dec 1, 2021

Arkatufus commented Dec 1, 2021 •

edited

Loading