code: def gen_submit_tileable_graph(
      session: AbstractSession,
      result_tileables: List[TileableType],
      warn_duplicated_execution: bool = False,
  ) -> Tuple[TileableGraph, List[TileableType]]:
      tileable_to_copied = dict()
      indexer = itertools.count()
      result_to_index = {
          t: i for t, i in zip(result_tileables, indexer)
      }
      result = list()
      to_execute_tileables = list()
      graph = TileableGraph(result)
  
      q = list(result_tileables)
      while q:
          tileable = q.pop()
          if tileable in tileable_to_copied:
              continue
          if tileable.cache and tileable not in result_to_index:
              result_to_index[tileable] = next(indexer)
          outputs = tileable.op.outputs
          inputs = (
              tileable.inputs
              if session not in tileable._executed_sessions
              else []
          )
          new_inputs = []
          all_inputs_processed = True
          for inp in inputs:
              if inp in tileable_to_copied:
                  new_inputs.append(tileable_to_copied[inp])
              elif session in inp._executed_sessions:
                  # executed, gen fetch
                  fetch_input = build_fetch(inp).data
                  tileable_to_copied[inp] = fetch_input
                  graph.add_node(fetch_input)
                  new_inputs.append(fetch_input)
              else:
                  # some input not processed before
                  all_inputs_processed = False
                  # put back tileable
                  q.append(tileable)
                  q.append(inp)
                  break
          if all_inputs_processed:
              if isinstance(tileable.op, Fetch):
                  new_outputs = [tileable]
              elif session in tileable._executed_sessions:
                  new_outputs = []
                  for out in outputs:
                      fetch_out = tileable_to_copied.get(
                          out, build_fetch(out).data
                      )
                      new_outputs.append(fetch_out)
              else:
                  new_outputs = [
                      t.data
                      for t in copy_tileables(
                          outputs, inputs=new_inputs
                      )
                  ]
              for out, new_out in zip(outputs, new_outputs):
                  tileable_to_copied[out] = new_out
                  graph.add_node(new_out)
                  for new_inp in new_inputs:
                      graph.add_edge(new_inp, new_out)
  
      # process results
      result.extend([None] * len(result_to_index))
      for t, i in result_to_index.items():
          result[i] = tileable_to_copied[t]
          to_execute_tileables.append(t)
  
      if warn_duplicated_execution:
          for n, c in tileable_to_copied.items():
              if (
                  not isinstance(c.op, Fetch)
                  and n in _submitted_tileables
              ):
                  warnings.warn(
                      f"Tileable {repr(n)} has been submitted before",
                      RuntimeWarning,
                  )
          # add all nodes into submitted tileables
          _submitted_tileables.update(
              n
              for n, c in tileable_to_copied.items()
              if not isinstance(c.op, Fetch)
          )
  
      return graph, to_execute_tileables
  
imports: import asyncio
  import itertools
  import logging
  import json
  import time
  import warnings
  from collections import defaultdict
  from dataclasses import dataclass
  from numbers import Integral
  from urllib.parse import urlparse
  from weakref import WeakKeyDictionary, WeakSet
  from typing import Any, Callable, Dict, List, Optional, Tuple, Union
  import numpy
  from ...None import oscar
  from ...utils import Timer, build_fetch, copy_tileables
  from ...core import ChunkType, TileableType, enter_mode, TileableGraph
  from ...core.operand import Fetch
  from ...lib.aio import alru_cache
  from ...metrics import Metrics
  from ...services.cluster import AbstractClusterAPI, ClusterAPI
  from ...services.lifecycle import AbstractLifecycleAPI, LifecycleAPI
  from ...services.meta import MetaAPI, AbstractMetaAPI
  from ...services.session import AbstractSessionAPI, SessionAPI
  from ...services.mutable import MutableAPI
  from ...services.storage import StorageAPI
  from ...services.task import AbstractTaskAPI, TaskAPI, TaskResult
  from ...services.task.execution.api import Fetcher
  from ...services.web import OscarWebAPI
  from ...session import AbstractAsyncSession, AbstractSession, IsolatedAsyncSession, ExecutionInfo, Progress, Profiling
  from ...typing import ClientType, BandType
  from ...utils import implements, merge_chunks, merged_chunk_as_tileable_type, register_asyncio_task_timeout_detector
label_desc: Use image node_id 1 for calling a global function with example usage: gen_submit_tileable_graph(session, result_tileables, warn_duplicated_execution) and returns: graph,  to_execute_tileables