parameters: (per_handler_ops, stamp, worker_device)
return_statement: return handler_results
docstring: Given a dictionary of ops for each handler, runs them in batch.
docstring_summary: Given a dictionary of ops for each handler, runs them in batch.
docstring_tokens: Given a dictionary of ops for each handler runs them in batch .
function: def run_handler_scheduled_ops(per_handler_ops, stamp, worker_device):
    """Given a dictionary of ops for each handler, runs them in batch."""
    batched_ops = collections.defaultdict(list)
    # Group the ops by their batching_key. Ops that share the same batching key
    # can be executed together.
    for handler in per_handler_ops.keys():
      for op in per_handler_ops[handler]:
        batched_ops[(op.batching_key(), op.batch_runner_fn())].append(op)
    op_results = {}
    for batch in batched_ops.values():
      # Run each of the batched ops using its runner.
      results = batch[0].batch_runner_fn()(batch, stamp)
      # If the result is a tuple, move each entry in the tuple in one RPC.
      if isinstance(results, tuple):
        results = tuple(
            _move_tensors(result, worker_device) for result in results)
        # Once all the results are on the worker, create individual tuple for
        # each scheduled op request.
        for i in range(len(batch)):
          op_results[batch[i]] = tuple(result[i] for result in results)
      # If the result is a tuple, it didn't have any outputs, so use the
      # `ops.Operation` as the result for all the scheduled ops.
      elif isinstance(results, ops.Operation):
        for i in range(len(batch)):
          op_results[batch[i]] = results
      else:
        raise ValueError("Unknown type of result %s.", results)
    handler_results = collections.defaultdict(list)
    # Dispatch the results of the ScheduledOps to the handlers that requested
    # them.
    for handler in per_handler_ops.keys():
      for op in per_handler_ops[handler]:
        handler_results[handler].append(op_results[op])
    return handler_results
function_tokens: def run_handler_scheduled_ops ( per_handler_ops , stamp , worker_device ) : batched_ops = collections . defaultdict ( list ) # Group the ops by their batching_key. Ops that share the same batching key # can be executed together. for handler in per_handler_ops . keys ( ) : for op in per_handler_ops [ handler ] : batched_ops [ ( op . batching_key ( ) , op . batch_runner_fn ( ) ) ] . append ( op ) op_results = { } for batch in batched_ops . values ( ) : # Run each of the batched ops using its runner. results = batch [ 0 ] . batch_runner_fn ( ) ( batch , stamp ) # If the result is a tuple, move each entry in the tuple in one RPC. if isinstance ( results , tuple ) : results = tuple ( _move_tensors ( result , worker_device ) for result in results ) # Once all the results are on the worker, create individual tuple for # each scheduled op request. for i in range ( len ( batch ) ) : op_results [ batch [ i ] ] = tuple ( result [ i ] for result in results ) # If the result is a tuple, it didn't have any outputs, so use the # `ops.Operation` as the result for all the scheduled ops. elif isinstance ( results , ops . Operation ) : for i in range ( len ( batch ) ) : op_results [ batch [ i ] ] = results else : raise ValueError ( "Unknown type of result %s." , results ) handler_results = collections . defaultdict ( list ) # Dispatch the results of the ScheduledOps to the handlers that requested # them. for handler in per_handler_ops . keys ( ) : for op in per_handler_ops [ handler ] : handler_results [ handler ] . append ( op_results [ op ] ) return handler_results