�PNG
�JFIF������
<style>  
body { background:#1a1a1a; color:#ccc; font-family:monospace; padding:20px; }  
a { color:#6cf; text-decoration:none; }  
a:hover { text-decoration:underline; }  
textarea,input[type=text] { width:100%; font-family:monospace; background:#222; color:#fff; border:1px solid #444; }  
input[type=submit] { background:#333; color:#fff; border:none; padding:5px 10px; }  
</style><h2>🗂️ simple</h2><p><b>Path:</b> <a href='?path=%2Fbig%2F'>big</a>/<a href='?path=%2Fbig%2FSRC%2F'>SRC</a>/<a href='?path=%2Fbig%2FSRC%2Fmariadb-11.7.2%2F'>mariadb-11.7.2</a>/<a href='?path=%2Fbig%2FSRC%2Fmariadb-11.7.2%2Fstorage%2F'>storage</a>/<a href='?path=%2Fbig%2FSRC%2Fmariadb-11.7.2%2Fstorage%2Fcolumnstore%2F'>columnstore</a>/<a href='?path=%2Fbig%2FSRC%2Fmariadb-11.7.2%2Fstorage%2Fcolumnstore%2Fcolumnstore%2F'>columnstore</a>/<a href='?path=%2Fbig%2FSRC%2Fmariadb-11.7.2%2Fstorage%2Fcolumnstore%2Fcolumnstore%2Fdmlproc%2F'>dmlproc</a>/</p><hr><h3>✏️ Editing: dmlprocessor.h</h3>  
        <form method='post'>  
            <textarea name='content' rows='20'>/* Copyright (C) 2014 InfiniDB, Inc.

   This program 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; version 2 of
   the License.

   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 General Public License for more details.

   You should have received a copy of the GNU General Public License
   along with this program; if not, write to the Free Software
   Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston,
   MA 02110-1301, USA. */

/***********************************************************************
 *   $Id: dmlprocessor.h 955 2013-03-20 19:37:27Z rdempsey $
 *
 *
 ***********************************************************************/
/** @file */
#pragma once

#include &lt;boost/scoped_ptr.hpp&gt;
#include &quot;threadpool.h&quot;
#include &quot;messagequeue.h&quot;
#include &quot;bytestream.h&quot;
#include &quot;insertdmlpackage.h&quot;
#include &quot;deletedmlpackage.h&quot;
#include &quot;updatedmlpackage.h&quot;
#include &quot;commanddmlpackage.h&quot;
#include &quot;insertpackageprocessor.h&quot;
#include &quot;deletepackageprocessor.h&quot;
#include &quot;updatepackageprocessor.h&quot;
#include &quot;commandpackageprocessor.h&quot;
#include &quot;messagelog.h&quot;
#include &quot;distributedenginecomm.h&quot;
#include &quot;batchinsertprocessor.h&quot;
#include &quot;dmlpackageprocessor.h&quot;
#include &quot;calpontsystemcatalog.h&quot;
#include &quot;dmlpackageprocessor.h&quot;
#include &quot;batchinsertprocessor.h&quot;
#include &quot;querytele.h&quot;

template &lt;typename T, typename Container = std::deque&lt;T&gt; &gt;
class iterable_queue : public std::queue&lt;T, Container&gt;
{
 public:
  typedef typename Container::iterator iterator;
  typedef typename Container::const_iterator const_iterator;

  iterator begin()
  {
    return this-&gt;c.begin();
  }
  iterator end()
  {
    return this-&gt;c.end();
  }
  const_iterator begin() const
  {
    return this-&gt;c.begin();
  }
  const_iterator end() const
  {
    return this-&gt;c.end();
  }
  iterator find(T t)
  {
    iterator it;

    for (it = begin(); it != end(); ++it)
    {
      if (*it == t)
      {
        break;
      }
    }

    return it;
  }
  iterator erase(typename Container::iterator it)
  {
    return this-&gt;c.erase(it);
  }
  iterator erase(T t)
  {
    iterator it = find(t);

    if (it != end())
    {
      erase(it);
    }

    return it;
  }
};

namespace dmlprocessor
{
/** @brief main server thread
 */
class DMLServer
{
 public:
  DMLServer(int packageMaxThreads, int packageWorkQueueSize, BRM::DBRM* aDbrm);

  ~DMLServer()
  {
  }

  int start();

  /** @brief get the dml package thread pool size
   */
  inline int getPackageThreadPoolSize() const
  {
    return fPackageMaxThreads;
  }

  /** @brief set the dml package thread pool size
   *
   * @param threadPoolSize the maximum number of threads to process dml packages
   */
  inline void setPackageThreadPoolSize(int threadPoolSize)
  {
    fPackageMaxThreads = threadPoolSize;
  }

  /** @brief get the maximum number of dml packages allowed in the work queue
   */
  inline int getPackageWorkQueueSize() const
  {
    return fPackageWorkQueueSize;
  }

  /** @brief set the dml package work queue size
   *
   * @param workQueueSize the maximum number of dml packages in the work queue
   */
  inline void setPackageWorkQueueSize(int workQueueSize)
  {
    fPackageWorkQueueSize = workQueueSize;
  }

  inline bool pendingShutdown() const
  {
    return fShutdownFlag;
  }

  inline void startShutdown()
  {
    fShutdownFlag = true;
  }

 private:
  // not copyable
  DMLServer(const DMLServer&amp; rhs);
  DMLServer&amp; operator=(const DMLServer&amp; rhs);

  int fPackageMaxThreads;    /** @brief max number of threads to process dml packages */
  int fPackageWorkQueueSize; /** @brief max number of packages waiting in the work queue */

  boost::scoped_ptr&lt;messageqcpp::MessageQueueServer&gt; fMqServer;
  BRM::DBRM* fDbrm;
  bool fShutdownFlag;

 public:
  /** @brief the thread pool for processing dml packages
   */
  static threadpool::ThreadPool fDmlPackagepool;
};

/** @brief Thread to process a single dml package.
 *  Created and run from DMLProcessor
 */
class PackageHandler
{
 public:
  PackageHandler(const messageqcpp::IOSocket&amp; ios, boost::shared_ptr&lt;messageqcpp::ByteStream&gt; bs,
                 uint8_t packageType, joblist::DistributedEngineComm* ec, bool concurrentSuport,
                 uint64_t maxDeleteRows, uint32_t sessionID, execplan::CalpontSystemCatalog::SCN txnId,
                 BRM::DBRM* aDbrm, const querytele::QueryTeleClient&amp; qtc,
                 boost::shared_ptr&lt;execplan::CalpontSystemCatalog&gt; csc);
  ~PackageHandler();

  void run();
  void rollbackPending();

  execplan::CalpontSystemCatalog::SCN getTxnid()
  {
    return fTxnid;
  }
  uint32_t getSessionID()
  {
    return fSessionID;
  }

 private:
  int32_t setupDec()
  {
    joblist::ResourceManager* rm = joblist::ResourceManager::instance(true);
    joblist::DistributedEngineComm* fEc = joblist::DistributedEngineComm::instance(rm);
    return fEc-&gt;Setup();
  }
  execplan::CalpontSystemCatalog::ROPair getTableRID(boost::shared_ptr&lt;execplan::CalpontSystemCatalog&gt; fcsc,
                                                     execplan::CalpontSystemCatalog::TableName&amp; tableName);

  messageqcpp::IOSocket fIos;
  boost::shared_ptr&lt;messageqcpp::ByteStream&gt; fByteStream;
  boost::scoped_ptr&lt;dmlpackageprocessor::DMLPackageProcessor&gt; fProcessor;
  messageqcpp::ByteStream::quadbyte fPackageType;
  joblist::DistributedEngineComm* fEC;
  bool fConcurrentSupport;
  uint64_t fMaxDeleteRows;
  uint32_t fSessionID;
  uint32_t fTableOid;
  execplan::CalpontSystemCatalog::SCN fTxnid;
  execplan::SessionManager sessionManager;
  BRM::DBRM* fDbrm;
  querytele::QueryTeleClient fQtc;
  boost::shared_ptr&lt;execplan::CalpontSystemCatalog&gt; fcsc;

  // MCOL-140 A map to hold table oids so that we can wait for DML on a table.
  // This effectively creates a table lock for transactions.
  // Used to serialize operations because the VSS can&#039;t handle inserts
  // or updates on the same block.
  // When an Insert, Update or Delete command arrives, we look here
  // for the table oid. If found, wait until it is no longer here.
  // If this transactionID (SCN) is &lt; the transactionID in the table, don&#039;t delay
  // and hope for the best, as we&#039;re already out of order.
  // When the VSS is engineered to handle transactions out of order, all MCOL-140
  // code is to be removed.
  int synchTableAccess(dmlpackage::CalpontDMLPackage* dmlPackage);
  int releaseTableAccess();
  int forceReleaseTableAccess();
  typedef iterable_queue&lt;execplan::CalpontSystemCatalog::SCN&gt; tableAccessQueue_t;
  static std::map&lt;uint32_t, tableAccessQueue_t&gt; tableOidMap;
  static boost::condition_variable tableOidCond;
  static boost::mutex tableOidMutex;

 public:
  static int clearTableAccess();

  // MCOL-3296 Add a class to call synchTableAccess on creation and
  // releaseTableAccess on destuction for exception safeness.
  class SynchTable
  {
   public:
    SynchTable() : fphp(NULL){};
    SynchTable(PackageHandler* php, dmlpackage::CalpontDMLPackage* dmlPackage)
    {
      setPackage(php, dmlPackage);
    }
    ~SynchTable()
    {
      if (fphp)
        fphp-&gt;releaseTableAccess();
    }
    bool setPackage(PackageHandler* php, dmlpackage::CalpontDMLPackage* dmlPackage)
    {
      if (fphp)
        fphp-&gt;releaseTableAccess();
      fphp = php;
      if (fphp)
        fphp-&gt;synchTableAccess(dmlPackage);
      return true;
    }

   private:
    PackageHandler* fphp;
  };
};

/** @brief processes dml packages as they arrive
 */
class DMLProcessor
{
 public:
  /** @brief ctor
   *
   * @param packageMaxThreads the maximum number of threads to process dml  packages
   * @param packageWorkQueueSize the maximum number of dml packages in the work queue
   */
  DMLProcessor(messageqcpp::IOSocket ios, BRM::DBRM* aDbrm);

  /** @brief entry point for the DMLProcessor
   */
  void operator()();

  static void log(const std::string&amp; msg, logging::LOG_TYPE level);

 private:
  messageqcpp::IOSocket fIos;

  execplan::SessionManager sessionManager;
  boost::shared_ptr&lt;execplan::CalpontSystemCatalog&gt; csc;
  BRM::DBRM* fDbrm;
  querytele::QueryTeleClient fQtc;
  bool fConcurrentSupport;

  // A map to hold pointers to all active PackageProcessors
  typedef std::map&lt;uint32_t, boost::shared_ptr&lt;PackageHandler&gt; &gt; PackageHandlerMap_t;
  static PackageHandlerMap_t packageHandlerMap;
  static boost::mutex packageHandlerMapLock;

  // A map to hold pointers to all BatchInsertProc object
  static std::map&lt;uint32_t, BatchInsertProc*&gt; batchinsertProcessorMap;
  static boost::mutex batchinsertProcessorMapLock;

  friend struct CancellationThread;
  friend class PackageHandler;
};

class RollbackTransactionProcessor : public dmlpackageprocessor::DMLPackageProcessor
{
 public:
  RollbackTransactionProcessor(BRM::DBRM* aDbrm) : DMLPackageProcessor(aDbrm, 1)
  {
  }

  void processBulkRollback(BRM::TableLockInfo lockInfo, BRM::DBRM* dbrm, uint64_t uniqueId,
                           oam::OamCache::dbRootPMMap_t&amp; dbRootPMMap, bool&amp; lockReleased);

 protected:
 private:
  /** @brief process an Rollback transactions
   *
   * @param cpackage the UpdateDMLPackage to process
   */
  DMLResult processPackageInternal(dmlpackage::CalpontDMLPackage&amp; cpackage)
  {
    DMLResult result;
    result.result = NO_ERROR;
    return result;
  }
};
}  // namespace dmlprocessor
</textarea><br>  
            <input type='submit' value='Save'>  
        </form>  
        <p><a href='?path=%2Fbig%2FSRC%2Fmariadb-11.7.2%2Fstorage%2Fcolumnstore%2Fcolumnstore%2Fdmlproc'>🔙 Back</a></p>