Do not busy wait in Python. (#3892)

Instead, keep a stack of outstanding RPCs and await each in turn. This
allows the main loop to block instead of spin.

Fixes #3759.

CHANGELOG.md

@@ -1,6 +1,10 @@
 CHANGELOG
 =========
 
+## HEAD (unreleased)
+- Improve CPU utilization in the Python SDK when waiting for resource operations.
+  [#3892](https://github.com/pulumi/pulumi/pull/3892)
+
 ## 1.10.1 (2020-02-06)
 - Support stack references in the Go SDK.
   [#3829](https://github.com/pulumi/pulumi/pull/3829)
@@ -19,7 +23,8 @@ CHANGELOG
 - Add support for aliases in the Go SDK
   [3853](https://github.com/pulumi/pulumi/pull/3853)
 
-- Fix Python Dynamic Providers on Windows. [#3855](https://github.com/pulumi/pulumi/pull/3855)
+- Fix Python Dynamic Providers on Windows.
+  [#3855](https://github.com/pulumi/pulumi/pull/3855)
 
 ## 1.9.1 (2020-01-27)
 - Fix a stack reference regression in the Python SDK.

sdk/python/lib/pulumi/runtime/rpc_manager.py

@@ -14,7 +14,7 @@
 import asyncio
 import sys
 import traceback
-from typing import Callable, Awaitable, Tuple, Any, Optional
+from typing import Callable, Awaitable, Tuple, Any, Optional, List
 from .. import log
 
 
@@ -26,9 +26,9 @@ class RPCManager:
     outstanding RPCs.
     """
 
-    count: int
+    rpcs: List[Awaitable]
     """
-    The number of active RPCs.
+    The active RPCs.
     """
 
     unhandled_exception: Optional[Exception]
@@ -42,7 +42,7 @@ class RPCManager:
     """
 
     def __init__(self):
-        self.count = 0
+        self.rpcs = []
         self.unhandled_exception = None
         self.exception_traceback = None
 
@@ -60,11 +60,11 @@ class RPCManager:
         """
         async def rpc_wrapper(*args, **kwargs):
             log.debug(f"beginning rpc {name}")
-            self.count += 1
-            log.debug(f"recorded new RPC, {self.count} RPCs outstanding")
 
+            rpc = asyncio.ensure_future(rpc_function(*args, **kwargs))
+            self.rpcs.append(rpc)
             try:
-                result = await rpc_function(*args, **kwargs)
+                result = await rpc
                 exception = None
             except Exception as exn:
                 log.debug(f"RPC failed with exception:")
@@ -75,9 +75,6 @@ class RPCManager:
                 result = None
                 exception = exn
 
-            self.count -= 1
-            log.debug(f"recorded RPC completion, {self.count} RPCs outstanding")
-
             return result, exception
 
         return rpc_wrapper

sdk/python/lib/pulumi/runtime/stack.py

@@ -44,10 +44,14 @@ async def run_in_stack(func: Callable):
         #
         # Note that "asyncio.sleep(0)" is the blessed way to do this:
         # https://github.com/python/asyncio/issues/284#issuecomment-154180935
+        #
+        # We await each RPC in turn so that this loop will actually block rather than busy-wait.
         while True:
             await asyncio.sleep(0)
-            if RPC_MANAGER.count == 0:
+            if len(RPC_MANAGER.rpcs) == 0:
                 break
+            log.debug(f"waiting for quiescence; {len(RPC_MANAGER.rpcs)} RPCs outstanding")
+            await RPC_MANAGER.rpcs.pop()
 
         # Asyncio event loops require that all outstanding tasks be completed by the time that the
         # event loop closes. If we're at this point and there are no outstanding RPCs, we should