Fix async actor recursion limitation (#6672)

* Do not start threadpool when using async

* Turn function_executor into a generator

* Add new test for high concurrency and bump the default

* Set direct call

src/ray/core_worker/transport/direct_actor_transport.cc

@@ -186,7 +186,7 @@ void CoreWorkerDirectTaskReceiver::SetMaxActorConcurrency(int max_concurrency) {
   }
 }
 
-void CoreWorkerDirectTaskReceiver::SetActorAsAsync() {
+void CoreWorkerDirectTaskReceiver::SetActorAsAsync(int max_concurrency) {
   if (!is_asyncio_) {
     RAY_LOG(DEBUG) << "Setting direct actor as async, creating new fiber thread.";
 
@@ -204,7 +204,8 @@ void CoreWorkerDirectTaskReceiver::SetActorAsAsync() {
       // immediately start working on any ready fibers.
       fiber_shutdown_event_.Wait();
     });
-    fiber_rate_limiter_.reset(new FiberRateLimiter(max_concurrency_));
+    fiber_rate_limiter_.reset(new FiberRateLimiter(max_concurrency));
+    max_concurrency_ = max_concurrency;
     is_asyncio_ = true;
   }
 };
@@ -220,9 +221,13 @@ void CoreWorkerDirectTaskReceiver::HandlePushTask(
                         nullptr, nullptr);
     return;
   }
-  SetMaxActorConcurrency(worker_context_.CurrentActorMaxConcurrency());
+
+  // Only call SetMaxActorConcurrency to configure threadpool size when the
+  // actor is not async actor. Async actor is single threaded.
   if (worker_context_.CurrentActorIsAsync()) {
-    SetActorAsAsync();
+    SetActorAsAsync(worker_context_.CurrentActorMaxConcurrency());
+  } else {
+    SetMaxActorConcurrency(worker_context_.CurrentActorMaxConcurrency());
   }
 
   std::vector<ObjectID> dependencies;

src/ray/core_worker/transport/direct_actor_transport.h

@@ -285,7 +285,7 @@ class FiberRateLimiter {
  private:
   boost::fibers::condition_variable cond_;
   boost::fibers::mutex mutex_;
-  int num_;
+  int num_ = 1;
 };
 
 /// Used to ensure serial order of task execution per actor handle.
@@ -472,7 +472,8 @@ class CoreWorkerDirectTaskReceiver {
   /// Set the max concurrency at runtime. It cannot be changed once set.
   void SetMaxActorConcurrency(int max_concurrency);
 
-  void SetActorAsAsync();
+  /// Set the max concurrency and start async actor context.
+  void SetActorAsAsync(int max_concurrency);
 
  private:
   // Worker context.