code: def retrieve_results_concurrently(
      client: cx.Client,
      api_name: str,
      concurrency: int,
      event_stopper: td.Event,
      job_ids: List[str],
      responses: List[Tuple[str, Dict[str, Any]]] = [],
      poll_sleep_seconds: int = 1,
      timeout: Optional[int] = None,
  ):
      api_info = client.get_api(api_name)
      task_kind = api_info["spec"]["kind"] == "TaskAPI"
      async_kind = api_info["spec"]["kind"] == "AsyncAPI"
      if not task_kind and not async_kind:
          raise ValueError(
              "function can only be called for TaskAPI/AsyncAPI kinds"
          )
  
      exec = futures.ThreadPoolExecutor(concurrency)
      thread_local = td.local()
  
      def _get_session() -> requests.Session:
          if not hasattr(thread_local, "session"):
              thread_local.session = requests.Session()
          return thread_local.session
  
      def _retriever(request_id: str):
          session = _get_session()
  
          while not event_stopper.is_set():
              if task_kind:
                  result_response = session.get(
                      f"{api_info['endpoint']}?jobID={request_id}"
                  )
              if async_kind:
                  result_response = session.get(
                      f"{api_info['endpoint']}/{request_id}"
                  )
  
              if result_response.status_code != HTTPStatus.OK:
                  content = result_response.content.decode("utf-8")
                  if "error" in content:
                      event_stopper.set()
                      raise RuntimeError(
                          f"received {result_response.status_code} status code with the following message: {content}"
                      )
                  time.sleep(poll_sleep_seconds)
                  continue
  
              result_response_json = result_response.json()
              if async_kind and "status" in result_response_json:
                  if result_response_json["status"] == "completed":
                      break
                  if result_response_json["status"] not in [
                      "in_progress",
                      "in_queue",
                  ]:
                      raise RuntimeError(
                          f"status for request ID {request_id} got set to {result_response_json['status']}"
                      )
  
              if (
                  task_kind
                  and "job_status" in result_response_json
                  and "status" in result_response_json["job_status"]
              ):
                  if (
                      result_response_json["job_status"]["status"]
                      == "succeeded"
                  ):
                      break
                  if result_response_json["job_status"][
                      "status"
                  ] not in [
                      "pending",
                      "enqueuing",
                      "running",
                  ]:
                      raise RuntimeError(
                          f"status for job ID {request_id} got set to {result_response_json['job_status']['status']}"
                      )
  
          if event_stopper.is_set():
              return
  
          responses.append((request_id, result_response_json))
  
      # will throw an exception if something failed in any thread
      for _ in exec.map(_retriever, job_ids, timeout=timeout):
          pass
  
imports: import os
  import threading
  import time
  from concurrent import futures
  from http import HTTPStatus
  from typing import Any, List, Optional, Tuple, Union, Dict, Callable
  import cortex
  import requests
  import yaml
label_desc: Use image node_id 15 for calling a global function with example usage: retrieve_results_concurrently(client, api_name, concurrency, event_stopper, job_ids, responses, poll_sleep_seconds, timeout) and returns: thread_local