Change latch behavior so that it can coexist with other IO events

git-svn-id: https://svn.apache.org/repos/asf/tomcat/tc6.0.x/trunk@532411 13f79535-47bb-0310-9956-ffa450edef68

diff --git a/java/org/apache/tomcat/util/net/NioBlockingSelector.java b/java/org/apache/tomcat/util/net/NioBlockingSelector.java
index 94321ab..e6eb9b0 100644 (file)
--- a/java/org/apache/tomcat/util/net/NioBlockingSelector.java
+++ b/java/org/apache/tomcat/util/net/NioBlockingSelector.java
@@ -66,9 +66,10 @@ public class NioBlockingSelector {
                 if ( key == null ) throw new IOException("Key no longer registered");
                 KeyAttachment att = (KeyAttachment) key.attachment();
                 try {
-                    if ( att.getLatch()==null || att.getLatch().getCount()==0) att.startLatch(1);
-                    if ( att.interestOps() == 0) socket.getPoller().add(socket,SelectionKey.OP_WRITE);
-                    att.getLatch().await(writeTimeout,TimeUnit.MILLISECONDS);
+                    if ( att.getLatch()==null || att.getLatch().getCount()==0) att.startLatch(1,SelectionKey.OP_WRITE);
+                    //only register for write if a write has not yet been issued
+                    if ( (att.interestOps() & SelectionKey.OP_WRITE) == 0) socket.getPoller().add(socket,SelectionKey.OP_WRITE);
+                    att.awaitLatch(writeTimeout,TimeUnit.MILLISECONDS,SelectionKey.OP_WRITE);
                 }catch (InterruptedException ignore) {
                     Thread.interrupted();
                 }
@@ -134,9 +135,9 @@ public class NioBlockingSelector {
                 }
                 KeyAttachment att = (KeyAttachment) key.attachment();
                 try {
-                    if ( att.getLatch()==null || att.getLatch().getCount()==0) att.startLatch(1);
+                    if ( att.getLatch()==null || att.getLatch().getCount()==0) att.startLatch(1,SelectionKey.OP_READ);
                     if ( att.interestOps() == 0) socket.getPoller().add(socket,SelectionKey.OP_READ);
-                    att.getLatch().await(readTimeout,TimeUnit.MILLISECONDS);
+                    att.awaitLatch(readTimeout,TimeUnit.MILLISECONDS, SelectionKey.OP_READ);
                 }catch (InterruptedException ignore) {
                     Thread.interrupted();
                 }

diff --git a/java/org/apache/tomcat/util/net/NioEndpoint.java b/java/org/apache/tomcat/util/net/NioEndpoint.java
index f2d1ff2..2c1f956 100644 (file)
--- a/java/org/apache/tomcat/util/net/NioEndpoint.java
+++ b/java/org/apache/tomcat/util/net/NioEndpoint.java
@@ -1494,12 +1494,15 @@ public class NioEndpoint {
                     sk.attach(attachment);//cant remember why this is here
                     NioChannel channel = attachment.getChannel();
                     if (sk.isReadable() || sk.isWritable() ) {
-                        if ( attachment.getSendfileData() != null ) {
+                        if ( attachment.getLatch() != null ) {
+                            unreg(sk, attachment,attachment.getLatchOps());
+                            attachment.getLatch().countDown();
+                        } else if ( attachment.getSendfileData() != null ) {
                             processSendfile(sk,attachment,true);
                         } else if ( attachment.getComet() ) {
                             //check if thread is available
                             if ( isWorkerAvailable() ) {
-                                unreg(sk, attachment);
+                                unreg(sk, attachment, sk.readyOps());
                                 if (!processSocket(channel, SocketStatus.OPEN))
                                     processSocket(channel, SocketStatus.DISCONNECT);
                                 attachment.setFairness(0);
@@ -1508,13 +1511,10 @@ public class NioEndpoint {
                                 attachment.incFairness();
                                 result = false;
                             }
-                        } else if ( attachment.getLatch() != null ) {
-                            unreg(sk, attachment);
-                            attachment.getLatch().countDown();
                         } else {
                             //later on, improve latch behavior
                             if ( isWorkerAvailable() ) {
-                                unreg(sk, attachment);
+                                unreg(sk, attachment,sk.readyOps());
                                 boolean close = (!processSocket(channel));
                                 if (close) {
                                     cancelledKey(sk,SocketStatus.DISCONNECT,false);
@@ -1578,9 +1578,9 @@ public class NioEndpoint {
             return true;
         }
 
-        protected void unreg(SelectionKey sk, KeyAttachment attachment) {
+        protected void unreg(SelectionKey sk, KeyAttachment attachment, int readyOps) {
             //this is a must, so that we don't have multiple threads messing with the socket
-            reg(sk,attachment,0);
+            reg(sk,attachment,sk.interestOps()& (~readyOps));
         }
         
         protected void reg(SelectionKey sk, KeyAttachment attachment, int intops) {
@@ -1649,6 +1649,9 @@ public class NioEndpoint {
             fairness = 0;
             lastRegistered = 0;
             sendfileData = null;
+            if ( latch!=null ) try {latch.countDown();}catch (Exception ignore){}
+            latch = null;
+            latchOps = 0;
         }
         
         public void reset() {
@@ -1676,11 +1679,24 @@ public class NioEndpoint {
         public int interestOps() { return interestOps;}
         public int interestOps(int ops) { this.interestOps  = ops; return ops; }
         public CountDownLatch getLatch() { return latch; }
-        public void resetLatch() { if ( latch.getCount() == 0 ) latch = null; else throw new IllegalStateException("Latch must be at count 0");}
-        public void startLatch(int cnt) { 
-            if ( latch == null || latch.getCount() == 0 ) this.latch = new CountDownLatch(cnt); 
+        public void resetLatch() { 
+            if ( latch.getCount() == 0 ) latch = null; 
+            else throw new IllegalStateException("Latch must be at count 0");
+            latchOps = 0;
+        }
+        public void startLatch(int cnt, int latchOps) { 
+            if ( latch == null || latch.getCount() == 0 ) {
+                this.latch = new CountDownLatch(cnt);
+                this.latchOps = latchOps;
+            }
             else throw new IllegalStateException("Latch must be at count 0 or null.");
         }
+        public void awaitLatch(long timeout, TimeUnit unit, int latchOps) throws InterruptedException {
+            if ( latch == null ) throw new IllegalStateException("Latch cannot be null");
+            this.latchOps = this.latchOps | latchOps;
+            latch.await(timeout,unit);
+        }
+        public int getLatchOps() { return latchOps;}
         public int getFairness() { return fairness; }
         public void setFairness(int f) { fairness = f;}
         public void incFairness() { fairness++; }
@@ -1698,6 +1714,7 @@ public class NioEndpoint {
         protected boolean error = false;
         protected NioChannel channel = null;
         protected CountDownLatch latch = null;
+        protected int latchOps = 0;
         protected int fairness = 0;
         protected long lastRegistered = 0;
         protected SendfileData sendfileData = null;