Repository: dwavesystems/dwave-cloud-client
Pull request #137 (2018-04-24 00:00:08)

Exponential back-off in status polling
Estimated completion time propagated from API to Future,
exponential back-off for status polls, via priority queue time-based
scheduler.


Patch:
diff --git a/dwave/cloud/client.py b/dwave/cloud/client.py
index 3526da4..cb4661a 100644
--- a/dwave/cloud/client.py
+++ b/dwave/cloud/client.py
@@ -104,6 +104,13 @@ class Client(object):
     _POLL_THREAD_COUNT = 2
     _LOAD_THREAD_COUNT = 5
 
+    # Poll back-off interval [sec]
+    _POLL_BACKOFF_MIN = 1
+    _POLL_BACKOFF_MAX = 60
+
+    # Poll grouping time frame; two scheduled polls are grouped if closer than [sec]:
+    _POLL_GROUP_TIMEFRAME = 2
+
     @classmethod
     def from_config(cls, config_file=None, profile=None, client=None,
                     endpoint=None, token=None, solver=None, proxy=None,
@@ -427,7 +434,7 @@ class Client(object):
             self._cancel_workers.append(worker)
 
         # Build the problem status polling queue, start its workers
-        self._poll_queue = queue.Queue()
+        self._poll_queue = queue.PriorityQueue()
         self._poll_workers = []
         for _ in range(self._POLL_THREAD_COUNT):
             worker = threading.Thread(target=self._do_poll_problems)
@@ -479,7 +486,7 @@ class Client(object):
         for _ in self._cancel_workers:
             self._cancel_queue.put(None)
         for _ in self._poll_workers:
-            self._poll_queue.put(None)
+            self._poll_queue.put((-1, None))
         for _ in self._load_workers:
             self._load_queue.put(None)
 
@@ -643,7 +650,7 @@ class Client(object):
                         break
 
                 # Submit the problems
-                _LOGGER.debug("submitting {} problems".format(len(ready_problems)))
+                _LOGGER.debug("Submitting %d problems", len(ready_problems))
                 body = '[' + ','.join(mess.body for mess in ready_problems) + ']'
                 try:
                     response = self.session.post(posixpath.join(self.endpoint, 'problems/'), body)
@@ -653,8 +660,9 @@ class Client(object):
                     response.raise_for_status()
 
                     message = response.json()
-                    _LOGGER.debug("Finished submitting {} problems".format(len(ready_problems)))
+                    _LOGGER.debug("Finished submitting %d problems", len(ready_problems))
                 except BaseException as exception:
+                    _LOGGER.debug("Submit failed for %d problems", len(ready_problems))
                     if not isinstance(exception, SolverAuthenticationError):
                         exception = IOError(exception)
 
@@ -665,7 +673,7 @@ class Client(object):
 
                 # Pass on the information
                 for submission, res in zip(ready_problems, message):
-                    self._handle_problem_status(res, submission.future, False)
+                    self._handle_problem_status(res, submission.future)
                     self._submission_queue.task_done()
 
                 # this is equivalent to a yield to scheduler in other threading libraries
@@ -674,7 +682,7 @@ class Client(object):
         except BaseException as err:
             _LOGGER.exception(err)
 
-    def _handle_problem_status(self, message, future, in_poll):
+    def _handle_problem_status(self, message, future):
         """Handle the results of a problem submission or results request.
 
         This method checks the status of the problem and puts it in the correct queue.
@@ -682,10 +690,6 @@ class Client(object):
         Args:
             message (dict): Update message from the SAPI server wrt. this problem.
             future `Future`: future corresponding to the problem
-            in_poll (bool): Flag set to true if the problem is in the poll loop already.
-
-        Returns:
-            true if the problem has been processed out of the status poll loop
 
         Note:
             This method is always run inside of a daemon thread.
@@ -716,6 +720,12 @@ class Client(object):
             if not future.time_solved and message.get('solved_on'):
                 future.time_solved = parse_timestamp(message['solved_on'])
 
+            if not future.eta_min and message.get('earliest_completion_time'):
+                future.eta_min = parse_timestamp(message['earliest_completion_time'])
+
+            if not future.eta_max and message.get('latest_completion_time'):
+                future.eta_max = parse_timestamp(message['latest_completion_time'])
+
             if status == self.STATUS_COMPLETE:
                 # If the message is complete, forward it to the future object
                 if 'answer' in message:
@@ -726,9 +736,7 @@ class Client(object):
                     self._load(future)
             elif status in self.ANY_STATUS_ONGOING:
                 # If the response is pending add it to the queue.
-                if not in_poll:
-                    self._poll(future)
-                return False
+                self._poll(future)
             elif status == self.STATUS_CANCELLED:
                 # If canceled return error
                 future._set_error(CanceledFutureError())
@@ -740,7 +748,6 @@ class Client(object):
             # If there were any unhandled errors we need to release the
             # lock in the future, otherwise deadlock occurs.
             future._set_error(error, sys.exc_info())
-        return True
 
     def _cancel(self, id_, future):
         """Enqueue a problem to be canceled.
@@ -795,7 +802,18 @@ class Client(object):
 
         This method is threadsafe.
         """
-        self._poll_queue.put(future)
+        # update exponential poll back-off, clipped to a range
+        future._poll_backoff = \
+            max(self._POLL_BACKOFF_MIN,
+                min(future._poll_backoff * 2, self._POLL_BACKOFF_MAX))
+
+        # for poll priority we use timestamp of next scheduled poll
+        at = time.time() + future._poll_backoff
+
+        _LOGGER.debug("_poll_backoff is %.2f sec (scheduled at %.2f) for %s",
+                      future._poll_backoff, at, future.id)
+
+        self._poll_queue.put((at, future))
 
     def _do_poll_problems(self):
         """Poll the server for the status of a set of problems.
@@ -804,44 +822,66 @@ class Client(object):
             This method is always run inside of a daemon thread.
         """
         try:
-            # Maintain an active group of queries
-            futures = {}
-            active_queries = set()
+            # grouped futures (all scheduled within _POLL_GROUP_TIMEFRAME)
+            frame_futures = {}
+
+            def task_done():
+                self._poll_queue.task_done()
+
+            def add(future):
+                # add future to query frame_futures
+                # returns: worker lives on?
 
-            # Add a query to the active queries
-            def add(ftr):
                 # `None` task signifies thread termination
-                if ftr is None:
+                if future is None:
+                    task_done()
                     return False
-                if ftr.id not in futures and not ftr.done():
-                    active_queries.add(ftr.id)
-                    futures[ftr.id] = ftr
+
+                if future.id not in frame_futures and not future.done():
+                    frame_futures[future.id] = future
                 else:
-                    self._poll_queue.task_done()
-                return True
+                    task_done()
 
-            # Remove a query from the active set
-            def remove(id_):
-                del futures[id_]
-                active_queries.remove(id_)
-                self._poll_queue.task_done()
+                return True
 
             while True:
-                try:
-                    # If we have no active queries, wait on the status queue
-                    while len(active_queries) == 0:
-                        if not add(self._poll_queue.get()):
-                            return
-                    # Once there is any active queries try to fill up the set and move on
-                    while len(active_queries) < self._STATUS_QUERY_SIZE:
-                        if not add(self._poll_queue.get_nowait()):
+                frame_futures.clear()
+
+                # blocking add first scheduled
+                frame_earliest, future = self._poll_queue.get()
+                if not add(future):
+                    return
+
+                # try grouping if scheduled within grouping timeframe
+                while len(frame_futures) < self._STATUS_QUERY_SIZE:
+                    try:
+                        task = self._poll_queue.get_nowait()
+                    except queue.Empty:
+                        break
+
+                    at, future = task
+                    if at - frame_earliest <= self._POLL_GROUP_TIMEFRAME:
+                        if not add(future):
                             return
-                except queue.Empty:
-                    pass
+                    else:
+                        task_done()
+                        self._poll_queue.put(task)
+                        break
+
+                # build a query string with ids of all futures in this frame
+                ids = [future.id for future in frame_futures.values()]
+                _LOGGER.debug("Polling for status of futures: %s", ids)
+                query_string = 'problems/?id=' + ','.join(ids)
 
-                # Build a query string with block of ids
-                _LOGGER.debug("Query on futures: %s", ', '.join(active_queries))
-                query_string = 'problems/?id=' + ','.join(active_queries)
+                # if futures were cancelled while `add`ing, skip empty frame
+                if not ids:
+                    continue
+
+                # wait until `frame_earliest` before polling
+                delay = frame_earliest - time.time()
+                if delay > 0:
+                    _LOGGER.debug("Pausing polling %.2f sec for futures: %s", delay, ids)
+                    time.sleep(delay)
 
                 try:
                     response = self.session.get(posixpath.join(self.endpoint, query_string))
@@ -850,28 +890,20 @@ class Client(object):
                         raise SolverAuthenticationError()
                     response.raise_for_status()
 
-                    message = response.json()
+                    statuses = response.json()
+                    for status in statuses:
+                        self._handle_problem_status(status, frame_futures[status['id']])
+
                 except BaseException as exception:
                     if not isinstance(exception, SolverAuthenticationError):
                         exception = IOError(exception)
 
-                    for id_ in list(active_queries):
-                        futures[id_]._set_error(IOError(exception), sys.exc_info())
-                        remove(id_)
-                    continue
-
-                # If problems are removed from the polling by _handle_problem_status
-                # remove them from the active set
-                for single_message in message:
-                    if self._handle_problem_status(single_message, futures[single_message['id']], True):
-                        remove(single_message['id'])
+                    for id_ in frame_futures.keys():
+                        frame_futures[id_]._set_error(IOError(exception), sys.exc_info())
 
-                # Remove the finished queries
-                for id_ in list(active_queries):
-                    if futures[id_].done():
-                        remove(id_)
+                for id_ in frame_futures.keys():
+                    task_done()
 
-                # this is equivalent to a yield to scheduler in other threading libraries
                 time.sleep(0)
 
         except Exception as err:
@@ -902,7 +934,7 @@ class Client(object):
                 # `None` task signifies thread termination
                 if future is None:
                     break
-                _LOGGER.debug("Query for results: %s", future.id)
+                _LOGGER.debug("Loading results of: %s", future.id)
 
                 # Submit the query
                 query_string = 'problems/{}/'.format(future.id)
@@ -922,7 +954,7 @@ class Client(object):
                     continue
 
                 # Dispatch the results, mark the task complete
-                self._handle_problem_status(message, future, False)
+                self._handle_problem_status(message, future)
                 self._load_queue.task_done()
 
                 # this is equivalent to a yield to scheduler in other threading libraries
diff --git a/dwave/cloud/computation.py b/dwave/cloud/computation.py
index 1fc6bbd..5854b1c 100644
--- a/dwave/cloud/computation.py
+++ b/dwave/cloud/computation.py
@@ -64,6 +64,12 @@ class Future(object):
         #: `datetime` corresponding to the time when the problem was completed by the server (None before then)
         self.time_solved = None
 
+        # estimated `earliest_completion_time` as returned on problem submit
+        self.eta_min = None
+
+        # estimated `latest_completion_time` as returned on problem submit
+        self.eta_max = None
+
         # Track how long it took us to parse the data
         self.parse_time = None
 
@@ -81,6 +87,9 @@ class Future(object):
         self._results_ready_event = threading.Event()
         self._other_events = []
 
+        # current poll back-off interval, in seconds
+        self._poll_backoff = 0
+
     def _set_message(self, message):
         """Complete the future with a message from the server.
 
diff --git a/perf/multiple.py b/perf/multiple.py
new file mode 100644
index 0000000..697aece
--- /dev/null
+++ b/perf/multiple.py
@@ -0,0 +1,38 @@
+from __future__ import absolute_import
+
+import logging
+
+from dwave.cloud.qpu import Client
+from dwave.cloud.computation import Future
+
+
+# increase logging verbosity for root logger
+logging.getLogger('dwave.cloud').setLevel(logging.DEBUG)
+
+# setup local logger
+formatter = logging.Formatter('%(asctime)s %(name)s %(levelname)s %(message)s')
+handler = logging.StreamHandler()
+handler.setFormatter(formatter)
+logger = logging.getLogger(__name__)
+logger.setLevel(logging.DEBUG)
+logger.addHandler(handler)
+
+
+with Client.from_config(profile='prod') as client:
+    solvers = client.get_solvers()
+    logger.info("Solvers available: %r", solvers.keys())
+
+    solver = client.get_solver()
+    comps = [solver.sample_qubo({}, num_reads=1) for _ in range(20)]
+
+    for comp in Future.as_completed(comps):
+        try:
+            result = comp.result()
+        except Exception as e:
+            logger.info("Computation %s failed: %r", comp.id, e)
+
+        logger.info("Computation %s succeeded:", comp.id)
+        logger.info(" - time received: %s", comp.time_received)
+        logger.info(" - time solved: %s", comp.time_solved)
+        logger.info(" - parse time: %s", comp.parse_time)
+        logger.info(" - remote status: %s", comp.remote_status)
diff --git a/perf/simple.py b/perf/simple.py
index a6042b5..a9dd7f5 100644
--- a/perf/simple.py
+++ b/perf/simple.py
@@ -1,15 +1,8 @@
 from __future__ import absolute_import
 
-import sys
 import logging
-import unittest
 
-from dwave.cloud.config import load_config
 from dwave.cloud.qpu import Client
-from dwave.cloud.exceptions import SolverAuthenticationError
-from dwave.cloud.testing import mock
-
-from tests import config
 
 
 # increase logging verbosity for root logger


Test patch:
diff --git a/tests/test_mock_submission.py b/tests/test_mock_submission.py
index 4a990de..dbb7785 100644
--- a/tests/test_mock_submission.py
+++ b/tests/test_mock_submission.py
@@ -6,6 +6,8 @@ import unittest
 import itertools
 import threading
 
+from dateutil.parser import parse as parse_timestamp
+
 from dwave.cloud.utils import evaluate_ising
 from dwave.cloud.qpu import Client, Solver
 from dwave.cloud.exceptions import SolverFailureError, CanceledFutureError
@@ -90,6 +92,9 @@ def cancel_reply(id_, solver_name):
     })
 
 
+continue_reply_eta_min = parse_timestamp("2013-01-18T10:25:59.941674Z")
+continue_reply_eta_max = parse_timestamp("2013-01-18T10:27:51.123456Z")
+
 def continue_reply(id_, solver_name):
     """A reply saying a problem is still in the queue."""
     return json.dumps({
@@ -97,6 +102,8 @@ def continue_reply(id_, solver_name):
         "solved_on": None,
         "solver": solver_name,
         "submitted_on": "2013-01-18T10:06:10.025064",
+        "earliest_completion_time": continue_reply_eta_min.isoformat(),
+        "latest_completion_time": continue_reply_eta_max.isoformat(),
         "type": "ising",
         "id": id_
     })
@@ -123,6 +130,7 @@ class _QueryTest(unittest.TestCase):
             self.assertTrue(energy == evaluate_ising(linear, quad, state))
 
 
+@mock.patch('time.sleep', lambda *x: None)
 class MockSubmission(_QueryTest):
     """Test connecting and some related failure modes."""
 
@@ -207,6 +215,10 @@ class MockSubmission(_QueryTest):
 
             self._check(results, linear, quad, 100)
 
+            # test future has eta_min and eta_max parsed correctly
+            self.assertEqual(results.eta_min, continue_reply_eta_min)
+            self.assertEqual(results.eta_max, continue_reply_eta_max)
+
     def test_submit_continue_then_error_reply(self):
         """Handle polling for an error message."""
         with Client('endpoint', 'token') as client:
@@ -272,6 +284,46 @@ class MockSubmission(_QueryTest):
 
         Client._POLL_THREAD_COUNT = old_value
 
+    def test_exponential_backoff_polling(self):
+        "After each poll, back-off should double"
+
+        # Reduce the number of poll threads to 1 so that the system can be tested
+        old_value = Client._POLL_THREAD_COUNT
+        Client._POLL_THREAD_COUNT = 1
+
+        with Client('endpoint', 'token') as client:
+            client.session = mock.Mock()
+            # on submit, return status pending
+            client.session.post = lambda a, _: choose_reply(a, {
+                'endpoint/problems/': '[%s]' % continue_reply('123', 'abc123')
+            })
+            # on first and second status poll, return pending
+            # on third status poll, return completed
+            def continue_then_complete(path, state={'count': 0}):
+                state['count'] += 1
+                if state['count'] < 3:
+                    return choose_reply(path, {
+                        'endpoint/problems/?id=123': '[%s]' % continue_reply('123', 'abc123'),
+                        'endpoint/problems/123/': continue_reply('123', 'abc123')
+                    })
+                else:
+                    return choose_reply(path, {
+                        'endpoint/problems/?id=123': '[%s]' % complete_no_answer_reply('123', 'abc123'),
+                        'endpoint/problems/123/': complete_reply('123', 'abc123')
+                    })
+
+            client.session.get = continue_then_complete
+
+            solver = Solver(client, solver_data('abc123'))
+
+            future = solver.sample_qubo({})
+            future.result()
+
+            # after third poll, back-off interval should be 4
+            self.assertEqual(future._poll_backoff, 4)
+
+        Client._POLL_THREAD_COUNT = old_value
+
 
 class DeleteEvent(Exception):
     """Throws exception when mocked client submits an HTTP DELETE request."""