ignite-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From ptupit...@apache.org
Subject [ignite] branch master updated: IGNITE-13226 .NET: Fix ClientNotificationHandler leak in ClientSocket
Date Wed, 08 Jul 2020 09:22:08 GMT
This is an automated email from the ASF dual-hosted git repository.

ptupitsyn pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/ignite.git


The following commit(s) were added to refs/heads/master by this push:
     new 69c5b51  IGNITE-13226 .NET: Fix ClientNotificationHandler leak in ClientSocket
69c5b51 is described below

commit 69c5b516686924834876d3f52d349ac1880b20f3
Author: Pavel Tupitsyn <ptupitsyn@apache.org>
AuthorDate: Wed Jul 8 12:21:44 2020 +0300

    IGNITE-13226 .NET: Fix ClientNotificationHandler leak in ClientSocket
    
    Fix race condition in ClientSocket.AddNotificationHandler: SetHandler call was causing
RemoveNotificationHandler call before ConcurrentDictionary.AddOrUpdate has finished, so AddOrUpdate
added the handler back, causing the leak.
---
 .../Common/TestRunner.cs                           |  2 +-
 .../Client/Compute/ComputeClientTests.cs           |  2 ++
 .../Apache.Ignite.Core.Tests/TestUtils.Common.cs   | 33 ++++++++++++++++++++--
 .../Impl/Client/ClientNotificationHandler.cs       | 25 ++++++++++------
 .../Apache.Ignite.Core/Impl/Client/ClientSocket.cs | 25 +++++++++++-----
 .../Impl/Client/Compute/ComputeClient.cs           |  4 +--
 6 files changed, 69 insertions(+), 22 deletions(-)

diff --git a/modules/platforms/dotnet/Apache.Ignite.Core.Tests.DotNetCore/Common/TestRunner.cs
b/modules/platforms/dotnet/Apache.Ignite.Core.Tests.DotNetCore/Common/TestRunner.cs
index 1b0d45c..85bbda7 100644
--- a/modules/platforms/dotnet/Apache.Ignite.Core.Tests.DotNetCore/Common/TestRunner.cs
+++ b/modules/platforms/dotnet/Apache.Ignite.Core.Tests.DotNetCore/Common/TestRunner.cs
@@ -30,4 +30,4 @@ namespace Apache.Ignite.Core.Tests.DotNetCore.Common
             new IgnitionStartTest().TestIgniteStartsFromAppConfig();
         }
     }
-}
\ No newline at end of file
+}
diff --git a/modules/platforms/dotnet/Apache.Ignite.Core.Tests/Client/Compute/ComputeClientTests.cs
b/modules/platforms/dotnet/Apache.Ignite.Core.Tests/Client/Compute/ComputeClientTests.cs
index ceffb11..289468a 100644
--- a/modules/platforms/dotnet/Apache.Ignite.Core.Tests/Client/Compute/ComputeClientTests.cs
+++ b/modules/platforms/dotnet/Apache.Ignite.Core.Tests/Client/Compute/ComputeClientTests.cs
@@ -80,6 +80,8 @@ namespace Apache.Ignite.Core.Tests.Client.Compute
                     Assert.Fail(entry.Message);
                 }
             }
+
+            Assert.IsEmpty(Client.GetActiveNotificationListeners());
         }
 
         /// <summary>
diff --git a/modules/platforms/dotnet/Apache.Ignite.Core.Tests/TestUtils.Common.cs b/modules/platforms/dotnet/Apache.Ignite.Core.Tests/TestUtils.Common.cs
index d7a31e7..f8b14d2 100644
--- a/modules/platforms/dotnet/Apache.Ignite.Core.Tests/TestUtils.Common.cs
+++ b/modules/platforms/dotnet/Apache.Ignite.Core.Tests/TestUtils.Common.cs
@@ -18,6 +18,7 @@
 namespace Apache.Ignite.Core.Tests
 {
     using System;
+    using System.Collections;
     using System.Collections.Concurrent;
     using System.Collections.Generic;
     using System.Diagnostics;
@@ -27,11 +28,13 @@ namespace Apache.Ignite.Core.Tests
     using System.Threading;
     using Apache.Ignite.Core.Binary;
     using Apache.Ignite.Core.Cache.Affinity;
+    using Apache.Ignite.Core.Client;
     using Apache.Ignite.Core.Cluster;
     using Apache.Ignite.Core.Discovery.Tcp;
     using Apache.Ignite.Core.Discovery.Tcp.Static;
     using Apache.Ignite.Core.Impl;
     using Apache.Ignite.Core.Impl.Binary;
+    using Apache.Ignite.Core.Impl.Client;
     using NUnit.Framework;
 
     /// <summary>
@@ -472,7 +475,7 @@ namespace Apache.Ignite.Core.Tests
                 }
             }
         }
-        
+
         /// <summary>
         /// Gets the dot net source dir.
         /// </summary>
@@ -491,7 +494,7 @@ namespace Apache.Ignite.Core.Tests
 
             throw new InvalidOperationException("Could not resolve Ignite.NET source directory.");
         }
-        
+
         /// <summary>
         /// Gets a value indicating whether specified partition is reserved.
         /// </summary>
@@ -499,7 +502,7 @@ namespace Apache.Ignite.Core.Tests
         {
             Debug.Assert(ignite != null);
             Debug.Assert(cacheName != null);
-            
+
             const string taskName = "org.apache.ignite.platform.PlatformIsPartitionReservedTask";
 
             return ignite.GetCompute().ExecuteJavaTask<bool>(taskName, new object[]
{cacheName, part});
@@ -517,5 +520,29 @@ namespace Apache.Ignite.Core.Tests
 
             return ex;
         }
+
+        /// <summary>
+        /// Gets the private field value.
+        /// </summary>
+        public static T GetPrivateField<T>(object obj, string name)
+        {
+            Assert.IsNotNull(obj);
+
+            var field = obj.GetType().GetField(name, BindingFlags.Instance | BindingFlags.NonPublic);
+
+            Assert.IsNotNull(field);
+
+            return (T) field.GetValue(obj);
+        }
+
+        /// <summary>
+        /// Gets active notification listeners.
+        /// </summary>
+        public static ICollection GetActiveNotificationListeners(this IIgniteClient client)
+        {
+            var failoverSocket = GetPrivateField<ClientFailoverSocket>(client, "_socket");
+            var socket = GetPrivateField<ClientSocket>(failoverSocket, "_socket");
+            return GetPrivateField<ICollection>(socket, "_notificationListeners");
+        }
     }
 }
diff --git a/modules/platforms/dotnet/Apache.Ignite.Core/Impl/Client/ClientNotificationHandler.cs
b/modules/platforms/dotnet/Apache.Ignite.Core/Impl/Client/ClientNotificationHandler.cs
index cc031e8..d5d5c0e 100644
--- a/modules/platforms/dotnet/Apache.Ignite.Core/Impl/Client/ClientNotificationHandler.cs
+++ b/modules/platforms/dotnet/Apache.Ignite.Core/Impl/Client/ClientNotificationHandler.cs
@@ -35,12 +35,12 @@ namespace Apache.Ignite.Core.Impl.Client
     {
         /** Handler delegate. */
         public delegate void Handler(IBinaryStream stream, Exception ex);
-        
+
         /** Logger. */
         private readonly ILogger _logger;
 
         /** Nested handler. */
-        private Handler _handler;
+        private volatile Handler _handler;
 
         /** Queue. */
         private List<KeyValuePair<IBinaryStream, Exception>> _queue;
@@ -51,12 +51,20 @@ namespace Apache.Ignite.Core.Impl.Client
         public ClientNotificationHandler(ILogger logger, Handler handler = null)
         {
             Debug.Assert(logger != null);
-            
+
             _logger = logger;
             _handler = handler;
         }
 
         /// <summary>
+        /// Gets a value indicating whether handler is set for this instance.
+        /// </summary>
+        public bool HasHandler
+        {
+            get { return _handler != null; }
+        }
+
+        /// <summary>
         /// Handles the notification.
         /// </summary>
         /// <param name="stream">Notification data.</param>
@@ -69,7 +77,7 @@ namespace Apache.Ignite.Core.Impl.Client
             {
                 // NOTE: Back pressure control should be added here when needed (e.g. for
Continuous Queries).
                 var handler = _handler;
-                
+
                 if (handler != null)
                 {
                     ThreadPool.QueueUserWorkItem(_ => Handle(handler, stream, exception));
@@ -86,10 +94,11 @@ namespace Apache.Ignite.Core.Impl.Client
         /// Sets the handler.
         /// </summary>
         /// <param name="handler">Handler.</param>
-        public ClientNotificationHandler SetHandler(Handler handler)
+        public void SetHandler(Handler handler)
         {
             Debug.Assert(handler != null);
-            
+            Debug.Assert(_handler == null);
+
             lock (this)
             {
                 _handler = handler;
@@ -102,8 +111,6 @@ namespace Apache.Ignite.Core.Impl.Client
                     ThreadPool.QueueUserWorkItem(_ => Drain(handler, queue));
                 }
             }
-
-            return this;
         }
 
         /// <summary>
@@ -134,4 +141,4 @@ namespace Apache.Ignite.Core.Impl.Client
             }
         }
     }
-}
\ No newline at end of file
+}
diff --git a/modules/platforms/dotnet/Apache.Ignite.Core/Impl/Client/ClientSocket.cs b/modules/platforms/dotnet/Apache.Ignite.Core/Impl/Client/ClientSocket.cs
index ba13550..bec7c10 100644
--- a/modules/platforms/dotnet/Apache.Ignite.Core/Impl/Client/ClientSocket.cs
+++ b/modules/platforms/dotnet/Apache.Ignite.Core/Impl/Client/ClientSocket.cs
@@ -252,12 +252,18 @@ namespace Apache.Ignite.Core.Impl.Client
         /// Adds a notification handler.
         /// </summary>
         /// <param name="notificationId">Notification id.</param>
-        /// <param name="handler">Handler delegate.</param>
-        public void AddNotificationHandler(long notificationId, ClientNotificationHandler.Handler
handler)
+        /// <param name="handlerDelegate">Handler delegate.</param>
+        public void AddNotificationHandler(long notificationId, ClientNotificationHandler.Handler
handlerDelegate)
         {
-            _notificationListeners.AddOrUpdate(notificationId,
-                _ => new ClientNotificationHandler(_logger, handler),
-                (_, oldHandler) => oldHandler.SetHandler(handler));
+            var handler = _notificationListeners.GetOrAdd(notificationId,
+                _ => new ClientNotificationHandler(_logger, handlerDelegate));
+
+            if (!handler.HasHandler)
+            {
+                // We could use AddOrUpdate, but SetHandler must be called outside of Update
call,
+                // because it causes handler execution, which, in turn, may call RemoveNotificationHandler.
+                handler.SetHandler(handlerDelegate);
+            }
 
             _listenerEvent.Set();
         }
@@ -269,6 +275,11 @@ namespace Apache.Ignite.Core.Impl.Client
         /// <returns>True when removed, false otherwise.</returns>
         public void RemoveNotificationHandler(long notificationId)
         {
+            if (IsDisposed)
+            {
+                return;
+            }
+
             ClientNotificationHandler unused;
             var removed = _notificationListeners.TryRemove(notificationId, out unused);
             Debug.Assert(removed);
@@ -440,8 +451,8 @@ namespace Apache.Ignite.Core.Impl.Client
                 throw new IgniteClientException("Unexpected thin client notification: " +
requestId);
             }
 
-            _notificationListeners.GetOrAdd(requestId, _ => new ClientNotificationHandler(_logger))
-                .Handle(stream, null);
+            var handler = _notificationListeners.GetOrAdd(requestId, _ => new ClientNotificationHandler(_logger));
+            handler.Handle(stream, null);
 
             return true;
         }
diff --git a/modules/platforms/dotnet/Apache.Ignite.Core/Impl/Client/Compute/ComputeClient.cs
b/modules/platforms/dotnet/Apache.Ignite.Core/Impl/Client/Compute/ComputeClient.cs
index a02608f..6e1fdca 100644
--- a/modules/platforms/dotnet/Apache.Ignite.Core/Impl/Client/Compute/ComputeClient.cs
+++ b/modules/platforms/dotnet/Apache.Ignite.Core/Impl/Client/Compute/ComputeClient.cs
@@ -191,14 +191,14 @@ namespace Apache.Ignite.Core.Impl.Client.Compute
 
             ctx.Socket.AddNotificationHandler(taskId, (stream, ex) =>
             {
+                ctx.Socket.RemoveNotificationHandler(taskId);
+
                 if (ex != null)
                 {
                     tcs.TrySetException(ex);
                     return;
                 }
 
-                ctx.Socket.RemoveNotificationHandler(taskId);
-
                 var reader = ctx.Marshaller.StartUnmarshal(stream,
                     keepBinary ? BinaryMode.ForceBinary : BinaryMode.Deserialize);
 


Mime
View raw message