github-actions[bot] commented on code in PR #67352:
URL: https://github.com/apache/doris/pull/67352#discussion_r3930542197
##########
be/test/udf/python/python_server_test.cpp:
##########
@@ -401,19 +353,31 @@ TEST_F(PythonServerTest,
EnsurePoolInitializedWithInvalidVersionFails) {
result.error().to_string().find("Timed out") !=
std::string::npos);
}
-TEST_F(PythonServerTest,
EnsurePoolInitializedReturnsImmediatelyWhenAllWorkersFail) {
+TEST_F(PythonServerTest, EnsurePoolInitializedRetriesAfterRuntimeIsRepaired) {
+ setup_doris_home();
+ config::max_python_process_num = 1;
+
+ // The first executable starts but never publishes its Flight socket. This
reproduces a real
+ // worker-start failure and verifies that the pool does not remain stuck
in INITIALIZING.
+ std::string python_path =
create_fake_python_without_socket_creation("python3", "3.9.16");
+ PythonVersion version("3.9.16", test_dir_, python_path);
PythonServerManager mgr;
- config::max_python_process_num = 2;
- PythonVersion invalid_version("3.9.16", test_dir_, test_dir_ +
"/missing_python");
+ auto failed_result = mgr._ensure_pool_initialized(version);
+ ASSERT_FALSE(failed_result.has_value());
- auto start = std::chrono::steady_clock::now();
- auto result = mgr._ensure_pool_initialized(invalid_version);
- auto elapsed = std::chrono::duration_cast<std::chrono::milliseconds>(
- std::chrono::steady_clock::now() - start);
+ // Repair the runtime in place and retry the same version key. Production
can hit this when an
+ // environment or server entry is fixed after a transient initialization
failure.
+ ASSERT_EQ(create_fake_python_with_socket_creation("3.9.16"), python_path);
+ auto recovered_result = mgr._ensure_pool_initialized(version);
Review Comment:
**[P2] Wait for the failed generation before retrying**
The first caller and the detached coordinator each start their own 1 s wait,
but the coordinator cannot acquire `versioned_pool->mutex` and start its timer
until the caller enters `wait_for`. If the caller deadline wins under load,
`failed_result` returns while the pool is still `INITIALIZING`. This immediate
retry then skips worker creation; when the coordinator later resets the state
to `UNINITIALIZED`, the retry wakes and returns another error, so the test
fails on correct code. Synchronize on the failed generation leaving
`INITIALIZING` before rewriting the runtime and retrying.
##########
be/test/udf/python/python_server_test.cpp:
##########
@@ -924,33 +832,74 @@ TEST_F(PythonServerTest,
EnsurePoolInitializedForDifferentVersionsDoesNotShareVe
config::max_python_process_num = 1;
- std::string python39_path =
- create_fake_python_with_delay_and_socket_creation("python3.9",
"3.9.16", 50);
- std::string python310_path =
- create_fake_python_with_delay_and_socket_creation("python3.10",
"3.10.0", 50);
+ std::string python39_path =
create_fake_python_with_socket_creation("3.9.16");
+ std::string python310_path = test_dir_ + "/bin/python3.10";
+ ASSERT_TRUE(fs::copy_file(python39_path, python310_path));
+ fs::permissions(python310_path, fs::perms::owner_all);
PythonServerManager mgr;
PythonVersion version39("3.9.16", test_dir_, python39_path);
PythonVersion version310("3.10.0", test_dir_, python310_path);
- auto start = std::chrono::steady_clock::now();
+ struct ForkBarrier {
+ std::mutex mutex;
+ std::condition_variable cv;
+ int entries = 0;
+ bool released = false;
+ };
+ auto fork_barrier = std::make_shared<ForkBarrier>();
+ auto* sync_point = SyncPoint::get_instance();
+ Defer clear_sync_point {[fork_barrier, sync_point]() {
+ {
+ std::lock_guard lock(fork_barrier->mutex);
+ fork_barrier->released = true;
+ }
+ fork_barrier->cv.notify_all();
+ sync_point->disable_processing();
+
sync_point->clear_call_back("PythonServerManager::fork:before_process_start");
+ sync_point->clear_call_back(
+
"PythonServerManager::_ensure_pool_initialized:process_pool_init_timeout");
+ }};
+ // Keep callers inside initialization longer than the fork barrier. A
manager-wide lock must
+ // therefore fail the barrier instead of serializing through the short
BE_TEST pool timeout.
+ sync_point->set_call_back(
+
"PythonServerManager::_ensure_pool_initialized:process_pool_init_timeout",
+ [](auto&& args) {
+ auto* timeout =
try_any_cast<std::chrono::milliseconds*>(args.front());
+ *timeout = std::chrono::seconds(10);
+ });
+ sync_point->set_call_back(
+ "PythonServerManager::fork:before_process_start",
[fork_barrier](auto&&) {
Review Comment:
**[P2] Count only the two target version forks**
`SyncPoint` is process-global, this callback receives no version or manager
identity, and `_ensure_pool_initialized` detaches its slot workers without
joining them in `shutdown()`. A worker left from an earlier manager/test can
therefore resume here and increment `entries`; one such entry plus only one
target version satisfies `entries >= 2`, so this test can pass even if the
other target is still serialized behind a manager-wide lock. Pass the
`PythonVersion` (and, if needed, a manager/pool token) through the hook and
count/block only `version39` and `version310`.
--
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]
