liaoxin01 commented on PR #65584:
URL: https://github.com/apache/doris/pull/65584#issuecomment-5302697310
Reviewed the whole close path against master. The direction is right —
pulling "register" and "all-received check" into one lock scope does eliminate
the orphaned stream, and moving `brpc::StreamClose` out of `_lock` onto a local
vector is a real improvement on its own (master iterates the *member*
`_closing_stream_ids` while holding `_lock`, and `on_closed` -> `clear_load`
deletes `this`). A few things I'd like addressed before merge.
**1. [Medium] The `_all_close_load_received` latch is unnecessary, and it
can bypass the fence.**
Re-evaluating `_close_load_cnt >= _total_streams` inside
`mark_eos_sent_and_collect()` is already sufficient, because registration and
the check now share one lock:
- T1 marks first (cnt < N, parks A, no drain) -> T2 marks (cnt == N, parks
B, drains A+B). OK
- T2 marks first (cnt == N, drains to B) -> T1 marks (parks A, re-checks cnt
>= N, drains A). OK
The latch, however, is write-once and never reset, while `_total_streams` is
*not* constant: for a LoadStream created by an incremental open
(`_is_incremental = (_total_streams == 0)`), `add_source()` keeps incrementing
it (load_stream.h:131-137). So:
- source A opens 1 incremental stream -> `_total_streams = 1`; A's
CLOSE_LOAD -> cnt=1 >= 1 -> **latch set**
- sources B and C open later -> `_total_streams = 3`
- B's CLOSE_LOAD -> cnt=2 < 3. master would park B and wait for C; with the
sticky latch B is `StreamClose`d immediately, skipping the #56120 fence.
Suggest dropping `_all_close_load_received` entirely and just doing:
```cpp
if (_close_load_cnt >= _total_streams) {
to_close.insert(to_close.end(), _eos_sent_stream_ids.begin(),
_eos_sent_stream_ids.end());
_eos_sent_stream_ids.clear();
}
```
**2. [Medium] `is_incremental` is a misnomer, and the new comment built on
it is inverted.**
`hdr.num_incremental_streams()` means "how many incremental destinations
*this source* opened", not "this stream is incremental".
`LoadStreamMap::close_load()` passes `_num_incremental_streams` for **both**
groups (load_stream_map_pool.cpp:125), and vtablet_writer_v2.cpp:737-743 states
explicitly that a source's *non-incremental* CLOSE_LOAD also carries
`num_incremental_streams > 0` so the destination defers StreamClose.
So this comment is the opposite of what happens:
> `// A non-incremental stream is closed as soon as its own CLOSE_LOAD (and
EOS) is handled -- ... not subject to fencing.`
Behavior matches master (no bug introduced), but the comment will mislead
the next person touching this code. Please rename to something like
`sender_has_incremental_streams` / `needs_close_fence` and rewrite the comment.
**3. [Medium] The PR description's "second hazard" (close-before-EOS) does
not exist on master.**
master already does `push_back(id)` *after* `_report_result(id, ...)`, so
every drained id had its EOS sent; this PR keeps the exact same ordering. The
`Stream closed without EOS` reports are more likely the *tail* of the same bug:
the orphaned stream eventually hits `on_idle_timeout`, which calls
`brpc::StreamClose` without EOS (load_stream.cpp:795). Worth correcting in the
description so the backports aren't evaluated against a hazard that isn't there.
**Minor**
- `mark_eos_sent_and_collect` is declared `public` but only `_dispatch` uses
it (no test calls it). Should be private, and per BE convention `_`-prefixed.
Also the name doesn't quite fit — nothing is "marked", it registers + collects.
- `_eos_sent_stream_ids`: every stream reaching this function has had its
EOS sent, including the ones returned directly. The vector actually holds
fence-deferred ids — `_deferred_close_stream_ids` reads better.
- `close()`'s return value is now consumed *only* by the debug point
(load_stream.cpp:799 / :812). Either move the condition inside the
`DBUG_EXECUTE_IF` block or add a note, otherwise it looks like it still carries
business meaning.
- UT name typo: `incremental_close_race_orhpans_streams` -> `orphans`.
- In the UT, `config::enable_debug_points` / debug point cleanup isn't RAII;
the `ASSERT_TRUE(debug_points->is_enable(...))` in the middle returns early on
failure and leaves the global config set to `true` for the rest of the test
binary. A `Defer` would fix it.
- The debug point hardcodes `bthread_usleep(3000000)`. Doris convention is a
parameterized sleep, e.g. `auto sleep_sec = dp->param<int32_t>("sleep_sec",
1);` (vtablet_writer_v2.cpp:674). As written the UT always costs ~7s.
- The docker suite only covers `cloudMode = false`; the receiver-side path
is identical in cloud mode. Either cover both or note why not.
- `deadline = 60000` for "create 50 partitions on a 3-BE docker cluster with
a 3s sleep per incremental stream" may be tight on a loaded CI. Consider
raising it, or attaching load state to the timeout message so a flake is
diagnosable.
- Optional observability: log the drain (`to_close.size()` and remaining
parked count). Orphaned-stream issues are very hard to diagnose from logs today.
**Compatibility**: no issues found — BE-process-local state only, no
RPC/thrift/protobuf/metadata/config changes, and the debug point is inert
unless `enable_debug_points=true`.
**Backport note**: all three labelled branches contain
`_closing_stream_ids`, so the fix is needed on all of them, but on branch-3.1 /
branch-4.0 the file lives at `be/src/runtime/load_stream.cpp` (only 4.1 and
master use `be/src/load/channel/`), so the cherry-pick will conflict.
--
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]
