YugabyteDB (2.13.0.0-b42, bfc6a6643e7399ac8a0e81d06a3ee6d6571b33ab)

// Licensed to the Apache Software Foundation (ASF) under one

// or more contributor license agreements.  See the NOTICE file

// distributed with this work for additional information

// regarding copyright ownership.  The ASF licenses this file

// to you under the Apache License, Version 2.0 (the

// "License"); you may not use this file except in compliance

// with the License.  You may obtain a copy of the License at

//

//   http://www.apache.org/licenses/LICENSE-2.0

//

// Unless required by applicable law or agreed to in writing,

// software distributed under the License is distributed on an

// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY

// KIND, either express or implied.  See the License for the

// specific language governing permissions and limitations

// under the License.

//

// The following only applies to changes made to this file as part of YugaByte development.

//

// Portions Copyright (c) YugaByte, Inc.

//

// Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except

// in compliance with the License.  You may obtain a copy of the License at

//

// http://www.apache.org/licenses/LICENSE-2.0

//

// Unless required by applicable law or agreed to in writing, software distributed under the License

// is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express

// or implied.  See the License for the specific language governing permissions and limitations

// under the License.

//

#ifndef YB_TABLET_TABLET_PEER_H_

#define YB_TABLET_TABLET_PEER_H_

#include <atomic>

#include <future>

#include <map>

#include <memory>

#include <mutex>

#include <string>

#include <vector>

#include "yb/consensus/consensus_fwd.h"

#include "yb/consensus/consensus_context.h"

#include "yb/consensus/consensus_meta.h"

#include "yb/consensus/consensus_types.h"

#include "yb/gutil/callback.h"

#include "yb/gutil/ref_counted.h"

#include "yb/gutil/thread_annotations.h"

#include "yb/rpc/rpc_fwd.h"

#include "yb/tablet/tablet_fwd.h"

#include "yb/tablet/metadata.pb.h"

#include "yb/tablet/mvcc.h"

#include "yb/tablet/transaction_coordinator.h"

#include "yb/tablet/transaction_participant_context.h"

#include "yb/tablet/operations/operation_tracker.h"

#include "yb/tablet/preparer.h"

#include "yb/tablet/tablet_options.h"

#include "yb/tablet/write_query_context.h"

#include "yb/util/atomic.h"

#include "yb/util/semaphore.h"

using yb::consensus::StateChangeContext;

namespace yb {

namespace tserver {

class CatchUpServiceTest;

class UpdateTransactionResponsePB;

}

class MaintenanceManager;

class MaintenanceOp;

class ThreadPool;

namespace tablet {

struct TabletOnDiskSizeInfo {

  int64_t consensus_metadata_disk_size = 0;

  int64_t wal_files_disk_size = 0;

  int64_t sst_files_disk_size = 0;

  int64_t uncompressed_sst_files_disk_size = 0;

  int64_t sum_on_disk_size = 0;

  template <class PB>

  static TabletOnDiskSizeInfo FromPB(const PB& pb) {

    return {

      .consensus_metadata_disk_size = pb.consensus_metadata_disk_size(),

      .wal_files_disk_size = pb.wal_files_disk_size(),

      .sst_files_disk_size = pb.sst_files_disk_size(),

      .uncompressed_sst_files_disk_size = pb.uncompressed_sst_files_disk_size(),

      .sum_on_disk_size = pb.estimated_on_disk_size()

    };

  }

  template <class PB>

  void ToPB(PB* pb) const {

    pb->set_consensus_metadata_disk_size(consensus_metadata_disk_size);

    pb->set_wal_files_disk_size(wal_files_disk_size);

    pb->set_sst_files_disk_size(sst_files_disk_size);

    pb->set_uncompressed_sst_files_disk_size(uncompressed_sst_files_disk_size);

    pb->set_estimated_on_disk_size(sum_on_disk_size);

  }

  void operator+=(const TabletOnDiskSizeInfo& other) {

    consensus_metadata_disk_size += other.consensus_metadata_disk_size;

    wal_files_disk_size += other.wal_files_disk_size;

    sst_files_disk_size += other.sst_files_disk_size;

    uncompressed_sst_files_disk_size += other.uncompressed_sst_files_disk_size;

    sum_on_disk_size += other.sum_on_disk_size;

  }

  void RecomputeTotalSize() {

    sum_on_disk_size =

        consensus_metadata_disk_size +

        sst_files_disk_size +

        wal_files_disk_size;

  }

};

// A peer is a tablet consensus configuration, which coordinates writes to tablets.

// Each time Write() is called this class appends a new entry to a replicated

// state machine through a consensus algorithm, which makes sure that other

// peers see the same updates in the same order. In addition to this, this

// class also splits the work and coordinates multi-threaded execution.

class TabletPeer : public consensus::ConsensusContext,

                   public TransactionParticipantContext,

                   public TransactionCoordinatorContext,

                   public WriteQueryContext {

 public:

  typedef std::map<int64_t, int64_t> MaxIdxToSegmentSizeMap;

  // Creates TabletPeer.

  // `tablet_splitter` will be used for applying split tablet Raft operation.

  TabletPeer(

      const RaftGroupMetadataPtr& meta,

      const consensus::RaftPeerPB& local_peer_pb,

      const scoped_refptr<server::Clock>& clock,

      const std::string& permanent_uuid,

      Callback<void(std::shared_ptr<StateChangeContext> context)> mark_dirty_clbk,

      MetricRegistry* metric_registry,

      TabletSplitter* tablet_splitter,

      const std::shared_future<client::YBClient*>& client_future);

  ~TabletPeer();

  // Initializes the TabletPeer, namely creating the Log and initializing

  // Consensus.

  // split_op_id is the ID of split tablet Raft operation requesting split of this tablet or unset.

  CHECKED_STATUS InitTabletPeer(

      const TabletPtr& tablet,

      const std::shared_ptr<MemTracker>& server_mem_tracker,

      rpc::Messenger* messenger,

      rpc::ProxyCache* proxy_cache,

      const scoped_refptr<log::Log>& log,

      const scoped_refptr<MetricEntity>& table_metric_entity,

      const scoped_refptr<MetricEntity>& tablet_metric_entity,

      ThreadPool* raft_pool,

      ThreadPool* tablet_prepare_pool,

      consensus::RetryableRequests* retryable_requests,

      consensus::MultiRaftManager* multi_raft_manager);

  // Starts the TabletPeer, making it available for Write()s. If this

  // TabletPeer is part of a consensus configuration this will connect it to other peers

  // in the consensus configuration.

  CHECKED_STATUS Start(const consensus::ConsensusBootstrapInfo& info);

  // Starts shutdown process.

  // Returns true if shutdown was just initiated, false if shutdown was already running.

  MUST_USE_RESULT bool StartShutdown(IsDropTable is_drop_table = IsDropTable::kFalse);

  // Completes shutdown process and waits for it's completeness.

  void CompleteShutdown(IsDropTable is_drop_table = IsDropTable::kFalse);

  // Abort active transactions on the tablet after shutdown is initiated.

  CHECKED_STATUS AbortSQLTransactions();

  CHECKED_STATUS Shutdown(IsDropTable is_drop_table = IsDropTable::kFalse);

  // Check that the tablet is in a RUNNING state.

  CHECKED_STATUS CheckRunning() const;

  // Returns whether shutdown started. If shutdown already completed returns true as well.

  bool IsShutdownStarted() const;

  // Check that the tablet is in a SHUTDOWN/NOT_STARTED state.

  CHECKED_STATUS CheckShutdownOrNotStarted() const;

  // Wait until the tablet is in a RUNNING state or if there's a timeout.

  // TODO have a way to wait for any state?

  CHECKED_STATUS WaitUntilConsensusRunning(const MonoDelta& timeout);

  // Submits a write to a tablet and executes it asynchronously.

  // The caller is expected to build and pass a WriteOperation that points

  // to the RPC WriteRequest, WriteResponse, RpcContext and to the tablet's

  // MvccManager.

  // The operation_state is deallocated after use by this function.

  void WriteAsync(std::unique_ptr<WriteQuery> query);

  void Submit(std::unique_ptr<Operation> operation, int64_t term) override;

  void UpdateClock(HybridTime hybrid_time) override;

  std::unique_ptr<UpdateTxnOperation> CreateUpdateTransaction(

      TransactionStatePB* request) override;

  void SubmitUpdateTransaction(

      std::unique_ptr<UpdateTxnOperation> operation, int64_t term) override;

  HybridTime SafeTimeForTransactionParticipant() override;

  Result<HybridTime> WaitForSafeTime(HybridTime safe_time, CoarseTimePoint deadline) override;

  void GetLastReplicatedData(RemoveIntentsData* data) override;

  void GetLastCDCedData(RemoveIntentsData* data) override;

  void GetTabletStatusPB(TabletStatusPB* status_pb_out);

  // Used by consensus to create and start a new ReplicaOperation.

  CHECKED_STATUS StartReplicaOperation(

      const scoped_refptr<consensus::ConsensusRound>& round,

      HybridTime propagated_safe_time) override;

  // This is an override of a ConsensusContext method. This is called from

  // UpdateReplica -> EnqueuePreparesUnlocked on Raft heartbeats.

  void SetPropagatedSafeTime(HybridTime ht) override;

  // Returns false if it is preferable to don't apply write operation.

  bool ShouldApplyWrite() override;

  consensus::Consensus* consensus() const;

  consensus::RaftConsensus* raft_consensus() const;

  std::shared_ptr<consensus::Consensus> shared_consensus() const;

  std::shared_ptr<consensus::RaftConsensus> shared_raft_consensus() const;

  Tablet* tablet() const EXCLUDES(lock_) {

    std::lock_guard<simple_spinlock> lock(lock_);

    return tablet_.get();

  }

  TabletPtr shared_tablet() const {

    std::lock_guard<simple_spinlock> lock(lock_);

    return tablet_;

  }

  RaftGroupStatePB state() const {

    return state_.load(std::memory_order_acquire);

  }

  TabletDataState data_state() const;

  // Returns the current Raft configuration.

  consensus::RaftConfigPB RaftConfig() const;

  TabletStatusListener* status_listener() const {

    return status_listener_.get();

  }

  // Sets the tablet to a BOOTSTRAPPING state, indicating it is starting up.

  CHECKED_STATUS SetBootstrapping() {

    return UpdateState(RaftGroupStatePB::NOT_STARTED, RaftGroupStatePB::BOOTSTRAPPING, "");

  }

  CHECKED_STATUS UpdateState(RaftGroupStatePB expected, RaftGroupStatePB new_state,

                             const std::string& error_message);

  // sets the tablet state to FAILED additionally setting the error to the provided

  // one.

  void SetFailed(const Status& error);

  // Returns the error that occurred, when state is FAILED.

  CHECKED_STATUS error() const {

    Status *error;

    if ((error = error_.get(std::memory_order_acquire)) != nullptr) {

      // Once the error_ is set, we do not reset it to nullptr

      return *error;

    }

    return Status::OK();

  }

  // Returns a human-readable string indicating the state of the tablet.

  // Typically this looks like "NOT_STARTED", "TABLET_DATA_COPYING",

  // etc. For use in places like the Web UI.

  std::string HumanReadableState() const;

  // Adds list of transactions in-flight at the time of the call to

  // 'out'. OperationStatusPB objects are used to allow this method

  // to be used by both the web-UI and ts-cli.

  void GetInFlightOperations(Operation::TraceType trace_type,

                             std::vector<consensus::OperationStatusPB>* out) const;

  // Returns the minimum known log index that is in-memory or in-flight.

  // Used for selection of log segments to delete during Log GC.

  // If details is specified then this function appends explanation of how index was calculated

  // to it.

  Result<int64_t> GetEarliestNeededLogIndex(std::string* details = nullptr) const;

  // Returns the amount of bytes that would be GC'd if RunLogGC() was called.

  //

  // Returns a non-ok status if the tablet isn't running.

  CHECKED_STATUS GetGCableDataSize(int64_t* retention_size) const;

  // Returns true if it is safe to retrieve the log pointer using the log() function from this

  // tablet peer. Once the log pointer is initialized, it will stay valid for the lifetime of the

  // TabletPeer.

  bool log_available() const {

    return log_atomic_.load(std::memory_order_acquire) != nullptr;

  }

  // Return a pointer to the Log. TabletPeer keeps a reference to Log after Init(). This function

  // will crash if the log has not been initialized yet.

  log::Log* log() const;

  // Returns the OpId of the latest entry in the log, or a zero OpId if the log has not been

  // initialized.

  yb::OpId GetLatestLogEntryOpId() const;

  const server::ClockPtr& clock_ptr() const override {

    return clock_;

  }

  void Enqueue(rpc::ThreadPoolTask* task);

  void StrandEnqueue(rpc::StrandTask* task) override;

  const std::shared_future<client::YBClient*>& client_future() const override {

    return client_future_;

  }

  int64_t LeaderTerm() const override;

  consensus::LeaderStatus LeaderStatus(bool allow_stale = false) const;

  Result<HybridTime> LeaderSafeTime() const override;

  HybridTime HtLeaseExpiration() const override;

  const scoped_refptr<log::LogAnchorRegistry>& log_anchor_registry() const {

    return log_anchor_registry_;

  }

  // Returns the tablet_id of the tablet managed by this TabletPeer.

  // Returns the correct tablet_id even if the underlying tablet is not available

  // yet.

  const std::string& tablet_id() const override { return tablet_id_; }

  // Convenience method to return the permanent_uuid of this peer.

  const std::string& permanent_uuid() const override {

    return permanent_uuid_;

  }

  Result<OperationDriverPtr> NewOperationDriver(std::unique_ptr<Operation>* operation,

                                                int64_t term);

  Result<OperationDriverPtr> NewLeaderOperationDriver(

      std::unique_ptr<Operation>* operation, int64_t term);

  Result<OperationDriverPtr> NewReplicaOperationDriver(std::unique_ptr<Operation>* operation);

  // Tells the tablet's log to garbage collect.

  CHECKED_STATUS RunLogGC();

  // Register the maintenance ops associated with this peer's tablet, also invokes

  // Tablet::RegisterMaintenanceOps().

  void RegisterMaintenanceOps(MaintenanceManager* maintenance_manager);

  // Unregister the maintenance ops associated with this peer's tablet.

  // This method is not thread safe.

  void UnregisterMaintenanceOps();

  // Return pointer to the transaction tracker for this peer.

  const OperationTracker* operation_tracker() const { return &operation_tracker_; }

  const RaftGroupMetadataPtr& tablet_metadata() const {

    return meta_;

  }

  CHECKED_STATUS set_cdc_min_replicated_index(int64_t cdc_min_replicated_index);

  CHECKED_STATUS set_cdc_min_replicated_index_unlocked(int64_t cdc_min_replicated_index);

  CHECKED_STATUS reset_cdc_min_replicated_index_if_stale();

  TableType table_type();

  // Returns the number of segments in log_.

  size_t GetNumLogSegments() const;

  // Might update the can_be_deleted_.

  bool CanBeDeleted();

  std::string LogPrefix() const;

 protected:

  friend class RefCountedThreadSafe<TabletPeer>;

  friend class TabletPeerTest;

  FRIEND_TEST(TabletPeerTest, TestDMSAnchorPreventsLogGC);

  FRIEND_TEST(TabletPeerTest, TestActiveOperationPreventsLogGC);

  // Wait until the TabletPeer is fully in SHUTDOWN state.

  void WaitUntilShutdown();

  // After bootstrap is complete and consensus is setup this initiates the transactions

  // that were not complete on bootstrap.

  // Not implemented yet. See .cc file.

  CHECKED_STATUS StartPendingOperations(PeerRole my_role,

                                        const consensus::ConsensusBootstrapInfo& bootstrap_info);

  scoped_refptr<OperationDriver> CreateOperationDriver();

  virtual std::unique_ptr<Operation> CreateOperation(consensus::ReplicateMsg* replicate_msg);

  const RaftGroupMetadataPtr meta_;

  const std::string tablet_id_;

  const consensus::RaftPeerPB local_peer_pb_;

  // The atomics state_, error_ and has_consensus_ maintain information about the tablet peer.

  // While modifying the other fields in tablet peer, state_ is modified last.

  // error_ is set before state_ is set to an error state.

  std::atomic<enum RaftGroupStatePB> state_;

  AtomicUniquePtr<Status> error_;

  std::atomic<bool> has_consensus_ = {false};

  OperationTracker operation_tracker_;

  scoped_refptr<log::Log> log_;

  std::atomic<log::Log*> log_atomic_{nullptr};

  TabletPtr tablet_;

  rpc::ProxyCache* proxy_cache_;

  std::shared_ptr<consensus::RaftConsensus> consensus_;

  std::unique_ptr<TabletStatusListener> status_listener_;

  simple_spinlock prepare_replicate_lock_;

  // Lock protecting state_ as well as smart pointers to collaborating

  // classes such as tablet_ and consensus_.

  mutable simple_spinlock lock_;

  // Lock taken during Init/Shutdown which ensures that only a single thread

  // attempts to perform major lifecycle operations (Init/Shutdown) at once.

  // This must be acquired before acquiring lock_ if they are acquired together.

  // We don't just use lock_ since the lifecycle operations may take a while

  // and we'd like other threads to be able to quickly poll the state_ variable

  // during them in order to reject RPCs, etc.

  mutable simple_spinlock state_change_lock_;

  std::unique_ptr<Preparer> prepare_thread_;

  scoped_refptr<server::Clock> clock_;

  scoped_refptr<log::LogAnchorRegistry> log_anchor_registry_;

  // Function to mark this TabletPeer's tablet as dirty in the TSTabletManager.

  // This function must be called any time the cluster membership or cluster

  // leadership changes. Note that this function is called synchronously on the followers

  // or leader via the consensus round completion callback of NonTrackedRoundReplicationFinished.

  // Hence this should be a relatively lightweight function - e.g., update in-memory only state

  // and defer any other heavy duty operations to a thread pool.

  Callback<void(std::shared_ptr<consensus::StateChangeContext> context)> mark_dirty_clbk_;

  // List of maintenance operations for the tablet that need information that only the peer

  // can provide.

  std::vector<std::unique_ptr<MaintenanceOp>> maintenance_ops_;

  // Cache the permanent of the tablet UUID to retrieve it without a lock in the common case.

  const std::string permanent_uuid_;

  std::atomic<rpc::ThreadPool*> service_thread_pool_{nullptr};

  AtomicUniquePtr<rpc::Strand> strand_;

  OperationCounter preparing_operations_counter_;

  // Serializes access to set_cdc_min_replicated_index and reset_cdc_min_replicated_index_if_stale

  // and protects cdc_min_replicated_index_refresh_time_ for reads and writes.

  mutable simple_spinlock cdc_min_replicated_index_lock_;

  MonoTime cdc_min_replicated_index_refresh_time_ = MonoTime::Min();

 private:

  Result<HybridTime> ReportReadRestart() override;

  Result<FixedHybridTimeLease> HybridTimeLease(HybridTime min_allowed, CoarseTimePoint deadline);

  Result<HybridTime> PreparePeerRequest() override;

  void MajorityReplicated() override;

  void ChangeConfigReplicated(const consensus::RaftConfigPB& config) override;

  uint64_t NumSSTFiles() override;

  void ListenNumSSTFilesChanged(std::function<void()> listener) override;

  rpc::Scheduler& scheduler() const override;

  CHECKED_STATUS CheckOperationAllowed(

      const OpId& op_id, consensus::OperationType op_type) override;

  // Return granular types of on-disk size of this tablet replica, in bytes.

  TabletOnDiskSizeInfo GetOnDiskSizeInfo() const REQUIRES(lock_);

  MetricRegistry* metric_registry_;

  bool IsLeader() override {

    return LeaderTerm() != OpId::kUnknownTerm;

  }

  TabletSplitter* tablet_splitter_;

  std::shared_future<client::YBClient*> client_future_;

  rpc::Messenger* messenger_;

  DISALLOW_COPY_AND_ASSIGN(TabletPeer);

};

}  // namespace tablet

}  // namespace yb

#endif /* YB_TABLET_TABLET_PEER_H_ */