gt:                     more = more || cxn.m_callbacksToInvoke.get() > 0;
context: /* This file is part of VoltDB.
   * Copyright (C) 2008-2011 VoltDB Inc.
   *
   * VoltDB is free software: you can redistribute it and/or modify
   * it under the terms of the GNU General Public License as published by
   * the Free Software Foundation, either version 3 of the License, or
   * (at your option) any later version.
   *
   * VoltDB 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 for more details.
   *
   * You should have received a copy of the GNU General Public License
   * along with VoltDB.  If not, see <http://www.gnu.org/licenses/>.
   */
  
  package org.voltdb.client;
  
  import java.io.IOException;
  import java.net.UnknownHostException;
  import java.nio.ByteBuffer;
  import java.nio.channels.SocketChannel;
  import java.sql.SQLException;
  import java.util.ArrayList;
  import java.util.HashMap;
  import java.util.Map;
  import java.util.concurrent.atomic.AtomicInteger;
  
  import org.voltdb.ClientResponseImpl;
  import org.voltdb.VoltTable;
  import org.voltdb.VoltTable.ColumnInfo;
  import org.voltdb.VoltType;
  import org.voltdb.messaging.FastDeserializer;
  import org.voltdb.messaging.FastSerializable;
  import org.voltdb.messaging.FastSerializer;
  import org.voltdb.network.Connection;
  import org.voltdb.network.QueueMonitor;
  import org.voltdb.network.VoltNetwork;
  import org.voltdb.network.VoltProtocolHandler;
  import org.voltdb.utils.DBBPool;
  import org.voltdb.utils.DBBPool.BBContainer;
  import org.voltdb.utils.Pair;
  
  /**
   *   De/multiplexes transactions across a cluster
   *
   *   It is safe to synchronized on an individual connection and then the distributer, but it is always unsafe
   *   to synchronized on the distributer and then an individual connection.
   */
  class Distributer {
      // collection of connections to the cluster
      private final ArrayList<NodeConnection> m_connections = new ArrayList<NodeConnection>();
  
      private final ArrayList<ClientStatusListener> m_listeners = new ArrayList<ClientStatusListener>();
  
      //Selector and connection handling, does all work in blocking selection thread
      private final VoltNetwork m_network;
  
      // Temporary until a distribution/affinity algorithm is written
      private int m_nextConnection = 0;
  
      private final int m_expectedOutgoingMessageSize;
  
      private final DBBPool m_pool;
  
      private final boolean m_useMultipleThreads;
  
      private final String m_hostname;
  
      /**
       * Server's instances id. Unique for the cluster
       */
      private Object m_clusterInstanceId[];
  
      private final ClientStatsLoader m_statsLoader;
      private String m_buildString;
  
      private static class ProcedureStats {
          private final String m_name;
  
          private long m_invocationsCompleted = 0;
          private long m_lastInvocationsCompleted = 0;
          private long m_invocationAborts = 0;
          private long m_lastInvocationAborts = 0;
          private long m_invocationErrors = 0;
          private long m_lastInvocationErrors = 0;
  
          // cumulative latency measured by client, used to calculate avg. lat.
          private long m_roundTripTime = 0;
          private long m_lastRoundTripTime = 0;
  
          private int m_maxRoundTripTime = Integer.MIN_VALUE;
          private int m_lastMaxRoundTripTime = Integer.MIN_VALUE;
          private int m_minRoundTripTime = Integer.MAX_VALUE;
          private int m_lastMinRoundTripTime = Integer.MAX_VALUE;
  
          // cumulative latency measured by the cluster, used to calculate avg lat.
          private long m_clusterRoundTripTime = 0;
          private long m_lastClusterRoundTripTime = 0;
  
          // 10ms buckets. Last bucket is all transactions > 190ms.
          static int m_numberOfBuckets = 20;
          private long m_clusterRoundTripTimeBuckets[] = new long[m_numberOfBuckets];
          private long m_roundTripTimeBuckets[] = new long[m_numberOfBuckets];
  
          private int m_maxClusterRoundTripTime = Integer.MIN_VALUE;
          private int m_lastMaxClusterRoundTripTime = Integer.MIN_VALUE;
          private int m_minClusterRoundTripTime = Integer.MAX_VALUE;
          private int m_lastMinClusterRoundTripTime = Integer.MAX_VALUE;
  
          public ProcedureStats(String name) {
              m_name = name;
          }
  
          public void update(int roundTripTime, int clusterRoundTripTime, boolean abort, boolean error) {
              m_maxRoundTripTime = Math.max(roundTripTime, m_maxRoundTripTime);
              m_lastMaxRoundTripTime = Math.max( roundTripTime, m_lastMaxRoundTripTime);
              m_minRoundTripTime = Math.min( roundTripTime, m_minRoundTripTime);
              m_lastMinRoundTripTime = Math.max( roundTripTime, m_lastMinRoundTripTime);
  
              m_maxClusterRoundTripTime = Math.max( clusterRoundTripTime, m_maxClusterRoundTripTime);
              m_lastMaxClusterRoundTripTime = Math.max( clusterRoundTripTime, m_lastMaxClusterRoundTripTime);
              m_minClusterRoundTripTime = Math.min( clusterRoundTripTime, m_minClusterRoundTripTime);
              m_lastMinClusterRoundTripTime = Math.min( clusterRoundTripTime, m_lastMinClusterRoundTripTime);
  
              m_invocationsCompleted++;
              if (abort) {
                  m_invocationAborts++;
              }
              if (error) {
                  m_invocationErrors++;
              }
              m_roundTripTime += roundTripTime;
              m_clusterRoundTripTime += clusterRoundTripTime;
  
              // calculate the latency buckets to increment and increment.
              int rttBucket = (int)(Math.floor(roundTripTime / 10));
              if (rttBucket >= m_roundTripTimeBuckets.length) {
                  rttBucket = m_roundTripTimeBuckets.length - 1;
              }
              m_roundTripTimeBuckets[rttBucket] += 1;
  
              int rttClusterBucket = (int)(Math.floor(clusterRoundTripTime / 10));
              if (rttClusterBucket >= m_clusterRoundTripTimeBuckets.length) {
                  rttClusterBucket = m_clusterRoundTripTimeBuckets.length - 1;
              }
              m_clusterRoundTripTimeBuckets[rttClusterBucket] += 1;
  
          }
      }
  
      class NodeConnection extends VoltProtocolHandler implements org.voltdb.network.QueueMonitor {
          private final AtomicInteger m_callbacksToInvoke = new AtomicInteger(0);
          private final HashMap<Long, Object[]> m_callbacks;
          private final HashMap<String, ProcedureStats> m_stats
              = new HashMap<String, ProcedureStats>();
          private final int m_hostId;
          private final long m_connectionId;
          private Connection m_connection;
          private String m_hostname;
          private boolean m_isConnected = true;
  
          private long m_invocationsCompleted = 0;
          private long m_lastInvocationsCompleted = 0;
          private long m_invocationAborts = 0;
          private long m_lastInvocationAborts = 0;
          private long m_invocationErrors = 0;
          private long m_lastInvocationErrors = 0;
  
          public NodeConnection(long ids[]) {
              m_callbacks = new HashMap<Long, Object[]>();
              m_hostId = (int)ids[0];
              m_connectionId = ids[1];
          }
  
          public void createWork(long handle, String name, BBContainer c, ProcedureCallback callback) {
              synchronized (this) {
                  if (!m_isConnected) {
                      final ClientResponse r = new ClientResponseImpl(
                              ClientResponse.CONNECTION_LOST, new VoltTable[0],
                              "Connection to database host (" + m_hostname +
                              ") was lost before a response was received");
                      try {
                          callback.clientCallback(r);
                      } catch (Exception e) {
                          uncaughtException(callback, r, e);
                      }
                      c.discard();
                      return;
                  }
                  m_callbacks.put(handle, new Object[] { System.currentTimeMillis(), callback, name });
                  m_callbacksToInvoke.incrementAndGet();
              }
              m_connection.writeStream().enqueue(c);
          }
  
          public void createWork(long handle, String name, FastSerializable f, ProcedureCallback callback) {
              synchronized (this) {
                  if (!m_isConnected) {
                      final ClientResponse r = new ClientResponseImpl(
                              ClientResponse.CONNECTION_LOST, new VoltTable[0],
                              "Connection to database host (" + m_hostname +
                              ") was lost before a response was received");
                      try {
                          callback.clientCallback(r);
                      } catch (Exception e) {
                          uncaughtException(callback, r, e);
                      }
                      return;
                  }
                  m_callbacks.put(handle, new Object[]{ System.currentTimeMillis(), callback, name });
                  m_callbacksToInvoke.incrementAndGet();
              }
              m_connection.writeStream().enqueue(f);
          }
  
          /**
           * Update the procedures statistics
           * @param name Name of procedure being updated
           * @param roundTrip round trip from client queued to client response callback invocation
           * @param clusterRoundTrip round trip measured within the VoltDB cluster
           * @param abort true of the procedure was aborted
           * @param failure true if the procedure failed
           */
          private void updateStats(
                  String name,
                  int roundTrip,
                  int clusterRoundTrip,
                  boolean abort,
                  boolean failure) {
              ProcedureStats stats = m_stats.get(name);
              if (stats == null) {
                  stats = new ProcedureStats(name);
                  m_stats.put( name, stats);
              }
              stats.update(roundTrip, clusterRoundTrip, abort, failure);
          }
  
          @Override
          public void handleMessage(ByteBuffer buf, Connection c) {
              long now = System.currentTimeMillis();
              ClientResponseImpl response = null;
              FastDeserializer fds = new FastDeserializer(buf);
              try {
                  response = fds.readObject(ClientResponseImpl.class);
              } catch (IOException e) {
                  e.printStackTrace();
              }
              ProcedureCallback cb = null;
              long callTime = 0;
              int delta = 0;
              synchronized (this) {
                  Object stuff[] = m_callbacks.remove(response.getClientHandle());
  
                  callTime = (Long)stuff[0];
                  delta = (int)(now - callTime);
                  cb = (ProcedureCallback)stuff[1];
                  m_invocationsCompleted++;
                  final byte status = response.getStatus();
                  boolean abort = false;
                  boolean error = false;
                  if (status == ClientResponse.USER_ABORT || status == ClientResponse.GRACEFUL_FAILURE) {
                      m_invocationAborts++;
                      abort = true;
                  } else if (status != ClientResponse.SUCCESS) {
                      m_invocationErrors++;
                      error = true;
                  }
                  updateStats((String)stuff[2], delta, response.getClusterRoundtrip(), abort, error);
              }
  
              if (cb != null) {
                  response.setClientRoundtrip(delta);
                  try {
                      cb.clientCallback(response);
                  } catch (Exception e) {
                      uncaughtException(cb, response, e);
                  }
                  m_callbacksToInvoke.decrementAndGet();
              }
              else if (m_isConnected) {
                  // TODO: what's the right error path here?
                  assert(false);
                  System.err.println("Invalid response: no callback");
              }
          }
  
          /**
           * A number specify the expected size of the majority of outgoing messages.
           * Used to determine the tipping point between where a heap byte buffer vs. direct byte buffer will be
           * used. Also effects the usage of gathering writes.
           */
          @Override
          public int getExpectedOutgoingMessageSize() {
              return m_expectedOutgoingMessageSize;
          }
  
          @Override
          public int getMaxRead() {
              return Integer.MAX_VALUE;
          }
  
          public boolean hadBackPressure() {
              return m_connection.writeStream().hadBackPressure();
          }
  
  
          @Override
          public void stopping(Connection c) {
              super.stopping(c);
              synchronized (this) {
                  //Prevent queueing of new work to this connection
                  synchronized (Distributer.this) {
                      m_connections.remove(this);
                      //Notify listeners that a connection has been lost
                      for (ClientStatusListener s : m_listeners) {
                          s.connectionLost(m_hostname, m_connections.size());
                      }
                  }
                  m_isConnected = false;
  
                  //Invoke callbacks for all queued invocations with a failure response
                  final ClientResponse r =
                      new ClientResponseImpl(
                          ClientResponse.CONNECTION_LOST, new VoltTable[0],
                          "Connection to database host (" + m_hostname +
                          ") was lost before a response was received");
                  for (final Object stuff[] : m_callbacks.values()) {
                      try {
                          ((ProcedureCallback)stuff[1]).clientCallback(r);
                      } catch (Exception e) {
                          uncaughtException(((ProcedureCallback)stuff[1]), r, e);
                      }
                      m_callbacksToInvoke.decrementAndGet();
                  }
              }
          }
  
          @Override
          public Runnable offBackPressure() {
              return new Runnable() {
                  @Override
                  public void run() {
                      /*
                       * Synchronization on Distributer.this is critical to ensure that queue
                       * does not report backpressure AFTER the write stream reports that backpressure
                       * has ended thus resulting in a lost wakeup.
                       */
                      synchronized (Distributer.this) {
                          for (final ClientStatusListener csl : m_listeners) {
                              csl.backpressure(false);
                          }
                      }
                  }
              };
          }
  
          @Override
          public Runnable onBackPressure() {
              return null;
          }
  
          @Override
          public QueueMonitor writestreamMonitor() {
              return this;
          }
  
          /**
           * Get counters for invocations completed, aborted, errors. In that order.
           */
          public synchronized long[] getCounters() {
              return new long[] { m_invocationsCompleted, m_invocationAborts, m_invocationErrors };
          }
  
          /**
           * Get counters for invocations completed, aborted, errors. In that order.
           * Count returns count since this method was last invoked
           */
          public synchronized long[] getCountersInterval() {
              final long invocationsCompletedThisTime = m_invocationsCompleted - m_lastInvocationsCompleted;
              m_lastInvocationsCompleted = m_invocationsCompleted;
  
              final long invocationsAbortsThisTime = m_invocationAborts - m_lastInvocationAborts;
              m_lastInvocationAborts = m_invocationAborts;
  
              final long invocationErrorsThisTime = m_invocationErrors - m_lastInvocationErrors;
              m_lastInvocationErrors = m_invocationErrors;
              return new long[] {
                      invocationsCompletedThisTime,
                      invocationsAbortsThisTime,
                      invocationErrorsThisTime
              };
          }
  
          private int m_queuedBytes = 0;
          private final int m_maxQueuedBytes = 262144;
  
          @Override
          public boolean queue(int bytes) {
              m_queuedBytes += bytes;
              if (m_queuedBytes > m_maxQueuedBytes) {
                  return true;
              }
              return false;
          }
      }
  
      void drain() throws NoConnectionsException, InterruptedException {
          boolean more;
          do {
              more = false;
              synchronized (this) {
                  for (NodeConnection cxn : m_connections) {
  