From a906dda02f038358f1c2cff217e07284ad98a67f Mon Sep 17 00:00:00 2001
From: pveentjer <alarmnummer@gmail.com>
Date: Tue, 26 Jun 2018 07:17:47 +0300
Subject: [PATCH] Fixed close problems in NioOutboundPipeline

There are 2 problems:
* tmp stall because a task is execute on the io thread from
  that same io thread and then the io thread waits for it. Because
  the thread will not execute its own task (yet) you have a 3s stall
  in the io pipeline. This is a big problem because is blocks all traffic
  for that io thread.

* the close task is scheduled as a 'packet'. If the pipeline already is
  scheduled, scheduling the close task will not get picked up as part
  of an optimization.

This pr fixes the above problems.
---
 .../networking/nio/NioOutboundPipeline.java   | 26 +++++++++++++++++--
 1 file changed, 24 insertions(+), 2 deletions(-)

diff --git a/hazelcast/src/main/java/com/hazelcast/internal/networking/nio/NioOutboundPipeline.java b/hazelcast/src/main/java/com/hazelcast/internal/networking/nio/NioOutboundPipeline.java
index 73e1ed9bbcc7..87d0b67d5fd0 100644
--- a/hazelcast/src/main/java/com/hazelcast/internal/networking/nio/NioOutboundPipeline.java
+++ b/hazelcast/src/main/java/com/hazelcast/internal/networking/nio/NioOutboundPipeline.java
@@ -358,8 +358,24 @@ public void close() {
         urgentWriteQueue.clear();
 
         CloseTask closeTask = new CloseTask();
-        write(new TaskFrame(closeTask));
-        closeTask.awaitCompletion();
+        NioThread owner = this.owner;
+        Thread currentThread = Thread.currentThread();
+        if (currentThread instanceof NioThread) {
+            if (currentThread == owner) {
+                // we don't schedule the task, we execute it immediately
+                // This will prevent waiting on a task this thread is
+                // supposed to execute, but can't. And therefor runs into
+                // a temporary stall.
+                closeTask.run();
+            }
+
+            // if the currentThread isn't the owner, there
+            // is a migration happening and we can't close
+        } else {
+            // closing is executed from a non-io thread, so we can block
+            owner.addTaskAndWakeup(closeTask);
+            closeTask.awaitCompletion();
+        }
     }
 
     @Override
@@ -438,6 +454,12 @@ private class CloseTask implements Runnable {
 
         @Override
         public void run() {
+            NioThread owner = NioOutboundPipeline.this.owner;
+            if (owner != Thread.currentThread()) {
+                owner.addTaskAndWakeup(this);
+                return;
+            }
+
             try {
                 channel.closeOutbound();
             } catch (IOException e) {