[CIVIS-1949] PERF more efficient polling at future objects (#492)

* REF _ResultPollingThread doesn't actually take poller and poller_args

* ENH exponential polling

* ENH switch to backoff factor of 1.2

* TST geometric polling

* DOC update docstrings and changelog

* DOC update docs

* DOC update docs

Author: jacksonlee-civis

CHANGELOG.md

@@ -20,6 +20,10 @@ and this project adheres to [Semantic Versioning](http://semver.org/).
 - Improved the startup time of `import civis` with a 5x speed boost. (#490)
 - The downloaded API spec due to the `civis.APIClient` instantiation is now
   a time-to-live cache in memory (15 minutes for interactive Python, or 24 hours in scripts). (#491)
+- Polling at `PollableResult` (and consequently its subclasses as well: `CivisFuture`,
+  `ContainerFuture`, and `ModelFuture`) now defaults to geometrically increased polling
+  intervals. Short-running jobs' `future.result()` can now return faster, while
+  longer-running jobs have a capped polling interval of 15 seconds. (#492)
 
 ### Deprecated
 ### Removed

docs/source/user_guide.rst

@@ -34,6 +34,7 @@ uploads the data back into Civis:
    ...                                   table="my_schema.my_correlations")
    >>> fut.result()
 
+.. _civis_futures:
 
 Civis Futures
 =============
@@ -157,28 +158,17 @@ data to export. Then we create the export job and run it.
                                             remote_host_id=db_id,
                                             credential_id=cred_id,
                                             sql=generate_table)
-   >>> export_run = client.scripts.post_sql_runs(export_job.id)
+   >>> export_future = civis.utils.run_job(export_job.id)
 
-We can then poll and wait for the export to be completed.
+``export_future`` is a :class:`CivisFuture <civis.futures.CivisFuture>` object
+(see :ref:`civis_futures`  above). Calling ``.result()`` on ``export_future``
+blocks the program until the SQL run has completed.
 
 .. code:: python
 
-   >>> import time
-   >>> export_state = client.scripts.get_sql_runs(export_job.id,
-   ...                                            export_run.id)
-   >>> while export_state.state in ['queued', 'running']:
-   ...    time.sleep(60)
-   ...    export_state = client.scripts.get_sql_runs(export_job.id,
-   ...                                               export_run.id)
-
-Now, we can get the URL of the exported csv. First, we grab the result of our
-export job.
-
-.. code:: python
-
-   >>> export_result = client.scripts.get_sql_runs(export_job.id,
-   ...                                             export_run.id)
+   >>> export_result = export_future.result()
 
+Now, we can get the URL of the exported csv.
 In the future, a script may export multiple jobs, so the output of this is a
 list.
 

src/civis/__init__.py

@@ -19,6 +19,7 @@ def _lazy_import(name):
     return module
 
 
+futures = _lazy_import("civis.futures")
 io = _lazy_import("civis.io")
 ml = _lazy_import("civis.ml")
 parallel = _lazy_import("civis.parallel")

src/civis/base.py

@@ -173,40 +173,8 @@ class CivisAsyncResultBase(futures.Future):
     the result. The `_result_` object needs to be set to an object with a
     ``state`` attribute. Alternatively the `_check_result` method can be
     overwritten to change how the state of the object is returned.
-
-    Parameters
-    ----------
-    poller : func
-        A function which returns an object that has a ``state`` attribute.
-    poller_args : tuple
-        The arguments with which to call the poller function.
-    polling_interval : int or float
-        The number of seconds between API requests to check whether a result
-        is ready.
-    client : :class:`civis.APIClient`, optional
-        If not provided, an :class:`civis.APIClient` object will be
-        created from the :envvar:`CIVIS_API_KEY`.
-    poll_on_creation : bool, optional
-        If ``True`` (the default), it will poll upon calling ``result()`` the
-        first time. If ``False``, it will wait the number of seconds specified
-        in `polling_interval` from object creation before polling.
     """
 
-    def __init__(
-        self,
-        poller,
-        poller_args,
-        polling_interval=None,
-        client=None,
-        poll_on_creation=True,
-    ):
-        super().__init__()
-        self.poller = poller
-        self.poller_args = poller_args
-        self.polling_interval = polling_interval
-        self.client = client
-        self.poll_on_creation = poll_on_creation
-
     def __repr__(self):
         # Almost the same as the superclass's __repr__, except we use
         # the `_civis_state` rather than the `_state`.

src/civis/futures.py

@@ -32,7 +32,12 @@ class CivisFuture(PollableResult):
         The arguments with which to call the poller function.
     polling_interval : int or float, optional
         The number of seconds between API requests to check whether a result
-        is ready.
+        is ready. If an integer or float is provided, this number will be used
+        as the polling interval. If ``None`` (the default), the polling interval will
+        start at 1 second and increase geometrically up to 15 seconds. The ratio of
+        the increase is 1.2, resulting in polling intervals in seconds of
+        1, 1.2, 1.44, 1.728, etc. This default behavior allows for a faster return for
+        a short-running job and a capped polling interval for longer-running jobs.
     client : :class:`civis.APIClient`, optional
     poll_on_creation : bool, optional
         If ``True`` (the default), it will poll upon calling ``result()`` the
@@ -231,9 +236,14 @@ class ContainerFuture(CivisFuture):
         The ID for the run to monitor
     max_n_retries : int, optional
         If the job generates an exception, retry up to this many times
-    polling_interval: int or float, optional
+    polling_interval : int or float, optional
         The number of seconds between API requests to check whether a result
-        is ready.
+        is ready. If an integer or float is provided, this number will be used
+        as the polling interval. If ``None`` (the default), the polling interval will
+        start at 1 second and increase geometrically up to 15 seconds. The ratio of
+        the increase is 1.2, resulting in polling intervals in seconds of
+        1, 1.2, 1.44, 1.728, etc. This default behavior allows for a faster return for
+        a short-running job and a capped polling interval for longer-running jobs.
     client : :class:`civis.APIClient`, optional
         If not provided, an :class:`civis.APIClient` object will be
         created from the :envvar:`CIVIS_API_KEY`.

src/civis/ml/_model.py

@@ -1,7 +1,6 @@
 """Run CivisML jobs and retrieve the results"""
 
 import builtins
-from builtins import super
 import collections
 from functools import lru_cache
 import io
@@ -394,7 +393,12 @@ class ModelFuture(ContainerFuture):
         run, and ``train_run_id`` will equal ``run_id``.
     polling_interval : int or float, optional
         The number of seconds between API requests to check whether a result
-        is ready.
+        is ready. If an integer or float is provided, this number will be used
+        as the polling interval. If ``None`` (the default), the polling interval will
+        start at 1 second and increase geometrically up to 15 seconds. The ratio of
+        the increase is 1.2, resulting in polling intervals in seconds of
+        1, 1.2, 1.44, 1.728, etc. This default behavior allows for a faster return for
+        a short-running job and a capped polling interval for longer-running jobs.
     client : :class:`civis.APIClient`, optional
         If not provided, an :class:`civis.APIClient` object will be
         created from the :envvar:`CIVIS_API_KEY`.
@@ -555,6 +559,8 @@ def __setstate__(self, state):
         self._condition = threading.Condition()
         self.client = APIClient()
         self.poller = self.client.scripts.get_containers_runs
+        self._next_polling_interval = 1
+        self._use_geometric_polling = True
         self._begin_tracking()
         self.add_done_callback(self._set_job_exception)
 

src/civis/polling.py

@@ -5,19 +5,17 @@
 from civis.response import Response
 
 
-_DEFAULT_POLLING_INTERVAL = 15
+_MAX_POLLING_INTERVAL = 15
 
 
 class _ResultPollingThread(threading.Thread):
     """Poll a function until it returns a Response with a DONE state"""
 
     # Inspired by `threading.Timer`
 
-    def __init__(self, poller, poller_args, polling_interval):
+    def __init__(self, pollable_result):
         super().__init__(daemon=True)
-        self.polling_interval = polling_interval
-        self.poller = poller
-        self.poller_args = poller_args
+        self.pollable_result = pollable_result
         self.finished = threading.Event()
 
     def cancel(self):
@@ -31,11 +29,11 @@ def join(self, timeout=None):
 
     def run(self):
         """Poll until done."""
-        while not self.finished.wait(self.polling_interval):
+        while not self.finished.wait(self.pollable_result._next_polling_interval):
             # Spotty internet connectivity can result in polling functions
             # returning None. This treats None responses like responses which
             # have a non-DONE state.
-            poller_result = self.poller(*self.poller_args)
+            poller_result = self.pollable_result._check_result()
             if poller_result is not None and poller_result.state in DONE:
                 self.finished.set()
 
@@ -53,9 +51,14 @@ class PollableResult(CivisAsyncResultBase):
         A function which returns an object that has a ``state`` attribute.
     poller_args : tuple
         The arguments with which to call the poller function.
-    polling_interval : int or float
+    polling_interval : int or float, optional
         The number of seconds between API requests to check whether a result
-        is ready.
+        is ready. If an integer or float is provided, this number will be used
+        as the polling interval. If ``None`` (the default), the polling interval will
+        start at 1 second and increase geometrically up to 15 seconds. The ratio of
+        the increase is 1.2, resulting in polling intervals in seconds of
+        1, 1.2, 1.44, 1.728, etc. This default behavior allows for a faster return for
+        a short-running job and a capped polling interval for longer-running jobs.
     client : :class:`civis.APIClient`, optional
         If not provided, an :class:`civis.APIClient` object will be
         created from the :envvar:`CIVIS_API_KEY`.
@@ -103,18 +106,20 @@ def __init__(
         client=None,
         poll_on_creation=True,
     ):
-        if polling_interval is None:
-            polling_interval = _DEFAULT_POLLING_INTERVAL
-        super().__init__(
-            poller=poller,
-            poller_args=poller_args,
-            polling_interval=polling_interval,
-            client=client,
-            poll_on_creation=poll_on_creation,
-        )
-        if self.polling_interval <= 0:
+        super().__init__()
+
+        self.poller = poller
+        self.poller_args = poller_args
+        self.polling_interval = polling_interval
+        self.client = client
+        self.poll_on_creation = poll_on_creation
+
+        if self.polling_interval is not None and self.polling_interval <= 0:
             raise ValueError("The polling interval must be positive.")
 
+        self._next_polling_interval = 1
+        self._use_geometric_polling = True
+
         # Polling arguments. Never poll more often than the requested interval.
         if poll_on_creation:
             self._last_polled = None
@@ -153,8 +158,20 @@ def _check_result(self):
             now = time.time()
             if (
                 not self._last_polled
-                or (now - self._last_polled) >= self.polling_interval
+                or (now - self._last_polled) >= self._next_polling_interval
             ):
+                if self._use_geometric_polling:
+                    # Choosing a common ratio of 1.2 for these polling intervals:
+                    #   1, 1.2, 1.44, 1.73, 2.07, 2.49, 2.99, ..., and capped at 15.
+                    # Within the first 15 secs by wall time, we call the poller 7 times,
+                    # which gives a short-running job's future.result()
+                    # a higher chance to return faster.
+                    # For longer running jobs, the polling interval will be capped
+                    # at 15 secs when by wall time 87 secs have passed.
+                    self._next_polling_interval *= 1.2
+                    if self._next_polling_interval > _MAX_POLLING_INTERVAL:
+                        self._next_polling_interval = _MAX_POLLING_INTERVAL
+                        self._use_geometric_polling = False
                 # Poll for a new result
                 self._last_polled = now
                 try:
@@ -204,18 +221,16 @@ def cleanup(self):
             if self._polling_thread.is_alive():
                 self._polling_thread.cancel()
 
-    def _reset_polling_thread(
-        self, polling_interval=_DEFAULT_POLLING_INTERVAL, start_thread=False
-    ):
+    def _reset_polling_thread(self, polling_interval, start_thread=False):
         with self._condition:
             if (
                 getattr(self, "_polling_thread", None) is not None
                 and self._polling_thread.is_alive()
             ):
                 self._polling_thread.cancel()
             self.polling_interval = polling_interval
-            self._polling_thread = _ResultPollingThread(
-                self._check_result, (), polling_interval
-            )
+            self._next_polling_interval = 1 if (pi := polling_interval) is None else pi
+            self._use_geometric_polling = polling_interval is None
+            self._polling_thread = _ResultPollingThread(self)
             if start_thread:
                 self._polling_thread.start()

tests/test_polling.py

@@ -76,10 +76,15 @@ def test_poll_on_creation(self):
         assert poller.call_count > 0
 
     def test_poller_returns_none(self):
-        poller = mock.Mock(side_effect=[None, None, Response({"state": "success"})])
-        polling_thread = _ResultPollingThread(poller, (), polling_interval=0.01)
+        check_result = mock.Mock(
+            side_effect=[None, None, Response({"state": "success"})]
+        )
+        pollable_result = mock.Mock()
+        pollable_result._check_result = check_result
+        pollable_result._next_polling_interval = 0.01
+        polling_thread = _ResultPollingThread(pollable_result)
         polling_thread.run()
-        assert poller.call_count == 3
+        assert check_result.call_count == 3
 
     def test_reset_polling_thread(self):
         pollable = PollableResult(
@@ -89,16 +94,45 @@ def test_reset_polling_thread(self):
         )
         initial_polling_thread = pollable._polling_thread
         assert pollable.polling_interval == 0.1
-        assert pollable._polling_thread.polling_interval == 0.1
+        assert pollable._next_polling_interval == 0.1
         pollable._reset_polling_thread(0.2)
         # Check that the polling interval was updated
         assert pollable.polling_interval == 0.2
-        assert pollable._polling_thread.polling_interval == 0.2
+        assert pollable._next_polling_interval == 0.2
         # Check that the _polling_thread is a new thread
         assert pollable._polling_thread != initial_polling_thread
         # Check that the old thread was stopped
         assert not initial_polling_thread.is_alive()
 
+    def test_geometric_polling(self):
+        # To test polling, we make the poller function spit out a timestamp every time
+        # it is called. Then we check if these timestamps are what we'd expect.
+        poller_timestamps = []
+
+        def append_new_timestamp(*args, **kwargs):
+            nonlocal poller_timestamps
+            poller_timestamps.append(time.time())
+            if len(poller_timestamps) < 5:
+                return Response({"state": "running"})
+            else:
+                return Response({"state": "succeeded"})
+
+        poller = mock.Mock()
+        poller.side_effect = append_new_timestamp
+
+        pollable = PollableResult(poller, (), poll_on_creation=False)
+        start_time = time.time()
+        pollable.result()
+
+        assert len(poller_timestamps) == 5
+        expected_intervals = [1, 1.2, 1.44, 1.728, 2.0736]
+        actual_intervals = []
+        for i, timestamp in enumerate(poller_timestamps):
+            actual_intervals.append(
+                timestamp - (poller_timestamps[i - 1] if i else start_time)
+            )
+        assert actual_intervals == pytest.approx(expected_intervals, abs=0.02)
+
 
 if __name__ == "__main__":
     unittest.main()