code: def serialize_numpy_ndarray(x, context=None):
      if x.dtype.hasobject or (
          x.dtype.flags & np_core.multiarray.LIST_PICKLE
      ):
          header = {"pickle": True}
          frames = [None]
  
          def buffer_callback(f):
              frames.append(memoryview(f))
  
          frames[0] = pickle.dumps(
              x,
              buffer_callback=buffer_callback,
              protocol=(context or {}).get("pickle-protocol", None),
          )
          return header, frames
  
      # We cannot blindly pickle the dtype as some may fail pickling,
      # so we have a mixture of strategies.
      if x.dtype.kind == "V":
          # Preserving all the information works best when pickling
          try:
              # Only use stdlib pickle as cloudpickle is slow when failing
              # (microseconds instead of nanoseconds)
              dt = (
                  1,
                  pickle.pickle.dumps(
                      x.dtype,
                      protocol=(context or {}).get(
                          "pickle-protocol", None
                      ),
                  ),
              )
              pickle.loads(dt[1])  # does it unpickle fine?
          except Exception:
              # dtype fails pickling => fall back on the descr if reasonable.
              if x.dtype.type is not np.void or x.dtype.alignment != 1:
                  raise
              else:
                  dt = (0, x.dtype.descr)
      else:
          dt = (0, x.dtype.str)
  
      # Only serialize broadcastable data for arrays with zero strided axes
      broadcast_to = None
      if 0 in x.strides:
          broadcast_to = x.shape
          strides = x.strides
          writeable = x.flags.writeable
          x = x[
              tuple(
                  slice(None) if s != 0 else slice(1) for s in strides
              )
          ]
          if not x.flags.c_contiguous and not x.flags.f_contiguous:
              # Broadcasting can only be done with contiguous arrays
              x = np.ascontiguousarray(x)
              x = np.lib.stride_tricks.as_strided(
                  x,
                  strides=[
                      j if i != 0 else i
                      for i, j in zip(strides, x.strides)
                  ],
                  writeable=writeable,
              )
  
      if not x.shape:
          # 0d array
          strides = x.strides
          data = x.ravel()
      elif x.flags.c_contiguous or x.flags.f_contiguous:
          # Avoid a copy and respect order when unserializing
          strides = x.strides
          data = x.ravel(order="K")
      else:
          x = np.ascontiguousarray(x)
          strides = x.strides
          data = x.ravel()
  
      if data.dtype.fields or data.dtype.itemsize > 8:
          data = data.view("u%d" % math.gcd(x.dtype.itemsize, 8))
  
      try:
          data = data.data
      except ValueError:
          # "ValueError: cannot include dtype 'M' in a buffer"
          data = data.view("u%d" % math.gcd(x.dtype.itemsize, 8)).data
  
      header = {
          "dtype": dt,
          "shape": x.shape,
          "strides": strides,
          "writeable": [x.flags.writeable],
      }
  
      if broadcast_to is not None:
          header["broadcast_to"] = broadcast_to
  
      frames = [data]
      return header, frames
  
file_path: distributed/distributed/protocol/numpy.py
imports: from __future__ import annotations
  import math
  import numpy
  from packaging.version import parse
  from distributed.protocol import pickle
  from distributed.protocol.serialize import dask_deserialize, dask_serialize
  from distributed.utils import log_errors
label_desc: Use image node_id 2 for calling a global function with example usage: serialize_numpy_ndarray(x, context) and returns: header,  frames, header,  frames