class_docstr: simple cache for WarmProcess-es
      
class_docstr_tok: ["simple","cache","for","WarmProcess-es"]
docstr: run new task. If possible - do it in existing process, if not - start new one.
  
  Args: TODO rewrite!
      handler (object): handler class
      func (Callable): function to run
      model_marker (tuple): if any of processes processed task with same marker - new task will be sent to it
      args (tuple): args to be passed to function
      kwargs (dict): kwargs to be passed to function
  
  Returns:
      Future
docstr_tok: ["run","new","task",".","If","possible","-","do","it","in","existing","process",",","if","not","-","start","new","one",".","Args",":","TODO","rewrite","!","handler","(","object",")",":","handler","class","func","(","Callable",")",":","function","to","run","model_marker","(","tuple",")",":","if","any","of","processes","processed","task","with","same","marker","-","new","task","will","be","sent","to","it","args","(","tuple",")",":","args","to","be","passed","to","function","kwargs","(","dict",")",":","kwargs","to","be","passed","to","function","Returns",":","Future"]
code: def apply_async(
      self,
      task_type: ML_TASK_TYPE,
      model_id: int,
      payload: dict,
      dataframe: DataFrame = None,
  ) -> Future:
      """run new task. If possible - do it in existing process, if not - start new one.
  
      Args: TODO rewrite!
          handler (object): handler class
          func (Callable): function to run
          model_marker (tuple): if any of processes processed task with same marker - new task will be sent to it
          args (tuple): args to be passed to function
          kwargs (dict): kwargs to be passed to function
  
      Returns:
          Future
      """
      if task_type in (ML_TASK_TYPE.LEARN, ML_TASK_TYPE.FINETUNE):
          func = learn_process
      elif task_type == ML_TASK_TYPE.PREDICT:
          func = predict_process
      else:
          raise Exception(f"Unknown ML task type: {task_type}")
  
      handler_module_path = payload["handler_meta"]["module_path"]
      handler_name = payload["handler_meta"]["engine"]
      model_marker = (model_id, payload["context"]["company_id"])
      with self._lock:
          if handler_name not in self.cache:
              warm_process = WarmProcess(
                  init_ml_handler, (handler_module_path,)
              )
              self.cache[handler_name] = {
                  "last_usage_at": None,
                  "handler_module": handler_module_path,
                  "processes": [warm_process],
              }
          else:
              warm_process = None
              if model_marker is not None:
                  try:
                      warm_process = next(
                          p
                          for p in self.cache[handler_name]["processes"]
                          if p.ready() and p.has_marker(model_marker)
                      )
                  except StopIteration:
                      pass
              if warm_process is None:
                  try:
                      warm_process = next(
                          p
                          for p in self.cache[handler_name]["processes"]
                          if p.ready()
                      )
                  except StopIteration:
                      pass
              if warm_process is None:
                  warm_process = WarmProcess(
                      init_ml_handler, (handler_module_path,)
                  )
                  self.cache[handler_name]["processes"].append(
                      warm_process
                  )
  
          task = warm_process.apply_async(
              warm_function,
              func,
              payload["context"],
              payload,
              dataframe,
          )
          self.cache[handler_name]["last_usage_at"] = time.time()
          warm_process.add_marker(model_marker)
      return task
  
imports: import sys
  import time
  import threading
  from typing import Optional, Callable
  from concurrent.futures import ProcessPoolExecutor, Future
  from pandas import DataFrame
  from mindsdb.utilities.config import Config
  from mindsdb.utilities.context import context
  from mindsdb.utilities.ml_task_queue.const import ML_TASK_TYPE
  from mindsdb.integrations.libs.learn_process import learn_process, predict_process
label_desc: Use image node_id 6 for calling the ProcessCache obj's underlying member method code with example usage: obj.apply_async(task_type, model_id, payload, dataframe) and returns: task