Finish the connect phase and potentially verify its liveness.

@param sk the selection key for the node.
@param node the actual node.
@throws IOException if something goes wrong during reading/writing.
963486
private void finishConnect(final SelectionKey sk, final MemcachedNode node)
    throws IOException {
    if (verifyAliveOnConnect) {
      final CountDownLatch latch = new CountDownLatch(1);
      final OperationFuture<Boolean> rv = new OperationFuture<Boolean>("noop",
        latch, 2500, listenerExecutorService);
      NoopOperation testOp = opFact.noop(new OperationCallback() {
        public void receivedStatus(OperationStatus status) {
          rv.set(status.isSuccess(), status);
        }

        @Override
        public void complete() {
          latch.countDown();
        }
      });

      testOp.setHandlingNode(node);
      testOp.initialize();
      checkState();
      insertOperation(node, testOp);
      node.copyInputQueue();

      boolean done = false;
      if (sk.isValid()) {
        long timeout = TimeUnit.MILLISECONDS.toNanos(
          connectionFactory.getOperationTimeout());

        long stop = System.nanoTime() + timeout;
        while (stop > System.nanoTime()) {
          handleWrites(node);
          handleReads(node);
          if(done = (latch.getCount() == 0)) {
            break;
          }
        }
      }

      if (!done || testOp.isCancelled() || testOp.hasErrored()
        || testOp.isTimedOut()) {
        throw new ConnectException("Could not send noop upon connect! "
          + "This may indicate a running, but not responding memcached "
          + "instance.");
      }
    }

    connected(node);
    addedQueue.offer(node);
    if (node.getWbuf().hasRemaining()) {
      handleWrites(node);
    }
  }
884581
private void processNode(Node node) throws IOException {
		for (Node cursor = node.getFirstChild(); cursor != null; cursor = cursor.getNext()) {
			int lineno = cursor.getLineno()-1;	// adjust for rhino line numbers being 1 based
			int charno = cursor.getCharno();
			if (lineno > currentLine || lineno == currentLine && charno > currentChar) {
				currentLine = lineno;
				currentChar = charno;
			}
			FilePosition endPosition = getEndPosition(cursor);
			sgen.addMapping(name, null,
					new FilePosition(currentLine, currentChar),
					new FilePosition(currentLine, currentChar),
					endPosition);
			if (cursor.hasChildren()) {
				processNode(cursor);
			}
		}
	}
434230
private void link(Node<K, V> y, Node<K, V> x) {
        // remove from root list
        y.prev.next = y.next;
        y.next.prev = y.prev;

        // one less root
        roots--;

        // clear if marked
        y.mark = false;

        // hang as x's child
        x.degree++;
        y.parent = x;

        Node<K, V> child = x.child;
        if (child == null) {
            x.child = y;
            y.next = y;
            y.prev = y;
        } else {
            y.prev = child;
            y.next = child.next;
            child.next = y;
            y.next.prev = y;
        }
    }
1081253
void connectTo(final Node<T> o) {
    if (!this.graph.equals(o.graph))
      throw new IllegalArgumentException("Cannot connect to nodes of two different graphs.");
    if (this.equals(o)) {
      return;
    }
    neighbours.add(o);
  }
1320995
private void connected(final MemcachedNode node) {
    assert node.getChannel().isConnected() : "Not connected.";
    int rt = node.getReconnectCount();
    node.connected();

    for (ConnectionObserver observer : connObservers) {
      observer.connectionEstablished(node.getSocketAddress(), rt);
    }
  }
290160
public void onNodeStarted(final IExecutionNode node) {

        final Description description = descriptionMap.get(Long.valueOf(node.getId()));

        final boolean b = description != null;
        log.debug("notifyTestStarted nodeid: " + node.getId() + " description: " + b);

        notifyTestStarted(description);

    }
928977
public void pushNode(Node node) {
        initNode(node);

        // Reset in stream
        inStream = null;

        synchronized (nodeStack) {

            // Clear popped stack
            poppedNodes.clear();

            // Check if fragment is in suppression mode
            if (suppress) {
                suppressedNodeStack.push(node);
                return;
            }

            if (nodeStack.isEmpty()) {
                if (log.isLoggable(Level.FINEST)) {
                    log.finest("Pushing top level node: " + node + " for txn: " + trace);
                }
                trace.getNodes().add(node);
            } else {
                Node parent = nodeStack.peek();

                if (parent instanceof ContainerNode) {
                    if (log.isLoggable(Level.FINEST)) {
                        log.finest("Add node: " + node + " to parent: " + parent + " in txn: " + trace);
                    }
                    ((ContainerNode) parent).getNodes().add(node);
                } else {
                    log.severe("Attempt to add node '" + node + "' under non-container node '" + parent + "'");
                }
            }
            nodeStack.push(node);
        }
    }
89067
private void handleReadsAndWrites(final SelectionKey sk, final MemcachedNode node) throws IOException {
      if (sk.isValid() && sk.isReadable()) {
          handleReads(node);
      }

      if (sk.isValid() && sk.isWritable()) {
          handleWrites(node);
      }
  }
315039
public void onNodeStarted(final IExecutionNode node) {

        if (!node.equals(this.theLastNode)) {

            if (this.theLastNode != null) {
                printStarted("Starting: " + format(this.theLastNode));
            }

            this.theLastNode = node;
        } else {
            // not sure how this would happen
            printStarted("Starting: " + format(node));
        }

    }
421487
public void insertOperation(final MemcachedNode node, final Operation o) {
    o.setHandlingNode(node);
    o.initialize();
    node.insertOp(o);
    addedQueue.offer(node);
    metrics.markMeter(OVERALL_REQUEST_METRIC);

    Selector s = selector.wakeup();
    assert s == selector : "Wakeup returned the wrong selector.";
    getLogger().debug("Added %s to %s", o, node);
  }
1269200
public void nodeSelected(SiteNode node) {
		if (node != null) {
			siteModel.setSelectedItem(ScanPanel.cleanSiteName(node, true));
		}
	}
1146735
void afterRead(Node<K, V> node) {
        final int bufferIndex = readBufferIndex();
        final long writeCount = recordRead(bufferIndex, node);
        drainOnReadIfNeeded(bufferIndex, writeCount);
        notifyListener();
    }
386355
public void serialize(Node node) throws IOException
    {

        try
        {
            TreeWalker walker =
                new TreeWalker(this);

            walker.traverse(node);
        }
        catch (org.xml.sax.SAXException se)
        {
            throw new WrappedRuntimeException(se);
        }
    }
1215928
private void delete(Node<K, V> n) {
        if (root == n) {
            deleteMin();
            n.o_c = null;
            n.y_s = null;
            n.o_s = null;
            return;
        }

        if (n.o_s == null) {
            throw new IllegalArgumentException("Invalid handle!");
        }

        // unlink from parent
        if (n.y_s != null) {
            n.y_s.o_s = n.o_s;
        }
        if (n.o_s.o_c == n) { // I am the oldest :(
            n.o_s.o_c = n.y_s;
        } else { // I have an older sibling!
            n.o_s.y_s = n.y_s;
        }
        n.y_s = null;
        n.o_s = null;

        // perform delete-min at tree rooted at this
        Node<K, V> t = combine(cutChildren(n));

        // and merge with other cut tree
        if (comparator == null) {
            root = link(root, t);
        } else {
            root = linkWithComparator(root, t);
        }

        size--;
    }
1267076
private void handleKey(SelectionKey key) throws IOException {
        // We could have a !isValid() key here if the connection is already closed at this point
        if (key.isValid() && key.isConnectable()) { // ie a client connection which has finished the initial connect process
            // Create a ConnectionHandler and hook everything together
            PendingConnect data = (PendingConnect) key.attachment();
            StreamConnection connection = data.connection;
            SocketChannel sc = (SocketChannel) key.channel();
            ConnectionHandler handler = new ConnectionHandler(connection, key, connectedHandlers);
            try {
                if (sc.finishConnect()) {
                    log.info("Connected to {}", sc.socket().getRemoteSocketAddress());
                    key.interestOps((key.interestOps() | SelectionKey.OP_READ) & ~SelectionKey.OP_CONNECT).attach(handler);
                    connection.connectionOpened();
                    data.future.set(data.address);
                } else {
                    log.warn("Failed to connect to {}", sc.socket().getRemoteSocketAddress());
                    handler.closeConnection(); // Failed to connect for some reason
                    data.future.setException(new ConnectException("Unknown reason"));
                    data.future = null;
                }
            } catch (Exception e) {
                // If e is a CancelledKeyException, there is a race to get to interestOps after finishConnect() which
                // may cause this. Otherwise it may be any arbitrary kind of connection failure.
                // Calling sc.socket().getRemoteSocketAddress() here throws an exception, so we can only log the error itself
                Throwable cause = Throwables.getRootCause(e);
                log.warn("Failed to connect with exception: {}: {}", cause.getClass().getName(), cause.getMessage(), e);
                handler.closeConnection();
                data.future.setException(cause);
                data.future = null;
            }
        } else // Process bytes read
            ConnectionHandler.handleKey(key);
    }
45784
protected void setNode(Node node) {
        assert node!=null;
        if(node instanceof Slave)
            this.nodeName = node.getNodeName();
        else
            this.nodeName = null;

        setNumExecutors(node.getNumExecutors());
        if (this.temporarilyOffline) {
            // When we get a new node, push our current temp offline
            // status to it (as the status is not carried across
            // configuration changes that recreate the node).
            // Since this is also called the very first time this
            // Computer is created, avoid pushing an empty status
            // as that could overwrite any status that the Node
            // brought along from its persisted config data.
            node.setTemporaryOfflineCause(this.offlineCause);
        }
    }
72662
private void putNodeToStore(final Node<K, V> node) {
		final int nodeid = node.id;
		final int index = (nodeid < 0 ? -nodeid : nodeid);
		final WriteBuffer wbuf = storage.set(index);
		final ByteBuffer buf = wbuf.buf();
		if (node.isDeleted()) { 	// This block is for delete
			if (cleanBlocksOnFree) {
				buf.clear();
				int cx = (blockSize >> 3); // division by 8
				while (cx-- > 0) {
					buf.putLong(0);	// Fill with zeroes
				}
				buf.flip();
			} else {
				node.clean(buf); 	// Generate zeroed minimal buffer
			}
			freeBlocks.set(index); 	// Mark block as free
		} else {
			node.serialize(buf);
		}
		wbuf.save();
		if (enableDirtyCheck) {
			dirtyCheck.clear(index);
		}
		if (enableIOStats) {
			getIOStat(nodeid).incPhysWrite();
		}
	}