Fix to avoid idle connections from blocking worker threads

lib/grpcxx/worker.cpp

@@ -2,42 +2,37 @@
 
 namespace grpcxx {
 namespace detail {
-worker::worker() : _cv(), _handles(), _loop(), _mutex() {
+worker::worker() : _async(), _handles(), _loop(), _mutex() {
 	uv_loop_init(&_loop);
 }
 
 void worker::enqueue(std::coroutine_handle<> h) noexcept {
-	std::lock_guard lock(_mutex);
-	_handles.emplace(h);
-	_cv.notify_one();
+	{
+		std::lock_guard lock(_mutex);
+		_handles.emplace(h);
+	}
+
+	uv_async_send(&_async);
 }
 
 void worker::run() {
-	while (true) {
-		if (uv_run(&_loop, UV_RUN_ONCE) != 0) {
-			// Give extra cpu time if the loop is "active"
-			for (uint8_t r = _loop.active_handles; r > 0; r--) {
-				if (uv_run(&_loop, UV_RUN_NOWAIT) == 0) {
-					break;
-				}
-			}
-		}
+	_async.data = this;
+	uv_async_init(&_loop, &_async, [](uv_async_t *handle) {
+		auto *w = static_cast<worker *>(handle->data);
 
-		std::unique_lock lock(_mutex);
-		while (uv_loop_alive(&_loop) == 0 && _handles.empty()) {
-			// FIXME: make timeout configurable
-			_cv.wait_for(lock, std::chrono::microseconds(100));
-		}
-
-		while (!_handles.empty()) {
-			auto &h = _handles.front();
-			_handles.pop();
+		while (!w->_handles.empty()) {
+			handles_t::value_type h;
+			{
+				std::lock_guard lock(w->_mutex);
+				h = w->_handles.front();
+				w->_handles.pop();
+			}
 
-			lock.unlock();
 			h.resume();
-			lock.lock();
 		}
-	}
+	});
+
+	uv_run(&_loop, UV_RUN_DEFAULT);
 }
 } // namespace detail
 } // namespace grpcxx

lib/grpcxx/worker.h

@@ -1,6 +1,5 @@
 #pragma once
 
-#include <condition_variable>
 #include <coroutine>
 #include <queue>
 #include <thread>
@@ -22,11 +21,11 @@ class worker {
 private:
 	using handles_t = std::queue<std::coroutine_handle<>>;
 
-	std::condition_variable _cv;
-	handles_t               _handles;
-	std::mutex              _mutex;
+	handles_t  _handles;
+	std::mutex _mutex;
 
-	uv_loop_t _loop;
+	uv_async_t _async;
+	uv_loop_t  _loop;
 };
 } // namespace detail
 } // namespace grpcxx