From 45e4a1de71cd67258b171636bdccde26468822ad Mon Sep 17 00:00:00 2001 From: fhanik Date: Wed, 25 Apr 2007 17:19:18 +0000 Subject: [PATCH] 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 --- .../tomcat/util/net/NioBlockingSelector.java | 11 +++--- java/org/apache/tomcat/util/net/NioEndpoint.java | 39 ++++++++++++++++------ 2 files changed, 34 insertions(+), 16 deletions(-) diff --git a/java/org/apache/tomcat/util/net/NioBlockingSelector.java b/java/org/apache/tomcat/util/net/NioBlockingSelector.java index 94321abb4..e6eb9b050 100644 --- 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 f2d1ff21e..2c1f956e2 100644 --- 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; -- 2.11.0