code: def benchmark(fn, args, filetype=None):
      """Benchmark when "fn" function gets called on "args" tuple.
      "args" may have a Kwargs instance at the end.
      If "filetype" is provided, it may be used to convert columns to
      categorical dtypes after reading (the "loading" is assumed).
      """
      posargs = list(args)
      kwargs = {}
      # Remove Kwargs instance at end of posargs list, if one exists
      if posargs and isinstance(posargs[-1], Kwargs):
          lastarg = posargs.pop()
          kwargs.update(lastarg)
  
      if DEBUG:
          printable_posargs = ", ".join(
              [
                  str(posarg.head())
                  if hasattr(posarg, "head")
                  else str(posarg)
                  for posarg in posargs
              ]
          )
          printable_kwargs = ", ".join(
              ["{}={}".format(k, v) for k, v in kwargs.items()]
          )
          print(
              "DEBUG: {}({}{})".format(
                  fn.__name__,
                  printable_posargs,
                  ", " + printable_kwargs if printable_kwargs else "",
              ),
              flush=True,
          )
  
      # Benchmark fn when run on posargs and kwargs
      start = time.time()
      res = fn(*posargs, **kwargs)
  
      # If we're loading data
      if filetype is not None:
          if filetype not in filetypes_storing_categories:
              opts = {}
              if p.dftype == "pandas":
                  opts["copy"] = False
              for c in p.categories:
                  res[c] = res[c].astype("category", **opts)
  
          # Force loading (--cache=persist was provided)
          if p.dftype == "dask" and DD_FORCE_LOAD:
              if DASK_CLIENT is not None:
                  # 2017-04-28: This combination leads to a large drop in
                  #   aggregation performance (both --distributed and
                  #   --cache=persist were provided)
                  res = DASK_CLIENT.persist(res)
                  distributed.wait(res)
              else:
                  if DEBUG:
                      print(
                          "DEBUG: Force-loading Dask dataframe",
                          flush=True,
                      )
                  res = res.persist()
  
      end = time.time()
  
      return end - start, res
  
imports: from __future__ import annotations
  import time
  import os, os.path, sys, glob, argparse, resource, multiprocessing
  import pandas
  import dask.dataframe
  import numpy
  import datashader
  import feather
  import fastparquet
  from datashader.utils import export_image
  from datashader import transfer_functions
  from dask import distributed
label_desc: Use image node_id 1 for calling a global function with example usage: benchmark(fn, args, filetype) and returns: unknown,  res