class_docstr: Worker pool entry point.
  
  The worker pool exposes an RPC service that is used in MiniCluster to start and stop the Python
  SDK workers.
  
  The worker pool uses child thread for parallelism
class_docstr_tok: ["Worker","pool","entry","point",".","The","worker","pool","exposes","an","RPC","service","that","is","used","in","MiniCluster","to","start","and","stop","the","Python","SDK","workers",".","The","worker","pool","uses","child","thread","for","parallelism"]
code: def _start_sdk_worker_main(
      self, start_worker_request: beam_fn_api_pb2.StartWorkerRequest
  ):
      params = start_worker_request.params
      self._parse_param_lock.acquire()
      # The first thread to start is responsible for preparing all execution environment.
      if not self._ref_cnt:
          if "PYTHONPATH" in params:
              self._old_python_path = sys.path[:]
              python_path_list = params["PYTHONPATH"].split(":")
              python_path_list.reverse()
              for path in python_path_list:
                  sys.path.insert(0, path)
          if "_PYTHON_WORKING_DIR" in params:
              self._old_working_dir = os.getcwd()
              os.chdir(params["_PYTHON_WORKING_DIR"])
          os.environ.update(params)
      self._ref_cnt += 1
      self._parse_param_lock.release()
  
      # read job information from provision stub
      metadata = [("worker_id", start_worker_request.worker_id)]
      provision_endpoint = start_worker_request.provision_endpoint.url
      with grpc.insecure_channel(provision_endpoint) as channel:
          client = ProvisionServiceStub(channel=channel)
          info = client.GetProvisionInfo(
              GetProvisionInfoRequest(), metadata=metadata
          ).info
          options = json_format.MessageToJson(info.pipeline_options)
          logging_endpoint = info.logging_endpoint.url
          control_endpoint = info.control_endpoint.url
  
      try:
          logging_service_descriptor = (
              endpoints_pb2.ApiServiceDescriptor(url=logging_endpoint)
          )
  
          # Send all logs to the runner.
          fn_log_handler = FnApiLogRecordHandler(
              logging_service_descriptor
          )
          logging.getLogger().setLevel(logging.INFO)
          # Remove all the built-in log handles
          logging.getLogger().handlers = []
          logging.getLogger().addHandler(fn_log_handler)
          logging.info("Starting up Python worker in loopback mode.")
      except Exception:
          _LOGGER.error(
              "Failed to set up logging handler, continuing without.",
              exc_info=True,
          )
          fn_log_handler = None
  
      sdk_pipeline_options = sdk_worker_main._parse_pipeline_options(
          options
      )
  
      _worker_id = start_worker_request.worker_id
  
      try:
          control_service_descriptor = (
              endpoints_pb2.ApiServiceDescriptor(url=control_endpoint)
          )
          status_service_descriptor = (
              endpoints_pb2.ApiServiceDescriptor()
          )
  
          experiments = (
              sdk_pipeline_options.view_as(DebugOptions).experiments
              or []
          )
          enable_heap_dump = "enable_heap_dump" in experiments
          SdkHarness(
              control_address=control_service_descriptor.url,
              status_address=status_service_descriptor.url,
              worker_id=_worker_id,
              state_cache_size=sdk_worker_main._get_state_cache_size(
                  experiments
              ),
              data_buffer_time_limit_ms=sdk_worker_main._get_data_buffer_time_limit_ms(
                  experiments
              ),
              profiler_factory=profiler.Profile.factory_from_options(
                  sdk_pipeline_options.view_as(ProfilingOptions)
              ),
              enable_heap_dump=enable_heap_dump,
          ).run()
      except:  # pylint: disable=broad-except
          _LOGGER.exception("Python sdk harness failed: ")
          raise
      finally:
          self._parse_param_lock.acquire()
          self._ref_cnt -= 1
          # The last thread to exit is responsible for reverting working directory and sys.path.
          if self._ref_cnt == 0:
              if self._old_python_path is not None:
                  sys.path.clear()
                  for item in self._old_python_path:
                      sys.path.append(item)
                  self._old_python_path = None
              if self._old_working_dir is not None:
                  os.chdir(self._old_working_dir)
                  self._old_working_dir = None
          self._parse_param_lock.release()
          if fn_log_handler:
              fn_log_handler.close()
  
imports: import atexit
  import functools
  import logging
  import os
  import sys
  import threading
  import traceback
  import grpc
  from apache_beam.options.pipeline_options import DebugOptions
  from apache_beam.options.pipeline_options import ProfilingOptions
  from apache_beam.portability.api import beam_fn_api_pb2
  from apache_beam.portability.api import beam_fn_api_pb2_grpc
  from apache_beam.portability.api import endpoints_pb2
  from apache_beam.portability.api.org.apache.beam.model.fn_execution.v1.beam_provision_api_pb2 import GetProvisionInfoRequest
  from apache_beam.portability.api.org.apache.beam.model.fn_execution.v1.beam_provision_api_pb2_grpc import ProvisionServiceStub
  from apache_beam.runners.worker import sdk_worker_main
  from apache_beam.runners.worker.log_handler import FnApiLogRecordHandler
  from apache_beam.runners.worker.sdk_worker import SdkHarness
  from apache_beam.utils import thread_pool_executor, profiler
  from google.protobuf import json_format
  from pyflink.fn_execution.beam import beam_sdk_worker_main
label_desc: Use image node_id 6 for calling the BeamFnLoopbackWorkerPoolServicer obj's underlying member method code with example usage: obj._start_sdk_worker_main(start_worker_request) without return types