470727 - Thread Starvation with EWYK

Implemented NonBlockingCallbacks and fallback to PEC scheduling

4 files changed, 33 insertions, 2 deletions

diff --git a/jetty-io/src/main/java/org/eclipse/jetty/io/FillInterest.java b/jetty-io/src/main/java/org/eclipse/jetty/io/FillInterest.java
index 34f7385ee9..758601bfe3 100644
--- a/jetty-io/src/main/java/org/eclipse/jetty/io/FillInterest.java
+++ b/jetty-io/src/main/java/org/eclipse/jetty/io/FillInterest.java
@@ -104,6 +104,12 @@ public abstract class FillInterest
     {
         return _interested.get() != null;
     }
+    
+    public boolean isCallbackNonBlocking()
+    {
+        Callback callback = _interested.get();
+        return callback instanceof Callback.NonBlocking;
+    }
 
     /**
      * Call to signal a failure to a registered interest
diff --git a/jetty-io/src/main/java/org/eclipse/jetty/io/ManagedSelector.java b/jetty-io/src/main/java/org/eclipse/jetty/io/ManagedSelector.java
index ae5522c3fe..d2a33bff57 100644
--- a/jetty-io/src/main/java/org/eclipse/jetty/io/ManagedSelector.java
+++ b/jetty-io/src/main/java/org/eclipse/jetty/io/ManagedSelector.java
@@ -269,8 +269,7 @@ public class ManagedSelector extends AbstractLifeCycle implements Runnable, Dump
                         if (attachment instanceof SelectableEndPoint)
                         {
                             // Try to produce a task
-                            SelectableEndPoint selectable = (SelectableEndPoint)attachment;
-                            Runnable task = selectable.onSelected();
+                            Runnable task = ((SelectableEndPoint)attachment).onSelected();
                             if (task != null)
                                 return task;
                         }
diff --git a/jetty-io/src/main/java/org/eclipse/jetty/io/SelectChannelEndPoint.java b/jetty-io/src/main/java/org/eclipse/jetty/io/SelectChannelEndPoint.java
index dff8d4092b..f15343d26f 100644
--- a/jetty-io/src/main/java/org/eclipse/jetty/io/SelectChannelEndPoint.java
+++ b/jetty-io/src/main/java/org/eclipse/jetty/io/SelectChannelEndPoint.java
@@ -100,8 +100,23 @@ public class SelectChannelEndPoint extends ChannelEndPoint implements ManagedSel
         if (LOG.isDebugEnabled())
             LOG.debug("onSelected {}->{} for {}", oldInterestOps, newInterestOps, this);
 
+        
         boolean readable = (readyOps & SelectionKey.OP_READ) != 0;
         boolean writable = (readyOps & SelectionKey.OP_WRITE) != 0;
+        
+        // Call non blocking directly
+        if (readable && getFillInterest().isCallbackNonBlocking())
+        {
+            getFillInterest().fillable();
+            readable=false;
+        }
+        if (writable && getWriteFlusher().isCallbackNonBlocking())
+        {
+            getWriteFlusher().completeWrite();
+            writable=false;
+        }
+        
+        // return task to complete the job
         return readable ? (writable ? _runFillableCompleteWrite : _runFillable)
                         : (writable ? _runCompleteWrite : null);
     }
diff --git a/jetty-io/src/main/java/org/eclipse/jetty/io/WriteFlusher.java b/jetty-io/src/main/java/org/eclipse/jetty/io/WriteFlusher.java
index 2d5dc6bb91..2705355465 100644
--- a/jetty-io/src/main/java/org/eclipse/jetty/io/WriteFlusher.java
+++ b/jetty-io/src/main/java/org/eclipse/jetty/io/WriteFlusher.java
@@ -268,8 +268,19 @@ abstract public class WriteFlusher
             if (_callback!=null)
                 _callback.succeeded();
         }
+        
+        boolean isCallbackNonBlocking()
+        {
+            return _callback instanceof Callback.NonBlocking;
+        }
     }
 
+    public boolean isCallbackNonBlocking()
+    {
+        State s = _state.get();
+        return (s instanceof PendingState) && ((PendingState)s).isCallbackNonBlocking();
+    }
+    
     /**
      * Abstract call to be implemented by specific WriteFlushers. It should schedule a call to {@link #completeWrite()}
      * or {@link #onFail(Throwable)} when appropriate.