| 214 | logger.debug('%s done waiting for dependent futures', self) |
| 215 | |
| 216 | def _get_all_main_kwargs(self): |
| 217 | # Copy over all of the kwargs that we know is available. |
| 218 | kwargs = copy.copy(self._main_kwargs) |
| 219 | |
| 220 | # Iterate through the kwargs whose values are pending on the result |
| 221 | # of a future. |
| 222 | for key, pending_value in self._pending_main_kwargs.items(): |
| 223 | # If the value is a list of futures, iterate though the list |
| 224 | # appending on the result from each future. |
| 225 | if isinstance(pending_value, list): |
| 226 | result = [] |
| 227 | for future in pending_value: |
| 228 | result.append(future.result()) |
| 229 | # Otherwise if the pending_value is a future, just wait for it. |
| 230 | else: |
| 231 | result = pending_value.result() |
| 232 | # Add the retrieved value to the kwargs to be sent to the |
| 233 | # main() call. |
| 234 | kwargs[key] = result |
| 235 | return kwargs |
| 236 | |
| 237 | |
| 238 | class SubmissionTask(Task): |