Add Realm execution support for parallel operators (Replicate, Repartition, Combine, Reduction)

[seemamirch]

Summary

This PR implements support for all four parallel operators (Replicate, Repartition, Combine, Reduction) CPU based in the Realm-based execution backend.

Motivation

Parallel operators are fundamental to distributed training in FlexFlow — they handle data redistribution between devices (repartition), replication (replicate), gathering (combine), and partial sum reduction (reduction). Previously only the graph pass infrastructure existed; this PR adds the Realm execution layer.

Approach

Each parallel op is executed as a Realm copy operation rather than a regular op task, since they involve data movement between devices rather than computation.

Data Movement

Op	FWD	BWD
Replicate	broadcast copy to all replicas	sum-reduce replica gradients
Repartition	scatter full tensor into shards	gather shards into full tensor
Combine	gather shards into full tensor	scatter gradient into shards
Reduction	sum-reduce partial tensors	broadcast gradient to all partials

Key Design Decisions

Offset index space allocation: Each shard instance is allocated with a non-zero origin rect reflecting its position in the full tensor (e.g. shard 1 of a [10,16] tensor split along dim 0 is allocated at [5..9, 0..15]). This allows plain Realm copies between shards and combined tensors to work correctly.

CopyDomain: A new CopyDomain enum (SRC/DST) selects which instance's index space is used as the copy domain. SRC is used when the source is the smaller piece; DST when the destination is the smaller piece.

get_per_device_shape: A new function that correctly computes the per-device tensor size by dividing each dimension by its shard degree, used for instance allocation.

ManyToOne collision fix: Combine FWD and Reduction FWD produce multiple invocations with the same output DynamicValueAttrs. Only the first is registered in the producer map to avoid collision while still marking the value as having a producer.

Changes

New files

• lib/task-spec/include/task-spec/dynamic_graph/parallel_op_utils.h — shared is_parallel_op_attrs helper
• lib/realm-execution/test/src/realm-execution/test_op_combine.cc — combine op e2e test
• lib/realm-execution/test/src/realm-execution/test_op_reduce.cc — reduction op e2e test
• lib/realm-execution/test/src/realm-execution/test_op_repartition.cc — repartition op e2e test

Modified files

• shard_expansion.cc — add perform_shard_expansion_one_to_many and perform_shard_expansion_many_to_one generic functions covering all eight FWD/BWD combinations
• pass_expansion.cc — add perform_pass_expansion_for_parallel_op
• copy_insertion.cc — guard parallel ops in copy insertion
• make_dynamic_open_dataflow_graph_from_mapped_pcg.cc — dispatch parallel ops to build_parallel_op_invocation
• dynamic_open_dataflow_graph.cc — fix ManyToOne collision for combine/reduction FWD
• pcg_instance.cc — add Realm copy dispatch for all parallel ops
• realm_context.cc/h — add CopyDomain, create_instance_with_offset, update issue_copy to use get_indexspace()
• instance_allocation.cc — use offset index spaces for sharded tensor allocation
• parallel_tensor_dims.cc/h — add get_per_device_dims
• parallel_tensor_shape.cc/h — add get_per_device_shape
• task_id_t.cc — return nullopt for all parallel ops in get_init_task_id_for_op_attrs

Testing

Added e2e tests for each parallel op running on 2 CPU devices:

• test_op_combine.cc — repartition → combine → relu
• test_op_reduce.cc — repartition → linear → reduction → relu
• test_op_repartition.cc — repartition → relu

---

This change is 

[seemamirch]

@lockshaw @elliottslaughter - please review