gt:                     long fragmentId = m_connection.readLong();
context: /* This file is part of VoltDB.
   * Copyright (C) 2008-2018 VoltDB Inc.
   *
   * This program is free software: you can redistribute it and/or modify
   * it under the terms of the GNU Affero General Public License as
   * published by the Free Software Foundation, either version 3 of the
   * License, or (at your option) any later version.
   *
   * This program 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 Affero General Public License for more details.
   *
   * You should have received a copy of the GNU Affero General Public License
   * along with VoltDB.  If not, see <http://www.gnu.org/licenses/>.
   */
  
  package org.voltdb.jni;
  
  import java.io.EOFException;
  import java.io.IOException;
  import java.io.UnsupportedEncodingException;
  import java.lang.reflect.InvocationTargetException;
  import java.net.InetSocketAddress;
  import java.net.Socket;
  import java.nio.ByteBuffer;
  import java.nio.channels.SocketChannel;
  import java.util.List;
  import java.util.logging.Level;
  import java.util.logging.Logger;
  
  import org.voltcore.utils.DBBPool.BBContainer;
  import org.voltcore.utils.Pair;
  import org.voltdb.BackendTarget;
  import org.voltdb.ParameterSet;
  import org.voltdb.PrivateVoltTableFactory;
  import org.voltdb.StatsSelector;
  import org.voltdb.TableStreamType;
  import org.voltdb.TheHashinator.HashinatorConfig;
  import org.voltdb.UserDefinedFunctionManager.UserDefinedFunctionRunner;
  import org.voltdb.VoltTable;
  import org.voltdb.common.Constants;
  import org.voltdb.exceptions.EEException;
  import org.voltdb.exceptions.SerializableException;
  import org.voltdb.export.ExportManager;
  import org.voltdb.iv2.DeterminismHash;
  import org.voltdb.messaging.FastDeserializer;
  import org.voltdb.messaging.FastSerializer;
  import org.voltdb.sysprocs.saverestore.SnapshotUtil;
  import org.voltdb.utils.SerializationHelper;
  import org.voltdb.utils.CompressionService;
  
  import com.google_voltpatches.common.base.Charsets;
  import com.google_voltpatches.common.base.Throwables;
  
  /* Serializes data over a connection that presumably is being read
   * by a voltdb execution engine. The serialization is currently a
   * a packed binary big endian buffer. Each buffer has a header
   * that provides the total length (as an int) and the command being
   * serialized (also as an int). Commands are described by the Commands
   * enum.
   *
   * These serializations are implemented as required since a lot of
   * the native interface isn't needed in many cases.
   *
   * The serialization could be more robust. Probably the right way to
   * get that robustness would be to create FastSerializable classes
   * for the more complex calls and read them through the fast serializable
   * C++ interface.
   *
   * (The fastserializer still requires hand-authoring the ser/deser
   * logic in java and C and doesn't result in any more favorable
   * packing; in fact the resulting wire content might be the same in
   * most cases... And you can't fast serialize an array of primitive
   * types without writing a wrapper object.)
   *
   * Responses to the IPC interface start with a 1 byte result code. If the
   * result code is failure then no message follows. If the result code
   * is success and a response is warranted it will follow and is length prefixed if necessary.
   * Unlike requests to the EE, the length prefix in responses does not include the 1 byte
   * result code nor the 4 byte length field. When we correct the 1 and 4 byte network reads and writes
   * we can this. A length will not prefix the response
   * if the length of the response can be deduced from the original request.
   *
   * The return message format for DMLPlanFragments is all big endian:
   * 1 byte result code
   * 8 byte results codes. Same number of results as numPlanFragments.
   *
   * The return message format for QueryPlanFragments is all big endian:
   * 1 byte result code
   * 4 byte result length. Does not include the length of the result code or result length field.
   * X bytes of serialized VoltTables. Same number of tables as numPlanFragments
   *
   * The return message format for PlanFragment is all big endian:
   * 1 byte result code
   * 4 byte result length. Does not include the length of the result code or result length field.
   * 4 byte result indicating number of dependencies returned
   * The dependency tables
   *
   * The return message format for ReceiveDependency consists of a single byte result code.
   *
   * The return message format for Load table consists of a single byte result code.
   */
  
  public class ExecutionEngineIPC extends ExecutionEngine {
  
      /** Commands are serialized over the connection */
      private enum Commands {
          Initialize(0),
          LoadCatalog(2),
          ToggleProfiler(3),
          Tick(4),
          GetStats(5),
          QueryPlanFragments(6),
          PlanFragment(7),
          LoadTable(9),
          releaseUndoToken(10),
          undoUndoToken(11),
          CustomPlanFragment(12),
          SetLogLevels(13),
          Quiesce(16),
          ActivateTableStream(17),
          TableStreamSerializeMore(18),
          UpdateCatalog(19),
          ExportAction(20),
          RecoveryMessage(21),
          TableHashCode(22),
          Hashinate(23),
          GetPoolAllocations(24),
          GetUSOs(25),
          updateHashinator(27),
          executeTask(28),
          applyBinaryLog(29);
          Commands(final int id) {
              m_id = id;
          }
  
          int m_id;
      }
  
      /**
       * One connection per ExecutionEngineIPC. This connection also interfaces
       * with Valgrind to report any problems that Valgrind may find including
       * reachable heap blocks on exit. Unfortunately it is not enough to inspect
       * the Valgrind return code as it considers reachable blocks to not be an
       * error.
       **/
      private class Connection {
          private Socket m_socket = null;
          private SocketChannel m_socketChannel = null;
          Connection(BackendTarget target, int port) {
              boolean connected = false;
              int retries = 0;
              while (!connected) {
                  try {
                      System.out.println("Connecting to localhost:" + port);
                      m_socketChannel = SocketChannel.open(new InetSocketAddress(
                              "localhost", port));
                      m_socketChannel.configureBlocking(true);
                      m_socket = m_socketChannel.socket();
                      m_socket.setTcpNoDelay(true);
                      connected = true;
                  } catch (final Exception e) {
                      System.out.println(e.getMessage());
                      if (retries++ <= 10) {
                          if (retries > 1) {
                              System.out.printf("Failed to connect to IPC EE on port %d. Retry #%d of 10\n", port, retries-1);
                              try {
                                  Thread.sleep(10000);
                              }
                              catch (InterruptedException e1) {}
                          }
                      }
                      else {
                          System.out.printf("Failed to initialize IPC EE connection on port %d. Quitting.\n", port);
                          System.exit(-1);
                      }
                  }
                  if (!connected && retries == 1 && target == BackendTarget.NATIVE_EE_IPC) {
                      System.out.println("Ready to connect to voltdbipc process on port " + port);
                      System.out.println("Press Enter after you have started the EE process to initiate the connection to the EE");
                      try {
                          System.in.read();
                      } catch (final IOException e1) {
                          e1.printStackTrace();
                      }
                  }
              }
              System.out.println("Created IPC connection for site.");
          }
  
          /* Close the socket indicating to the EE it should terminate */
          public void close() throws InterruptedException {
              if (m_socketChannel != null) {
                  try {
                      m_socketChannel.close();
                  } catch (final IOException e) {
                      throw new RuntimeException(e);
                  }
                  m_socketChannel = null;
                  m_socket = null;
              }
          }
  
          /** blocking write of all m_data to outputstream */
          void write() throws IOException {
              // write 4 byte length (which includes its own 4 bytes) in big-endian
              // order. this hurts .. but I'm not sure of a better way.
              m_dataNetwork.clear();
              final int amt = m_data.remaining();
              m_dataNetwork.putInt(4 + amt);
              if (m_dataNetwork.capacity() < (4 + amt)) {
                  throw new IOException("Catalog data size (" + (4 + amt) +
                                        ") exceeds ExecutionEngineIPC's hard-coded data buffer capacity (" +
                                        m_dataNetwork.capacity() + ")");
              }
              m_dataNetwork.limit(4 + amt);
              m_dataNetwork.rewind();
              while (m_dataNetwork.hasRemaining()) {
                  m_socketChannel.write(m_dataNetwork);
              }
          }
  
          /**
           * An error code specific to the IPC backend that indicates
           * that as part of fulfilling a previous request the IPC
           * backend requires a dependency table. This is not really
           * an error code.
           */
          static final int kErrorCode_RetrieveDependency = 100;
  
          /**
           * An error code to be sent in a response to an RetrieveDependency request.
           * Indicates that a dependency table was found and that it follows
           */
          static final int kErrorCode_DependencyFound = 101;
  
          /**
           * An error code to be sent in a response to an RetrieveDependency request.
           * Indicates that no dependency tables could be found and that no data follows.
           */
          static final int kErrorCode_DependencyNotFound = 102 ;
  
          /**
           * An error code that can be sent at any time indicating that
           * an export buffer follows
           */
          static final int kErrorCode_pushExportBuffer = 103;
  
          /**
           * Invoke crash VoltDB
           */
          static final int kErrorCode_CrashVoltDB = 104;
  
          /**
           * Retrieve value from Java for stats
           */
          static final int kErrorCode_getQueuedExportBytes = 105;
  
          /**
           * An error code that can be sent at any time indicating that
           * a per-fragment statistics buffer follows
           */
          static final int kErrorCode_pushPerFragmentStatsBuffer = 106;
  
          /**
           * Instruct the Java side to invoke a user-defined function and
           * return the result.
           */
          static final int kErrorCode_callJavaUserDefinedFunction = 107;
  
          /**
           * An error code that can be sent at any time indicating that
           * an export stream is dropped
           */
          static final int kErrorCode_pushEndOfStream = 113;
  
          ByteBuffer getBytes(int size) throws IOException {
              ByteBuffer header = ByteBuffer.allocate(size);
              while (header.hasRemaining()) {
                  final int read = m_socket.getChannel().read(header);
                  if (read == -1) {
                      throw new EOFException();
                  }
              }
              header.flip();
              return header;
          }
  
          // Read per-fragment stats from the wire.
          void extractPerFragmentStatsInternal() {
              try {
                  int bufferSize = m_connection.readInt();
                  final ByteBuffer perFragmentStatsBuffer = ByteBuffer.allocate(bufferSize);
                  while (perFragmentStatsBuffer.hasRemaining()) {
                      int read = m_socketChannel.read(perFragmentStatsBuffer);
                      if (read == -1) {
                          throw new EOFException();
                      }
                  }
                  perFragmentStatsBuffer.flip();
                  // Skip the perFragmentTimingEnabled flag.
                  perFragmentStatsBuffer.get();
                  m_succeededFragmentsCount = perFragmentStatsBuffer.getInt();
                  if (m_perFragmentTimingEnabled) {
                      for (int i = 0; i < m_succeededFragmentsCount; i++) {
                          m_executionTimes[i] = perFragmentStatsBuffer.getLong();
                      }
                      // This is the time for the failed fragment.
                      if (m_succeededFragmentsCount < m_executionTimes.length) {
                          m_executionTimes[m_succeededFragmentsCount] = perFragmentStatsBuffer.getLong();
                      }
                  }
              }
              catch (IOException e) {
                  throw new RuntimeException(e);
              }
          }
  
          // Internal function to receive and execute the UDF invocation request.
          void callJavaUserDefinedFunctionInternal() {
              try {
                  // Read the request content from the wire.
                  int bufferSize = m_connection.readInt();
                  final ByteBuffer udfBuffer = ByteBuffer.allocate(bufferSize);
                  while (udfBuffer.hasRemaining()) {
                      int read = m_socketChannel.read(udfBuffer);
                      if (read == -1) {
                          throw new EOFException();
                      }
                  }
                  udfBuffer.flip();
  
                  int functionId = udfBuffer.getInt();
                  UserDefinedFunctionRunner udfRunner = m_functionManager.getFunctionRunnerById(functionId);
                  assert(udfRunner != null);
                  Throwable throwable = null;
                  Object returnValue = null;
                  try {
                      // Call the user-defined function.
                      returnValue = udfRunner.call(udfBuffer);
                      m_data.clear();
                      // Put the status code for success (zero) into the buffer.
                      m_data.putInt(0);
                      // Write the result to the buffer.
                      UserDefinedFunctionRunner.writeValueToBuffer(m_data, udfRunner.getReturnType(), returnValue);
                      m_data.flip();
                      m_connection.write();
                      return;
                  }
                  catch (InvocationTargetException ex1) {
                      // Exceptions thrown during Java reflection will be wrapped into this InvocationTargetException.
                      // We need to get its cause and throw that to the user.
                      throwable = ex1.getCause();
                  }
                  catch (Throwable ex2) {
                      throwable = ex2;
                  }
                  // Getting here means the execution was not successful.
                  m_data.clear();
                  if (throwable != null) {
                      // Exception thrown, put return code = -1.
                      m_data.putInt(-1);
                      byte[] errorMsg = throwable.toString().getBytes(Constants.UTF8ENCODING);
                      SerializationHelper.writeVarbinary(errorMsg, m_data);
                  }
                  m_data.flip();
                  m_connection.write();
              }
              catch (IOException e) {
                  throw new RuntimeException(e);
              }
          }
  
          /**
           * Read a single byte indicating a return code. This method has evolved
           * to include providing dependency tables necessary for the completion of previous
           * request. The method loops ready status bytes instead of recursing to avoid
           * excessive recursion in the case where a large number of dependency tables
           * must be fetched before the request can be satisfied.
           *
           * Facing further evolutionary pressure, readStatusByte has  grown
           * EL buffer reading fins. EL buffers arrive here mid-command execution.
           * @return
           * @throws IOException
           */
          int readStatusByte() throws IOException {
              int status = kErrorCode_RetrieveDependency;
  
              while (true) {
                  status = m_socket.getInputStream().read();
                  if (status == kErrorCode_RetrieveDependency) {
                      final ByteBuffer dependencyIdBuffer = ByteBuffer.allocate(4);
                      while (dependencyIdBuffer.hasRemaining()) {
                          final int read = m_socketChannel.read(dependencyIdBuffer);
                          if (read == -1) {
                              throw new IOException("Unable to read enough bytes for dependencyId in order to " +
                              " satisfy IPC backend request for a dependency table");
                          }
                      }
                      dependencyIdBuffer.rewind();
                      sendDependencyTable(dependencyIdBuffer.getInt());
                  }
                  else if (status == ExecutionEngine.ERRORCODE_PROGRESS_UPDATE) {
                      m_history.append("GOT PROGRESS_UPDATE... ");
                      int batchIndex = m_connection.readInt();
                      int planNodeTypeAsInt = m_connection.readInt();
                      long tuplesFound = m_connection.readLong();
                      long currMemoryInBytes = m_connection.readLong();
                      long peakMemoryInBytes = m_connection.readLong();
                      long nextStep = fragmentProgressUpdate(batchIndex, planNodeTypeAsInt, tuplesFound,
                              currMemoryInBytes, peakMemoryInBytes);
                      m_history.append("...RESPONDING TO PROGRESS_UPDATE...nextStep=" + nextStep);
                      m_data.clear();
                      m_data.putLong(nextStep);
                      m_data.flip();
                      m_connection.write();
                      m_history.append(" WROTE RESPONSE TO PROGRESS_UPDATE\n");
                  }
                  else if (status == kErrorCode_pushExportBuffer) {
                      // Message structure:
                      // pushExportBuffer error code - 1 byte
                      // partition id - 4 bytes
                      // signature length (in bytes) - 4 bytes
                      // signature - signature length bytes
                      // uso - 8 bytes
                      // sync - 1 byte
                      // export buffer length - 4 bytes
                      // export buffer - export buffer length bytes
                      int partitionId = getBytes(4).getInt();
                      int signatureLength = getBytes(4).getInt();
                      byte signatureBytes[] = new byte[signatureLength];
                      getBytes(signatureLength).get(signatureBytes);
                      String signature = new String(signatureBytes, "UTF-8");
                      long uso = getBytes(8).getLong();
                      boolean sync = getBytes(1).get() == 1 ? true : false;
                      int length = getBytes(4).getInt();
                      ExportManager.pushExportBuffer(
                              partitionId,
                              signature,
                              uso,
                              0,
                              length == 0 ? null : getBytes(length),
                              sync);
                  }
                  else if (status == kErrorCode_getQueuedExportBytes) {
                      ByteBuffer header = ByteBuffer.allocate(8);
                      while (header.hasRemaining()) {
                          final int read = m_socket.getChannel().read(header);
                          if (read == -1) {
                              throw new EOFException();
                          }
                      }
                      header.flip();
  
                      int partitionId = header.getInt();
                      int signatureLength = header.getInt();
                      ByteBuffer sigbuf = ByteBuffer.allocate(signatureLength);
                      while (sigbuf.hasRemaining()) {
                          final int read = m_socket.getChannel().read(sigbuf);
                          if (read == -1) {
                              throw new EOFException();
                          }
                      }
                      sigbuf.flip();
                      byte signatureBytes[] = new byte[signatureLength];
                      sigbuf.get(signatureBytes);
                      String signature = new String(signatureBytes, "UTF-8");
  
                      long retval = ExportManager.getQueuedExportBytes(partitionId, signature);
                      ByteBuffer buf = ByteBuffer.allocate(8);
                      buf.putLong(retval).flip();
  
                      while (buf.hasRemaining()) {
                          m_socketChannel.write(buf);
                      }
                  }
                  else if (status == kErrorCode_pushEndOfStream) {
                      ByteBuffer header = ByteBuffer.allocate(8);
                      while (header.hasRemaining()) {
                          final int read = m_socket.getChannel().read(header);
                          if (read == -1) {
                              throw new EOFException();
                          }
                      }
                      header.flip();
  
                      int partitionId = header.getInt();
                      int signatureLength = header.getInt();
                      ByteBuffer sigbuf = ByteBuffer.allocate(signatureLength);
                      while (sigbuf.hasRemaining()) {
                          final int read = m_socket.getChannel().read(sigbuf);
                          if (read == -1) {
                              throw new EOFException();
                          }
                      }
                      sigbuf.flip();
                      byte signatureBytes[] = new byte[signatureLength];
                      sigbuf.get(signatureBytes);
                      String signature = new String(signatureBytes, "UTF-8");
  
                      ExportManager.pushEndOfStream(partitionId, signature);
                  }
                  else if (status == ExecutionEngine.ERRORCODE_DECODE_BASE64_AND_DECOMPRESS) {
                      int dataLength = m_connection.readInt();
                      String data = m_connection.readString(dataLength);
                      byte[] decodedDecompressedData = CompressionService.decodeBase64AndDecompressToBytes(data);
                      m_data.clear();
                      m_data.put(decodedDecompressedData);
                      m_data.flip();
                      m_connection.write();
                  }
                  else if (status == kErrorCode_CrashVoltDB) {
                      ByteBuffer lengthBuffer = ByteBuffer.allocate(4);
                      while (lengthBuffer.hasRemaining()) {
                          final int read = m_socket.getChannel().read(lengthBuffer);
                          if (read == -1) {
                              throw new EOFException();
                          }
                      }
                      lengthBuffer.flip();
                      ByteBuffer messageBuffer = ByteBuffer.allocate(lengthBuffer.getInt());
                      while (messageBuffer.hasRemaining()) {
                          final int read = m_socket.getChannel().read(messageBuffer);
                          if (read == -1) {
                              throw new EOFException();
                          }
                      }
                      messageBuffer.flip();
                      final int reasonLength = messageBuffer.getInt();
                      final byte reasonBytes[] = new byte[reasonLength];
                      messageBuffer.get(reasonBytes);
                      final String message = new String(reasonBytes, "UTF-8");
  
                      final int filenameLength = messageBuffer.getInt();
                      final byte filenameBytes[] = new byte[filenameLength];
                      messageBuffer.get(filenameBytes);
                      final String filename = new String(filenameBytes, "UTF-8");
  
                      final int lineno = messageBuffer.getInt();
  
  
                      final int numTraces = messageBuffer.getInt();
                      final String traces[] = new String[numTraces];
  
                      for (int ii = 0; ii < numTraces; ii++) {
                          final int traceLength = messageBuffer.getInt();
                          final byte traceBytes[] = new byte[traceLength];
                          messageBuffer.get(traceBytes);
                          traces[ii] = new String(traceBytes, "UTF-8");
                      }
  
                      ExecutionEngine.crashVoltDB(message, traces, filename, lineno);
                  }
                  else if (status == kErrorCode_pushPerFragmentStatsBuffer) {
                      // The per-fragment stats are in the very beginning.
                      extractPerFragmentStatsInternal();
                  }
                  else if (status == kErrorCode_callJavaUserDefinedFunction) {
                      callJavaUserDefinedFunctionInternal();
                  }
                  else {
                      break;
                  }
              }
  
              try {
                  checkErrorCode(status);
                  return status;
              }
              catch (SerializableException e) {
                  throw e;
              }
              catch (RuntimeException e) {
                  throw (IOException)e.getCause();
              }
          }
  
  
          /**
           * Read and deserialize some number of tables from the wire. Assumes that the message is length prefixed.
           * @param tables Output array as well as indicator of exactly how many tables to read off of the wire
           * @throws IOException
           */
          public void readResultTables(final VoltTable tables[]) throws IOException {
              final ByteBuffer resultTablesLengthBytes = ByteBuffer.allocate(4);
  
              //resultTablesLengthBytes.order(ByteOrder.LITTLE_ENDIAN);
              while (resultTablesLengthBytes.hasRemaining()) {
                  int read = m_socketChannel.read(resultTablesLengthBytes);
                  if (read == -1) {
                      throw new EOFException();
                  }
              }
              resultTablesLengthBytes.flip();
  
              final int resultTablesLength = resultTablesLengthBytes.getInt();
  
              // check the dirty-ness of the batch
              final ByteBuffer dirtyBytes = ByteBuffer.allocate(1);
              while (dirtyBytes.hasRemaining()) {
                  int read = m_socketChannel.read(dirtyBytes);
                  if (read == -1) {
                      throw new EOFException();
                  }
              }
              dirtyBytes.flip();
              // check if anything was changed
              final boolean dirty  = dirtyBytes.get() > 0;
              if (dirty)
                  m_dirty = true;
  
              if (resultTablesLength <= 0)
                  return;
  
              final ByteBuffer resultTablesBuffer = ByteBuffer
                      .allocate(resultTablesLength);
              //resultTablesBuffer.order(ByteOrder.LITTLE_ENDIAN);
              while (resultTablesBuffer.hasRemaining()) {
                  int read = m_socketChannel.read(resultTablesBuffer);
                  if (read == -1) {
                      throw new EOFException();
                  }
              }
              resultTablesBuffer.flip();
  
              for (int ii = 0; ii < tables.length; ii++) {
                  final int dependencyCount = resultTablesBuffer.getInt(); // ignore the table count
                  assert(dependencyCount == 1); //Expect one dependency generated per plan fragment
                  resultTablesBuffer.getInt(); // ignore the dependency ID
                  tables[ii] = PrivateVoltTableFactory.createVoltTableFromSharedBuffer(resultTablesBuffer);
              }
          }
  
          public ByteBuffer readResultsBuffer() throws IOException {
              // check the dirty-ness of the batch
              final ByteBuffer dirtyBytes = ByteBuffer.allocate(1);
              while (dirtyBytes.hasRemaining()) {
                  int read = m_socketChannel.read(dirtyBytes);
                  if (read == -1) {
                      throw new EOFException();
                  }
              }
              dirtyBytes.flip();
              // check if anything was changed
              m_dirty |= dirtyBytes.get() > 0;
  
              final ByteBuffer resultTablesLengthBytes = ByteBuffer.allocate(4);
              //resultTablesLengthBytes.order(ByteOrder.LITTLE_ENDIAN);
              while (resultTablesLengthBytes.hasRemaining()) {
                  int read = m_socketChannel.read(resultTablesLengthBytes);
                  if (read == -1) {
                      throw new EOFException();
                  }
              }
              resultTablesLengthBytes.flip();
              final int resultTablesLength = resultTablesLengthBytes.getInt();
  
              if (resultTablesLength <= 0)
                  return resultTablesLengthBytes;
  
              final ByteBuffer resultTablesBuffer = ByteBuffer
                      .allocate(resultTablesLength+4);
              //resultTablesBuffer.order(ByteOrder.LITTLE_ENDIAN);
              resultTablesBuffer.putInt(resultTablesLength);
              while (resultTablesBuffer.hasRemaining()) {
                  int read = m_socketChannel.read(resultTablesBuffer);
                  if (read == -1) {
                      throw new EOFException();
                  }
              }
              resultTablesBuffer.flip();
              return resultTablesBuffer;
          }
  
          /**
           * Read and deserialize a long from the wire.
           */
          public long readLong() throws IOException {
              final ByteBuffer longBytes = ByteBuffer.allocate(8);
  
              //resultTablesLengthBytes.order(ByteOrder.LITTLE_ENDIAN);
              while (longBytes.hasRemaining()) {
                  int read = m_socketChannel.read(longBytes);
                  if (read == -1) {
                      throw new EOFException();
                  }
              }
              longBytes.flip();
  
              final long retval = longBytes.getLong();
              return retval;
          }
  
          /**
           * Read and deserialize an int from the wire.
           */
          public int readInt() throws IOException {
              final ByteBuffer intBytes = ByteBuffer.allocate(4);
  
              //resultTablesLengthBytes.order(ByteOrder.LITTLE_ENDIAN);
              while (intBytes.hasRemaining()) {
                  int read = m_socketChannel.read(intBytes);
                  if (read == -1) {
                      throw new EOFException();
                  }
              }
              intBytes.flip();
  
              final int retval = intBytes.getInt();
              return retval;
          }
  
          /**
           * Read and deserialize a short from the wire.
           */
          public short readShort() throws IOException {
              final ByteBuffer shortBytes = ByteBuffer.allocate(2);
  
              //resultTablesLengthBytes.order(ByteOrder.LITTLE_ENDIAN);
              while (shortBytes.hasRemaining()) {
                  int read = m_socketChannel.read(shortBytes);
                  if (read == -1) {
                      throw new EOFException();
                  }
              }
              shortBytes.flip();
  
              final short retval = shortBytes.getShort();
              return retval;
          }
  
          /**
           * Read and deserialize a byte from the wire.
           */
          public byte readByte() throws IOException {
              final ByteBuffer bytes = ByteBuffer.allocate(1);
  
              //resultTablesLengthBytes.order(ByteOrder.LITTLE_ENDIAN);
              while (bytes.hasRemaining()) {
                  int read = m_socketChannel.read(bytes);
                  if (read == -1) {
                      throw new EOFException();
                  }
              }
              bytes.flip();
  
              final byte retval = bytes.get();
              return retval;
          }
  
          /**
           * Read and deserialize a string from the wire.
           */
          public String readString(int size) throws IOException {
              final ByteBuffer stringBytes = ByteBuffer.allocate(size);
  
              //resultTablesLengthBytes.order(ByteOrder.LITTLE_ENDIAN);
              while (stringBytes.hasRemaining()) {
                  int read = m_socketChannel.read(stringBytes);
                  if (read == -1) {
                      throw new EOFException();
                  }
              }
              stringBytes.flip();
  
              final String retval = new String(stringBytes.array(), Constants.UTF8ENCODING);
              return retval;
          }
  
          public void throwException(final int errorCode) throws IOException {
              final ByteBuffer lengthBuffer = ByteBuffer.allocate(4);
              while (lengthBuffer.hasRemaining()) {
                  int read = m_socketChannel.read(lengthBuffer);
                  if (read == -1) {
                      throw new EOFException();
                  }
              }
              lengthBuffer.flip();
              final int exceptionLength = lengthBuffer.getInt();//Length is only between EE and Java.
              if (exceptionLength == 0) {
                  throw new EEException(errorCode);
              } else {
                  final ByteBuffer exceptionBuffer = ByteBuffer.allocate(exceptionLength + 4);
                  exceptionBuffer.putInt(exceptionLength);
                  while(exceptionBuffer.hasRemaining()) {
                      int read = m_socketChannel.read(exceptionBuffer);
                      if (read == -1) {
                          throw new EOFException();
                      }
                  }
                  assert(!exceptionBuffer.hasRemaining());
                  exceptionBuffer.rewind();
                  throw SerializableException.deserializeFromBuffer(exceptionBuffer);
              }
          }
      }
  
      /** Local m_data */
      private final int m_clusterIndex;
      private final long m_siteId;
      private final int m_partitionId;
      private final int m_hostId;
      private final String m_hostname;
      // private final FastSerializer m_fser;
      private final Connection m_connection;
      private BBContainer m_dataNetworkOrigin;
      private ByteBuffer m_dataNetwork;
      private ByteBuffer m_data;
  
      // private int m_counter;
  
      private void verifyDataCapacity(int size) {
          if (size+4 > m_dataNetwork.capacity()) {
              m_dataNetworkOrigin.discard();
              m_dataNetworkOrigin = org.voltcore.utils.DBBPool.allocateDirect(size+4);
              m_dataNetwork = m_dataNetworkOrigin.b();
              m_dataNetwork.position(4);
              m_data = m_dataNetwork.slice();
          }
      }
  
      public ExecutionEngineIPC(
              final int clusterIndex,
              final long siteId,
              final int partitionId,
              final int hostId,
              final String hostname,
              final int drClusterId,
              final int defaultDrBufferSize,
              final int tempTableMemory,
              final BackendTarget target,
              final int port,
              final HashinatorConfig hashinatorConfig,
              final boolean createDrReplicatedStream) {
          super(siteId, partitionId);
  
          // m_counter = 0;
          m_clusterIndex = clusterIndex;
          m_siteId = siteId;
          m_partitionId = partitionId;
          m_hostId = hostId;
          m_hostname = hostname;
          // m_fser = new FastSerializer(false, false);
          m_connection = new Connection(target, port);
  
          // voltdbipc assumes host byte order everywhere
          // Arbitrarily set to 20MB when 10MB crashed for an arbitrarily scaled unit test.
          m_dataNetworkOrigin = org.voltcore.utils.DBBPool.allocateDirect(1024 * 1024 * 20);
          m_dataNetwork = m_dataNetworkOrigin.b();
          m_dataNetwork.position(4);
          m_data = m_dataNetwork.slice();
  
          initialize(
                  m_clusterIndex,
                  m_siteId,
                  m_partitionId,
                  m_hostId,
                  m_hostname,
                  drClusterId,
                  defaultDrBufferSize,
                  1024 * 1024 * tempTableMemory,
                  hashinatorConfig,
                  createDrReplicatedStream);
      }
  
      /** Utility method to generate an EEXception that can be overriden by derived classes**/
      @Override
      protected void throwExceptionForError(final int errorCode) {
          try {
              m_connection.throwException(errorCode);
          } catch (final IOException e) {
              throw new RuntimeException(e);
          }
      }
  
      private StringBuffer m_history = new StringBuffer();
      @Override
      public void release() throws EEException, InterruptedException {
          System.out.println("Shutdown IPC connection in progress.");
          System.out.println("But first, a little history:\n" + m_history );
          m_connection.close();
          System.out.println("Shutdown IPC connection done.");
          m_dataNetworkOrigin.discard();
      }
  
      private static final Object printLockObject = new Object();
      /**
       * the abstract api assumes construction initializes but here initialization
       * is just another command.
       */
      public void initialize(
              final int clusterIndex,
              final long siteId,
              final int partitionId,
              final int hostId,
              final String hostname,
              final int drClusterId,
              final int defaultDrBufferSize,
              final long tempTableMemory,
              final HashinatorConfig hashinatorConfig,
              final boolean createDrReplicatedStream)
      {
          synchronized(printLockObject) {
              System.out.println("Initializing an IPC EE " + this + " for hostId " + hostId + " siteId " + siteId + " from thread " + Thread.currentThread().getId());
          }
          int result = ExecutionEngine.ERRORCODE_ERROR;
          m_data.clear();
          m_data.putInt(Commands.Initialize.m_id);
          m_data.putInt(clusterIndex);
          m_data.putLong(siteId);
          m_data.putInt(partitionId);
          m_data.putInt(hostId);
          m_data.putInt(drClusterId);
          m_data.putInt(defaultDrBufferSize);
          m_data.putLong(EELoggers.getLogLevels());
          m_data.putLong(tempTableMemory);
          m_data.putInt(createDrReplicatedStream ? 1 : 0);
          m_data.putInt((short)hostname.length());
          m_data.put(hostname.getBytes(Charsets.UTF_8));
          try {
              m_data.flip();
              m_connection.write();
              result = m_connection.readStatusByte();
          } catch (final IOException e) {
              System.out.println("Exception: " + e.getMessage());
              throw new RuntimeException(e);
          }
          checkErrorCode(result);
          updateHashinator(hashinatorConfig);
      }
  
      /** write the catalog as a UTF-8 byte string via connection */
      @Override
      protected void coreLoadCatalog(final long timestamp, final byte[] catalogBytes) throws EEException {
          int result = ExecutionEngine.ERRORCODE_ERROR;
          verifyDataCapacity(catalogBytes.length + 100);
          m_data.clear();
          m_data.putInt(Commands.LoadCatalog.m_id);
          m_data.putLong(timestamp);
          m_data.put(catalogBytes);
          m_data.put((byte)'\0');
  
          try {
              m_data.flip();
              m_connection.write();
              result = m_connection.readStatusByte();
          } catch (final IOException e) {
              System.out.println("Exception: " + e.getMessage());
              throw new RuntimeException(e);
          }
          checkErrorCode(result);
      }
  
      /** write the diffs as a UTF-8 byte string via connection */
      @Override
      public void coreUpdateCatalog(final long timestamp, final boolean isStreamUpdate, final String catalogDiffs) throws EEException {
          int result = ExecutionEngine.ERRORCODE_ERROR;
  
          try {
              final byte catalogBytes[] = catalogDiffs.getBytes("UTF-8");
              verifyDataCapacity(catalogBytes.length + 100);
              m_data.clear();
              m_data.putInt(Commands.UpdateCatalog.m_id);
              m_data.putLong(timestamp);
              m_data.putInt(isStreamUpdate ? 1 : 0);
              m_data.put(catalogBytes);
              m_data.put((byte)'\0');
          } catch (final UnsupportedEncodingException ex) {
              Logger.getLogger(ExecutionEngineIPC.class.getName()).log(
                      Level.SEVERE, null, ex);
          }
  
          try {
              m_data.flip();
              m_connection.write();
              result = m_connection.readStatusByte();
          } catch (final IOException e) {
              System.out.println("Exception: " + e.getMessage());
              throw new RuntimeException(e);
          }
          checkErrorCode(result);
      }
  
      @Override
      public void tick(final long time, final long lastCommittedSpHandle) {
          int result = ExecutionEngine.ERRORCODE_ERROR;
          m_data.clear();
          m_data.putInt(Commands.Tick.m_id);
          m_data.putLong(time);
          m_data.putLong(lastCommittedSpHandle);
          try {
              m_data.flip();
              m_connection.write();
              result = m_connection.readStatusByte();
          } catch (final IOException e) {
              System.out.println("Exception: " + e.getMessage());
              throw new RuntimeException(e);
          }
          checkErrorCode(result);
          // no return code for tick.
      }
  
      @Override
      public void quiesce(long lastCommittedSpHandle) {
          int result = ExecutionEngine.ERRORCODE_ERROR;
          m_data.clear();
          m_data.putInt(Commands.Quiesce.m_id);
          m_data.putLong(lastCommittedSpHandle);
          try {
              m_data.flip();
              m_connection.write();
              result = m_connection.readStatusByte();
          } catch (final IOException e) {
              System.out.println("Excpeption: " + e.getMessage());
              throw new RuntimeException();
          }
          checkErrorCode(result);
      }
  
      private void sendPlanFragmentsInvocation(final Commands cmd,
              final int numFragmentIds,
              final long[] planFragmentIds,
              long[] inputDepIdsIn,
              final Object[] parameterSets,
              DeterminismHash determinismHash,
              boolean[] isWriteFrags,
              int[] sqlCRCs,
              final long txnId,
              final long spHandle,
              final long lastCommittedSpHandle,
              final long uniqueId,
              final long undoToken)
      {
          // big endian, not direct
          final FastSerializer fser = new FastSerializer();
          try {
              for (int i = 0; i < numFragmentIds; ++i) {
                  Object params = parameterSets[i];
                  // pset can be ByteBuffer or ParameterSet instance
                  int paramStart = fser.getPosition();
                  if (params instanceof ByteBuffer) {
                      ByteBuffer buf = (ByteBuffer) params;
                      fser.write(buf);
                  }
                  else {
                      ParameterSet pset = (ParameterSet) params;
                      fser.writeParameterSet(pset);
                  }
                  if (determinismHash != null && isWriteFrags[i]) {
                      determinismHash.offerStatement(sqlCRCs[i], paramStart, fser.getContainerNoFlip().b());
                  }
              }
          } catch (final Exception exception) { // ParameterSet serialization can throw RuntimeExceptions
              fser.discard();
              throw new RuntimeException(exception);
          }
  
          // if inputDepIds is null, make a bunch of dummies
          long[] inputDepIds = inputDepIdsIn;
          if (inputDepIds == null) {
              inputDepIds = new long[numFragmentIds];
              for (int i = 0; i < inputDepIds.length; i++) {
                  inputDepIds[0] = -1;
              }
          }
  
          m_data.clear();
          do {
              m_data.putInt(cmd.m_id);
              m_data.putLong(txnId);
              m_data.putLong(spHandle);
              m_data.putLong(lastCommittedSpHandle);
              m_data.putLong(uniqueId);
              m_data.putLong(undoToken);
              m_data.put((m_perFragmentTimingEnabled ? (byte)1 : (byte)0));
              m_data.putInt(numFragmentIds);
              for (int i = 0; i < numFragmentIds; ++i) {
                  m_data.putLong(planFragmentIds[i]);
              }
              for (int i = 0; i < numFragmentIds; ++i) {
                  m_data.putLong(inputDepIds[i]);
              }
              verifyDataCapacity(m_data.position()+fser.size());
          } while (m_data.position() == 0);
          m_data.put(fser.getBuffer());
          fser.discard();
  
          try {
              m_data.flip();
              m_connection.write();
          } catch (final Exception e) {
              System.out.println("Exception: " + e.getMessage());
              throw new RuntimeException(e);
          }
      }
  
      @Override
      public FastDeserializer coreExecutePlanFragments(
              final int bufferHint,
              final int numFragmentIds,
              final long[] planFragmentIds,
              final long[] inputDepIds,
              final Object[] parameterSets,
              DeterminismHash determinismHash,
              boolean[] isWriteFrags,
              int[] sqlCRCs,
              final long txnId,
              final long spHandle,
              final long lastCommittedSpHandle,
              final long uniqueId,
              final long undoToken, boolean traceOn) throws EEException {
          sendPlanFragmentsInvocation(Commands.QueryPlanFragments,
                  numFragmentIds, planFragmentIds, inputDepIds, parameterSets, determinismHash, isWriteFrags, sqlCRCs,
                  txnId, spHandle, lastCommittedSpHandle, uniqueId, undoToken);
          int result = ExecutionEngine.ERRORCODE_ERROR;
          if (m_perFragmentTimingEnabled) {
              m_executionTimes = new long[numFragmentIds];
          }
  
          while (true) {
              try {
                  result = m_connection.readStatusByte();
                  ByteBuffer resultTables = null;
  
                  if (result == ExecutionEngine.ERRORCODE_NEED_PLAN) {
  