Support for control replication.

Author: elliottslaughter

src/fuzzer.cc

@@ -51,6 +51,7 @@ enum ProjectionIDs {
 static Logger log_fuzz("fuzz");
 
 static RngSeed root_seed;
+static uint64_t replicate_levels = 0;
 
 static long long parse_long_long(const std::string &flag, const std::string &arg) {
   long long result;
@@ -80,6 +81,7 @@ struct FuzzerConfig {
   uint64_t region_tree_branch_factor = 4;
   uint64_t region_tree_size_factor = 4;
   uint64_t region_tree_num_fields = 4;
+  uint64_t replicate_levels = 0;
   uint64_t num_ops = 1;
   uint64_t skip_ops = 0;
 
@@ -105,6 +107,9 @@ struct FuzzerConfig {
       } else if (flag == "-fuzz:fields") {
         std::string arg(argv[++i]);
         config.region_tree_num_fields = parse_uint64_t(flag, arg);
+      } else if (flag == "-fuzz:replicate") {
+        std::string arg(argv[++i]);
+        config.replicate_levels = parse_uint64_t(flag, arg);
       } else if (flag == "-fuzz:ops") {
         std::string arg(argv[++i]);
         config.num_ops = parse_uint64_t(flag, arg);
@@ -127,6 +132,7 @@ struct FuzzerConfig {
                               << region_tree_size_factor);
     LOG_ONCE(log_fuzz.print() << "  config.region_tree_num_fields = "
                               << region_tree_num_fields);
+    LOG_ONCE(log_fuzz.print() << "  config.replicate_levels = " << replicate_levels);
     LOG_ONCE(log_fuzz.print() << "  config.num_ops = " << num_ops);
     LOG_ONCE(log_fuzz.print() << "  config.skip_ops = " << skip_ops);
   }
@@ -1205,8 +1211,9 @@ void top_level(const Task *task, const std::vector<PhysicalRegion> &regions, Con
 static void create_mappers(Machine machine, Runtime *runtime,
                            const std::set<Processor> &local_procs) {
   for (Processor proc : local_procs) {
-    FuzzMapper::FuzzMapper *mapper = new FuzzMapper::FuzzMapper(
-        runtime->get_mapper_runtime(), machine, proc, root_seed.make_stream());
+    FuzzMapper::FuzzMapper *mapper =
+        new FuzzMapper::FuzzMapper(runtime->get_mapper_runtime(), machine, proc,
+                                   root_seed.make_stream(), replicate_levels);
     runtime->replace_default_mapper(mapper, proc);
   }
 }
@@ -1219,6 +1226,7 @@ int main(int argc, char **argv) {
   Runtime::initialize(&argc, &argv, true /* filter */);
   FuzzerConfig config = FuzzerConfig::parse_args(argc, argv);
   root_seed = RngSeed(config.initial_seed);
+  replicate_levels = config.replicate_levels;
 
   Runtime::preregister_projection_functor(PROJECTION_OFFSET_1_ID,
                                           new OffsetProjection(1));

src/mapper.cc

@@ -31,13 +31,15 @@ enum MapperCallIDs {
 
 static Logger log_map("fuzz_mapper");
 
-FuzzMapper::FuzzMapper(MapperRuntime *rt, Machine machine, Processor local, RngStream st)
+FuzzMapper::FuzzMapper(MapperRuntime *rt, Machine machine, Processor local, RngStream st,
+                       uint64_t replicate)
     : NullMapper(rt, machine),
       stream(st),
       select_tasks_to_map_channel(st.make_channel(int32_t(SELECT_TASKS_TO_MAP))),
       map_inline_channel(st.make_channel(int32_t(MAP_INLINE))),
       select_inline_sources_channel(st.make_channel(int32_t(SELECT_INLINE_SOURCES))),
-      local_proc(local) {
+      local_proc(local),
+      replicate_levels(replicate) {
   // TODO: something other than CPU processor
   {
     Machine::ProcessorQuery query(machine);
@@ -85,7 +87,7 @@ void FuzzMapper::select_task_options(const MapperContext ctx, const Task &task,
   output.map_locally = false;  // TODO
   output.valid_instances = false;
   output.memoize = true;
-  output.replicate = task.get_depth() == 0;  // TODO: replicate other tasks
+  output.replicate = task.get_depth() < static_cast<int64_t>(replicate_levels);
   // output.parent_priority = ...; // Leave parent at current priority.
   // output.check_collective_regions.insert(...); // TODO
 }
@@ -149,6 +151,8 @@ void FuzzMapper::map_task(const MapperContext ctx, const Task &task,
 void FuzzMapper::replicate_task(MapperContext ctx, const Task &task,
                                 const ReplicateTaskInput &input,
                                 ReplicateTaskOutput &output) {
+  if (task.get_depth() > static_cast<int64_t>(replicate_levels)) return;
+
   // TODO: cache this?
   std::vector<VariantID> variants;
   runtime->find_valid_variants(ctx, task.task_id, variants);
@@ -158,7 +162,28 @@ void FuzzMapper::replicate_task(MapperContext ctx, const Task &task,
     abort();
   }
   output.chosen_variant = variants.at(0);
-  // TODO: actually replicate
+
+  bool is_replicable =
+      runtime->is_replicable_variant(ctx, task.task_id, output.chosen_variant);
+  // For now assume we always have replicable variants at this level.
+  if (!is_replicable) {
+    log_map.fatal() << "Bad variants in replicate_task: variant is not replicable";
+    abort();
+  }
+
+  std::map<AddressSpace, Processor> targets;
+  for (Processor proc : global_procs) {
+    AddressSpace space = proc.address_space();
+    if (!targets.count(space)) {
+      targets[space] = proc;
+    }
+  }
+
+  if (targets.size() > 1) {
+    for (auto &target : targets) {
+      output.target_processors.push_back(target.second);
+    }
+  }
 }
 
 void FuzzMapper::select_task_sources(const MapperContext ctx, const Task &task,

src/mapper.h

@@ -28,7 +28,8 @@ using namespace Legion::Mapping;
 
 class FuzzMapper : public NullMapper {
 public:
-  FuzzMapper(MapperRuntime *runtime, Machine machine, Processor local, RngStream stream);
+  FuzzMapper(MapperRuntime *runtime, Machine machine, Processor local, RngStream stream,
+             uint64_t replicate);
 
 public:
   const char *get_mapper_name(void) const override;
@@ -97,6 +98,8 @@ class FuzzMapper : public NullMapper {
   Processor local_proc;
   std::vector<Processor> local_procs;
   std::vector<Processor> global_procs;
+
+  uint64_t replicate_levels;
 };
 
 }  // namespace FuzzMapper