code: def _estimate_sizes(cls, subtask: Subtask, input_sizes: Dict):
      size_context = dict(input_sizes.items())
      graph = subtask.chunk_graph
  
      key_to_ops = defaultdict(set)
      chunk_key_to_sizes = defaultdict(lambda: 0)
      for n in graph:
          key_to_ops[n.op.key].add(n.op)
          chunk_key_to_sizes[n.key] += 1
      key_to_ops = {k: list(v) for k, v in key_to_ops.items()}
  
      # condense op key graph
      op_key_graph = DAG()
      for n in graph.topological_iter():
          if n.key in subtask.pure_depend_keys:
              continue
          if n.op.key not in op_key_graph:
              op_key_graph.add_node(n.op.key)
          for succ in graph.iter_successors(n):
              if succ.op.key not in op_key_graph:
                  op_key_graph.add_node(succ.op.key)
              op_key_graph.add_edge(n.op.key, succ.op.key)
  
      key_stack = list(op_key_graph.iter_indep())
      pred_ref_count = {
          k: op_key_graph.count_predecessors(k) for k in op_key_graph
      }
      succ_ref_count = {
          k: op_key_graph.count_successors(k) for k in op_key_graph
      }
  
      visited_op_keys = set()
      total_memory_cost = 0
      max_memory_cost = sum(
          calc_size for _, calc_size in size_context.values()
      )
      while key_stack:
          key = key_stack.pop()
          op = key_to_ops[key][0]
  
          if not isinstance(op, Fetch):
              op.estimate_size(size_context, op)
  
          calc_cost = sum(
              size_context[out.key][1] for out in op.outputs
          )
          total_memory_cost += calc_cost
          max_memory_cost = max(total_memory_cost, max_memory_cost)
  
          if not isinstance(op, Fetch):
              # when calculation result is stored, memory cost of calculation
              #  can be replaced with result memory cost
              result_cost = sum(
                  size_context[out.key][0] for out in op.outputs
              )
              total_memory_cost += result_cost - calc_cost
  
          visited_op_keys.add(key)
  
          for succ_op_key in op_key_graph.iter_successors(key):
              pred_ref_count[succ_op_key] -= 1
              if pred_ref_count[succ_op_key] == 0:
                  key_stack.append(succ_op_key)
  
          for pred_op_key in op_key_graph.iter_predecessors(key):
              succ_ref_count[pred_op_key] -= 1
              if succ_ref_count[pred_op_key] == 0:
                  pred_op = key_to_ops[pred_op_key][0]
                  outs = key_to_ops[pred_op_key][0].outputs
                  for out in outs:
                      chunk_key_to_sizes[out.key] -= 1
                  # when clearing fetches, subtract memory size, otherwise subtract store size
                  account_idx = 1 if isinstance(pred_op, Fetch) else 0
                  pop_result_cost = 0
                  for out in outs:
                      # corner case exist when a fetch op and another op has same chunk key
                      # but their op keys are different
                      if chunk_key_to_sizes[out.key] == 0:
                          pop_result_cost += size_context.pop(
                              out.key, (0, 0)
                          )[account_idx]
                      else:
                          pop_result_cost += size_context.get(
                              out.key, (0, 0)
                          )[account_idx]
                  total_memory_cost -= pop_result_cost
      return sum(t[0] for t in size_context.values()), max_memory_cost
  
imports: import asyncio
  import functools
  import logging
  import operator
  import pprint
  import sys
  from collections import defaultdict
  from dataclasses import dataclass, field
  from typing import Dict, Optional
  from ....None import oscar
  from ....core import ExecutionError
  from ....core.graph import DAG
  from ....core.operand import Fetch, FetchShuffle
  from ....lib.aio import alru_cache
  from ....metrics import Metrics
  from ....oscar.errors import MarsError
  from ....storage import StorageLevel
  from ....utils import dataslots, get_chunk_key_to_data_keys, wrap_exception
  from ...cluster import ClusterAPI
  from ...meta import MetaAPI
  from ...storage import StorageAPI
  from ...subtask import Subtask, SubtaskAPI, SubtaskResult, SubtaskStatus
  from .workerslot import BandSlotManagerActor
  from .quota import QuotaActor
label_desc: Use image node_id 2 for calling the SubtaskExecutionActor obj's underlying member method code with example usage: obj._estimate_sizes(cls, subtask, input_sizes) and returns: sum,  max_memory_cost