Optimize lf.concurrent_execute.

When `executor` is empty and `max_worker` is 1, we could avoid the overhead of creating a short-lived thread pool.

PiperOrigin-RevId: 615501183

langfun/core/concurrent.py

@@ -215,6 +215,12 @@ def concurrent_execute(
         exponential_backoff=exponential_backoff,
     )
 
+  # NOTE(daiyip): when executor is not specified and max_worker is 1,
+  # we don't need to create a executor pool. Instead, the inputs will be
+  # processed by the user function in sequence within the current thread.
+  if executor is None and max_workers == 1:
+    return [func(i) for i in parallel_inputs]
+
   shutdown_after_finish = executor is None
   executor = _executor_pool.executor_from(executor, max_workers=max_workers)
 

langfun/core/concurrent_test.py

@@ -163,6 +163,16 @@ def fun(a):
     with component.context(y=2):
       self.assertEqual(concurrent.concurrent_execute(fun, [A(1), A(2)]), [2, 4])
 
+  def test_concurrent_execute_with_a_single_worker(self):
+    def fun(a):
+      return a.x * a.y
+
+    with component.context(y=2):
+      self.assertEqual(
+          concurrent.concurrent_execute(fun, [A(1), A(2)], max_workers=1),
+          [2, 4],
+      )
+
   def test_concurrent_execute_with_external_executor(self):
     def fun(a):
       return a.x * a.y