code: def dumps(x, *, buffer_callback=None, protocol=HIGHEST_PROTOCOL):
      """Manage between cloudpickle and pickle
  
      1.  Try pickle
      2.  If it is short then check if it contains __main__
      3.  If it is long, then first check type, then check __main__
      """
      buffers = []
      dump_kwargs = {"protocol": protocol or HIGHEST_PROTOCOL}
      if dump_kwargs["protocol"] >= 5 and buffer_callback is not None:
          dump_kwargs["buffer_callback"] = buffers.append
      try:
          try:
              result = pickle.dumps(x, **dump_kwargs)
          except Exception:
              f = io.BytesIO()
              pickler = _DaskPickler(f, **dump_kwargs)
              buffers.clear()
              pickler.dump(x)
              result = f.getvalue()
          if b"__main__" in result or (
              CLOUDPICKLE_GE_20
              and getattr(inspect.getmodule(x), "__name__", None)
              in cloudpickle.list_registry_pickle_by_value()
          ):
              if len(result) < 1000 or not _always_use_pickle_for(x):
                  buffers.clear()
                  result = cloudpickle.dumps(x, **dump_kwargs)
      except Exception:
          try:
              buffers.clear()
              result = cloudpickle.dumps(x, **dump_kwargs)
          except Exception:
              logger.exception("Failed to serialize %s.", x)
              raise
      if buffer_callback is not None:
          for b in buffers:
              buffer_callback(b)
      return result
  
file_path: distributed/distributed/protocol/pickle.py
imports: from __future__ import annotations
  import inspect
  import io
  import logging
  import pickle
  import cloudpickle
  from packaging.version import parse
  from distributed.protocol.serialize import dask_deserialize, dask_serialize
label_desc: Use image node_id 2 for calling a global function with example usage: dumps(x) and returns: result