docstr: Cull a SimpleShuffleLayer HighLevelGraph layer.
  
  The underlying graph will only include the necessary
  tasks to produce the keys (indices) included in `parts_out`.
  Therefore, "culling" the layer only requires us to reset this
  parameter.
docstr_tok: ["Cull","a","SimpleShuffleLayer","HighLevelGraph","layer",".","The","underlying","graph","will","only","include","the","necessary","tasks","to","produce","the","keys","(","indices",")","included","in","`","parts_out","`",".","Therefore",",","``","culling","''","the","layer","only","requires","us","to","reset","this","parameter","."]
code: def cull(
      self, keys: Iterable[str], all_keys: Any
  ) -> tuple[HashJoinP2PLayer, dict]:
      """Cull a SimpleShuffleLayer HighLevelGraph layer.
  
      The underlying graph will only include the necessary
      tasks to produce the keys (indices) included in `parts_out`.
      Therefore, "culling" the layer only requires us to reset this
      parameter.
      """
      parts_out = self._keys_to_parts(keys)
      culled_deps = self._cull_dependencies(keys, parts_out=parts_out)
      if parts_out != set(self.parts_out):
          culled_layer = self._cull(parts_out)
          return culled_layer, culled_deps
      else:
          return self, culled_deps
  
imports: from __future__ import annotations
  from collections.abc import Iterable, Sequence
  from typing import TYPE_CHECKING, Any
  import dask
  from dask.base import is_dask_collection, tokenize
  from dask.highlevelgraph import HighLevelGraph
  from dask.layers import Layer
  from distributed.shuffle._arrow import check_minimal_arrow_version
  from distributed.shuffle._core import ShuffleId, barrier_key, get_worker_plugin
  from distributed.shuffle._shuffle import shuffle_barrier, shuffle_transfer
label_desc: Use image node_id 12 for calling the HashJoinP2PLayer obj's underlying member method code with example usage: obj.cull(keys, all_keys) and returns: culled_layer,  culled_deps, self,  culled_deps