[MapServer]: ThreadPool Executor leads to CPU thrashing when combined with readBatchSize and concurrency

[AdityaSoni19031997]

Summary

Python versions less than 3.14t have GIL in place which ensures that only one thread can execute at a time on one core and Python doesn't use all the cores out of the box. Now, if the SyncMapServer code, we use ThreadPool Executor to trigger the map's handler concurrently which turn very easily leads to CPU thrashing depending on the end user's actual UDF core logic (whether it's pure I/O or CPU)

Asyncio helps Python execute I/O-bound workloads concurrently, but does not help work around the Python GIL and provide CPU parallelism.

What change needs making?

• We can likely revisit and decide if we want to control a singleTon like ProcessPoolExecutor as well to give end users the capacity to choose

• Code Reference:

  numaflow-python/packages/pynumaflow/pynumaflow/mapper/_servicer/_sync_servicer.py

  Lines 44 to 80 in 5fd1823

  	try:
  	# The first message to be received should be a valid handshake
  	req = next(request_iterator)
  	# check if it is a valid handshake req
  	if not (req.handshake and req.handshake.sot):
  	raise MapError("MapFn: expected handshake as the first message")
  	yield map_pb2.MapResponse(handshake=map_pb2.Handshake(sot=True))
  	# result queue to stream messages from the user code back to the client
  	result_queue = SyncIterator()
  	# Reader thread to keep reading from the request iterator and schedule
  	# execution for each of them
  	reader_thread = threading.Thread(
  	target=self._process_requests, args=(context, request_iterator, result_queue)
  	)
  	reader_thread.start()
  	# Read the result queue and keep forwarding them upstream
  	for res in result_queue.read_iterator():
  	# if error handler accordingly
  	if isinstance(res, BaseException):
  	if isinstance(res, grpc.RpcError):
  	# Client disconnected mid-stream — the reader thread
  	# surfaced the error via the queue. Not a UDF fault.
  	_LOGGER.warning("gRPC stream closed, shutting down the server.")
  	result_queue.close()
  	self.shutdown_event.set()
  	return
  	err_msg = f"{ERR_UDF_EXCEPTION_STRING}: {repr(res)}"
  	update_context_err(context, res, err_msg)
  	# Unblock the reader thread if it is waiting on queue.put()
  	result_queue.close()
  	self.error = res
  	self.shutdown_event.set()
  	return
  	# return the result
  	yield res

Use Cases

When would you use this?

• CPU Bound use-cases would benefit from this as people might NOT realize what's happening and assume that they're processing things concurrently when it's sequential in reality (Not saying it's hard to spot, it's just they need to realize it sooner than later)

---

Message from the maintainers:

If you wish to see this enhancement implemented please add a 👍 reaction to this issue! We often sort issues this way to know what to prioritize.

[AdityaSoni19031997]

PS: I know that we've different alternatives at hand (AsyncServer, BatchMapServer etc) but we can maybe add a warning etc and be a bit more explicit as to what to expect maybe?

Edits: The code is not protected with Semaphores, so it will lead to a very high trigger on UDF side.

Thanks!