xiangfu0 commented on code in PR #20022:
URL: https://github.com/apache/pinot/pull/20022#discussion_r4369189152
##########
pinot-broker/src/main/java/org/apache/pinot/broker/routing/manager/BaseBrokerRoutingManager.java:
##########
@@ -235,11 +245,32 @@ int getPartitionPruningPreparationThreshold() {
return _partitionPruningPreparationThreshold;
}
- /// Sets a callback to be invoked when a server becomes routable: when it
starts, and when it is enabled again after
- /// it was disabled (for example while it restarted) or excluded. This is
useful for resetting gRPC channel state to
- /// avoid exponential backoff delays and connections to a stale address.
- public void setServerEnabledCallback(Consumer<ServerInstance> callback) {
+ /// Registers a callback invoked with the [ServerInstance] of a server that
becomes routable: when it starts, and when
+ /// it is enabled again after it was disabled (for example while it
restarted) or excluded. Callers use it to drop
+ /// state that may be tied to a previous process of the server: gRPC
reconnection backoff and connections to a stale
+ /// address, cached Netty channels that may be half-open. Callbacks run on
the Helix callback thread while the routing
+ /// manager holds its write lock, so they must be quick and must not block.
+ public void addServerEnabledCallback(Consumer<ServerInstance> callback) {
+ _serverEnabledCallbacks.add(callback);
+ }
+
+ /// Sets a callback to be invoked when a server becomes routable, see
[#addServerEnabledCallback(Consumer)]. Setting
+ /// it again replaces the callback set before through this method, and only
that one, so callbacks registered
+ /// through `addServerEnabledCallback` are never dropped.
+ public synchronized void setServerEnabledCallback(Consumer<ServerInstance>
callback) {
Review Comment:
It is benign even with dynamic registration, which is why only the setter is
synchronized. Each `CopyOnWriteArrayList` operation is atomic on its own, so an
`add` that interleaves between the setter's `remove` and `add` simply ends up
registered next to the setter's new callback: no callback is lost and none is
left behind. What `synchronized` protects is the setter's own slot bookkeeping
(`_serverEnabledCallback`) against two concurrent `set` calls, which is the
only case where remove-then-add could leave the wrong callback in place.
Registration order between a concurrent `add` and `set` is unspecified either
way, and nothing depends on it.
##########
pinot-core/src/main/java/org/apache/pinot/core/transport/ServerChannels.java:
##########
@@ -182,7 +293,90 @@ public void preConnect(ServerRoutingInstance
serverRoutingInstance, long timeout
_serverToChannelMap.computeIfAbsent(serverRoutingInstance,
ServerChannel::new).preConnect(timeoutMs);
}
+ /// Marks the cached channel to the given routing instance suspect: the
server stopped answering queries on it
+ /// while the socket still looked active. The next query to it, or the next
[#probe], replaces it with a fresh
+ /// connection instead of trusting `isActive()`.
+ public void markSuspect(ServerRoutingInstance serverRoutingInstance) {
+ ServerChannel serverChannel =
_serverToChannelMap.get(serverRoutingInstance);
+ if (serverChannel != null) {
+ serverChannel._suspect = true;
+ }
+ }
+
+ /// The failure detector's retry probe for one channel, when pings are off.
Does nothing and returns `false` when
+ /// this broker has not queried the server over this channel (see
[#hasChannel]); otherwise reconnects the channel if
+ /// it is down, replaces it with a fresh connection if it was marked
suspect, and returns `true`, throwing if the
+ /// connection cannot be established.
+ public boolean probe(ServerRoutingInstance serverRoutingInstance)
+ throws InterruptedException, TimeoutException {
+ ServerChannel serverChannel =
_serverToChannelMap.get(serverRoutingInstance);
+ if (serverChannel == null || !serverChannel._openedByQuery) {
+ return false;
+ }
+ serverChannel.connect();
+ return true;
+ }
+
+ /// Records a response from the server on the given channel: it is alive,
whatever it answered.
+ public void recordResponse(ServerRoutingInstance serverRoutingInstance) {
+ ServerChannel serverChannel =
_serverToChannelMap.get(serverRoutingInstance);
+ if (serverChannel != null) {
+ serverChannel.recordResponse();
+ }
+ }
+
+ /// Records that a query written at `submitTimeMs` to `channel` (when known)
timed out unanswered. The verdict waits
+ /// out the grace period: the query only counts if the entry still holds
that channel and the server has still sent
+ /// nothing on it since the query was written. That lets the server's own
late timeout response through, so a burst
+ /// of slow queries, or a broker pause that delays reading the responses,
does not look like a dead channel: only a
+ /// channel the server has gone silent on does. At the threshold the channel
is marked suspect and the server is
+ /// reported through [QueryRouter#reportUnansweredServer]. No-op when the
check is disabled.
+ public void recordUnansweredQuery(ServerRoutingInstance
serverRoutingInstance, long submitTimeMs,
+ @Nullable Channel channel) {
+ if (_livenessExecutor == null) {
+ return;
+ }
+ ServerChannel serverChannel =
_serverToChannelMap.get(serverRoutingInstance);
+ if (serverChannel == null) {
+ return;
+ }
+ if (serverChannel._numPendingJudgments.incrementAndGet() >
_maxPendingJudgmentsPerChannel) {
Review Comment:
Right, and I left it that way on purpose: the bound that matters is exact. A
caller only schedules a judgment when its own increment came out at or below
the cap, and a caller whose increment overshoots undoes it without scheduling,
so the number of queued or running judgments per channel never exceeds the cap.
Only a concurrent reader of the counter can see it briefly above, by at most
the number of callers racing at that instant, and the counter is read only by
the cap check itself and a test getter. A CAS loop would make that transient
reading exact at the cost of retries under exactly the burst the cap exists for.
##########
pinot-core/src/main/java/org/apache/pinot/core/transport/AsyncQueryResponse.java:
##########
@@ -116,6 +121,20 @@ public Map<ServerRoutingInstance, ServerResponse>
getFinalResponses()
}
}
+ /// On a timeout, hands every server the request reached but that never
answered to the transport, with the time
+ /// and the channel the request was written to; the transport judges after a
grace period whether the server has gone
+ /// silent on that channel (see [ServerChannels#recordUnansweredQuery]).
+ private void recordUnansweredServers() {
+ for (Map.Entry<ServerRoutingInstance, ServerResponse> entry :
_responseMap.entrySet()) {
+ ServerResponse response = entry.getValue();
+ int submitDelayMs = response.getSubmitDelayMs();
+ // Only servers the request was actually submitted to can be blamed for
not answering it
+ if (response.getDataTable() == null && submitDelayMs >= 0) {
Review Comment:
That case does not reach the judgment. `QueryRouter#submitQuery` writes
every request synchronously before the caller starts waiting on the response,
and a request is only marked submitted (`submitDelayMs >= 0`, the condition
`recordUnansweredServers` checks) after `ServerChannels#sendRequest` returned,
by which point `markRequestChannel` has recorded its channel. A query still
blocked on the channel lock is therefore not submitted yet, and if the lock
wait runs out the send throws, the query is marked FAILED rather than
TIMED_OUT, and `recordUnansweredServers` never runs. So with the current router
every judged query carries its channel; the time-only fallback for a null
channel only covers callers that submit without recording one.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
