Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
19 changes: 15 additions & 4 deletions driver/host_selector/round_robin_host_selector.cpp
Original file line number Diff line number Diff line change
Expand Up @@ -15,6 +15,7 @@
#include "round_robin_host_selector.h"

#include <algorithm>
#include <optional>
#include <sstream>
#include <stdexcept>

Expand Down Expand Up @@ -73,7 +74,12 @@ HostInfo RoundRobinHostSelector::GetHost(std::vector<HostInfo> hosts, bool is_wr

CreateCacheEntries(selection, properties);
const std::string cluster_id_key = selection.at(0).GetHost();
const std::shared_ptr<RoundRobinProperty::RoundRobinClusterInfo> cluster_info = round_robin_cache_.Get(cluster_id_key);
const std::optional<std::shared_ptr<RoundRobinProperty::RoundRobinClusterInfo>> cached_info =
round_robin_cache_.Get(cluster_id_key);
if (!cached_info.has_value() || !cached_info.value()) {
throw std::runtime_error("No round robin cluster info cached for host: " + cluster_id_key);
}
const std::shared_ptr<RoundRobinProperty::RoundRobinClusterInfo>& cluster_info = cached_info.value();

const std::shared_ptr<HostInfo> last_host = cluster_info->last_host;
int last_host_idx = NO_HOST_IDX;
Expand Down Expand Up @@ -143,15 +149,20 @@ void RoundRobinHostSelector::CreateCacheEntries(const std::vector<HostInfo>& hos
return RoundRobinHostSelector::round_robin_cache_.Find(host.GetHost());
});

// An entry may expire between the Find above and the Get below, so treat a missing entry the same as having no cached entry at all.
std::optional<std::shared_ptr<RoundRobinProperty::RoundRobinClusterInfo>> cached_info;
if (!hosts_with_cached_entry.empty()) {
cached_info = round_robin_cache_.Get(hosts_with_cached_entry.at(0).GetHost());
}

// Update cache entries, else create new cache
if (hosts_with_cached_entry.empty()) {
if (!cached_info.has_value() || !cached_info.value()) {
const std::shared_ptr<RoundRobinProperty::RoundRobinClusterInfo> cluster_info = std::make_shared<RoundRobinProperty::RoundRobinClusterInfo>();
UpdatePropsDefaultWeight(cluster_info, props);
UpdatePropsHostWeight(cluster_info, props);
UpdateCache(hosts, cluster_info);
} else {
const std::string cluster_id_key = hosts_with_cached_entry.at(0).GetHost();
const std::shared_ptr<RoundRobinProperty::RoundRobinClusterInfo> cluster_info = round_robin_cache_.Get(cluster_id_key);
const std::shared_ptr<RoundRobinProperty::RoundRobinClusterInfo>& cluster_info = cached_info.value();
if (CheckPropChange(cluster_info->last_default_weight_str, RoundRobinProperty::DEFAULT_WEIGHT_KEY, props)) {
cluster_info->default_weight = 1;
UpdatePropsDefaultWeight(cluster_info, props);
Expand Down
8 changes: 4 additions & 4 deletions driver/plugin/blue_green/blue_green_plugin.cpp
Original file line number Diff line number Diff line change
Expand Up @@ -65,7 +65,7 @@ SQLRETURN BlueGreenPlugin::Connect(
// Get provider lock to check if there is an existing status
// This is to allow the plugin to increment the provider if it hasn't already
std::unique_lock<std::recursive_mutex> lock_guard(provider_lock_);
this->blue_green_status_ = status_map_->Get(this->blue_green_id_);
this->blue_green_status_ = status_map_->Get(this->blue_green_id_).value_or(BlueGreenStatus{});
if (this->blue_green_status_.GetCurrentPhase().GetPhase() == BlueGreenPhase::UNKNOWN) {
LOG(INFO) << "[FALLBACK-A] Default connection, no status found: " << this->blue_green_id_;
LOG(INFO) << "Default connection, no status found: " << this->blue_green_id_;
Expand Down Expand Up @@ -120,7 +120,7 @@ SQLRETURN BlueGreenPlugin::Connect(
rc = (*route_itr)->Connect(dbc, HostInfo(conn_host), this->odbc_helper_, BlueGreenPlugin::status_map_);
LOG(INFO) << "[ROUTE-LOOP] iteration=" << loop_iteration << ", returned rc=" << std::to_string(rc);
if (!SQL_SUCCEEDED(rc)) {
this->blue_green_status_ = status_map_->Get(this->blue_green_id_);
this->blue_green_status_ = status_map_->Get(this->blue_green_id_).value_or(BlueGreenStatus{});
BlueGreenPhase refetched_phase = this->blue_green_status_.GetCurrentPhase();
if (refetched_phase.GetPhase() == BlueGreenPhase::UNKNOWN) {
this->end_time_ = std::chrono::steady_clock::now();
Expand Down Expand Up @@ -182,7 +182,7 @@ SQLRETURN BlueGreenPlugin::Execute(
this->ResetRoutingTiming();
this->InitProvider();

this->blue_green_status_ = status_map_->Get(this->blue_green_id_);
this->blue_green_status_ = status_map_->Get(this->blue_green_id_).value_or(BlueGreenStatus{});
if (this->blue_green_status_.GetCurrentPhase().GetPhase() == BlueGreenPhase::UNKNOWN) {
LOG(INFO) << "Default execution, no status found: " << this->blue_green_id_;
return ExecuteNext(StatementHandle, StatementText, TextLength);
Expand Down Expand Up @@ -220,7 +220,7 @@ SQLRETURN BlueGreenPlugin::Execute(
rc = (*route_itr)->Execute(stmt, this->odbc_helper_, BlueGreenPlugin::status_map_);
LOG(INFO) << "Execute route returned: " << std::to_string(rc);
if (!SQL_SUCCEEDED(rc)) {
this->blue_green_status_ = status_map_->Get(this->blue_green_id_);
this->blue_green_status_ = status_map_->Get(this->blue_green_id_).value_or(BlueGreenStatus{});
if (this->blue_green_status_.GetCurrentPhase().GetPhase() == BlueGreenPhase::UNKNOWN) {
this->end_time_ = std::chrono::steady_clock::now();
LOG(WARNING) << "Default execution, statuses reset, routes cleared for role: " << host_role.ToString() << ", host: " << conn_host;
Expand Down
23 changes: 13 additions & 10 deletions driver/plugin/blue_green/blue_green_status_provider.cpp
Original file line number Diff line number Diff line change
Expand Up @@ -28,6 +28,7 @@
#include <algorithm>
#include <format>
#include <functional>
#include <optional>
#include <ranges>

static constexpr int HASH_MULTIPLIER = 31;
Expand Down Expand Up @@ -447,12 +448,12 @@ std::vector<std::shared_ptr<BaseConnectRouting>> BlueGreenStatusProvider::AddSub
for (const auto& itr : this->role_by_host_map_->GetMapCopy()) {
const std::string host = itr.first;
const BlueGreenRole role = itr.second;
const std::pair<HostInfo, HostInfo> host_pair = this->corresponding_nodes_->Get(host);
if (role != BlueGreenRole::SOURCE || host_pair.first.GetHost().empty()) {
const std::optional<std::pair<HostInfo, HostInfo>> host_pair = this->corresponding_nodes_->Get(host);
if (role != BlueGreenRole::SOURCE || !host_pair.has_value() || host_pair->first.GetHost().empty()) {
continue;
}
const HostInfo blue_host = host_pair.first;
const std::string blue_ip = this->host_ip_map_->Get(blue_host.GetHost());
const HostInfo blue_host = host_pair->first;
const std::string blue_ip = this->host_ip_map_->Get(blue_host.GetHost()).value_or("");
const HostInfo blue_ip_host = blue_ip.empty() ?
blue_host
: HostInfo(blue_ip, blue_host.GetPort(), blue_host.GetHostState(), blue_host.GetHostRole());
Expand Down Expand Up @@ -590,13 +591,14 @@ BlueGreenStatus BlueGreenStatusProvider::GetStatusOfPost() { // NOLINT(misc-no-
void BlueGreenStatusProvider::CreatePostRouting(std::vector<std::shared_ptr<BaseConnectRouting>>& connect_routing) {
if (!this->blue_dns_update_completed_ || !this->all_green_nodes_changed_) {
for (const auto& [host, role] : this->role_by_host_map_->GetMapCopy()) {
if (role != BlueGreenRole::SOURCE || !this->corresponding_nodes_->Contains(host)) {
const std::optional<std::pair<HostInfo, HostInfo>> node_pair = this->corresponding_nodes_->Get(host);
if (role != BlueGreenRole::SOURCE || !node_pair.has_value()) {
continue;
}
const std::string blue_host = host;
const bool is_blue_host_instance = RdsUtils::IsRdsInstance(blue_host);

const auto [blue_host_info, green_host_info] = this->corresponding_nodes_->Get(blue_host);
const auto& [blue_host_info, green_host_info] = *node_pair;

if (green_host_info.GetHost().empty()) {
connect_routing.push_back(std::make_shared<SuspendUntilNodeFoundConnectRouting>(blue_host, role, this->blue_green_id_));
Expand All @@ -607,7 +609,7 @@ void BlueGreenStatusProvider::CreatePostRouting(std::vector<std::shared_ptr<Base
}
} else {
const std::string green_host = green_host_info.GetHost();
const std::string green_host_ip = this->host_ip_map_->Get(green_host);
const std::string green_host_ip = this->host_ip_map_->Get(green_host).value_or("");
const HostInfo green_ip_host_info = HostInfo(green_host_ip, green_host_info.GetPort(), green_host_info.GetHostState(), green_host_info.GetHostRole());

// Check if green ghost connected with blue (non-prefixed) IAM host name
Expand Down Expand Up @@ -653,7 +655,7 @@ void BlueGreenStatusProvider::CreatePostRouting(std::vector<std::shared_ptr<Base

const HostInfo green_host_info = HostInfo(green_host, interim_status.port_);
const HostInfo blue_host_info = HostInfo(blue_host, interim_status.port_);
const HostInfo green_ip_host_info = HostInfo(this->host_ip_map_->Get(green_host), interim_status.port_);
const HostInfo green_ip_host_info = HostInfo(this->host_ip_map_->Get(green_host).value_or(""), interim_status.port_);

std::vector<HostInfo> iam_hosts;
if (this->IsAlreadySuccessfullyConnected(green_host, blue_host)) {
Expand Down Expand Up @@ -715,7 +717,7 @@ void BlueGreenStatusProvider::RegisterIamHost(const std::string& connect_host, c
}
{
const std::lock_guard<std::mutex> lock_guard(this->iam_mutex_);
std::set<std::string> iam_success_connects = this->iam_host_success_connects_map_->Get(connect_host);
std::set<std::string> iam_success_connects = this->iam_host_success_connects_map_->Get(connect_host).value_or(std::set<std::string>{});
iam_success_connects.insert(iam_host);
this->iam_host_success_connects_map_->InsertOrAssign(connect_host, iam_success_connects);
}
Expand Down Expand Up @@ -743,7 +745,8 @@ bool BlueGreenStatusProvider::IsAlreadySuccessfullyConnected(std::string connect
if (!this->iam_host_success_connects_map_) {
return false;
}
return this->iam_host_success_connects_map_->Get(connect_host).contains(iam_host);
const std::optional<std::set<std::string>> iam_success_connects = this->iam_host_success_connects_map_->Get(connect_host);
return iam_success_connects.has_value() && iam_success_connects->contains(iam_host);
}

void BlueGreenStatusProvider::StorePhaseTime(BlueGreenPhase phase) {
Expand Down
4 changes: 2 additions & 2 deletions driver/plugin/blue_green/routing/base_routing.cpp
Original file line number Diff line number Diff line change
Expand Up @@ -34,14 +34,14 @@ void BaseRouting::Delay(
const std::chrono::steady_clock::time_point end = start + delay_ms;
const std::chrono::milliseconds min_delay = delay_ms < MIN_SLEEP_MS ? delay_ms : MIN_SLEEP_MS;

BlueGreenStatus cached_status = status_cache->Get(id);
BlueGreenStatus cached_status = status_cache->Get(id).value_or(BlueGreenStatus{});

if (cached_status.GetCurrentPhase().GetPhase() == BlueGreenPhase::UNKNOWN) {
std::this_thread::sleep_for(delay_ms);
} else {
do {
std::this_thread::sleep_for(min_delay);
} while ((cached_status = status_cache->Get(id)) == status && GetCurrTime() < end);
} while ((cached_status = status_cache->Get(id).value_or(BlueGreenStatus{})) == status && GetCurrTime() < end);
}
}

Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -32,11 +32,11 @@ SQLRETURN SuspendConnectRouting::Connect(
const std::chrono::steady_clock::time_point start_time = GetCurrTime();
const std::chrono::steady_clock::time_point end_time = start_time + timeout;

BlueGreenStatus cached_status = status_cache->Get(this->blue_green_id_);
BlueGreenStatus cached_status = status_cache->Get(this->blue_green_id_).value_or(BlueGreenStatus{});

while (GetCurrTime() <= end_time && cached_status.GetCurrentPhase().GetPhase() == BlueGreenPhase::IN_PROGRESS) {
this->Delay(SLEEP_DURATION_MS, cached_status, status_cache, this->blue_green_id_);
cached_status = status_cache->Get(this->blue_green_id_);
cached_status = status_cache->Get(this->blue_green_id_).value_or(BlueGreenStatus{});
};

if (cached_status.GetCurrentPhase().GetPhase() == BlueGreenPhase::IN_PROGRESS) {
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -24,7 +24,7 @@ SQLRETURN SuspendUntilNodeFoundConnectRouting::Connect(
std::shared_ptr<OdbcHelper> odbc_helper,
std::shared_ptr<ConcurrentMap<std::string, BlueGreenStatus>> status_cache)
{
BlueGreenStatus cached_status = status_cache->Get(this->blue_green_id_);
BlueGreenStatus cached_status = status_cache->Get(this->blue_green_id_).value_or(BlueGreenStatus{});
const std::string host = info.GetHost();
const std::map<std::string, std::pair<HostInfo, HostInfo>> nodes = cached_status.GetCorrespondingNodes();

Expand All @@ -34,7 +34,7 @@ SQLRETURN SuspendUntilNodeFoundConnectRouting::Connect(

while (GetCurrTime() <= end_time && cached_status.GetCurrentPhase().GetPhase() != BlueGreenPhase::COMPLETED && !nodes.contains(host)) {
this->Delay(SLEEP_DURATION_MS, cached_status, status_cache, this->blue_green_id_);
cached_status = status_cache->Get(this->blue_green_id_);
cached_status = status_cache->Get(this->blue_green_id_).value_or(BlueGreenStatus{});
};

if (cached_status.GetCurrentPhase().GetPhase() == BlueGreenPhase::COMPLETED) {
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -32,11 +32,11 @@ SQLRETURN SuspendExecuteRouting::Execute(
const std::chrono::steady_clock::time_point start_time = GetCurrTime();
const std::chrono::steady_clock::time_point end_time = start_time + timeout;

BlueGreenStatus cached_status = status_cache->Get(this->blue_green_id_);
BlueGreenStatus cached_status = status_cache->Get(this->blue_green_id_).value_or(BlueGreenStatus{});

while (GetCurrTime() <= end_time && cached_status.GetCurrentPhase().GetPhase() == BlueGreenPhase::IN_PROGRESS) {
this->Delay(SLEEP_DURATION_MS, cached_status, status_cache, this->blue_green_id_);
cached_status = status_cache->Get(this->blue_green_id_);
cached_status = status_cache->Get(this->blue_green_id_).value_or(BlueGreenStatus{});
};

if (cached_status.GetCurrentPhase().GetPhase() == BlueGreenPhase::IN_PROGRESS) {
Expand Down
5 changes: 3 additions & 2 deletions driver/plugin/custom_endpoint/custom_endpoint_monitor.cpp
Original file line number Diff line number Diff line change
Expand Up @@ -26,6 +26,7 @@
#include "../../util/rds_utils.h"

#include <chrono>
#include <optional>

SlidingCacheMap<std::string, HostFilter> CustomEndpointMonitor::endpoint_cache_;

Expand Down Expand Up @@ -114,7 +115,7 @@ void CustomEndpointMonitor::Run() {
}
filter.endpoint_type = endpoint_info.GetEndpointType();

const HostFilter cached_filter = endpoint_cache_.Get(endpoint_identifier_);
const std::optional<HostFilter> cached_filter = endpoint_cache_.Get(endpoint_identifier_);
if (cached_filter != filter) {
LOG(INFO) << "Detected change in custom endpoint info for " << endpoint_identifier_;
if (const std::shared_ptr<PluginService> service = this->plugin_service_.lock()) {
Expand Down Expand Up @@ -148,7 +149,7 @@ void CustomEndpointMonitor::Run() {
}

bool CustomEndpointMonitor::HasInfo() {
return endpoint_cache_.Get(endpoint_identifier_) != HostFilter{};
return endpoint_cache_.Get(endpoint_identifier_).has_value();
}

void CustomEndpointMonitor::IncreaseDelay() {
Expand Down
5 changes: 3 additions & 2 deletions driver/util/concurrent_map.h
Original file line number Diff line number Diff line change
Expand Up @@ -17,6 +17,7 @@

#include <map>
#include <mutex>
#include <optional>

template <typename Key, typename Value>
class ConcurrentMap {
Expand Down Expand Up @@ -56,12 +57,12 @@ class ConcurrentMap {
map_.insert(other.begin(), other.end());
};

Value Get(const Key& key) const {
std::optional<Value> Get(const Key& key) const {
const std::lock_guard<std::mutex> lock(mutex_);
if (const auto itr = map_.find(key); itr != map_.end()) {
return itr->second;
}
return {};
return std::nullopt;
};

std::map<Key, Value> GetMapCopy() const {
Expand Down
6 changes: 3 additions & 3 deletions driver/util/plugin_service.cpp
Original file line number Diff line number Diff line change
Expand Up @@ -183,16 +183,16 @@ void PluginService::ForceRefreshHosts(bool verify_writer, std::chrono::milliseco
}

std::vector<HostInfo> PluginService::GetHosts() {
return topology_map_->Get(this->cluster_id_);
return topology_map_->Get(this->cluster_id_).value_or(std::vector<HostInfo>{});
}

void PluginService::SetHosts(const std::vector<HostInfo>& hosts) {
topology_map_->Put(this->cluster_id_, hosts);
}

std::vector<HostInfo> PluginService::GetFilteredHosts() {
std::vector<HostInfo> hosts = topology_map_->Get(this->cluster_id_);
HostFilter host_filter = host_filter_map_->Get(this->cluster_id_);
std::vector<HostInfo> hosts = topology_map_->Get(this->cluster_id_).value_or(std::vector<HostInfo>{});
HostFilter host_filter = host_filter_map_->Get(this->cluster_id_).value_or(HostFilter{});

if (host_filter.allowed_host_ids.empty() && host_filter.blocked_host_ids.empty()) {
return hosts;
Expand Down
12 changes: 1 addition & 11 deletions driver/util/rds_lib_loader.h
Original file line number Diff line number Diff line change
Expand Up @@ -82,18 +82,8 @@ class RdsLibLoader {
template <typename RdsFunc, typename... Args>
RdsLibResult RdsLibLoader::CallFunction(const std::string& func_name, Args... args)
{
FUNC_HANDLE driver_function = nullptr;
// Try retrieving from cache
{
if (function_cache_->Contains(func_name)) {
try {
driver_function = function_cache_->Get(func_name);
} catch (const std::out_of_range&) {
// Should not happen but done to satisfy clang-tidy
driver_function = nullptr;
}
}
}
FUNC_HANDLE driver_function = function_cache_->Get(func_name).value_or(nullptr);
// Cache miss
if (!driver_function) {
driver_function = GetFunction(func_name);
Expand Down
5 changes: 3 additions & 2 deletions driver/util/sliding_cache_map.h
Original file line number Diff line number Diff line change
Expand Up @@ -18,6 +18,7 @@
#include <chrono>
#include <map>
#include <mutex>
#include <optional>

template <typename V>
struct CacheEntry {
Expand Down Expand Up @@ -65,7 +66,7 @@ class SlidingCacheMap {
cache_[key] = CacheEntry{value, expiry_time, ms_ttl};
}

V Get(const K& key) {
std::optional<V> Get(const K& key) {
const std::lock_guard<std::mutex> lock(cache_lock_);
const std::chrono::steady_clock::time_point now = std::chrono::steady_clock::now();
if (const auto itr = cache_.find(key); itr != cache_.end()) {
Expand All @@ -78,7 +79,7 @@ class SlidingCacheMap {
// Expired, remove from cache
cache_.erase(itr);
}
return {};
return std::nullopt;
}

bool Find(const K& key) {
Expand Down
7 changes: 7 additions & 0 deletions test/unit_test/concurrent_map_test.cpp
Original file line number Diff line number Diff line change
Expand Up @@ -14,6 +14,7 @@

#include "../../driver/util/concurrent_map.h"

#include <optional>
#include <string>
#include <thread>

Expand Down Expand Up @@ -52,6 +53,12 @@ TEST_F(ConcurrentMapTest, sync_basic_operations) {
// Erase
cache.Erase(CACHE_KEY_A);
EXPECT_FALSE(cache.Contains(CACHE_KEY_A));
EXPECT_EQ(std::nullopt, cache.Get(CACHE_KEY_A));

// A stored default-constructed value is distinguishable from a miss
cache.InsertOrAssign(CACHE_KEY_A, 0);
EXPECT_EQ(0, cache.Get(CACHE_KEY_A));
cache.Erase(CACHE_KEY_A);

// Size
EXPECT_EQ(0, cache.Size());
Expand Down
Loading
Loading