tomcat-dev mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From fha...@apache.org
Subject svn commit: r385839 - in /tomcat/container/tc5.5.x/modules/groupcom/src/share/org/apache/catalina/tribes: ./ io/ tcp/bio/ tcp/nio/
Date Tue, 14 Mar 2006 17:22:07 GMT
Author: fhanik
Date: Tue Mar 14 09:22:04 2006
New Revision: 385839

URL: http://svn.apache.org/viewcvs?rev=385839&view=rev
Log:
Implemented a blocking IO receiver, threaded

Added:
    tomcat/container/tc5.5.x/modules/groupcom/src/share/org/apache/catalina/tribes/tcp/bio/BioReceiver.java
    tomcat/container/tc5.5.x/modules/groupcom/src/share/org/apache/catalina/tribes/tcp/bio/TcpReplicationThread.java
Modified:
    tomcat/container/tc5.5.x/modules/groupcom/src/share/org/apache/catalina/tribes/ChannelReceiver.java
    tomcat/container/tc5.5.x/modules/groupcom/src/share/org/apache/catalina/tribes/io/ObjectReader.java
    tomcat/container/tc5.5.x/modules/groupcom/src/share/org/apache/catalina/tribes/tcp/bio/BioSender.java
    tomcat/container/tc5.5.x/modules/groupcom/src/share/org/apache/catalina/tribes/tcp/bio/MultipointBioSender.java
    tomcat/container/tc5.5.x/modules/groupcom/src/share/org/apache/catalina/tribes/tcp/nio/NioReceiver.java
    tomcat/container/tc5.5.x/modules/groupcom/src/share/org/apache/catalina/tribes/tcp/nio/ThreadPool.java

Modified: tomcat/container/tc5.5.x/modules/groupcom/src/share/org/apache/catalina/tribes/ChannelReceiver.java
URL: http://svn.apache.org/viewcvs/tomcat/container/tc5.5.x/modules/groupcom/src/share/org/apache/catalina/tribes/ChannelReceiver.java?rev=385839&r1=385838&r2=385839&view=diff
==============================================================================
--- tomcat/container/tc5.5.x/modules/groupcom/src/share/org/apache/catalina/tribes/ChannelReceiver.java
(original)
+++ tomcat/container/tc5.5.x/modules/groupcom/src/share/org/apache/catalina/tribes/ChannelReceiver.java
Tue Mar 14 09:22:04 2006
@@ -44,7 +44,7 @@
      * set ack mode
      * @param isSendAck
      */
-    public void setSendAck(boolean isSendAck);
+    public void setSendAck(boolean sendack);
     
     /**
      * get the listing ip interface

Modified: tomcat/container/tc5.5.x/modules/groupcom/src/share/org/apache/catalina/tribes/io/ObjectReader.java
URL: http://svn.apache.org/viewcvs/tomcat/container/tc5.5.x/modules/groupcom/src/share/org/apache/catalina/tribes/io/ObjectReader.java?rev=385839&r1=385838&r2=385839&view=diff
==============================================================================
--- tomcat/container/tc5.5.x/modules/groupcom/src/share/org/apache/catalina/tribes/io/ObjectReader.java
(original)
+++ tomcat/container/tc5.5.x/modules/groupcom/src/share/org/apache/catalina/tribes/io/ObjectReader.java
Tue Mar 14 09:22:04 2006
@@ -20,6 +20,7 @@
 import java.nio.channels.SocketChannel;
 import org.apache.catalina.tribes.ChannelMessage;
 import java.io.IOException;
+import java.net.Socket;
 
 
 
@@ -62,6 +63,16 @@
             this.buffer = new XByteBuffer(43800,true);
         }
     }
+    public ObjectReader(Socket socket, ListenCallback callback) {
+        try{
+            this.buffer = new XByteBuffer(socket.getReceiveBufferSize(), true);
+        }catch ( IOException x ) {
+            //unable to get buffer size
+            log.warn("Unable to retrieve the socket receiver buffer size, setting to default
43800 bytes.");
+            this.buffer = new XByteBuffer(43800,true);
+        }
+        this.callback = callback;
+    }
 
     /**
      * get the current SimpleTcpCluster
@@ -125,6 +136,10 @@
         return pkgCnt;
     }
     
+    public int bufferSize() {
+        return buffer.getLength();
+    }
+    
     /**
      * Returns the number of packages that the reader has read
      * @return int
@@ -141,6 +156,12 @@
      */
     public int write(ByteBuffer buf) throws java.io.IOException {
         return getChannel().write(buf);
+    }
+    
+    public void close() {
+        this.callback = null;
+        this.channel = null;
+        this.buffer = null;
     }
 
 }

Added: tomcat/container/tc5.5.x/modules/groupcom/src/share/org/apache/catalina/tribes/tcp/bio/BioReceiver.java
URL: http://svn.apache.org/viewcvs/tomcat/container/tc5.5.x/modules/groupcom/src/share/org/apache/catalina/tribes/tcp/bio/BioReceiver.java?rev=385839&view=auto
==============================================================================
--- tomcat/container/tc5.5.x/modules/groupcom/src/share/org/apache/catalina/tribes/tcp/bio/BioReceiver.java
(added)
+++ tomcat/container/tc5.5.x/modules/groupcom/src/share/org/apache/catalina/tribes/tcp/bio/BioReceiver.java
Tue Mar 14 09:22:04 2006
@@ -0,0 +1,306 @@
+/*
+ * Copyright 1999,2006 The Apache Software Foundation.
+ * 
+ * Licensed under the Apache License, Version 2.0 (the "License"); you may not
+ * use this file except in compliance with the License. You may obtain a copy of
+ * the License at
+ * 
+ * http://www.apache.org/licenses/LICENSE-2.0
+ * 
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+ * WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+ * License for the specific language governing permissions and limitations under
+ * the License.
+ */
+package org.apache.catalina.tribes.tcp.bio;
+
+import java.io.IOException;
+
+import org.apache.catalina.tribes.ChannelReceiver;
+import org.apache.catalina.tribes.MessageListener;
+import java.net.InetAddress;
+import org.apache.catalina.tribes.tcp.nio.ThreadPool;
+import java.net.ServerSocket;
+import java.net.InetSocketAddress;
+import org.apache.catalina.tribes.io.ListenCallback;
+import org.apache.catalina.tribes.ChannelMessage;
+import java.net.Socket;
+import org.apache.catalina.tribes.io.ObjectReader;
+
+/**
+ * <p>Title: </p>
+ *
+ * <p>Description: </p>
+ *
+ * <p>Copyright: Copyright (c) 2005</p>
+ *
+ * <p>Company: </p>
+ *
+ * @author not attributable
+ * @version 1.0
+ */
+public class BioReceiver implements Runnable, ChannelReceiver, ListenCallback {
+
+    protected static org.apache.commons.logging.Log log = org.apache.commons.logging.LogFactory.getLog(BioReceiver.class);
+
+
+    protected MessageListener listener;
+    protected String host;
+    protected InetAddress bind;
+    protected int port;
+    protected boolean sendack;
+    protected boolean sync;
+    protected int rxBufSize = 43800;
+    protected int txBufSize = 25188;    
+    protected int tcpThreadCount;
+    protected ServerSocket serverSocket;
+    protected boolean doRun = true;
+    
+    protected ThreadPool pool;
+
+    public BioReceiver() {
+    }
+
+    /**
+     *
+     * @return The host
+     * @todo Implement this org.apache.catalina.tribes.ChannelReceiver method
+     */
+    public String getHost() {
+        return host;
+    }
+
+    /**
+     * getMessageListener
+     *
+     * @return MessageListener
+     * @todo Implement this org.apache.catalina.tribes.ChannelReceiver method
+     */
+    public MessageListener getMessageListener() {
+        return listener;
+    }
+
+    /**
+     *
+     * @return The port
+     * @todo Implement this org.apache.catalina.tribes.ChannelReceiver method
+     */
+    public int getPort() {
+        return port;
+    }
+
+    public int getRxBufSize() {
+        return rxBufSize;
+    }
+
+    public int getTxBufSize() {
+        return txBufSize;
+    }
+
+    public int getTcpThreadCount() {
+        return tcpThreadCount;
+    }
+
+    /**
+     *
+     * @return boolean
+     * @todo Implement this org.apache.catalina.tribes.ChannelReceiver method
+     */
+    public boolean getSendAck() {
+        return sendack;
+    }
+
+    /**
+     * setMessageListener
+     *
+     * @param listener MessageListener
+     * @todo Implement this org.apache.catalina.tribes.ChannelReceiver method
+     */
+    public void setMessageListener(MessageListener listener) {
+        this.listener = listener;
+    }
+    
+
+    /**
+     *
+     * @param isSendAck boolean
+     * @todo Implement this org.apache.catalina.tribes.ChannelReceiver method
+     */
+    public void setSendAck(boolean sendAck) {
+        this.sendack = sendAck;
+    }
+
+    /**
+     *
+     * @throws IOException
+     * @todo Implement this org.apache.catalina.tribes.ChannelReceiver method
+     */
+    public void start() throws IOException {
+        this.doRun = true;
+        try {
+            TcpReplicationThread[] receivers = new TcpReplicationThread[tcpThreadCount];
+            for ( int i=0; i<receivers.length; i++ ) {
+                receivers[i] = getReplicationThread();
+            }
+            pool = new ThreadPool(new Object(), receivers);
+        } catch (Exception e) {
+            log.error("ThreadPool can initilzed. Listener not started", e);
+            return;
+        }
+        try {
+            getBind();
+            bind();
+            Thread t = new Thread(this, "BioReceiver");
+            t.setDaemon(true);
+            t.start();
+        } catch (Exception x) {
+            log.fatal("Unable to start cluster receiver", x);
+        }
+    }
+    
+    protected TcpReplicationThread getReplicationThread() {
+        TcpReplicationThread result = new TcpReplicationThread();
+        result.setOptions(getWorkerThreadOptions());
+        return result;
+    }
+
+    /**
+     *
+     * @todo Implement this org.apache.catalina.tribes.ChannelReceiver method
+     */
+    public void stop() {
+        this.doRun = false;
+        try {
+            this.serverSocket.close();
+        }catch ( Exception x ) {}
+    }
+
+    public void setTcpListenPort(int tcpListenPort) {
+        this.port = tcpListenPort;
+    }
+
+    public void setTcpListenAddress(String tcpListenHost) {
+        this.host = tcpListenHost;
+    }
+
+    public void setSynchronized(boolean sync) {
+        this.sync = sync;
+    }
+
+    public void setRxBufSize(int rxBufSize) {
+        this.rxBufSize = rxBufSize;
+    }
+
+    public void setTxBufSize(int txBufSize) {
+        this.txBufSize = txBufSize;
+    }
+
+    public void setTcpThreadCount(int tcpThreadCount) {
+        this.tcpThreadCount = tcpThreadCount;
+    }
+    
+    public void setTcpSelectorTimeout(long timeout) {
+        //do nothing
+    }
+    
+    /**
+     * @return Returns the bind.
+     */
+    public InetAddress getBind() {
+        if (bind == null) {
+            try {
+                if ("auto".equals(host)) {
+                    host = java.net.InetAddress.getLocalHost().getHostAddress();
+                }
+                if (log.isDebugEnabled())
+                    log.debug("Starting replication listener on address:"+ host);
+                bind = java.net.InetAddress.getByName(host);
+            } catch (IOException ioe) {
+                log.error("Failed bind replication listener on address:"+ host, ioe);
+            }
+        }
+        return bind;
+    }
+    
+    protected int bind(ServerSocket socket, int portstart, int retries) throws IOException
{
+        while ( retries > 0 ) {
+            try {
+                InetSocketAddress addr = new InetSocketAddress(getBind(), portstart);
+                socket.bind(addr);
+                setTcpListenPort(portstart);
+                log.info("Bio Server Socket bound to:"+addr);
+                return 0;
+            }catch ( IOException x) {
+                retries--;
+                if ( retries <= 0 ) throw x;
+                portstart++;
+                retries = bind(socket,portstart,retries);
+            }
+        }
+        return retries;
+    }
+
+
+    
+    protected void bind() throws IOException {
+        // allocate an unbound server socket channel
+        serverSocket = new ServerSocket();
+        // set the port the server channel will listen to
+        //serverSocket.bind(new InetSocketAddress(getBind(), getTcpListenPort()));
+        bind(serverSocket,getPort(),10);
+    }
+    
+    public void messageDataReceived(ChannelMessage data) {
+        if ( this.listener != null ) {
+            listener.messageReceived(data);
+        }
+    }
+    
+    public void run() {
+        try {
+            listen();
+        } catch (Exception x) {
+            log.error("Unable to run replication listener.", x);
+        }
+
+    }
+    
+    public void listen() throws Exception {
+        while ( doRun ) {
+            Socket socket = null;
+            try {
+                socket = serverSocket.accept();
+            }catch ( Exception x ) {
+                if ( doRun ) throw x;
+            }
+            if ( !doRun ) break;
+            if ( socket == null ) continue;
+            socket.setReceiveBufferSize(rxBufSize);
+            socket.setSendBufferSize(txBufSize);
+            TcpReplicationThread thread = (TcpReplicationThread)pool.getWorker();
+            ObjectReader reader = new ObjectReader(socket,this);
+
+            if ( thread == null ) {
+                //we are out of workers, process the request on the listening thread
+                thread = getReplicationThread();
+                thread.socket = socket;
+                thread.reader = reader;
+                thread.run();
+            } else { 
+                thread.serviceSocket(socket,reader);
+            }//end if
+        }//while
+    }
+    
+    public int getWorkerThreadOptions() {
+        int options = 0;
+        if ( sync ) options = options |TcpReplicationThread.OPTION_SYNCHRONIZED;
+        if ( getSendAck() ) options = options |TcpReplicationThread.OPTION_SEND_ACK;
+        return options;
+    }
+
+
+
+
+}
\ No newline at end of file

Modified: tomcat/container/tc5.5.x/modules/groupcom/src/share/org/apache/catalina/tribes/tcp/bio/BioSender.java
URL: http://svn.apache.org/viewcvs/tomcat/container/tc5.5.x/modules/groupcom/src/share/org/apache/catalina/tribes/tcp/bio/BioSender.java?rev=385839&r1=385838&r2=385839&view=diff
==============================================================================
--- tomcat/container/tc5.5.x/modules/groupcom/src/share/org/apache/catalina/tribes/tcp/bio/BioSender.java
(original)
+++ tomcat/container/tc5.5.x/modules/groupcom/src/share/org/apache/catalina/tribes/tcp/bio/BioSender.java
Tue Mar 14 09:22:04 2006
@@ -1,5 +1,5 @@
 /*
- * Copyright 1999,2005 The Apache Software Foundation.
+ * Copyright 1999,2006 The Apache Software Foundation.
  * 
  * Licensed under the Apache License, Version 2.0 (the "License"); you may not
  * use this file except in compliance with the License. You may obtain a copy of

Modified: tomcat/container/tc5.5.x/modules/groupcom/src/share/org/apache/catalina/tribes/tcp/bio/MultipointBioSender.java
URL: http://svn.apache.org/viewcvs/tomcat/container/tc5.5.x/modules/groupcom/src/share/org/apache/catalina/tribes/tcp/bio/MultipointBioSender.java?rev=385839&r1=385838&r2=385839&view=diff
==============================================================================
--- tomcat/container/tc5.5.x/modules/groupcom/src/share/org/apache/catalina/tribes/tcp/bio/MultipointBioSender.java
(original)
+++ tomcat/container/tc5.5.x/modules/groupcom/src/share/org/apache/catalina/tribes/tcp/bio/MultipointBioSender.java
Tue Mar 14 09:22:04 2006
@@ -33,7 +33,7 @@
     protected long selectTimeout = 1000; 
     protected boolean waitForAck = false;
     protected int retryAttempts=0;
-    protected int keepAliveCount = Integer.MAX_VALUE;
+    protected int keepAliveCount = -1;
     protected HashMap bioSenders = new HashMap();
     protected boolean directBuf = false;
     protected int rxBufSize = 43800;

Added: tomcat/container/tc5.5.x/modules/groupcom/src/share/org/apache/catalina/tribes/tcp/bio/TcpReplicationThread.java
URL: http://svn.apache.org/viewcvs/tomcat/container/tc5.5.x/modules/groupcom/src/share/org/apache/catalina/tribes/tcp/bio/TcpReplicationThread.java?rev=385839&view=auto
==============================================================================
--- tomcat/container/tc5.5.x/modules/groupcom/src/share/org/apache/catalina/tribes/tcp/bio/TcpReplicationThread.java
(added)
+++ tomcat/container/tc5.5.x/modules/groupcom/src/share/org/apache/catalina/tribes/tcp/bio/TcpReplicationThread.java
Tue Mar 14 09:22:04 2006
@@ -0,0 +1,173 @@
+/*
+ * Copyright 1999,2004 The Apache Software Foundation.
+ * 
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ * 
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ * 
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.catalina.tribes.tcp.bio;
+import java.io.IOException;
+
+import org.apache.catalina.tribes.io.ObjectReader;
+import org.apache.catalina.tribes.tcp.Constants;
+import org.apache.catalina.tribes.tcp.nio.WorkerThread;
+import java.net.Socket;
+import java.io.InputStream;
+
+/**
+ * A worker thread class which can drain channels and echo-back the input. Each
+ * instance is constructed with a reference to the owning thread pool object.
+ * When started, the thread loops forever waiting to be awakened to service the
+ * channel associated with a SelectionKey object. The worker is tasked by
+ * calling its serviceChannel() method with a SelectionKey object. The
+ * serviceChannel() method stores the key reference in the thread object then
+ * calls notify() to wake it up. When the channel has been drained, the worker
+ * thread returns itself to its parent pool.
+ * 
+ * @author Filip Hanik
+ * 
+ * @version $Revision: 378050 $, $Date: 2006-02-15 12:30:02 -0600 (Wed, 15 Feb 2006) $
+ */
+public class TcpReplicationThread extends WorkerThread {
+    public static final int OPTION_SEND_ACK = 0x0001;
+    public static final int OPTION_SYNCHRONIZED = 0x0002;
+    public static final int OPTION_DIRECT_BUFFER = 0x0004;
+
+
+    protected static org.apache.commons.logging.Log log = org.apache.commons.logging.LogFactory.getLog(
TcpReplicationThread.class );
+    
+    protected Socket socket;
+    protected ObjectReader reader;
+    
+    public TcpReplicationThread ()
+    {
+    }
+
+    // loop forever waiting for work to do
+    public synchronized void run()
+    {
+        while (doRun) {
+            try {
+                // sleep and release object lock
+                this.wait();
+            } catch (InterruptedException e) {
+                if(log.isInfoEnabled())
+                    log.info("TCP worker thread interrupted in cluster",e);
+                // clear interrupt status
+                Thread.interrupted();
+            }
+            if ( this.socket != null ) {
+                try {
+                    drainSocket();
+                } catch ( Exception x ) {
+                    log.error("Unable to service bio socket");
+                }finally {
+                    try {reader.close();}catch ( Exception x){}
+                    try {socket.close();}catch ( Exception x){}
+                    reader = null;
+                    socket = null;
+                }
+            }
+            // done, ready for more, return to pool
+            if ( this.pool != null ) this.pool.returnWorker (this);
+            else doRun = false;
+        }
+    }
+
+    
+    public synchronized void serviceSocket(Socket socket, ObjectReader reader) {
+        this.socket = socket;
+        this.reader = reader;
+        this.notify();		// awaken the thread
+    }
+    
+    protected void execute(ObjectReader reader) throws Exception{
+        int pkgcnt = reader.count();
+        /**
+         * Use send ack here if you want to ack the request to the remote 
+         * server before completing the request
+         * This is considered an asynchronized request
+         */
+        if (sendAckAsync()) {
+            while ( pkgcnt > 0 ) {
+                sendAck();
+                pkgcnt--;
+            }
+        }
+        //check to see if any data is available
+        pkgcnt = reader.execute();
+
+        /**
+         * Use send ack here if you want the request to complete on this 
+         * server before sending the ack to the remote server
+         * This is considered a synchronized request
+         */
+        if (sendAckSync()) {
+            while ( pkgcnt > 0 ) {
+                sendAck();
+                pkgcnt--;
+            }
+        }        
+       
+    }
+
+    /**
+     * The actual code which drains the channel associated with
+     * the given key.  This method assumes the key has been
+     * modified prior to invocation to turn off selection
+     * interest in OP_READ.  When this method completes it
+     * re-enables OP_READ and calls wakeup() on the selector
+     * so the selector will resume watching this channel.
+     */
+    protected void drainSocket () throws Exception {
+        InputStream in = socket.getInputStream();
+        // loop while data available, channel is non-blocking
+        byte[] buf = new byte[1024];
+        int length = in.read(buf);
+        while ( length >= 0 ) {
+            int count = reader.append(buf,0,length,true);
+            if ( count > 0 ) execute(reader);
+            if ( in.available() == 0 && reader.bufferSize() == 0 ) length = -1;
+            else length = in.read(buf);
+        }
+    }
+
+
+    public boolean sendAckSync() {
+        int options = getOptions();
+        return ((OPTION_SEND_ACK & options) == OPTION_SEND_ACK) &&
+               ((OPTION_SYNCHRONIZED & options) == OPTION_SYNCHRONIZED);
+    }
+
+    public boolean sendAckAsync() {
+        int options = getOptions();
+        return ((OPTION_SEND_ACK & options) == OPTION_SEND_ACK) &&
+               ((OPTION_SYNCHRONIZED & options) != OPTION_SYNCHRONIZED);
+    }
+
+
+    /**
+     * send a reply-acknowledgement (6,2,3)
+     * @param key
+     * @param channel
+     */
+    protected void sendAck() {
+        try {
+            this.socket.getOutputStream().write(Constants.ACK_COMMAND);
+            if (log.isTraceEnabled()) {
+                log.trace("ACK sent to " + socket.getPort());
+            }
+        } catch ( java.io.IOException x ) {
+            log.warn("Unable to send ACK back through channel, channel disconnected?: "+x.getMessage());
+        }
+    }
+}

Modified: tomcat/container/tc5.5.x/modules/groupcom/src/share/org/apache/catalina/tribes/tcp/nio/NioReceiver.java
URL: http://svn.apache.org/viewcvs/tomcat/container/tc5.5.x/modules/groupcom/src/share/org/apache/catalina/tribes/tcp/nio/NioReceiver.java?rev=385839&r1=385838&r2=385839&view=diff
==============================================================================
--- tomcat/container/tc5.5.x/modules/groupcom/src/share/org/apache/catalina/tribes/tcp/nio/NioReceiver.java
(original)
+++ tomcat/container/tc5.5.x/modules/groupcom/src/share/org/apache/catalina/tribes/tcp/nio/NioReceiver.java
Tue Mar 14 09:22:04 2006
@@ -136,7 +136,7 @@
         }
         try {
             getBind();
-            this.bind();
+            bind();
             Thread t = new Thread(this, "NioReceiver");
             t.setDaemon(true);
             t.start();
@@ -310,7 +310,7 @@
         try {
             listen();
         } catch (Exception x) {
-            log.error("Unable to start replication listener.", x);
+            log.error("Unable to run replication listener.", x);
         }
     }
 

Modified: tomcat/container/tc5.5.x/modules/groupcom/src/share/org/apache/catalina/tribes/tcp/nio/ThreadPool.java
URL: http://svn.apache.org/viewcvs/tomcat/container/tc5.5.x/modules/groupcom/src/share/org/apache/catalina/tribes/tcp/nio/ThreadPool.java?rev=385839&r1=385838&r2=385839&view=diff
==============================================================================
--- tomcat/container/tc5.5.x/modules/groupcom/src/share/org/apache/catalina/tribes/tcp/nio/ThreadPool.java
(original)
+++ tomcat/container/tc5.5.x/modules/groupcom/src/share/org/apache/catalina/tribes/tcp/nio/ThreadPool.java
Tue Mar 14 09:22:04 2006
@@ -17,6 +17,7 @@
 package org.apache.catalina.tribes.tcp.nio;
 import java.util.LinkedList;
 import java.util.List;
+import java.util.Iterator;
 
 /**
  * @author not attributable
@@ -34,6 +35,7 @@
     List idle = new LinkedList();
     Object mutex = new Object();
     Object interestOpsMutex = null;
+    boolean running = true;
 
     public ThreadPool (Object interestOpsMutex, WorkerThread[] threads) throws Exception
{
         // fill up the pool with worker threads
@@ -98,12 +100,29 @@
      * idle pool.
      */
     public void returnWorker (WorkerThread worker) {
-        synchronized (mutex) {
-            idle.add (worker);
-            mutex.notify();
+        if ( running ) {
+            synchronized (mutex) {
+                idle.add(worker);
+                mutex.notify();
+            }
+        }else {
+            worker.doRun = false;
+            synchronized (worker){worker.notify();}
         }
     }
     public Object getInterestOpsMutex() {
         return interestOpsMutex;
+    }
+    
+    public void stop() {
+        running = false;
+        synchronized (mutex) {
+            Iterator i = idle.iterator();
+            while ( i.hasNext() ) {
+                WorkerThread worker = (WorkerThread)i.next();
+                returnWorker(worker);
+                i.remove();
+            }
+        }
     }
 }



---------------------------------------------------------------------
To unsubscribe, e-mail: dev-unsubscribe@tomcat.apache.org
For additional commands, e-mail: dev-help@tomcat.apache.org


Mime
View raw message