Pull problems from the submission queue and submit them. Note: This method is always run inside of a daemon thread.
(self)
| 1301 | _submit.Message = namedtuple('Message', ['body', 'future']) |
| 1302 | |
| 1303 | def _do_submit_problems(self): |
| 1304 | """Pull problems from the submission queue and submit them. |
| 1305 | |
| 1306 | Note: |
| 1307 | This method is always run inside of a daemon thread. |
| 1308 | """ |
| 1309 | |
| 1310 | def task_done(): |
| 1311 | self._submission_queue.task_done() |
| 1312 | |
| 1313 | def filter_ready(item): |
| 1314 | """Pass-through ready (encoded) problems, re-enqueue ones for which |
| 1315 | the encoding is in progress, and fail the ones for which encoding |
| 1316 | failed. |
| 1317 | """ |
| 1318 | |
| 1319 | # body is a `concurrent.futures.Future`, so make sure |
| 1320 | # it's ready for submitting |
| 1321 | if item.body.done(): |
| 1322 | exc = item.body.exception() |
| 1323 | if exc: |
| 1324 | # encoding failed, submit should fail as well |
| 1325 | logger.info("Problem encoding prior to submit " |
| 1326 | "failed with: %r", exc) |
| 1327 | item.future._set_exception(exc) |
| 1328 | self._jobs.dec() |
| 1329 | task_done() |
| 1330 | |
| 1331 | else: |
| 1332 | # problem ready for submit |
| 1333 | return [item] |
| 1334 | |
| 1335 | else: |
| 1336 | # body not ready, return the item to queue |
| 1337 | self._submission_queue.put(item) |
| 1338 | task_done() |
| 1339 | |
| 1340 | return [] |
| 1341 | |
| 1342 | session = self.create_session() |
| 1343 | session.set_accept(media_type='application/vnd.dwave.sapi.problems+json', |
| 1344 | accept_version='~=3.0', ask_version='3.0.0') |
| 1345 | try: |
| 1346 | while True: |
| 1347 | # Pull as many problems as we can, block on the first one, |
| 1348 | # but once we have one problem, switch to non-blocking then |
| 1349 | # submit without blocking again. |
| 1350 | |
| 1351 | # `None` task is used to signal thread termination |
| 1352 | item = self._submission_queue.get() |
| 1353 | |
| 1354 | if item is None: |
| 1355 | task_done() |
| 1356 | break |
| 1357 | |
| 1358 | ready_problems = filter_ready(item) |
| 1359 | while len(ready_problems) < self._SUBMIT_BATCH_SIZE: |
| 1360 | try: |
nothing calls this directly
no test coverage detected