title: Resettable CountdownLatch
body: <p>I need something which is directly equivalent to <code>CountDownLatch</code>, but is resettable (remaining thread-safe!). I can't use classic synchronisation constructs as they simply don't work in this situation (complex locking issues). At the moment, I'm creating many <code>CountDownLatch</code> objects, each replacing the previous one. I believe this is doing in the young generation in the GC (due to the sheer number of objects). You can see the code which uses the latches below (it's part of the <code>java.net</code> mock for a ns-3 network simulator interface).</p>
  
  <p>Some ideas might be to try <code>CyclicBarrier</code> (JDK5+) or <code>Phaser</code> (JDK7)</p>
  
  <p>I can test code and get back to anyone that finds a solution to this problem, since I'm the only one who can insert it into the running system to see what happens :)</p>
  
  <pre><code>/**
   *
   */
  package kokunet;
  
  import java.io.IOException;
  import java.nio.channels.ClosedSelectorException;
  import java.util.HashMap;
  import java.util.Map;
  import java.util.concurrent.CountDownLatch;
  import java.util.concurrent.TimeUnit;
  import kokuks.IConnectionSocket;
  import kokuks.KKSAddress;
  import kokuks.KKSSocket;
  import kokuks.KKSSocketListener;
  
  /**
   * KSelector
   * @version 1.0
   * @author Chris Dennett
   */
  public class KSelector extends SelectorImpl {
      // True if this Selector has been closed
      private volatile boolean closed = false;
  
      // Lock for close and cleanup
      final class CloseLock {}
      private final Object closeLock = new CloseLock();
  
      private volatile boolean selecting = false;
      private volatile boolean wakeup = false;
  
      class SocketListener implements KKSSocketListener {
          protected volatile CountDownLatch latch = null;
  
          /**
           *
           */
          public SocketListener() {
              newLatch();
          }
  
          protected synchronized CountDownLatch newLatch() {
              return this.latch = new CountDownLatch(1);
          }
  
          protected synchronized void refreshReady(KKSSocket socket) {
              if (!selecting) return;
  
              synchronized (socketToChannel) {
                  SelChImpl ch = socketToChannel.get(socket);
                  if (ch == null) {
                      System.out.println("ks sendCB: channel not found for socket: " + socket);
                      return;
                  }
                  synchronized (channelToKey) {
                      SelectionKeyImpl sk = channelToKey.get(ch);
                      if (sk != null) {
                          if (handleSelect(sk)) {
                              latch.countDown();
                          }
                      }
                  }
              }
          }
          @Override
          public void connectionSucceeded(KKSSocket socket) {
              refreshReady(socket);
          }
          @Override
          public void connectionFailed(KKSSocket socket) {
              refreshReady(socket);
          }
          @Override
          public void dataSent(KKSSocket socket, long bytesSent) {
              refreshReady(socket);
          }
          @Override
          public void sendCB(KKSSocket socket, long bytesAvailable) {
              refreshReady(socket);
          }
          @Override
          public void onRecv(KKSSocket socket) {
              refreshReady(socket);
          }
          @Override
          public void newConnectionCreated(KKSSocket socket, KKSSocket newSocket, KKSAddress remoteaddress) {
              refreshReady(socket);
          }
          @Override
          public void normalClose(KKSSocket socket) {
              wakeup();
          }
          @Override
          public void errorClose(KKSSocket socket) {
              wakeup();
          }
      }
  
      protected final Map&lt;KKSSocket, SelChImpl&gt;        socketToChannel = new HashMap&lt;KKSSocket, SelChImpl&gt;();
      protected final Map&lt;SelChImpl, SelectionKeyImpl&gt; channelToKey    = new HashMap&lt;SelChImpl, SelectionKeyImpl&gt;();
      protected final SocketListener currListener = new SocketListener();
      protected Thread selectingThread = null;
  
      SelChImpl getChannelForSocket(KKSSocket s) {
          synchronized (socketToChannel) {
              return socketToChannel.get(s);
          }
      }
  
      SelectionKeyImpl getSelKeyForChannel(KKSSocket s) {
          synchronized (channelToKey) {
              return channelToKey.get(s);
          }
      }
  
      protected boolean markRead(SelectionKeyImpl impl) {
          synchronized (impl) {
              if (!impl.isValid()) return false;
              impl.nioReadyOps(impl.readyOps() | SelectionKeyImpl.OP_READ);
              return selectedKeys.add(impl);
          }
      }
  
      protected boolean markWrite(SelectionKeyImpl impl) {
          synchronized (impl) {
              if (!impl.isValid()) return false;
              impl.nioReadyOps(impl.readyOps() | SelectionKeyImpl.OP_WRITE);
              return selectedKeys.add(impl);
          }
      }
  
      protected boolean markAccept(SelectionKeyImpl impl) {
          synchronized (impl) {
              if (!impl.isValid()) return false;
              impl.nioReadyOps(impl.readyOps() | SelectionKeyImpl.OP_ACCEPT);
              return selectedKeys.add(impl);
          }
      }
  
      protected boolean markConnect(SelectionKeyImpl impl) {
          synchronized (impl) {
              if (!impl.isValid()) return false;
              impl.nioReadyOps(impl.readyOps() | SelectionKeyImpl.OP_CONNECT);
              return selectedKeys.add(impl);
          }
      }
  
      /**
       * @param provider
       */
      protected KSelector(SelectorProvider provider) {
          super(provider);
      }
  
      /* (non-Javadoc)
       * @see kokunet.SelectorImpl#implClose()
       */
      @Override
      protected void implClose() throws IOException {
          provider().getApp().printMessage("implClose: closed: " + closed);
          synchronized (closeLock) {
              if (closed) return;
              closed = true;
              for (SelectionKey sk : keys) {
                  provider().getApp().printMessage("dereg1");
                  deregister((AbstractSelectionKey)sk);
                  provider().getApp().printMessage("dereg2");
                  SelectableChannel selch = sk.channel();
                  if (!selch.isOpen() &amp;&amp; !selch.isRegistered())
                      ((SelChImpl)selch).kill();
              }
              implCloseInterrupt();
          }
      }
  
      protected void implCloseInterrupt() {
          wakeup();
      }
  
      private boolean handleSelect(SelectionKey k) {
          synchronized (k) {
              boolean notify = false;
  
              if (!k.isValid()) {
                  k.cancel();
                  ((SelectionKeyImpl)k).channel.socket().removeListener(currListener);
                  return false;
              }
  
              SelectionKeyImpl ski = (SelectionKeyImpl)k;
  
              if ((ski.interestOps() &amp; SelectionKeyImpl.OP_READ) != 0) {
                  if (ski.channel.socket().getRxAvailable() &gt; 0) {
                      notify |= markRead(ski);
                  }
              }
  
              if ((ski.interestOps() &amp; SelectionKeyImpl.OP_WRITE) != 0) {
                  if (ski.channel.socket().getTxAvailable() &gt; 0) {
                      notify |= markWrite(ski);
                  }
              }
  
              if ((ski.interestOps() &amp; SelectionKeyImpl.OP_CONNECT) != 0) {
                  if (!ski.channel.socket().isConnectionless()) {
                      IConnectionSocket cs = (IConnectionSocket)ski.channel.socket();
                      if (!ski.channel.socket().isAccepting() &amp;&amp; !cs.isConnecting() &amp;&amp; !cs.isConnected()) {
                          notify |= markConnect(ski);
                      }
                  }
              }
  
              if ((ski.interestOps() &amp; SelectionKeyImpl.OP_ACCEPT) != 0) {
                  //provider().getApp().printMessage("accept check: ski: " + ski + ", connectionless: " + ski.channel.socket().isConnectionless() + ", listening: " + ski.channel.socket().isListening() + ", hasPendingConn: " + (ski.channel.socket().isConnectionless() ? "nope!" : ((IConnectionSocket)ski.channel.socket()).hasPendingConnections()));
                  if (!ski.channel.socket().isConnectionless() &amp;&amp; ski.channel.socket().isListening()) {
                      IConnectionSocket cs = (IConnectionSocket)ski.channel.socket();
                      if (cs.hasPendingConnections()) {
                          notify |= markAccept(ski);
                      }
                  }
              }
              return notify;
          }
      }
  
      private boolean handleSelect() {
          boolean notify = false;
  
          // get initial status
          for (SelectionKey k : keys) {
              notify |= handleSelect(k);
          }
  
          return notify;
      }
  
      /* (non-Javadoc)
       * @see kokunet.SelectorImpl#doSelect(long)
       */
      @Override
      protected int doSelect(long timeout) throws IOException {
          processDeregisterQueue();
  
          long timestartedms = System.currentTimeMillis();
  
          synchronized (selectedKeys) {
              synchronized (currListener) {
                  wakeup = false;
                  selectingThread = Thread.currentThread();
                  selecting = true;
              }
              try {
                  handleSelect();
  
                  if (!selectedKeys.isEmpty() || timeout == 0) {
                      return selectedKeys.size();
                  }
  
                  //TODO: useless op if we have keys available
                  for (SelectionKey key : keys) {
                      ((SelectionKeyImpl)key).channel.socket().addListener(currListener);
                  }
                  try {
                      while (!wakeup &amp;&amp; isOpen() &amp;&amp; selectedKeys.isEmpty()) {
                          CountDownLatch latch = null;
                          synchronized (currListener) {
                              if (wakeup || !isOpen() || !selectedKeys.isEmpty()) {
                                  break;
                              }
                              latch = currListener.newLatch();
                          }
                          try {
                              if (timeout &gt; 0) {
                                  long currtimems = System.currentTimeMillis();
                                  long remainingMS = (timestartedms + timeout) - currtimems;
  
                                  if (remainingMS &gt; 0) {
                                      latch.await(remainingMS, TimeUnit.MILLISECONDS);
                                  } else {
                                      break;
                                  }
                              } else {
                                  latch.await();
                              }
                          } catch (InterruptedException e) {
  
                          }
                      }
                      return selectedKeys.size();
                  } finally {
                      for (SelectionKey key : keys) {
                          ((SelectionKeyImpl)key).channel.socket().removeListener(currListener);
                      }
                  }
              } finally {
                  synchronized (currListener) {
                      selecting = false;
                      selectingThread = null;
                      wakeup = false;
                  }
              }
          }
      }
  
      /* (non-Javadoc)
       * @see kokunet.SelectorImpl#implRegister(kokunet.SelectionKeyImpl)
       */
      @Override
      protected void implRegister(SelectionKeyImpl ski) {
          synchronized (closeLock) {
              if (closed) throw new ClosedSelectorException();
              synchronized (channelToKey) {
                  synchronized (socketToChannel) {
                      keys.add(ski);
                      socketToChannel.put(ski.channel.socket(), ski.channel);
                      channelToKey.put(ski.channel, ski);
                  }
              }
          }
  
      }
  
      /* (non-Javadoc)
       * @see kokunet.SelectorImpl#implDereg(kokunet.SelectionKeyImpl)
       */
      @Override
      protected void implDereg(SelectionKeyImpl ski) throws IOException {
          synchronized (channelToKey) {
              synchronized (socketToChannel) {
                  keys.remove(ski);
                  socketToChannel.remove(ski.channel.socket());
                  channelToKey.remove(ski.channel);
  
                  SelectableChannel selch = ski.channel();
  
                  if (!selch.isOpen() &amp;&amp; !selch.isRegistered())
                      ((SelChImpl)selch).kill();
              }
          }
      }
  
      /* (non-Javadoc)
       * @see kokunet.SelectorImpl#wakeup()
       */
      @Override
      public Selector wakeup() {
          synchronized (currListener) {
              if (selecting) {
                  wakeup = true;
                  selecting = false;
                  selectingThread.interrupt();
                  selectingThread = null;
              }
          }
          return this;
      }
  }
  </code></pre>
  
  <p>Cheers,<br>
  Chris</p>