diff --git a/bazel/experiments.bzl b/bazel/experiments.bzl index 8dffc68bf2125..0a345c89bce41 100644 --- a/bazel/experiments.bzl +++ b/bazel/experiments.bzl @@ -43,7 +43,7 @@ EXPERIMENT_ENABLES = { "trace_record_callops": "trace_record_callops", "unconstrained_max_quota_buffer_size": "unconstrained_max_quota_buffer_size", "work_serializer_clears_time_cache": "work_serializer_clears_time_cache", - "work_serializer_dispatch": "event_engine_client,work_serializer_dispatch", + "work_serializer_dispatch": "work_serializer_dispatch", "call_v3": "call_v3,event_engine_client,event_engine_listener,work_serializer_dispatch", } diff --git a/src/core/lib/event_engine/thread_pool/work_stealing_thread_pool.cc b/src/core/lib/event_engine/thread_pool/work_stealing_thread_pool.cc index 7c0e421c4f7a3..89e8f9962e762 100644 --- a/src/core/lib/event_engine/thread_pool/work_stealing_thread_pool.cc +++ b/src/core/lib/event_engine/thread_pool/work_stealing_thread_pool.cc @@ -321,13 +321,18 @@ void WorkStealingThreadPool::WorkStealingThreadPoolImpl::PrepareFork() { LOG(INFO) << "WorkStealingThreadPoolImpl::PrepareFork"; SetForking(true); work_signal_.SignalAll(); + LOG(INFO) << "WorkStealingThreadPoolImpl::PrepareFork SignallAll() called"; auto threads_were_shut_down = living_thread_count_.BlockUntilThreadCount( 0, "forking", kBlockUntilThreadCountTimeout); + LOG(INFO) << "WorkStealingThreadPoolImpl::PrepareFork after blocking: " + "thread_shut_down: " + << threads_were_shut_down; if (!threads_were_shut_down.ok() && g_log_verbose_failures) { DumpStacksAndCrash(); } grpc_core::MutexLock lock(&lifeguard_ptr_mu_); lifeguard_.reset(); + LOG(INFO) << "Done WorkStealingThreadPoolImpl::PrepareFork"; } void WorkStealingThreadPool::WorkStealingThreadPoolImpl::Postfork() { diff --git a/src/core/lib/experiments/experiments.cc b/src/core/lib/experiments/experiments.cc index 8fc435e0f24ca..bcaedb6d91aa4 100644 --- a/src/core/lib/experiments/experiments.cc +++ b/src/core/lib/experiments/experiments.cc @@ -129,8 +129,6 @@ const char* const description_work_serializer_dispatch = "callback, instead of running things inline in the first thread that " "successfully enqueues work."; const char* const additional_constraints_work_serializer_dispatch = "{}"; -const uint8_t required_experiments_work_serializer_dispatch[] = { - static_cast(grpc_core::kExperimentIdEventEngineClient)}; const char* const description_call_v3 = "Promise-based call version 3."; const char* const additional_constraints_call_v3 = "{}"; const uint8_t required_experiments_call_v3[] = { @@ -208,8 +206,7 @@ const ExperimentMetadata g_experiment_metadata[] = { additional_constraints_work_serializer_clears_time_cache, nullptr, 0, true, true}, {"work_serializer_dispatch", description_work_serializer_dispatch, - additional_constraints_work_serializer_dispatch, - required_experiments_work_serializer_dispatch, 1, false, true}, + additional_constraints_work_serializer_dispatch, nullptr, 0, false, true}, {"call_v3", description_call_v3, additional_constraints_call_v3, required_experiments_call_v3, 3, false, false}, }; @@ -323,8 +320,6 @@ const char* const description_work_serializer_dispatch = "callback, instead of running things inline in the first thread that " "successfully enqueues work."; const char* const additional_constraints_work_serializer_dispatch = "{}"; -const uint8_t required_experiments_work_serializer_dispatch[] = { - static_cast(grpc_core::kExperimentIdEventEngineClient)}; const char* const description_call_v3 = "Promise-based call version 3."; const char* const additional_constraints_call_v3 = "{}"; const uint8_t required_experiments_call_v3[] = { @@ -402,8 +397,7 @@ const ExperimentMetadata g_experiment_metadata[] = { additional_constraints_work_serializer_clears_time_cache, nullptr, 0, true, true}, {"work_serializer_dispatch", description_work_serializer_dispatch, - additional_constraints_work_serializer_dispatch, - required_experiments_work_serializer_dispatch, 1, false, true}, + additional_constraints_work_serializer_dispatch, nullptr, 0, false, true}, {"call_v3", description_call_v3, additional_constraints_call_v3, required_experiments_call_v3, 3, false, false}, }; @@ -517,8 +511,6 @@ const char* const description_work_serializer_dispatch = "callback, instead of running things inline in the first thread that " "successfully enqueues work."; const char* const additional_constraints_work_serializer_dispatch = "{}"; -const uint8_t required_experiments_work_serializer_dispatch[] = { - static_cast(grpc_core::kExperimentIdEventEngineClient)}; const char* const description_call_v3 = "Promise-based call version 3."; const char* const additional_constraints_call_v3 = "{}"; const uint8_t required_experiments_call_v3[] = { @@ -596,8 +588,7 @@ const ExperimentMetadata g_experiment_metadata[] = { additional_constraints_work_serializer_clears_time_cache, nullptr, 0, true, true}, {"work_serializer_dispatch", description_work_serializer_dispatch, - additional_constraints_work_serializer_dispatch, - required_experiments_work_serializer_dispatch, 1, true, true}, + additional_constraints_work_serializer_dispatch, nullptr, 0, true, true}, {"call_v3", description_call_v3, additional_constraints_call_v3, required_experiments_call_v3, 3, false, false}, }; diff --git a/src/core/lib/experiments/rollouts.yaml b/src/core/lib/experiments/rollouts.yaml index fe679a5718e49..d719ab3047798 100644 --- a/src/core/lib/experiments/rollouts.yaml +++ b/src/core/lib/experiments/rollouts.yaml @@ -111,7 +111,6 @@ - name: work_serializer_clears_time_cache default: true - name: work_serializer_dispatch - requires: ["event_engine_client"] default: # TODO(ysseung): Not fully tested. ios: broken diff --git a/src/python/grpcio_tests/tests/fork/methods.py b/src/python/grpcio_tests/tests/fork/methods.py index d69815183489c..be5d4eb4e1dea 100644 --- a/src/python/grpcio_tests/tests/fork/methods.py +++ b/src/python/grpcio_tests/tests/fork/methods.py @@ -70,8 +70,21 @@ def _async_unary(stub): ) response_future = stub.UnaryCall.future(request, timeout=_RPC_TIMEOUT_S) + sys.stderr.write("_async_unary response_future built\n") + sys.stderr.flush() response = response_future.result() + sys.stderr.write( + "_async_unary response returned, validating exp:{}/{} act:{}/{}\n".format( + size, + messages_pb2.COMPRESSABLE, + len(response.payload.body), + response.payload.type, + ) + ) + sys.stderr.flush() _validate_payload_type_and_length(response, messages_pb2.COMPRESSABLE, size) + sys.stderr.write("_async_unary response successfully validated\n") + sys.stderr.flush() def _blocking_unary(stub): @@ -144,11 +157,17 @@ def _child_main(self): try: self._task(*self._args) except grpc.RpcError as rpc_error: + sys.stderr.write("I'm child, RpcError:\n") + sys.stderr.flush() traceback.print_exc() self._exceptions.put("RpcError: %s" % rpc_error) except Exception as e: # pylint: disable=broad-except + sys.stderr.write("I'm child, exception:\n") + sys.stderr.flush() traceback.print_exc() self._exceptions.put(e) + sys.stderr.write("I'm child, successful:\n") + sys.stderr.flush() sys.exit(0) def _orchestrate_child_gdb(self): @@ -173,18 +192,35 @@ def _orchestrate_child_gdb(self): def start(self): # NOTE: Try uncommenting the following line if the child is segfaulting. - # self._orchestrate_child_gdb() + self._orchestrate_child_gdb() ret = os.fork() + if ret == 0: + sys.stderr.write("I'm child, forked:\n") + sys.stderr.flush() self._child_main() else: + sys.stderr.write("I'm parent, child forked: {}\n".format(ret)) + sys.stderr.flush() self._child_pid = ret def wait(self, timeout): + sys.stderr.write("wait() called timeout:{}\n".format(timeout)) + sys.stderr.flush() total = 0.0 wait_interval = 1.0 while total < timeout: + sys.stderr.write( + "calling waitpid pid:{} total/timeout:{}/{}\n".format( + self._child_pid, total, timeout + ) + ) + sys.stderr.flush() ret, termination = os.waitpid(self._child_pid, os.WNOHANG) + sys.stderr.write( + "waitpid returned {},{}\n".format(ret, termination) + ) + sys.stderr.flush() if ret == self._child_pid: self._rc = termination return True @@ -231,8 +267,11 @@ def _print_backtraces(self): sys.stderr.flush() def finish(self): + sys.stderr.write("finish() called\n") + sys.stderr.flush() terminated = self.wait(_CHILD_FINISH_TIMEOUT_S) sys.stderr.write("Exit code: {}\n".format(self._rc)) + sys.stderr.flush() if not terminated: self._print_backtraces() raise RuntimeError("Child process did not terminate") @@ -250,12 +289,22 @@ def finish(self): def _async_unary_same_channel(channel): def child_target(): + sys.stderr.write("async_unary_same_channel::child_target() started.") + sys.stderr.flush() try: _async_unary(stub) + sys.stderr.write("async_unary returned without an exception") + sys.stderr.flush() raise Exception( "Child should not be able to re-use channel after fork" ) except ValueError as expected_value_error: + sys.stderr.write( + "async_unary returned with an exception: {}".format( + expected_value_error + ) + ) + sys.stderr.flush() pass stub = test_pb2_grpc.TestServiceStub(channel)