code: def __init__(
      self,
      queueing_ref,
      global_resource_ref,
      speculation_config: Dict[str, any],
  ):
      self._grouped_unfinished_subtasks = defaultdict(dict)
      self._grouped_finished_subtasks = defaultdict(dict)
      self._queueing_ref = queueing_ref
      self._global_resource_ref = global_resource_ref
      self._speculation_config = speculation_config
      self._subtask_speculation_enabled = speculation_config.get(
          "enabled", False
      )
      assert self._subtask_speculation_enabled in (True, False)
      self._subtask_speculation_dry = speculation_config.get(
          "dry", False
      )
      self._subtask_speculation_threshold = parse_readable_size(
          speculation_config.get(
              "threshold", DEFAULT_SUBTASK_SPECULATION_THRESHOLD
          )
      )[0]
      self._subtask_speculation_interval = speculation_config.get(
          "interval", DEFAULT_SUBTASK_SPECULATION_INTERVAL
      )
      self._subtask_speculation_min_task_runtime = (
          speculation_config.get(
              "min_task_runtime",
              DEFAULT_SUBTASK_SPECULATION_MIN_TASK_RUNTIME,
          )
      )
      self._subtask_speculation_multiplier = speculation_config.get(
          "multiplier", DEFAULT_SUBTASK_SPECULATION_MULTIPLIER
      )
      self._subtask_speculation_max_concurrent_run = (
          speculation_config.get(
              "max_concurrent_run", DEFAULT_SUBTASK_MAX_CONCURRENT_RUN
          )
      )
      if self._subtask_speculation_enabled:
          assert 1 >= self._subtask_speculation_threshold > 0
          assert self._subtask_speculation_interval > 0
          assert self._subtask_speculation_min_task_runtime > 0
          assert self._subtask_speculation_multiplier > 0
          assert self._subtask_speculation_max_concurrent_run > 0
      self._speculation_execution_task = None
  
file_path: mars/mars/services/scheduling/supervisor/speculation.py
imports: import asyncio
  import logging
  import time
  import numpy
  from collections import defaultdict
  from typing import Dict
  from ....utils import parse_readable_size, create_task_with_error_log
  from ..errors import NoAvailableBand
  from .manager import SubtaskScheduleInfo
label_desc: Use image node_id 1 to create a new SpeculativeScheduler object  with example: obj = SpeculativeScheduler(queueing_ref, global_resource_ref, speculation_config)