code: def invoke_udf_reduce(graph, func, msgdata, *, orig_nid=None):
      """Invoke user-defined reduce function on all the nodes in the graph.
  
      It analyzes the graph, groups nodes by their degrees and applies the UDF on each
      group -- a strategy called *degree-bucketing*.
  
      Parameters
      ----------
      graph : DGLGraph
          The input graph.
      func : callable
          The user-defined function.
      msgdata : dict[str, Tensor]
          Message data.
      orig_nid : Tensor, optional
          Original node IDs. Useful if the input graph is an extracted subgraph.
  
      Returns
      -------
      dict[str, Tensor]
          Results from running the UDF.
      """
      degs = graph.in_degrees()
      nodes = graph.dstnodes()
      if orig_nid is None:
          orig_nid = nodes
      ntype = graph.dsttypes[0]
      ntid = graph.get_ntype_id_from_dst(ntype)
      dstdata = graph._node_frames[ntid]
      msgdata = Frame(msgdata)
  
      # degree bucketing
      unique_degs, bucketor = _bucketing(degs)
      bkt_rsts = []
      bkt_nodes = []
      for deg, node_bkt, orig_nid_bkt in zip(
          unique_degs, bucketor(nodes), bucketor(orig_nid)
      ):
          if deg == 0:
              # skip reduce function for zero-degree nodes
              continue
          bkt_nodes.append(node_bkt)
          ndata_bkt = dstdata.subframe(node_bkt)
  
          # order the incoming edges per node by edge ID
          eid_bkt = F.zerocopy_to_numpy(
              graph.in_edges(node_bkt, form="eid")
          )
          assert len(eid_bkt) == deg * len(node_bkt)
          eid_bkt = np.sort(eid_bkt.reshape((len(node_bkt), deg)), 1)
          eid_bkt = F.zerocopy_from_numpy(eid_bkt.flatten())
  
          msgdata_bkt = msgdata.subframe(eid_bkt)
          # reshape all msg tensors to (num_nodes_bkt, degree, feat_size)
          maildata = {}
          for k, msg in msgdata_bkt.items():
              newshape = (len(node_bkt), deg) + F.shape(msg)[1:]
              maildata[k] = F.reshape(msg, newshape)
          # invoke udf
          nbatch = NodeBatch(
              graph, orig_nid_bkt, ntype, ndata_bkt, msgs=maildata
          )
          bkt_rsts.append(func(nbatch))
  
      # prepare a result frame
      retf = Frame(num_rows=len(nodes))
      retf._initializers = dstdata._initializers
      retf._default_initializer = dstdata._default_initializer
  
      # merge bucket results and write to the result frame
      if (
          len(bkt_rsts) != 0
      ):  # if all the nodes have zero degree, no need to merge results.
          merged_rst = {}
          for k in bkt_rsts[0].keys():
              merged_rst[k] = F.cat([rst[k] for rst in bkt_rsts], dim=0)
          merged_nodes = F.cat(bkt_nodes, dim=0)
          retf.update_row(merged_nodes, merged_rst)
  
      return retf
  
file_path: dgl/python/dgl/core.py
imports: import numpy
  from .None import backend, function, ops
  from .base import ALL, dgl_warning, DGLError, EID, is_all, NID
  from .frame import Frame
  from .udf import EdgeBatch, NodeBatch
label_desc: Use image node_id 4 for calling a global function with example usage: invoke_udf_reduce(graph, func, msgdata) and returns: retf