TestHttpsServer.java

/*
 * Copyright (c) 2002, 2019, Oracle and/or its affiliates. All rights reserved.
 * DO NOT ALTER OR REMOVE COPYRIGHT NOTICES OR THIS FILE HEADER.
 *
 * This code is free software; you can redistribute it and/or modify it
 * under the terms of the GNU General Public License version 2 only, as
 * published by the Free Software Foundation.
 *
 * This code is distributed in the hope that it will be useful, but WITHOUT
 * ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
 * FITNESS FOR A PARTICULAR PURPOSE.  See the GNU General Public License
 * version 2 for more details (a copy is included in the LICENSE file that
 * accompanied this code).
 *
 * You should have received a copy of the GNU General Public License version
 * 2 along with this work; if not, write to the Free Software Foundation,
 * Inc., 51 Franklin St, Fifth Floor, Boston, MA 02110-1301 USA.
 *
 * Please contact Oracle, 500 Oracle Parkway, Redwood Shores, CA 94065 USA
 * or visit www.oracle.com if you need additional information or have any
 * questions.
 */

import java.net.*;
import java.io.*;
import java.nio.*;
import java.nio.channels.*;
import sun.net.www.MessageHeader;
import java.util.*;
import javax.net.ssl.*;
import javax.net.ssl.SSLEngineResult.*;
import java.security.*;

/**
 * This class implements a simple HTTPS server. It uses multiple threads to
 * handle connections in parallel, and will spin off a new thread to handle
 * each request. (this is easier to implement with SSLEngine)
 * <p>
 * It must be instantiated with a {@link HttpCallback} object to which
 * requests are given and must be handled.
 * <p>
 * Simple synchronization between the client(s) and server can be done
 * using the {@link #waitForCondition(String)}, {@link #setCondition(String)} and
 * {@link #rendezvous(String,int)} methods.
 *
 * NOTE NOTE NOTE NOTE NOTE NOTE NOTE
 *
 * If you make a change in here, please don't forget to make the
 * corresponding change in the J2SE equivalent.
 *
 * NOTE NOTE NOTE NOTE NOTE NOTE NOTE
 */

public class TestHttpsServer {

    ServerSocketChannel schan;
    int threads;
    int cperthread;
    HttpCallback cb;
    Server[] servers;

    // ssl related fields
    static SSLContext sslCtx;

    /**
     * Create a <code>TestHttpsServer<code> instance with the specified callback object
     * for handling requests. One thread is created to handle requests,
     * and up to ten TCP connections will be handled simultaneously.
     * @param cb the callback object which is invoked to handle each
     *  incoming request
     */

    public TestHttpsServer(HttpCallback cb) throws IOException {
        this(cb, 1, 10, 0);
    }

    /**
     * Create a <code>TestHttpsServer<code> instance with the specified number of
     * threads and maximum number of connections per thread. This functions
     * the same as the 4 arg constructor, where the port argument is set to zero.
     * @param cb the callback object which is invoked to handle each
     *     incoming request
     * @param threads the number of threads to create to handle requests
     *     in parallel
     * @param cperthread the number of simultaneous TCP connections to
     *     handle per thread
     */

    public TestHttpsServer(HttpCallback cb, int threads, int cperthread)
        throws IOException {
        this(cb, threads, cperthread, 0);
    }

    /**
     * Create a <code>TestHttpsServer<code> instance with the specified number
     * of threads and maximum number of connections per thread and running on
     * the specified port. The specified number of threads are created to
     * handle incoming requests, and each thread is allowed
     * to handle a number of simultaneous TCP connections.
     * @param cb the callback object which is invoked to handle
     *  each incoming request
     * @param threads the number of threads to create to handle
     *  requests in parallel
     * @param cperthread the number of simultaneous TCP connections
     *  to handle per thread
     * @param port the port number to bind the server to. <code>Zero</code>
     *  means choose any free port.
     */
    public TestHttpsServer(HttpCallback cb, int threads, int cperthread, int port)
        throws IOException {
        this(cb, threads, cperthread, null, port);
    }

    /**
     * Create a <code>TestHttpsServer<code> instance with the specified number
     * of threads and maximum number of connections per thread and running on
     * the specified port. The specified number of threads are created to
     * handle incoming requests, and each thread is allowed
     * to handle a number of simultaneous TCP connections.
     * @param cb the callback object which is invoked to handle
     *  each incoming request
     * @param threads the number of threads to create to handle
     *  requests in parallel
     * @param cperthread the number of simultaneous TCP connections
     *  to handle per thread
     * @param address the InetAddress to bind to. {@code Null} means the
     *  wildcard address.
     * @param port the port number to bind the server to. {@code Zero}
     *  means choose any free port.
     */

    public TestHttpsServer(HttpCallback cb, int threads, int cperthread, InetAddress address, int port)
        throws IOException {
        schan = ServerSocketChannel.open();
        InetSocketAddress addr = new InetSocketAddress(address, port);
        schan.socket().bind(addr);
        this.threads = threads;
        this.cb = cb;
        this.cperthread = cperthread;

        try {
            // create and initialize a SSLContext
            KeyStore ks = KeyStore.getInstance("JKS");
            KeyStore ts = KeyStore.getInstance("JKS");
            char[] passphrase = "passphrase".toCharArray();

            ks.load(new FileInputStream(System.getProperty("javax.net.ssl.keyStore")), passphrase);
            ts.load(new FileInputStream(System.getProperty("javax.net.ssl.trustStore")), passphrase);

            KeyManagerFactory kmf = KeyManagerFactory.getInstance("SunX509");
            kmf.init(ks, passphrase);

            TrustManagerFactory tmf = TrustManagerFactory.getInstance("SunX509");
            tmf.init(ts);

            sslCtx = SSLContext.getInstance("TLS");

            sslCtx.init(kmf.getKeyManagers(), tmf.getTrustManagers(), null);

            servers = new Server[threads];
            for (int i=0; i<threads; i++) {
                servers[i] = new Server(cb, schan, cperthread);
                servers[i].start();
            }
        } catch (Exception ex) {
            throw new RuntimeException("test failed. cause: "+ex.getMessage());
        }
    }

    /** Tell all threads in the server to exit within 5 seconds.
     *  This is an abortive termination. Just prior to the thread exiting
     *  all channels in that thread waiting to be closed are forceably closed.
     */

    public void terminate() {
        for (int i=0; i<threads; i++) {
            servers[i].terminate ();
        }
    }

    /**
     * return the local port number to which the server is bound.
     * @return the local port number
     */

    public int getLocalPort () {
        return schan.socket().getLocalPort ();
    }

    public String getAuthority() {
        InetAddress address = schan.socket().getInetAddress();
        String hostaddr = address.getHostAddress();
        if (address.isAnyLocalAddress()) hostaddr = "localhost";
        if (hostaddr.indexOf(':') > -1) hostaddr = "[" + hostaddr + "]";
        return hostaddr + ":" + getLocalPort();
    }

    static class Server extends Thread {

        ServerSocketChannel schan;
        Selector selector;
        SelectionKey listenerKey;
        SelectionKey key; /* the current key being processed */
        HttpCallback cb;
        ByteBuffer consumeBuffer;
        int maxconn;
        int nconn;
        ClosedChannelList clist;
        boolean shutdown;

        Server(HttpCallback cb, ServerSocketChannel schan, int maxconn) {
            this.schan = schan;
            this.maxconn = maxconn;
            this.cb = cb;
            nconn = 0;
            consumeBuffer = ByteBuffer.allocate(512);
            clist = new ClosedChannelList();
            try {
                selector = Selector.open();
                schan.configureBlocking(false);
                listenerKey = schan.register(selector, SelectionKey.OP_ACCEPT);
            } catch (IOException e) {
                System.err.println("Server could not start: " + e);
            }
        }

        /* Stop the thread as soon as possible */
        public synchronized void terminate() {
            shutdown = true;
        }

        public void run()  {
            try {
                while (true) {
                    selector.select(1000);
                    Set selected = selector.selectedKeys();
                    Iterator iter = selected.iterator();
                    while (iter.hasNext()) {
                        key = (SelectionKey)iter.next();
                        if (key.equals (listenerKey)) {
                            SocketChannel sock = schan.accept();
                            if (sock == null) {
                                /* false notification */
                                iter.remove();
                                continue;
                            }
                            sock.configureBlocking(true);
                            SSLEngine sslEng = sslCtx.createSSLEngine();
                            sslEng.setUseClientMode(false);
                            new ServerWorker(cb, sock, sslEng).start();
                            nconn ++;
                            if (nconn == maxconn) {
                                /* deregister */
                                listenerKey.cancel();
                                listenerKey = null;
                            }
                        } else {
                            if (key.isReadable()) {
                                boolean closed = false;
                                SocketChannel chan = (SocketChannel)key.channel();
                                if (key.attachment() != null) {
                                    closed = consume(chan);
                                }

                                if (closed) {
                                    chan.close();
                                    key.cancel();
                                    if (nconn == maxconn) {
                                        listenerKey = schan.register(selector, SelectionKey.OP_ACCEPT);
                                    }
                                    nconn --;
                                }
                            }
                        }
                        iter.remove();
                    }
                    clist.check();

                    synchronized (this) {
                        if (shutdown) {
                            clist.terminate();
                            return;
                        }
                    }
                }
            } catch (IOException e) {
                System.out.println("Server exception: " + e);
                // TODO finish
            }
        }

        /* read all the data off the channel without looking at it
         * return true if connection closed
         */
        boolean consume(SocketChannel chan) {
            try {
                consumeBuffer.clear();
                int c = chan.read(consumeBuffer);
                if (c == -1)
                    return true;
            } catch (IOException e) {
                return true;
            }
            return false;
        }
    }

    static class ServerWorker extends Thread {
        private ByteBuffer inNetBB;
        private ByteBuffer outNetBB;
        private ByteBuffer inAppBB;
        private ByteBuffer outAppBB;

        SSLEngine sslEng;
        SocketChannel schan;
        HttpCallback cb;
        HandshakeStatus currentHSStatus;
        boolean initialHSComplete;
        boolean handshakeStarted;
        /*
         * All inbound data goes through this buffer.
         *
         * It might be nice to use a cache of ByteBuffers so we're
         * not alloc/dealloc'ing all over the place.
         */

        /*
         * Application buffers, also used for handshaking
         */
        private int appBBSize;

        ServerWorker(HttpCallback cb, SocketChannel schan, SSLEngine sslEng) {
            this.sslEng = sslEng;
            this.schan = schan;
            this.cb = cb;
            currentHSStatus = HandshakeStatus.NEED_UNWRAP;
            initialHSComplete = false;
            int netBBSize = sslEng.getSession().getPacketBufferSize();
            inNetBB =  ByteBuffer.allocate(netBBSize);
            outNetBB = ByteBuffer.allocate(netBBSize);
            appBBSize = sslEng.getSession().getApplicationBufferSize();
            inAppBB = ByteBuffer.allocate(appBBSize);
            outAppBB = ByteBuffer.allocate(appBBSize);
        }

        public SSLEngine getSSLEngine() {
            return sslEng;
        }

        public ByteBuffer outNetBB() {
            return outNetBB;
        }

        public ByteBuffer outAppBB() {
            return outAppBB;
        }

        public void run () {
            try {
                SSLEngineResult result;

                while (!initialHSComplete) {

                    switch (currentHSStatus) {

                    case NEED_UNWRAP:
                        int bytes = schan.read(inNetBB);
                        if (!handshakeStarted && bytes > 0) {
                            handshakeStarted = true;
                            int byte0 = inNetBB.get(0);
                            if (byte0 != 0x16) {
                                // first byte of a TLS connection is supposed to be
                                // 0x16. If not it may be a plain text connection.
                                //
                                // Sometime a rogue client may try to open a plain
                                // connection with our server. Calling this method
                                // gives a chance to the test logic to ignore such
                                // rogue connections.
                                //
                                if (cb.dropPlainTextConnections()) {
                                    try { schan.close(); } catch (IOException x) { };
                                    return;
                                }
                                // else sslEng.unwrap will throw later on...
                            }
                        }

needIO:
                        while (currentHSStatus == HandshakeStatus.NEED_UNWRAP) {
                            /*
                             * Don't need to resize requestBB, since no app data should
                             * be generated here.
                             */
                            inNetBB.flip();
                            result = sslEng.unwrap(inNetBB, inAppBB);
                            inNetBB.compact();
                            currentHSStatus = result.getHandshakeStatus();

                            switch (result.getStatus()) {

                            case OK:
                                switch (currentHSStatus) {
                                case NOT_HANDSHAKING:
                                    throw new IOException(
                                                          "Not handshaking during initial handshake");

                                case NEED_TASK:
                                    Runnable task;
                                    while ((task = sslEng.getDelegatedTask()) != null) {
                                        task.run();
                                        currentHSStatus = sslEng.getHandshakeStatus();
                                    }
                                    break;
                                }

                                break;

                            case BUFFER_UNDERFLOW:
                                break needIO;

                            default: // BUFFER_OVERFLOW/CLOSED:
                                throw new IOException("Received" + result.getStatus() +
                                                      "during initial handshaking");
                            }
                        }

                        /*
                         * Just transitioned from read to write.
                         */
                        if (currentHSStatus != HandshakeStatus.NEED_WRAP) {
                            break;
                        }

                        // Fall through and fill the write buffer.

                    case NEED_WRAP:
                        /*
                         * The flush above guarantees the out buffer to be empty
                         */
                        outNetBB.clear();
                        result = sslEng.wrap(inAppBB, outNetBB);
                        outNetBB.flip();
                        schan.write (outNetBB);
                        outNetBB.compact();
                        currentHSStatus = result.getHandshakeStatus();

                        switch (result.getStatus()) {
                        case OK:

                            if (currentHSStatus == HandshakeStatus.NEED_TASK) {
                                Runnable task;
                                while ((task = sslEng.getDelegatedTask()) != null) {
                                    task.run();
                                    currentHSStatus = sslEng.getHandshakeStatus();
                                }
                            }

                            break;

                        default: // BUFFER_OVERFLOW/BUFFER_UNDERFLOW/CLOSED:
                            throw new IOException("Received" + result.getStatus() +
                                                  "during initial handshaking");
                        }
                        break;

                    case FINISHED:
                        initialHSComplete = true;
                        break;
                    default: // NOT_HANDSHAKING/NEED_TASK
                        throw new RuntimeException("Invalid Handshaking State" +
                                                   currentHSStatus);
                    } // switch
                }
                // read the application data; using non-blocking mode
                schan.configureBlocking(false);
                read(schan, sslEng);
            } catch (Exception ex) {
                throw new RuntimeException(ex);
            }
        }

        /* return true if the connection is closed, false otherwise */

        private boolean read(SocketChannel chan, SSLEngine sslEng) {
            HttpTransaction msg;
            boolean res;
            try {
                InputStream is = new BufferedInputStream(new NioInputStream(chan, sslEng, inNetBB, inAppBB));
                String requestline = readLine(is);
                MessageHeader mhead = new MessageHeader(is);
                String clen = mhead.findValue("Content-Length");
                String trferenc = mhead.findValue("Transfer-Encoding");
                String data = null;
                if (trferenc != null && trferenc.equals("chunked"))
                    data = new String(readChunkedData(is));
                else if (clen != null)
                    data = new String(readNormalData(is, Integer.parseInt(clen)));
                String[] req = requestline.split(" ");
                if (req.length < 2) {
                    /* invalid request line */
                    return false;
                }
                String cmd = req[0];
                URI uri = null;
                try {
                    uri = new URI(req[1]);
                    msg = new HttpTransaction(this, cmd, uri, mhead, data, null, chan);
                    cb.request(msg);
                } catch (URISyntaxException e) {
                    System.err.println ("Invalid URI: " + e);
                    msg = new HttpTransaction(this, cmd, null, null, null, null, chan);
                    msg.sendResponse(501, "Whatever");
                }
                res = false;
            } catch (IOException e) {
                res = true;
            }
            return res;
        }

        byte[] readNormalData(InputStream is, int len) throws IOException {
            byte[] buf  = new byte[len];
            int c, off=0, remain=len;
            while (remain > 0 && ((c=is.read (buf, off, remain))>0)) {
                remain -= c;
                off += c;
            }
            return buf;
        }

        private void readCRLF(InputStream is) throws IOException {
            int cr = is.read();
            int lf = is.read();

            if (((cr & 0xff) != 0x0d) ||
                ((lf & 0xff) != 0x0a)) {
                throw new IOException(
                    "Expected <CR><LF>:  got '" + cr + "/" + lf + "'");
            }
        }

        byte[] readChunkedData(InputStream is) throws IOException {
            LinkedList l = new LinkedList();
            int total = 0;
            for (int len=readChunkLen(is); len!=0; len=readChunkLen(is)) {
                l.add(readNormalData(is, len));
                total += len;
                readCRLF(is); // CRLF at end of chunk
            }
            readCRLF(is); // CRLF at end of Chunked Stream.
            byte[] buf = new byte[total];
            Iterator i = l.iterator();
            int x = 0;
            while (i.hasNext()) {
                byte[] b = (byte[])i.next();
                System.arraycopy(b, 0, buf, x, b.length);
                x += b.length;
            }
            return buf;
        }

        private int readChunkLen(InputStream is) throws IOException {
            int c, len=0;
            boolean done=false, readCR=false;
            while (!done) {
                c = is.read();
                if (c == '\n' && readCR) {
                    done = true;
                } else {
                    if (c == '\r' && !readCR) {
                        readCR = true;
                    } else {
                        int x=0;
                        if (c >= 'a' && c <= 'f') {
                            x = c - 'a' + 10;
                        } else if (c >= 'A' && c <= 'F') {
                            x = c - 'A' + 10;
                        } else if (c >= '0' && c <= '9') {
                            x = c - '0';
                        }
                        len = len * 16 + x;
                    }
                }
            }
            return len;
        }

        private String readLine(InputStream is) throws IOException {
            boolean done=false, readCR=false;
            byte[] b = new byte[512];
            int c, l = 0;

            while (!done) {
                c = is.read();
                if (c == '\n' && readCR) {
                    done = true;
                } else {
                    if (c == '\r' && !readCR) {
                        readCR = true;
                    } else {
                        b[l++] = (byte)c;
                    }
                }
            }
            return new String(b);
        }

        /** close the channel associated with the current key by:
         * 1. shutdownOutput (send a FIN)
         * 2. mark the key so that incoming data is to be consumed and discarded
         * 3. After a period, close the socket
         */

        synchronized void orderlyCloseChannel(SocketChannel ch) throws IOException {
            ch.socket().shutdownOutput();
        }

        synchronized void abortiveCloseChannel(SocketChannel ch) throws IOException {
            Socket s = ch.socket();
            s.setSoLinger(true, 0);
            ch.close();
        }
    }


    /**
     * Implements blocking reading semantics on top of a non-blocking channel
     */

    static class NioInputStream extends InputStream {
        SSLEngine sslEng;
        SocketChannel channel;
        Selector selector;
        ByteBuffer inNetBB;
        ByteBuffer inAppBB;
        SelectionKey key;
        int available;
        byte[] one;
        boolean closed;
        ByteBuffer markBuf; /* reads may be satisifed from this buffer */
        boolean marked;
        boolean reset;
        int readlimit;

        public NioInputStream(SocketChannel chan, SSLEngine sslEng, ByteBuffer inNetBB, ByteBuffer inAppBB) throws IOException {
            this.sslEng = sslEng;
            this.channel = chan;
            selector = Selector.open();
            this.inNetBB = inNetBB;
            this.inAppBB = inAppBB;
            key = chan.register(selector, SelectionKey.OP_READ);
            available = 0;
            one = new byte[1];
            closed = marked = reset = false;
        }

        public synchronized int read(byte[] b) throws IOException {
            return read(b, 0, b.length);
        }

        public synchronized int read() throws IOException {
            return read(one, 0, 1);
        }

        public synchronized int read(byte[] b, int off, int srclen) throws IOException {

            int canreturn, willreturn;

            if (closed)
                return -1;

            if (reset) { /* satisfy from markBuf */
                canreturn = markBuf.remaining();
                willreturn = canreturn > srclen ? srclen : canreturn;
                markBuf.get(b, off, willreturn);
                if (canreturn == willreturn) {
                    reset = false;
                }
            } else { /* satisfy from channel */
                canreturn = available();
                if (canreturn == 0) {
                    block();
                    canreturn = available();
                }
                willreturn = canreturn > srclen ? srclen : canreturn;
                inAppBB.get(b, off, willreturn);
                available -= willreturn;

                if (marked) { /* copy into markBuf */
                    try {
                        markBuf.put(b, off, willreturn);
                    } catch (BufferOverflowException e) {
                        marked = false;
                    }
                }
            }
            return willreturn;
        }

        public synchronized int available() throws IOException {
            if (closed)
                throw new IOException("Stream is closed");

            if (reset)
                return markBuf.remaining();

            if (available > 0)
                return available;

            inAppBB.clear();
            int bytes = channel.read(inNetBB);

            int needed = sslEng.getSession().getApplicationBufferSize();
            if (needed > inAppBB.remaining()) {
                inAppBB = ByteBuffer.allocate(needed);
            }
            inNetBB.flip();
            SSLEngineResult result = sslEng.unwrap(inNetBB, inAppBB);
            inNetBB.compact();
            available = result.bytesProduced();

            if (available > 0)
                inAppBB.flip();
            else if (available == -1)
                throw new IOException("Stream is closed");
            return available;
        }

        /**
         * block() only called when available==0 and buf is empty
         */
        private synchronized void block() throws IOException {
            //assert available == 0;
            int n = selector.select();
            //assert n == 1;
            selector.selectedKeys().clear();
            available();
        }

        public void close() throws IOException {
            if (closed)
                return;
            channel.close();
            closed = true;
        }

        public synchronized void mark(int readlimit) {
            if (closed)
                return;
            this.readlimit = readlimit;
            markBuf = ByteBuffer.allocate(readlimit);
            marked = true;
            reset = false;
        }

        public synchronized void reset() throws IOException {
            if (closed )
                return;
            if (!marked)
                throw new IOException("Stream not marked");
            marked = false;
            reset = true;
            markBuf.flip();
        }
    }

    static class NioOutputStream extends OutputStream {
        SSLEngine sslEng;
        SocketChannel channel;
        ByteBuffer outNetBB;
        ByteBuffer outAppBB;
        SelectionKey key;
        Selector selector;
        boolean closed;
        byte[] one;

        public NioOutputStream(SocketChannel channel, SSLEngine sslEng, ByteBuffer outNetBB, ByteBuffer outAppBB) throws IOException {
            this.sslEng = sslEng;
            this.channel = channel;
            this.outNetBB = outNetBB;
            this.outAppBB = outAppBB;
            selector = Selector.open();
            key = channel.register(selector, SelectionKey.OP_WRITE);
            closed = false;
            one = new byte[1];
        }

        public synchronized void write(int b) throws IOException {
            one[0] = (byte)b;
            write(one, 0, 1);
        }

        public synchronized void write(byte[] b) throws IOException {
            write(b, 0, b.length);
        }

        public synchronized void write(byte[] b, int off, int len) throws IOException {
            if (closed)
                throw new IOException("stream is closed");

            outAppBB = ByteBuffer.allocate(len);
            outAppBB.put(b, off, len);
            outAppBB.flip();
            int n;
            outNetBB.clear();
            int needed = sslEng.getSession().getPacketBufferSize();
            if (outNetBB.capacity() < needed) {
                outNetBB = ByteBuffer.allocate(needed);
            }
            SSLEngineResult ret = sslEng.wrap(outAppBB, outNetBB);
            outNetBB.flip();
            int newLen = ret.bytesProduced();
            while ((n = channel.write (outNetBB)) < newLen) {
                newLen -= n;
                if (newLen == 0)
                    return;
                selector.select();
                selector.selectedKeys().clear();
            }
        }

        public void close() throws IOException {
            if (closed)
                return;
            channel.close();
            closed = true;
        }
    }

    /**
     * Utilities for synchronization. A condition is
     * identified by a string name, and is initialized
     * upon first use (ie. setCondition() or waitForCondition()). Threads
     * are blocked until some thread calls (or has called) setCondition() for the same
     * condition.
     * <P>
     * A rendezvous built on a condition is also provided for synchronizing
     * N threads.
     */

    private static HashMap conditions = new HashMap();

    /*
     * Modifiable boolean object
     */
    private static class BValue {
        boolean v;
    }

    /*
     * Modifiable int object
     */
    private static class IValue {
        int v;
        IValue(int i) {
            v =i;
        }
    }


    private static BValue getCond(String condition) {
        synchronized (conditions) {
            BValue cond = (BValue) conditions.get(condition);
            if (cond == null) {
                cond = new BValue();
                conditions.put(condition, cond);
            }
            return cond;
        }
    }

    /**
     * Set the condition to true. Any threads that are currently blocked
     * waiting on the condition, will be unblocked and allowed to continue.
     * Threads that subsequently call waitForCondition() will not block.
     * If the named condition did not exist prior to the call, then it is created
     * first.
     */

    public static void setCondition(String condition) {
        BValue cond = getCond(condition);
        synchronized (cond) {
            if (cond.v) {
                return;
            }
            cond.v = true;
            cond.notifyAll();
        }
    }

    /**
     * If the named condition does not exist, then it is created and initialized
     * to false. If the condition exists or has just been created and its value
     * is false, then the thread blocks until another thread sets the condition.
     * If the condition exists and is already set to true, then this call returns
     * immediately without blocking.
     */

    public static void waitForCondition(String condition) {
        BValue cond = getCond(condition);
        synchronized (cond) {
            if (!cond.v) {
                try {
                    cond.wait();
                } catch (InterruptedException e) {}
            }
        }
    }

    /* conditions must be locked when accessing this */
    static HashMap rv = new HashMap();

    /**
     * Force N threads to rendezvous (ie. wait for each other) before proceeding.
     * The first thread(s) to call are blocked until the last
     * thread makes the call. Then all threads continue.
     * <p>
     * All threads that call with the same condition name, must use the same value
     * for N (or the results may be not be as expected).
     * <P>
     * Obviously, if fewer than N threads make the rendezvous then the result
     * will be a hang.
     */

    public static void rendezvous(String condition, int N) {
        BValue cond;
        IValue iv;
        String name = "RV_"+condition;

        /* get the condition */

        synchronized (conditions) {
            cond = (BValue)conditions.get(name);
            if (cond == null) {
                /* we are first caller */
                if (N < 2) {
                    throw new RuntimeException("rendezvous must be called with N >= 2");
                }
                cond = new BValue();
                conditions.put(name, cond);
                iv = new IValue(N-1);
                rv.put(name, iv);
            } else {
                /* already initialised, just decrement the counter */
                iv = (IValue) rv.get(name);
                iv.v--;
            }
        }

        if (iv.v > 0) {
            waitForCondition(name);
        } else {
            setCondition(name);
            synchronized (conditions) {
                clearCondition(name);
                rv.remove(name);
            }
        }
    }

    /**
     * If the named condition exists and is set then remove it, so it can
     * be re-initialized and used again. If the condition does not exist, or
     * exists but is not set, then the call returns without doing anything.
     * Note, some higher level synchronization
     * may be needed between clear and the other operations.
     */

    public static void clearCondition(String condition) {
        BValue cond;
        synchronized (conditions) {
            cond = (BValue) conditions.get(condition);
            if (cond == null) {
                return;
            }
            synchronized (cond) {
                if (cond.v) {
                    conditions.remove(condition);
                }
            }
        }
    }
}