lizhimins opened a new pull request, #1336: URL: https://github.com/apache/rocketmq-clients/pull/1336
### Which Issue(s) This PR Fixes Fixes #1334 Prior reports of the same crash were auto-closed as stale rather than fixed: #421, #521. ### Brief Description `PushConsumer` has no public `shutdown()` / `close()`, so teardown depends on releasing the last `shared_ptr`. `ConsumeTask::process()` promotes the service's `weak_ptr` into an *owning* `shared_ptr` on a consume worker thread, so an application that drops its last reference while a message is being consumed makes that worker the last owner. `~PushConsumerImpl()` then runs on the worker thread, concurrently with `process()` and process/static teardown, and crashes inside spdlog (`EXC_BAD_ACCESS` at `0x18` on macOS). This PR: - **Adds `PushConsumer::shutdown()`.** Called from the owner thread it cancels the periodic tasks and joins the consume workers synchronously. Because the join blocks the owner until in-flight consume tasks finish, a worker can never be the last owner, so the final destruction always happens on the owner thread. It is idempotent and remains safe to call again from the destructor. - **Stops `ThreadPoolImpl::shutdown()` from joining the calling thread.** Self-join raises `std::system_error(EDEADLK)`, and since `PushConsumerImpl::shutdown()` is `noexcept` that terminated the process. The current worker is now detached instead, so it unwinds once `io_context::run()` returns. - **Removes the `SPDLOG_DEBUG` call from `~PushConsumerImpl()`**, which could touch the static default logger after it was destroyed. - **Null-checks the locked consumer in `ConsumeTask::process()`** so an expired `PushConsumerImpl` no longer causes a null dereference. Two notes for reviewers, so the scope is not overstated: 1. The `ThreadPoolImpl` self-join guard only *hardens* the unsupported path where teardown is driven from a worker thread; it downgrades a guaranteed `terminate` to a survivable unwind. Deterministic teardown still requires calling `PushConsumer::shutdown()` from a non-worker thread, and that is what the API documentation now states. 2. This aligns C++ with the other SDKs rather than adding a new concept — Java already has `PushConsumer extends Closeable` with `close()`, and Go has `GracefulStop()`. C++ was the only one without an explicit shutdown. The crash is not FIFO-specific: `ConsumeTask::process()`, `~PushConsumerImpl()` and `ThreadPoolImpl::shutdown()` are shared by all message types, and `fifo_` only selects a `NextStep` branch. FIFO just widens the window because consumption is serialized. ### How Did You Test This Change? Three regression tests were added, and the first two were confirmed to fail against the unfixed code before the fix was applied: - `ThreadPoolTest.shutdownFromWorkerThreadDoesNotThrowTest` — shuts the pool down from one of its own workers. Before the fix this aborts with `terminate called without an active exception`. - `ConsumeTaskTest.processWithExpiredConsumerTest` — processes a task whose owning consumer has expired. Before the fix this segfaults. - `PushConsumerImplTest.shutdownIsIdempotentTest` — covers repeated `shutdown()` calls plus the implicit one from the destructor. Full C++ suite on this branch, rebased directly on `master`: ``` $ bazel test //source/... Executed 36 out of 37 tests: 37 tests pass. ``` This branch contains only this fix and sits directly on `master`, so it can be reviewed and merged independently of my producer-side fix for #1335. -- 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]
