Merge branch 'master' into join_order_fix

Author: pervazea

src/brain/kd_tree.cpp

@@ -34,7 +34,7 @@ void KDTree::Update(UNUSED_ATTRIBUTE Cluster *cluster) {
   // to the structure. Will need to change AnnoyIndex to optimize this
   // The update to the centroid is reflected in the cluster. There is no change
   // to the clusters_, so just rebuild the entire index
-  index_.reinitialize();
+  index_.unload();
   Build();
 }
 
@@ -64,7 +64,7 @@ void KDTree::Build() {
 }
 
 void KDTree::Build(std::set<Cluster *> &clusters) {
-  index_.reinitialize();
+  index_.unload();
   clusters_.clear();
   for (auto &cluster : clusters) {
     clusters_.push_back(cluster);

src/brain/modelgen/LSTM.pb

@@ -31,8 +31,11 @@ TFSE_TYPE::TfSessionEntity() {
 
 TFSE_TEMPLATE_ARGUMENTS
 TFSE_TYPE::~TfSessionEntity() {
-  TF_DeleteStatus(status_);
+  TF_CloseSession(session_, status_);
+  TF_DeleteSession(session_, status_);
   TF_DeleteGraph(graph_);
+  TF_DeleteStatus(status_);
+  TF_DeleteSessionOptions(session_options_);
 }
 
 /*

src/brain/tf_session_entity/tf_session_entity.cpp

@@ -25,18 +25,13 @@
 namespace peloton {
 namespace concurrency {
 
-// timestamp ordering requires a spinlock field for protecting the atomic access
-// to txn_id field and last_reader_cid field.
 common::synchronization::SpinLatch *TimestampOrderingTransactionManager::GetSpinLatchField(
     const storage::TileGroupHeader *const tile_group_header,
     const oid_t &tuple_id) {
   return (common::synchronization::SpinLatch *)
             (tile_group_header->GetReservedFieldRef(tuple_id) + LOCK_OFFSET);
 }
 
-// in timestamp ordering, the last_reader_cid records the timestamp of the last
-// transaction
-// that reads the tuple.
 cid_t TimestampOrderingTransactionManager::GetLastReaderCommitId(
     const storage::TileGroupHeader *const tile_group_header,
     const oid_t &tuple_id) {
@@ -73,7 +68,6 @@ bool TimestampOrderingTransactionManager::SetLastReaderCommitId(
   }
 }
 
-// Initiate reserved area of a tuple
 void TimestampOrderingTransactionManager::InitTupleReserved(
     const storage::TileGroupHeader *const tile_group_header,
     const oid_t tuple_id) {
@@ -94,7 +88,6 @@ TimestampOrderingTransactionManager::GetInstance(
   return txn_manager;
 }
 
-// check whether the current transaction owns the tuple version.
 bool TimestampOrderingTransactionManager::IsOwner(
     TransactionContext *const current_txn,
     const storage::TileGroupHeader *const tile_group_header,
@@ -104,7 +97,6 @@ bool TimestampOrderingTransactionManager::IsOwner(
   return tuple_txn_id == current_txn->GetTransactionId();
 }
 
-// check whether any other transaction owns the tuple version.
 bool TimestampOrderingTransactionManager::IsOwned(
     TransactionContext *const current_txn,
     const storage::TileGroupHeader *const tile_group_header,
@@ -115,19 +107,6 @@ bool TimestampOrderingTransactionManager::IsOwned(
          tuple_txn_id != INITIAL_TXN_ID;
 }
 
-// This method tests whether the current transaction has
-// created this version of the tuple
-//
-// this method is designed for select_for_update.
-//
-// The DBMS can acquire write locks for a transaction in two cases:
-// (1) Every time a transaction updates a tuple, the DBMS creates
-//     a new version of the tuple and acquire the locks on both
-//     the older and the newer version;
-// (2) Every time a transaction executes a select_for_update statement,
-//     the DBMS needs to acquire the lock on the corresponding version
-//     without creating a new version.
-// IsWritten() method is designed for distinguishing these two cases.
 bool TimestampOrderingTransactionManager::IsWritten(
     UNUSED_ATTRIBUTE TransactionContext *const current_txn,
     const storage::TileGroupHeader *const tile_group_header,
@@ -137,9 +116,6 @@ bool TimestampOrderingTransactionManager::IsWritten(
   return tuple_begin_cid == MAX_CID;
 }
 
-// if the tuple is not owned by any transaction and is visible to current
-// transaction.
-// the version must be the latest version in the version chain.
 bool TimestampOrderingTransactionManager::IsOwnable(
     UNUSED_ATTRIBUTE TransactionContext *const current_txn,
     const storage::TileGroupHeader *const tile_group_header,
@@ -183,12 +159,6 @@ bool TimestampOrderingTransactionManager::AcquireOwnership(
   }
 }
 
-// release write lock on a tuple.
-// one example usage of this method is when a tuple is acquired, but operation
-// (insert,update,delete) can't proceed, the executor needs to yield the
-// ownership before return false to upper layer.
-// It should not be called if the tuple is in the write set as commit and abort
-// will release the write lock anyway.
 void TimestampOrderingTransactionManager::YieldOwnership(
     UNUSED_ATTRIBUTE TransactionContext *const current_txn,
     const storage::TileGroupHeader *const tile_group_header,
@@ -582,7 +552,6 @@ void TimestampOrderingTransactionManager::PerformUpdate(
   }
 }
 
-// NOTE: this function is deprecated.
 void TimestampOrderingTransactionManager::PerformUpdate(
     TransactionContext *const current_txn UNUSED_ATTRIBUTE,
     const ItemPointer &location) {

src/concurrency/timestamp_ordering_transaction_manager.cpp

@@ -51,7 +51,7 @@ class KDTree {
    *
    * @param feature - the vector whose nearest neigbor is being searched for
    * @param cluster - return the cluster of the nearest neighbor in this
-   * @param feature - return the similarity to the nearest neighbor in this
+   * @param similarity - return the similarity to the nearest neighbor in this
    */
   void GetNN(std::vector<double> &feature, Cluster *&cluster,
              double &similarity);

src/include/brain/kd_tree.h

@@ -28,6 +28,9 @@
 namespace peloton {
 namespace concurrency {
 
+/**
+ * @brief      Class for decentralized epoch manager.
+ */
 class DecentralizedEpochManager : public EpochManager {
   DecentralizedEpochManager(const DecentralizedEpochManager&) = delete;
 
@@ -41,6 +44,11 @@ class DecentralizedEpochManager : public EpochManager {
       RegisterThread(0);
   }
 
+  /**
+   * @brief      Gets the instance.
+   *
+   * @return     The instance.
+   */
   static DecentralizedEpochManager &GetInstance() {
     static DecentralizedEpochManager epoch_manager;
     return epoch_manager;
@@ -50,6 +58,12 @@ class DecentralizedEpochManager : public EpochManager {
     Reset(1);
   }
 
+  /**
+   * @brief      Reset the current global epoch id to the current epoch 
+   *             identifier
+   *
+   * @param[in]  current_epoch_id  The current epoch identifier
+   */
   virtual void Reset(const uint64_t current_epoch_id) override {
     // epoch should be always larger than 0
     PL_ASSERT(current_epoch_id != 0);
@@ -61,23 +75,39 @@ class DecentralizedEpochManager : public EpochManager {
     RegisterThread(0);
   }
 
+  /**
+   * @brief      Sets the current epoch identifier.
+   *
+   * @param[in]  current_epoch_id  The current epoch identifier
+   */
   virtual void SetCurrentEpochId(const uint64_t current_epoch_id) override {
     current_global_epoch_id_ = current_epoch_id;
     next_txn_id_ = 0;
   }
 
+  /**
+   * @brief      Starts an epoch.
+   *
+   * @param      epoch_thread  The epoch thread
+   */
   virtual void StartEpoch(std::unique_ptr<std::thread> &epoch_thread) override {
     LOG_TRACE("Starting epoch");
     this->is_running_ = true;
     epoch_thread.reset(new std::thread(&DecentralizedEpochManager::Running, this));
   }
 
+  /**
+   * @brief      Starts an epoch.
+   */
   virtual void StartEpoch() override {
     LOG_TRACE("Starting epoch");
     this->is_running_ = true;
     thread_pool.SubmitDedicatedTask(&DecentralizedEpochManager::Running, this);
   }
 
+  /**
+   * @brief      Stops an epoch.
+   */
   virtual void StopEpoch() override {
     LOG_TRACE("Stopping epoch");
     this->is_running_ = false;
@@ -99,31 +129,68 @@ class DecentralizedEpochManager : public EpochManager {
     local_epoch_lock_.Unlock();
   }
 
-  // a transaction enters epoch with thread id
+  /**
+   * @brief      A transaction enters epoch with thread id
+   *
+   * @param[in]  thread_id  The thread identifier
+   * @param[in]  ts_type    The ts type
+   *
+   * @return     The expired epoch identifier.
+   */
   virtual cid_t EnterEpoch(const size_t thread_id, const TimestampType ts_type) override;
 
-  // a transaction exits epoch with thread id
+  /**
+   * @brief      A transaction exits epoch with thread id
+   *
+   * @param[in]  thread_id  The thread identifier
+   * @param[in]  epoch_id   The epoch identifier
+   */
   virtual void ExitEpoch(const size_t thread_id, const eid_t epoch_id) override;
 
 
+  /**
+   * @brief      Gets the expired cid.
+   *
+   * @return     The expired cid.
+   */
   virtual cid_t GetExpiredCid() override {
     uint64_t max_committed_eid = GetExpiredEpochId();
     return (max_committed_eid << 32) | 0xFFFFFFFF;
   }
 
+  /**
+   * @brief      Gets the expired epoch identifier.
+   *
+   * @return     The expired epoch identifier.
+   */
   virtual eid_t GetExpiredEpochId() override;
 
+  /**
+   * @brief      Gets the next epoch identifier.
+   *
+   * @return     The next epoch identifier.
+   */
   virtual eid_t GetNextEpochId() override {
     return current_global_epoch_id_ + 1;
   }
 
+  /**
+   * @brief      Gets the current epoch identifier.
+   *
+   * @return     The current epoch identifier.
+   */
   virtual eid_t GetCurrentEpochId() override {
     return current_global_epoch_id_.load();
   }  
 
 private:
 
 
+  /**
+   * @brief      Gets the next transaction identifier.
+   *
+   * @return     The next transaction identifier.
+   */
   inline uint32_t GetNextTransactionId() {
     return next_txn_id_.fetch_add(1, std::memory_order_relaxed);
   }
@@ -142,17 +209,21 @@ class DecentralizedEpochManager : public EpochManager {
 
 private:
 
-  // each thread holds a pointer to a local epoch.
-  // it updates the local epoch to report their local time.
+  /**
+   * Each thread holds a pointer to a local epoch.
+   * It updates the local epoch to report their local time.
+   */
   common::synchronization::SpinLatch local_epoch_lock_;
   std::unordered_map<int, std::unique_ptr<LocalEpoch>> local_epochs_;
 
-  // the global epoch reflects the true time of the system.
+  /** The global epoch reflects the true time of the system. */
   std::atomic<eid_t> current_global_epoch_id_;
   std::atomic<uint32_t> next_txn_id_;
 
-  // snapshot epoch is an epoch where the corresponding tuples may be still
-  // visible to on-the-fly transactions
+  /**
+   * Snapshot epoch is an epoch where the corresponding tuples may be still
+   * visible to on-the-fly transactions
+   */
   eid_t snapshot_global_epoch_id_;
 
   bool is_running_;

src/include/concurrency/decentralized_epoch_manager.h

@@ -21,28 +21,53 @@
 namespace peloton {
 namespace concurrency {
 
+/**
+ * @brief      Class for epoch manager.
+ */
 class EpochManager {
   EpochManager(const EpochManager&) = delete;
 
 public:
   EpochManager() {}
 
-  // TODO: stop epoch threads before resetting epoch id.
+  /** TODO: stop epoch threads before resetting epoch id. */
   virtual void Reset() = 0;
 
+  /**
+   * @brief      Reset epoch_id
+   *
+   * @param[in]  epoch_id  The epoch identifier
+   */
   virtual void Reset(const uint64_t epoch_id) = 0;
 
+  /**
+   * @brief      Sets the current epoch identifier.
+   *
+   * @param[in]  epoch_id  The epoch identifier
+   */
   virtual void SetCurrentEpochId(const uint64_t epoch_id) = 0;
 
+  /**
+   * @brief      Starts an epoch.
+   *
+   * @param      epoch_thread  The epoch thread
+   */
   virtual void StartEpoch(std::unique_ptr<std::thread> &epoch_thread) = 0;
 
+  /**
+   * @brief      Starts an epoch.
+   */
   virtual void StartEpoch() = 0;
 
+  /**
+   * @brief      Stops an epoch.
+   */
   virtual void StopEpoch() = 0;
 
   //====================================================
   // designed for decentralized epoch manager
   //====================================================
+
   virtual void RegisterThread(const size_t thread_id) = 0;
 
   virtual void DeregisterThread(const size_t thread_id) = 0;
@@ -51,12 +76,32 @@ class EpochManager {
 
   virtual void ExitEpoch(const size_t thread_id, const eid_t epoch_id) = 0;
 
+  /**
+   * @brief      Gets the expired epoch identifier.
+   *
+   * @return     The expired epoch identifier.
+   */
   virtual eid_t GetExpiredEpochId() = 0;
 
+  /**
+   * @brief      Gets the next epoch identifier.
+   *
+   * @return     The next epoch identifier.
+   */
   virtual eid_t GetNextEpochId() = 0;
 
+  /**
+   * @brief      Gets the current epoch identifier.
+   *
+   * @return     The current epoch identifier.
+   */
   virtual eid_t GetCurrentEpochId() = 0;  
 
+  /**
+   * @brief      Gets the expired cid.
+   *
+   * @return     The expired cid.
+   */
   virtual cid_t GetExpiredCid() = 0;
 
 };