def METHOD_NAME(
    self,
    tasks: Iterable[TaskProtocol],
    params_handle: Any,
    cancel_id: Any,
    task_comm_handler: TaskCommHandler,
):
    tasks = list(tasks)
    def _id_to_task(task_id):
        return tasks[task_id]
    self._futures[cancel_id] = []
    msg_queue = SimpleWorkerQueue()
    for idx, wrapped_task in list(enumerate(tasks)):
        future = self._get_future(wrapped_task, idx, params_handle, msg_queue)
        self._futures[cancel_id].append(future)
    with task_comm_handler.monitor(msg_queue):
        try:
            as_completed = concurrent.futures.as_completed(self._futures[cancel_id])
            for future in as_completed:
                result_wrap = future.result()
                if future.cancelled():
                    del self._futures[cancel_id]
                    raise JobCancelledError()
                result = result_wrap['task_result']
                task = _id_to_task(result_wrap['task_id'])
                yield result, task
        finally:
            if cancel_id in self._futures:
                del self._futures[cancel_id]