YugabyteDB (2.13.1.0-b60, 21121d69985fbf76aa6958d8f04a9bfa936293b5)

// 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.

//

#include "yb/common/consistent_read_point.h"

#include "yb/common/common.pb.h"

namespace yb {

ConsistentReadPoint::ConsistentReadPoint(const scoped_refptr<ClockBase>& clock)

    : clock_(clock) {

}

void ConsistentReadPoint::SetReadTime(

    const ReadHybridTime& read_time, HybridTimeMap&& local_limits) {

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

  read_time_ = read_time;

  restart_read_ht_ = read_time_.read;

  local_limits_ = std::move(local_limits);

  restarts_.clear();

}

void ConsistentReadPoint::SetCurrentReadTime() {

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

  read_time_ = ReadHybridTime::FromHybridTimeRange(clock_->NowRange());

  restart_read_ht_ = read_time_.read;

  local_limits_.clear();

  restarts_.clear();

}

ReadHybridTime ConsistentReadPoint::GetReadTime(const TabletId& tablet) const {

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

  ReadHybridTime read_time = read_time_;

  if (read_time) {

    // Use the local limit for the tablet but no earlier than the read time we want.

    const auto it = local_limits_.find(tablet);

    if (it != local_limits_.end()) {

      read_time.local_limit = it->second;

    }

  }

  return read_time;

}

void ConsistentReadPoint::RestartRequired(const TabletId& tablet,

                                          const ReadHybridTime& restart_time) {

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

  RestartRequiredUnlocked(tablet, restart_time);

}

void ConsistentReadPoint::RestartRequiredUnlocked(

    const TabletId& tablet, const ReadHybridTime& restart_time) {

  DCHECK
(read_time_) << "Unexpected restart without a read time set"1
;

  restart_read_ht_.MakeAtLeast(restart_time.read);

  // We should inherit per-tablet restart time limits before restart, doing it lazily.

  if (restarts_.empty()) {

    restarts_ = local_limits_;

  }

  UpdateLimitsMapUnlocked(tablet, restart_time.local_limit, &restarts_);

}

void ConsistentReadPoint::UpdateLocalLimit(const TabletId& tablet, HybridTime local_limit) {

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

  UpdateLimitsMapUnlocked(tablet, local_limit, &local_limits_);

}

void ConsistentReadPoint::UpdateLimitsMapUnlocked(

    const TabletId& tablet, const HybridTime& local_limit, HybridTimeMap* map) {

  auto emplace_result = map->emplace(tablet, local_limit);

  bool inserted = emplace_result.second;

  if (!inserted) {

    auto& existing_local_limit = emplace_result.first->second;

    existing_local_limit = std::min(existing_local_limit, local_limit);

  }

}

bool ConsistentReadPoint::IsRestartRequired() const {

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

  return IsRestartRequiredUnlocked();

}

bool ConsistentReadPoint::IsRestartRequiredUnlocked() const {

  return !restarts_.empty();

}

void ConsistentReadPoint::Restart() {

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

  local_limits_.swap(restarts_);

  restarts_.clear();

  read_time_.read = restart_read_ht_;

}

void ConsistentReadPoint::Defer() {

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

  read_time_.read = read_time_.global_limit;

}

void ConsistentReadPoint::UpdateClock(HybridTime propagated_hybrid_time) {

  clock_->Update(propagated_hybrid_time);

}

HybridTime ConsistentReadPoint::Now() const {

  return clock_->Now();

}

void ConsistentReadPoint::PrepareChildTransactionData(ChildTransactionDataPB* data) const {

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

  read_time_.AddToPB(data);

  auto& local_limits = *data->mutable_local_limits();

  for (const auto& entry : local_limits_) {

    typedef std::remove_reference<decltype(*local_limits.begin())>::type PairType;

    local_limits.insert(PairType(entry.first, entry.second.ToUint64()));

  }

}

void ConsistentReadPoint::FinishChildTransactionResult(

    HadReadTime had_read_time, ChildTransactionResultPB* result) const {

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

  if (IsRestartRequiredUnlocked()) {

    result->set_restart_read_ht(restart_read_ht_.ToUint64());

    auto& restarts = *result->mutable_read_restarts();

    for (const auto& restart : restarts_) {

      typedef std::remove_reference<decltype(*restarts.begin())>::type PairType;

      restarts.insert(PairType(restart.first, restart.second.ToUint64()));

    }

  } else {

    result->set_restart_read_ht(HybridTime::kInvalid.ToUint64());

  }

  if (!had_read_time && 
read_time_0
) {

    read_time_.ToPB(result->mutable_used_read_time());

  }

}

void ConsistentReadPoint::ApplyChildTransactionResult(const ChildTransactionResultPB& result) {

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

  if (result.has_used_read_time()) {

    LOG_IF(DFATAL, read_time_)

        << "Read time already picked (" << read_time_

        << ", but child result contains used read time: "

        << result.used_read_time().ShortDebugString();

    read_time_ = ReadHybridTime::FromPB(result.used_read_time());

    restart_read_ht_ = read_time_.read;

  }

  HybridTime restart_read_ht(result.restart_read_ht());

  if (restart_read_ht.is_valid()) {

    ReadHybridTime read_time;

    read_time.read = restart_read_ht;

    for (const auto& restart : result.read_restarts()) {

      read_time.local_limit = HybridTime(restart.second);

      RestartRequiredUnlocked(restart.first, read_time);

    }

  }

}

void ConsistentReadPoint::SetInTxnLimit(HybridTime value) {

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

  read_time_.in_txn_limit = value;

}

ReadHybridTime ConsistentReadPoint::GetReadTime() const {

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

  return read_time_;

}

// NO_THREAD_SAFETY_ANALYSIS is required here because anylysis does not understand std::lock.

void ConsistentReadPoint::MoveFrom(ConsistentReadPoint* rhs) NO_THREAD_SAFETY_ANALYSIS {

  std::lock(mutex_, rhs->mutex_);

  std::lock_guard<simple_spinlock> lock1(mutex_, std::adopt_lock);

  std::lock_guard<simple_spinlock> lock2(rhs->mutex_, std::adopt_lock);

  read_time_ = rhs->read_time_;

  restart_read_ht_ = rhs->restart_read_ht_;

  local_limits_ = std::move(rhs->local_limits_);

  restarts_ = std::move(rhs->restarts_);

}

} // namespace yb