Skip to content

Commit 955a620

Browse files
BewareMyPowerBewareMyPower
committed
make probe async
1 parent 56f6296 commit 955a620

4 files changed

Lines changed: 219 additions & 117 deletions

File tree

‎include/pulsar/ServiceUrlProvider.h‎

Lines changed: 18 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -22,6 +22,7 @@
2222
#include <pulsar/Authentication.h>
2323
#include <pulsar/defines.h>
2424

25+
#include <functional>
2526
#include <memory>
2627
#include <string>
2728

@@ -50,24 +51,38 @@ class PULSAR_PUBLIC ServiceUrlProviderClient {
5051
* Update the TLS trust certificate file path.
5152
*/
5253
virtual void updateTlsTrustCertsFilePath(const std::string& tlsTrustCertsFilePath) = 0;
54+
55+
/**
56+
* Update the TLS trust store path and password.
57+
*/
58+
virtual void updateTlsTrustStorePathAndPassword(const std::string& tlsTrustStorePath,
59+
const std::string& tlsTrustStorePassword) = 0;
5360
};
5461

5562
/**
5663
* Interface for probing the availability of a Pulsar service URL.
5764
* Implemented internally by ClientImpl using its networking stack.
65+
*
66+
* The interface is asynchronous to avoid blocking the IO thread.
5867
*/
5968
class PULSAR_PUBLIC ServiceProber {
6069
public:
6170
virtual ~ServiceProber() {}
6271

6372
/**
64-
* Probe the availability of the given service URL.
73+
* Callback type for the probe operation.
74+
* @param success True if the service is reachable, false otherwise.
75+
*/
76+
using ProbeCallback = std::function<void(bool success)>;
77+
78+
/**
79+
* Asynchronously probe the availability of the given service URL.
6580
*
6681
* @param serviceUrl The URL to probe (e.g. "pulsar://host:6650")
6782
* @param timeoutMs The timeout in milliseconds
68-
* @return true if the service is reachable, false otherwise
83+
* @param callback The callback to be invoked when the probe completes
6984
*/
70-
virtual bool probe(const std::string& serviceUrl, int timeoutMs) = 0;
85+
virtual void probeAsync(const std::string& serviceUrl, int timeoutMs, ProbeCallback callback) = 0;
7186
};
7287

7388
/**

‎lib/AutoClusterFailover.cc‎

Lines changed: 138 additions & 73 deletions
Original file line numberDiff line numberDiff line change
@@ -65,10 +65,6 @@ class AutoClusterFailoverImpl : public std::enable_shared_from_this<AutoClusterF
6565
client_ = client;
6666
prober_ = prober;
6767

68-
// Capture the primary auth and TLS from the client's current config — these are already set
69-
// on the client before initialize() is called.
70-
// (The design doc says the primary auth is the initial auth of the client configuration.)
71-
7268
executor_ = ExecutorService::create();
7369
timer_ = executor_->createDeadlineTimer();
7470
scheduleProbe();
@@ -97,97 +93,166 @@ class AutoClusterFailoverImpl : public std::enable_shared_from_this<AutoClusterF
9793
timer_->expires_after(checkInterval_);
9894
timer_->async_wait([self](const ASIO::error_code& ec) {
9995
if (ec || self->closed_) return;
100-
self->probe();
101-
self->scheduleProbe();
96+
self->startProbeCheck();
10297
});
10398
}
10499

105-
void probe() {
100+
void startProbeCheck() {
106101
std::lock_guard<std::mutex> lock(mutex_);
107-
if (!prober_ || !client_ || closed_) return;
102+
if (!prober_ || !client_ || closed_) {
103+
scheduleProbe();
104+
return;
105+
}
108106

109107
if (currentServiceUrl_ == primary_) {
110-
// On primary — check if it's down, and if so try secondary
111-
probeAndFailover();
108+
// On primary — probe it to see if it's still up
109+
probeCurrentForFailover();
112110
} else {
113-
// On secondary — check if secondary is still up
114-
probeCurrentAndFailover();
115-
// Also check if primary has recovered for switch-back
116-
if (currentServiceUrl_ != primary_) {
117-
probeAndSwitchBack();
118-
}
111+
// On secondary — probe it to see if it's still up, then check switch-back
112+
probeCurrentOnSecondary();
119113
}
120114
}
121115

122-
void probeAndFailover() {
123-
if (prober_->probe(currentServiceUrl_, PROBE_TIMEOUT_MS)) {
124-
failedTimestamp_ = -1;
116+
// === When on primary: probe current, if down try secondaries ===
117+
118+
void probeCurrentForFailover() {
119+
auto self = shared_from_this();
120+
std::string url = currentServiceUrl_;
121+
prober_->probeAsync(url, PROBE_TIMEOUT_MS, [self, url](bool success) {
122+
std::lock_guard<std::mutex> lock(self->mutex_);
123+
if (self->closed_) return;
124+
125+
if (success) {
126+
self->failedTimestamp_ = -1;
127+
self->scheduleProbe();
128+
return;
129+
}
130+
131+
auto now = currentTimeMs();
132+
if (self->failedTimestamp_ == -1) {
133+
self->failedTimestamp_ = now;
134+
self->scheduleProbe();
135+
} else if (now - self->failedTimestamp_ >= self->failoverDelay_.count()) {
136+
// Delay elapsed — try secondaries
137+
self->trySecondary(0);
138+
} else {
139+
self->scheduleProbe();
140+
}
141+
});
142+
}
143+
144+
void trySecondary(size_t index) {
145+
if (index >= secondary_.size()) {
146+
// No secondaries are available
147+
scheduleProbe();
125148
return;
126149
}
127150

128-
auto now = currentTimeMs();
129-
if (failedTimestamp_ == -1) {
130-
failedTimestamp_ = now;
131-
} else if (now - failedTimestamp_ >= failoverDelay_.count()) {
132-
for (const auto& target : secondary_) {
133-
if (prober_->probe(target, PROBE_TIMEOUT_MS)) {
134-
LOG_INFO("Current service " << currentServiceUrl_ << " has been down for "
135-
<< (now - failedTimestamp_) << " ms, switching to "
136-
<< target);
137-
performSwitch(target);
138-
failedTimestamp_ = -1;
151+
auto self = shared_from_this();
152+
std::string target = secondary_[index];
153+
prober_->probeAsync(target, PROBE_TIMEOUT_MS, [self, target, index](bool success) {
154+
std::lock_guard<std::mutex> lock(self->mutex_);
155+
if (self->closed_) return;
156+
157+
if (success) {
158+
auto now = currentTimeMs();
159+
LOG_INFO("Current service " << self->currentServiceUrl_ << " has been down for "
160+
<< (now - self->failedTimestamp_) << " ms, switching to "
161+
<< target);
162+
self->performSwitch(target);
163+
self->failedTimestamp_ = -1;
164+
self->scheduleProbe();
165+
} else {
166+
auto now = currentTimeMs();
167+
LOG_WARN("Current service " << self->currentServiceUrl_ << " has been down for "
168+
<< (now - self->failedTimestamp_)
169+
<< " ms. Failed to switch to " << target
170+
<< " (not available), trying next.");
171+
self->trySecondary(index + 1);
172+
}
173+
});
174+
}
175+
176+
// === When on secondary: probe current, then check primary for switch-back ===
177+
178+
void probeCurrentOnSecondary() {
179+
auto self = shared_from_this();
180+
std::string url = currentServiceUrl_;
181+
prober_->probeAsync(url, PROBE_TIMEOUT_MS, [self, url](bool success) {
182+
std::lock_guard<std::mutex> lock(self->mutex_);
183+
if (self->closed_) return;
184+
185+
if (success) {
186+
self->failedTimestamp_ = -1;
187+
} else {
188+
auto now = currentTimeMs();
189+
if (self->failedTimestamp_ == -1) {
190+
self->failedTimestamp_ = now;
191+
} else if (now - self->failedTimestamp_ >= self->failoverDelay_.count()) {
192+
// Secondary is also down — try primary as recovery
193+
self->tryPrimaryRecovery();
139194
return;
140-
} else {
141-
LOG_WARN("Current service " << currentServiceUrl_ << " has been down for "
142-
<< (now - failedTimestamp_) << " ms. Failed to switch to "
143-
<< target << " (not available), trying next.");
144195
}
145196
}
146-
}
147-
}
148197

149-
// When on a secondary, probe it and failover to next secondary if it's also down
150-
void probeCurrentAndFailover() {
151-
if (prober_->probe(currentServiceUrl_, PROBE_TIMEOUT_MS)) {
152-
failedTimestamp_ = -1;
153-
return;
154-
}
198+
// Check if primary has recovered for switch-back
199+
if (self->currentServiceUrl_ != self->primary_) {
200+
self->probePrimaryForSwitchBack();
201+
} else {
202+
self->scheduleProbe();
203+
}
204+
});
205+
}
155206

156-
auto now = currentTimeMs();
157-
if (failedTimestamp_ == -1) {
158-
failedTimestamp_ = now;
159-
} else if (now - failedTimestamp_ >= failoverDelay_.count()) {
160-
// Try primary first
161-
if (prober_->probe(primary_, PROBE_TIMEOUT_MS)) {
162-
LOG_INFO("Secondary " << currentServiceUrl_ << " has been down for "
163-
<< (now - failedTimestamp_) << " ms, switching back to primary "
164-
<< primary_);
165-
performSwitch(primary_);
166-
failedTimestamp_ = -1;
167-
recoverTimestamp_ = -1;
168-
return;
207+
void tryPrimaryRecovery() {
208+
auto self = shared_from_this();
209+
prober_->probeAsync(primary_, PROBE_TIMEOUT_MS, [self](bool success) {
210+
std::lock_guard<std::mutex> lock(self->mutex_);
211+
if (self->closed_) return;
212+
213+
if (success) {
214+
auto now = currentTimeMs();
215+
LOG_INFO("Secondary " << self->currentServiceUrl_ << " has been down for "
216+
<< (now - self->failedTimestamp_)
217+
<< " ms, switching back to primary " << self->primary_);
218+
self->performSwitch(self->primary_);
219+
self->failedTimestamp_ = -1;
220+
self->recoverTimestamp_ = -1;
221+
} else {
222+
LOG_ERROR("Current service "
223+
<< self->currentServiceUrl_ << " has been down for "
224+
<< (currentTimeMs() - self->failedTimestamp_) << " ms. Primary "
225+
<< self->primary_ << " is also not available.");
169226
}
170-
LOG_ERROR("Current service " << currentServiceUrl_ << " has been down for "
171-
<< (now - failedTimestamp_) << " ms. Primary " << primary_
172-
<< " is also not available.");
173-
}
227+
self->scheduleProbe();
228+
});
174229
}
175230

176-
void probeAndSwitchBack() {
177-
if (!prober_->probe(primary_, PROBE_TIMEOUT_MS)) {
178-
recoverTimestamp_ = -1;
179-
return;
180-
}
231+
void probePrimaryForSwitchBack() {
232+
auto self = shared_from_this();
233+
prober_->probeAsync(primary_, PROBE_TIMEOUT_MS, [self](bool success) {
234+
std::lock_guard<std::mutex> lock(self->mutex_);
235+
if (self->closed_) return;
181236

182-
auto now = currentTimeMs();
183-
if (recoverTimestamp_ == -1) {
184-
recoverTimestamp_ = now;
185-
} else if (now - recoverTimestamp_ >= switchBackDelay_.count()) {
186-
LOG_INFO("Primary " << primary_ << " has been recovered for " << (now - recoverTimestamp_)
187-
<< " ms, switching back from secondary " << currentServiceUrl_);
188-
performSwitch(primary_);
189-
recoverTimestamp_ = -1;
190-
}
237+
if (!success) {
238+
self->recoverTimestamp_ = -1;
239+
self->scheduleProbe();
240+
return;
241+
}
242+
243+
auto now = currentTimeMs();
244+
if (self->recoverTimestamp_ == -1) {
245+
self->recoverTimestamp_ = now;
246+
} else if (now - self->recoverTimestamp_ >= self->switchBackDelay_.count()) {
247+
LOG_INFO("Primary " << self->primary_ << " has been recovered for "
248+
<< (now - self->recoverTimestamp_)
249+
<< " ms, switching back from secondary "
250+
<< self->currentServiceUrl_);
251+
self->performSwitch(self->primary_);
252+
self->recoverTimestamp_ = -1;
253+
}
254+
self->scheduleProbe();
255+
});
191256
}
192257

193258
void performSwitch(const std::string& target) {

0 commit comments

Comments
 (0)