gt:                 SiteTasker task = m_scheduler.take();
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.iv2;
  
  import java.util.List;
  import java.util.Map;
  import java.util.concurrent.Future;
  
  import org.voltcore.logging.VoltLogger;
  import org.voltcore.utils.CoreUtils;
  import org.voltcore.utils.DBBPool;
  import org.voltcore.utils.Pair;
  import org.voltdb.BackendTarget;
  import org.voltdb.CatalogContext;
  import org.voltdb.DRConsumerDrIdTracker;
  import org.voltdb.DRConsumerDrIdTracker.DRSiteDrIdTracker;
  import org.voltdb.DRIdempotencyResult;
  import org.voltdb.DependencyPair;
  import org.voltdb.HsqlBackend;
  import org.voltdb.LoadedProcedureSet;
  import org.voltdb.NonVoltDBBackend;
  import org.voltdb.ParameterSet;
  import org.voltdb.PartitionDRGateway;
  import org.voltdb.PostGISBackend;
  import org.voltdb.PostgreSQLBackend;
  import org.voltdb.ProcedureRunner;
  import org.voltdb.SiteProcedureConnection;
  import org.voltdb.SiteSnapshotConnection;
  import org.voltdb.StatsSelector;
  import org.voltdb.SystemProcedureExecutionContext;
  import org.voltdb.TableStreamType;
  import org.voltdb.TheHashinator;
  import org.voltdb.TupleStreamStateInfo;
  import org.voltdb.VoltDB;
  import org.voltdb.VoltProcedure.VoltAbortException;
  import org.voltdb.VoltTable;
  import org.voltdb.catalog.Cluster;
  import org.voltdb.catalog.Column;
  import org.voltdb.catalog.Database;
  import org.voltdb.catalog.Procedure;
  import org.voltdb.catalog.Table;
  import org.voltdb.dtxn.SiteTracker;
  import org.voltdb.dtxn.TransactionState;
  import org.voltdb.dtxn.UndoAction;
  import org.voltdb.exceptions.EEException;
  import org.voltdb.messaging.FastDeserializer;
  import org.voltdb.settings.ClusterSettings;
  import org.voltdb.settings.NodeSettings;
  import org.voltdb.sysprocs.NibbleDeleteBase.ComparisonConstant;
  
  /**
   * An implementation of Site which provides only the functionality
   * necessary to run read-only multi-partition transactions.  A pool
   * of these will be used to run multiple read-only transactions
   * concurrently.
   */
  public class MpRoSite implements Runnable, SiteProcedureConnection
  {
      @SuppressWarnings("unused")
      private static final VoltLogger tmLog = new VoltLogger("TM");
  
      // Set to false trigger shutdown.
      volatile boolean m_shouldContinue = true;
  
      // HSId of this site's initiator.
      final long m_siteId;
  
      // What type of EE is controlled
      final BackendTarget m_backend;
  
      // Manages pending tasks.
      final SiteTaskerQueue m_scheduler;
  
      // Still need m_non_voltdb_backend (formerly m_hsql) here
      NonVoltDBBackend m_non_voltdb_backend;
  
      // Current catalog
      volatile CatalogContext m_context;
  
      // Currently available procedure
      volatile LoadedProcedureSet m_loadedProcedures;
  
      // Current topology
      int m_partitionId;
  
      //a place holder for current running transaction on this site
      //the transaction will be terminated upon node shutdown.
      private TransactionState m_txnState = null;
  
      @Override
      public long getLatestUndoToken()
      {
          throw new RuntimeException("Not needed for RO MP Site, shouldn't be here.");
      }
  
      SiteProcedureConnection getSiteProcedureConnection()
      {
          return this;
      }
  
      /**
       * SystemProcedures are "friends" with ExecutionSites and granted
       * access to internal state via m_systemProcedureContext.
       *
       * The only sysproc which should run on the RO MP Site is Adhoc.  Everything
       * else will yell at you.
       */
      SystemProcedureExecutionContext m_sysprocContext = new SystemProcedureExecutionContext() {
          @Override
          public ClusterSettings getClusterSettings() {
              throw new RuntimeException("Not needed for RO MP Site, shouldn't be here.");
          }
  
          @Override
          public NodeSettings getPaths() {
              throw new RuntimeException("Not needed for RO MP Site, shouldn't be here.");
          }
  
          @Override
          public Database getDatabase() {
              throw new RuntimeException("Not needed for RO MP Site, shouldn't be here.");
          }
  
          @Override
          public Cluster getCluster() {
              throw new RuntimeException("Not needed for RO MP Site, shouldn't be here.");
          }
  
          @Override
          public long getSpHandleForSnapshotDigest() {
              throw new RuntimeException("Not needed for RO MP Site, shouldn't be here.");
          }
  
          @Override
          public long getSiteId() {
              throw new RuntimeException("Not needed for RO MP Site, shouldn't be here.");
          }
  
          @Override
          public int getLocalSitesCount() {
              throw new RuntimeException("Not needed for RO MP Site, shouldn't be here.");
          }
  
          @Override
          public boolean isLowestSiteId()
          {
              throw new RuntimeException("Not needed for RO MP Site, shouldn't be here.");
          }
  
          @Override
          public int getClusterId()
          {
              return getCorrespondingClusterId();
          }
  
          @Override
          public int getHostId() {
              throw new RuntimeException("Not needed for RO MP Site, shouldn't be here.");
          }
  
          @Override
          public int getPartitionId() {
              throw new RuntimeException("Not needed for RO MP Site, shouldn't be here.");
          }
  
          @Override
          public long getCatalogCRC() {
              throw new RuntimeException("Not needed for RO MP Site, shouldn't be here.");
          }
  
          @Override
          public byte[] getCatalogHash() {
              // AdHoc invocations need to be able to check the hash of the current catalog
              // against the hash of the catalog they were planned against.
              return m_context.getCatalogHash();
          }
  
          @Override
          public byte[] getDeploymentHash() {
              throw new RuntimeException("Not needed for RO MP Site, shouldn't be here.");
          }
  
          // Needed for Adhoc queries
          @Override
          public int getCatalogVersion() {
              return m_context.catalogVersion;
          }
  
          @Override
          public SiteTracker getSiteTrackerForSnapshot() {
              throw new RuntimeException("Not needed for RO MP Site, shouldn't be here.");
          }
  
          @Override
          public int getNumberOfPartitions() {
              throw new RuntimeException("Not needed for RO MP Site, shouldn't be here.");
          }
  
          @Override
          public void setNumberOfPartitions(int partitionCount) {
              throw new RuntimeException("Not needed for RO MP Site, shouldn't be here.");
          }
  
          @Override
          public SiteProcedureConnection getSiteProcedureConnection()
          {
              throw new RuntimeException("Not needed for RO MP Site, shouldn't be here.");
          }
  
          @Override
          public SiteSnapshotConnection getSiteSnapshotConnection()
          {
              throw new RuntimeException("Not needed for RO MP Site, shouldn't be here.");
          }
  
          @Override
          public void updateBackendLogLevels() {
              throw new RuntimeException("RO MP Site doesn't do this, shouldn't be here.");
          }
  
          @Override
          public boolean updateCatalog(String diffCmds, CatalogContext context,
                  boolean requiresSnapshotIsolation, long txnId, long uniqueId, long spHandle, boolean isReplay,
                  boolean requireCatalogDiffCmdsApplyToEE, boolean requiresNewExportGeneration)
          {
              throw new RuntimeException("RO MP Site doesn't do this, shouldn't be here.");
          }
  
          @Override
          public boolean updateSettings(CatalogContext context)
          {
              throw new RuntimeException("RO MP Site doesn't do this, shouldn't be here.");
          }
  
          @Override
          public TheHashinator getCurrentHashinator()
          {
              throw new RuntimeException("RO MP Site doesn't do this, shouldn't be here.");
          }
  
          @Override
          public void updateHashinator(TheHashinator hashinator)
          {
              throw new RuntimeException("RO MP Site doesn't do this, shouldn't be here.");
          }
  
          @Override
          public boolean activateTableStream(int tableId, TableStreamType type, boolean undo, byte[] predicates)
          {
              throw new RuntimeException("RO MP Site doesn't do this, shouldn't be here.");
          }
  
          @Override
          public void forceAllDRNodeBuffersToDisk(final boolean nofsync)
          {
              throw new RuntimeException("RO MP Site doesn't do this, shouldn't be here.");
          }
  
          @Override
          public DRIdempotencyResult isExpectedApplyBinaryLog(int producerClusterId, int producerPartitionId,
                                                              long logId)
          {
              throw new RuntimeException("RO MP Site doesn't do this, shouldn't be here.");
          }
  
          @Override
          public void appendApplyBinaryLogTxns(int producerClusterId, int producerPartitionId,
                                               long localUniqueId, DRConsumerDrIdTracker tracker)
          {
              throw new RuntimeException("RO MP Site doesn't do this, shouldn't be here.");
          }
  
          @Override
          public void recoverWithDrAppliedTrackers(Map<Integer, Map<Integer, DRSiteDrIdTracker>> trackers)
          {
              throw new RuntimeException("RO MP Site doesn't do this, shouldn't be here.");
          }
  
          @Override
          public void resetDrAppliedTracker() {
              throw new RuntimeException("RO MP Site doesn't do this, shouldn't be here.");
          }
  
          @Override
          public void resetDrAppliedTracker(byte clusterId) {
              throw new RuntimeException("RO MP Site doesn't do this, shouldn't be here.");
          }
  
          @Override
          public boolean hasRealDrAppliedTracker(byte clusterId) {
              throw new RuntimeException("RO MP Site doesn't do this, shouldn't be here.");
          }
  
          @Override
          public void initDRAppliedTracker(Map<Byte, Integer> clusterIdToPartitionCountMap, boolean hasReplicatedStream) {
              throw new RuntimeException("RO MP Site doesn't do this, shouldn't be here.");
          }
  
          @Override
          public Map<Integer, Map<Integer, DRSiteDrIdTracker>> getDrAppliedTrackers()
          {
              throw new RuntimeException("RO MP Site doesn't do this, shouldn't be here.");
          }
  
          @Override
          public Pair<Long, Long> getDrLastAppliedUniqueIds()
          {
              throw new RuntimeException("RO MP Site doesn't do this, shouldn't be here.");
          }
  
          @Override
          public Pair<Long, int[]> tableStreamSerializeMore(int tableId, TableStreamType type,
                  List<DBBPool.BBContainer> outputBuffers)
          {
              throw new RuntimeException("RO MP Site doesn't do this, shouldn't be here.");
          }
  
          @Override
          public Procedure ensureDefaultProcLoaded(String procName) {
              throw new RuntimeException("RO MP Site doesn't do this, shouldn't be here.");
          }
  
          @Override
          public InitiatorMailbox getInitiatorMailbox() {
              throw new RuntimeException("RO MP Site doesn't do this, shouldn't be here.");
          }
      };
  
      /** Create a new RO MP execution site */
      public MpRoSite(
              SiteTaskerQueue scheduler,
              long siteId,
              BackendTarget backend,
              CatalogContext context,
              int partitionId)
      {
          m_siteId = siteId;
          m_context = context;
          m_partitionId = partitionId;
          m_scheduler = scheduler;
          m_backend = backend;
      }
  
      /** Update the loaded procedures. */
      void setLoadedProcedures(LoadedProcedureSet loadedProcedure)
      {
          m_loadedProcedures = loadedProcedure;
      }
  
      /** Thread specific initialization */
      void initialize()
      {
          if (m_backend == BackendTarget.HSQLDB_BACKEND) {
              m_non_voltdb_backend = HsqlBackend.initializeHSQLBackend(m_siteId,
                                                         m_context);
          }
          else if (m_backend == BackendTarget.POSTGRESQL_BACKEND) {
              m_non_voltdb_backend = PostgreSQLBackend.initializePostgreSQLBackend(m_context);
          }
          else if (m_backend == BackendTarget.POSTGIS_BACKEND) {
              m_non_voltdb_backend = PostGISBackend.initializePostGISBackend(m_context);
          }
          else {
              m_non_voltdb_backend = null;
          }
      }
  
      @Override
      public void run()
      {
          initialize();
  
          try {
              while (m_shouldContinue) {
                  // Normal operation blocks the site thread on the sitetasker queue.
                  m_txnState = null;
  