From e88d3ab5e3cad861e6e8cd1bf078f214f09c73f0 Mon Sep 17 00:00:00 2001 From: Eduard Gilmutdinov Date: Thu, 10 Aug 2017 11:45:19 -0700 Subject: [PATCH 1/5] Fix memory leak --- .../Opc.Ua.Core/Stack/Tcp/TcpMessageSocket.cs | 56 +++++++++---------- 1 file changed, 26 insertions(+), 30 deletions(-) diff --git a/Stack/Opc.Ua.Core/Stack/Tcp/TcpMessageSocket.cs b/Stack/Opc.Ua.Core/Stack/Tcp/TcpMessageSocket.cs index 613141c47a..7ee6ea4e6c 100644 --- a/Stack/Opc.Ua.Core/Stack/Tcp/TcpMessageSocket.cs +++ b/Stack/Opc.Ua.Core/Stack/Tcp/TcpMessageSocket.cs @@ -282,8 +282,8 @@ namespace Opc.Ua.Bindings SocketError error = SocketError.NotInitialized; TaskCompletionSource tcs = new TaskCompletionSource(); - TcpMessageSocketAsyncEventArgs argsV4 = null; - TcpMessageSocketAsyncEventArgs argsV6 = null; + SocketAsyncEventArgs argsV4 = null; + SocketAsyncEventArgs argsV6 = null; m_socketResponses = 0; lock (m_socketLock) @@ -299,12 +299,14 @@ namespace Opc.Ua.Bindings // create sockets if IP address was provided if (addressV6 != null) { - argsV6 = new TcpMessageSocketAsyncEventArgs(); - argsV6.UserToken = state; - m_socketV6 = new Socket(AddressFamily.InterNetworkV6, SocketType.Stream, ProtocolType.Tcp); - argsV6.m_args.RemoteEndPoint = new IPEndPoint(addressV6, port); m_socketResponses++; - argsV6.m_args.Completed += (o, e) => + m_socketV6 = new Socket(AddressFamily.InterNetworkV6, SocketType.Stream, ProtocolType.Tcp); + argsV6 = new SocketAsyncEventArgs() + { + UserToken = state, + RemoteEndPoint = new IPEndPoint(addressV6, port), + }; + argsV6.Completed += (o, e) => { lock (m_socketLock) { @@ -316,30 +318,28 @@ namespace Opc.Ua.Bindings { m_socket = m_socketV6; tcs.SetResult(e.SocketError); + callback(o, new TcpMessageSocketAsyncEventArgs { UserToken = e.UserToken }); } else { m_socketV6.Dispose(); - e.UserToken = null; } m_socketV6 = null; } - else - { - e.UserToken = null; - } } + e.Dispose(); }; - argsV6.Completed += callback; } if (addressV4 != null) { - argsV4 = new TcpMessageSocketAsyncEventArgs(); - argsV4.UserToken = state; - m_socketV4 = new Socket(AddressFamily.InterNetwork, SocketType.Stream, ProtocolType.Tcp); - argsV4.m_args.RemoteEndPoint = new IPEndPoint(addressV4, port); m_socketResponses++; - argsV4.m_args.Completed += (o, e) => + m_socketV4 = new Socket(AddressFamily.InterNetwork, SocketType.Stream, ProtocolType.Tcp); + argsV4 = new SocketAsyncEventArgs() + { + UserToken = state, + RemoteEndPoint = new IPEndPoint(addressV4, port), + }; + argsV4.Completed += (o, e) => { lock (m_socketLock) { @@ -351,43 +351,39 @@ namespace Opc.Ua.Bindings { m_socket = m_socketV4; tcs.SetResult(e.SocketError); + callback(o, new TcpMessageSocketAsyncEventArgs { UserToken = e.UserToken }); } else { m_socketV4.Dispose(); - e.UserToken = null; } m_socketV4 = null; } - else - { - e.UserToken = null; - } } + e.Dispose(); }; - argsV4.Completed += callback; } bool connectV6Sync = true; bool connectV4Sync = true; if (m_socketV6 != null) { - connectV6Sync = !m_socketV6.ConnectAsync(argsV6.m_args); + connectV6Sync = !m_socketV6.ConnectAsync(argsV6); if (connectV6Sync) { // I/O completed synchronously - callback(this, argsV6); - error = argsV6.m_args.SocketError; + callback(m_socketV6, new TcpMessageSocketAsyncEventArgs { UserToken = state }); + error = argsV6.SocketError; } } if (m_socketV4 != null && error != SocketError.Success) { - connectV4Sync = !m_socketV4.ConnectAsync(argsV4.m_args); + connectV4Sync = !m_socketV4.ConnectAsync(argsV4); if (connectV4Sync) { // I/O completed synchronously - callback(this, argsV4); - error = argsV4.m_args.SocketError; + callback(m_socketV4, new TcpMessageSocketAsyncEventArgs { UserToken = state }); + error = argsV4.SocketError; } } From 6df6828828ca97f5cd4e91aa6ad3106668db3c88 Mon Sep 17 00:00:00 2001 From: Eduard Gilmutdinov Date: Thu, 10 Aug 2017 14:15:28 -0700 Subject: [PATCH 2/5] Refactoring in TcpMessageSocket --- .../Opc.Ua.Core/Stack/Tcp/TcpMessageSocket.cs | 252 ++++++------------ .../Stack/Transport/IMessageSocket.cs | 2 +- 2 files changed, 87 insertions(+), 167 deletions(-) diff --git a/Stack/Opc.Ua.Core/Stack/Tcp/TcpMessageSocket.cs b/Stack/Opc.Ua.Core/Stack/Tcp/TcpMessageSocket.cs index 7ee6ea4e6c..f464ae7d7d 100644 --- a/Stack/Opc.Ua.Core/Stack/Tcp/TcpMessageSocket.cs +++ b/Stack/Opc.Ua.Core/Stack/Tcp/TcpMessageSocket.cs @@ -11,6 +11,7 @@ */ using System; +using System.Linq; using System.Net; using System.Net.Sockets; using System.Threading.Tasks; @@ -171,8 +172,6 @@ namespace Opc.Ua.Bindings m_sink = sink; m_socket = null; - m_socketV4 = null; - m_socketV6 = null; m_bufferManager = bufferManager; m_receiveBufferSize = receiveBufferSize; m_incomingMessageSize = -1; @@ -242,7 +241,7 @@ namespace Opc.Ua.Bindings /// /// Connects to an endpoint. /// - public async Task BeginConnect(Uri endpointUrl, EventHandler callback, object state) + public async Task BeginConnect(Uri endpointUrl, EventHandler callback, object state) { if (endpointUrl == null) throw new ArgumentNullException("endpointUrl"); @@ -251,151 +250,92 @@ namespace Opc.Ua.Bindings throw new InvalidOperationException("The socket is already connected."); } - // Get DNS host information. + // Get DNS host information IPAddress[] hostAdresses = await Dns.GetHostAddressesAsync(endpointUrl.DnsSafeHost); - // try IPv4 and IPv6 address - IPAddress addressV4 = null; - IPAddress addressV6 = null; + // Get IPv4 and IPv6 address + IPAddress addressV4 = hostAdresses.FirstOrDefault(a => a.AddressFamily == AddressFamily.InterNetwork); + IPAddress addressV6 = hostAdresses.FirstOrDefault(a => a.AddressFamily == AddressFamily.InterNetworkV6); - foreach (IPAddress address in hostAdresses) + // Get port + int port = endpointUrl.Port; + if (port <= 0 || port > UInt16.MaxValue) { - if (address.AddressFamily == AddressFamily.InterNetworkV6) - { - if (addressV6 == null) - { - addressV6 = address; - } - } - if (address.AddressFamily == AddressFamily.InterNetwork) - { - if (addressV4 == null) - { - addressV4 = address; - } - } - if ((addressV4 != null) && (addressV6 != null)) - { - break; - } + port = Utils.UaTcpDefaultPort; } - SocketError error = SocketError.NotInitialized; - TaskCompletionSource tcs = new TaskCompletionSource(); - SocketAsyncEventArgs argsV4 = null; - SocketAsyncEventArgs argsV6 = null; - m_socketResponses = 0; + Action doCallback = () => callback(this, new TcpMessageSocketAsyncEventArgs { UserToken = state }); lock (m_socketLock) { - // ensure a valid port. - int port = endpointUrl.Port; - - if (port <= 0 || port > UInt16.MaxValue) - { - port = Utils.UaTcpDefaultPort; - } - - // create sockets if IP address was provided if (addressV6 != null) { - m_socketResponses++; - m_socketV6 = new Socket(AddressFamily.InterNetworkV6, SocketType.Stream, ProtocolType.Tcp); - argsV6 = new SocketAsyncEventArgs() - { - UserToken = state, - RemoteEndPoint = new IPEndPoint(addressV6, port), - }; - argsV6.Completed += (o, e) => - { - lock (m_socketLock) - { - m_socketResponses--; - if (m_socketV6 != null) - { - if (m_socket == null && - (m_socketResponses == 0 || e.SocketError == SocketError.Success)) - { - m_socket = m_socketV6; - tcs.SetResult(e.SocketError); - callback(o, new TcpMessageSocketAsyncEventArgs { UserToken = e.UserToken }); - } - else - { - m_socketV6.Dispose(); - } - m_socketV6 = null; - } - } - e.Dispose(); - }; - } - if (addressV4 != null) - { - m_socketResponses++; - m_socketV4 = new Socket(AddressFamily.InterNetwork, SocketType.Stream, ProtocolType.Tcp); - argsV4 = new SocketAsyncEventArgs() - { - UserToken = state, - RemoteEndPoint = new IPEndPoint(addressV4, port), - }; - argsV4.Completed += (o, e) => - { - lock (m_socketLock) - { - m_socketResponses--; - if (m_socketV4 != null) - { - if (m_socket == null && - (m_socketResponses == 0 || e.SocketError == SocketError.Success)) - { - m_socket = m_socketV4; - tcs.SetResult(e.SocketError); - callback(o, new TcpMessageSocketAsyncEventArgs { UserToken = e.UserToken }); - } - else - { - m_socketV4.Dispose(); - } - m_socketV4 = null; - } - } - e.Dispose(); - }; + BeginConnect(addressV6, AddressFamily.InterNetworkV6, port, doCallback); } - bool connectV6Sync = true; - bool connectV4Sync = true; - if (m_socketV6 != null) + if (addressV4 != null && m_socket == null) { - connectV6Sync = !m_socketV6.ConnectAsync(argsV6); - if (connectV6Sync) - { - // I/O completed synchronously - callback(m_socketV6, new TcpMessageSocketAsyncEventArgs { UserToken = state }); - error = argsV6.SocketError; - } - } - if (m_socketV4 != null && error != SocketError.Success) - { - connectV4Sync = !m_socketV4.ConnectAsync(argsV4); - if (connectV4Sync) - { - // I/O completed synchronously - callback(m_socketV4, new TcpMessageSocketAsyncEventArgs { UserToken = state }); - error = argsV4.SocketError; - } - } - - if (connectV4Sync && connectV6Sync) - { - return (error == SocketError.Success) ? true : false; + BeginConnect(addressV4, AddressFamily.InterNetwork, port, doCallback); } } + } - error = await tcs.Task; + /// + /// Try to connect to endpoint and do callback if connected successfully + /// + /// Endpoint address + /// Endpoint address family + /// Endpoint port + /// Callback that must be executed if the connection would be established + private void BeginConnect(IPAddress address, AddressFamily addressFamily, int port, Action callback) + { + var socket = new Socket(addressFamily, SocketType.Stream, ProtocolType.Tcp); + var args = new SocketAsyncEventArgs() + { + UserToken = callback, + RemoteEndPoint = new IPEndPoint(address, port), + }; + args.Completed += new EventHandler(OnSocketConnected); - return (error == SocketError.Success) ? true : false; + var result = !socket.ConnectAsync(args); + if (result) + { + // I/O completed synchronously + OnSocketConnected(socket, args); + } + } + + /// + /// Handle socket connection event + /// + /// + /// + private void OnSocketConnected(object sender, SocketAsyncEventArgs args) + { + lock (m_socketLock) + { + var socket = sender as Socket; + if (!m_closed && m_socket == null && args.SocketError == SocketError.Success) + { + m_socket = socket; + ((Action)args.UserToken)(); + } + else + { + try + { + if (socket.Connected) + { + socket.Shutdown(SocketShutdown.Both); + } + } + finally + { + socket.Dispose(); + } + } + } + args.Dispose(); } /// @@ -403,48 +343,27 @@ namespace Opc.Ua.Bindings /// public void Close() { - // get the socket. - Socket socket = null; - Socket socketV4 = null; - Socket socketV6 = null; - lock (m_socketLock) { - socket = m_socket; - socketV4 = m_socketV4; - socketV6 = m_socketV6; - m_socket = null; - m_socketV6 = null; - m_socketV4 = null; - } + m_closed = true; - // shutdown sockets which may still be active - // due to a timeout during ConnectAsync - if (socketV4 != null) - { - socketV4.Dispose(); - } - - if (socketV6 != null) - { - socketV6.Dispose(); - } - - // shutdown the socket. - if (socket != null) - { - try + // Shutdown the socket. + if (m_socket != null) { - if (socket.Connected) + try { - socket.Shutdown(SocketShutdown.Both); - } + if (m_socket.Connected) + { + m_socket.Shutdown(SocketShutdown.Both); + } - socket.Dispose(); - } - catch (Exception e) - { - Utils.Trace(e, "Unexpected error closing socket."); + m_socket.Dispose(); + m_socket = null; + } + catch (Exception e) + { + Utils.Trace(e, "Unexpected error closing socket."); + } } } } @@ -718,6 +637,7 @@ namespace Opc.Ua.Bindings private object m_socketLock = new object(); private Socket m_socket; + private bool m_closed = false; private object m_readLock = new object(); private byte[] m_receiveBuffer; diff --git a/Stack/Opc.Ua.Core/Stack/Transport/IMessageSocket.cs b/Stack/Opc.Ua.Core/Stack/Transport/IMessageSocket.cs index 4de9131bff..829c3d61e4 100644 --- a/Stack/Opc.Ua.Core/Stack/Transport/IMessageSocket.cs +++ b/Stack/Opc.Ua.Core/Stack/Transport/IMessageSocket.cs @@ -137,7 +137,7 @@ namespace Opc.Ua.Bindings /// /// Connects to an endpoint. /// - Task BeginConnect(Uri endpointUrl, EventHandler callback, object state); + Task BeginConnect(Uri endpointUrl, EventHandler callback, object state); /// /// Forcefully closes the socket. From da2d9cf2c1a4af59b7d962987aa643614871101e Mon Sep 17 00:00:00 2001 From: Eduard Gilmutdinov Date: Thu, 10 Aug 2017 14:21:16 -0700 Subject: [PATCH 3/5] Remove unnecessary fields in TcpMessageSocket --- Stack/Opc.Ua.Core/Stack/Tcp/TcpMessageSocket.cs | 3 --- 1 file changed, 3 deletions(-) diff --git a/Stack/Opc.Ua.Core/Stack/Tcp/TcpMessageSocket.cs b/Stack/Opc.Ua.Core/Stack/Tcp/TcpMessageSocket.cs index f464ae7d7d..34e53f7844 100644 --- a/Stack/Opc.Ua.Core/Stack/Tcp/TcpMessageSocket.cs +++ b/Stack/Opc.Ua.Core/Stack/Tcp/TcpMessageSocket.cs @@ -644,9 +644,6 @@ namespace Opc.Ua.Bindings private int m_bytesReceived; private int m_bytesToReceive; private int m_incomingMessageSize; - private Socket m_socketV4; - private Socket m_socketV6; - private int m_socketResponses; #endregion } } From d112888613fff96f86fbc88b2988c848e8f5ca21 Mon Sep 17 00:00:00 2001 From: Eduard Gilmutdinov Date: Thu, 10 Aug 2017 14:35:08 -0700 Subject: [PATCH 4/5] Remove unnecessary lock in TcpMessageSocket --- Stack/Opc.Ua.Core/Stack/Tcp/TcpMessageSocket.cs | 15 ++++++--------- 1 file changed, 6 insertions(+), 9 deletions(-) diff --git a/Stack/Opc.Ua.Core/Stack/Tcp/TcpMessageSocket.cs b/Stack/Opc.Ua.Core/Stack/Tcp/TcpMessageSocket.cs index 34e53f7844..8d1427df41 100644 --- a/Stack/Opc.Ua.Core/Stack/Tcp/TcpMessageSocket.cs +++ b/Stack/Opc.Ua.Core/Stack/Tcp/TcpMessageSocket.cs @@ -266,17 +266,14 @@ namespace Opc.Ua.Bindings Action doCallback = () => callback(this, new TcpMessageSocketAsyncEventArgs { UserToken = state }); - lock (m_socketLock) + if (addressV6 != null) { - if (addressV6 != null) - { - BeginConnect(addressV6, AddressFamily.InterNetworkV6, port, doCallback); - } + BeginConnect(addressV6, AddressFamily.InterNetworkV6, port, doCallback); + } - if (addressV4 != null && m_socket == null) - { - BeginConnect(addressV4, AddressFamily.InterNetwork, port, doCallback); - } + if (addressV4 != null && m_socket == null) + { + BeginConnect(addressV4, AddressFamily.InterNetwork, port, doCallback); } } From f323ad24d9b47453c20ffb1b648202663f2a6c8e Mon Sep 17 00:00:00 2001 From: Eduard Gilmutdinov Date: Thu, 10 Aug 2017 23:52:49 -0700 Subject: [PATCH 5/5] Make the socket closing more safely --- Stack/Opc.Ua.Core/Stack/Tcp/TcpMessageSocket.cs | 8 +++++--- 1 file changed, 5 insertions(+), 3 deletions(-) diff --git a/Stack/Opc.Ua.Core/Stack/Tcp/TcpMessageSocket.cs b/Stack/Opc.Ua.Core/Stack/Tcp/TcpMessageSocket.cs index 8d1427df41..820e40e7b9 100644 --- a/Stack/Opc.Ua.Core/Stack/Tcp/TcpMessageSocket.cs +++ b/Stack/Opc.Ua.Core/Stack/Tcp/TcpMessageSocket.cs @@ -353,14 +353,16 @@ namespace Opc.Ua.Bindings { m_socket.Shutdown(SocketShutdown.Both); } - - m_socket.Dispose(); - m_socket = null; } catch (Exception e) { Utils.Trace(e, "Unexpected error closing socket."); } + finally + { + m_socket.Dispose(); + m_socket = null; + } } } }