diff --git a/lib/realm-execution/include/realm-execution/tasks/impl/nccl_task.h b/lib/realm-execution/include/realm-execution/tasks/impl/nccl_task.h
new file mode 100644
index 0000000000..bc146a30f4
--- /dev/null
+++ b/lib/realm-execution/include/realm-execution/tasks/impl/nccl_task.h
@@ -0,0 +1,57 @@
+#ifndef _FLEXFLOW_LIB_REALM_EXECUTION_INCLUDE_REALM_EXECUTION_TASKS_IMPL_NCCL_TASK_H
+#define _FLEXFLOW_LIB_REALM_EXECUTION_INCLUDE_REALM_EXECUTION_TASKS_IMPL_NCCL_TASK_H
+
+#include "realm-execution/device_specific_managed_per_device_ff_handle.h"
+#include "realm-execution/tensor_instance_backing.dtg.h"
+#include "task-spec/dynamic_graph/dynamic_node_invocation.dtg.h"
+#include "kernels/device.h"
+#include <nccl.h>
+#include "realm-execution/realm.h"
+#include "realm-execution/realm_context.h"
+#include <string>
+#include <cstddef>
+
+namespace FlexFlow {
+
+ncclResult_t run_nccl_all_reduce(void const *send_buffer,
+                                void *receive_buffer,
+                                size_t count,
+                                ncclDataType_t data_type,
+                                ncclRedOp_t reduction_op,
+                                ncclComm_t communicator,
+                                ffStream_t stream);
+
+ncclResult_t run_nccl_broadcast(void const *send_buffer,
+                                void *receive_buffer,
+                                size_t count,
+                                ncclDataType_t data_type,
+                                int root_rank,
+                                ncclComm_t communicator,
+                                ffStream_t stream);
+
+ncclResult_t run_nccl_reduce(void const *send_buffer,
+                             void *receive_buffer,
+                             size_t count,
+                             ncclDataType_t data_type,
+                             ncclRedOp_t reduction_op,
+                             int root_rank,
+                             ncclComm_t communicator,
+                             ffStream_t stream);
+
+void nccl_task_body(void const *args,
+    size_t arglen,
+    void const *userdata,
+    size_t userdata_len,
+    Realm::Processor proc);
+
+Realm::Event spawn_nccl_task(
+    RealmContext &ctx,
+    Realm::Processor target_proc,
+    DynamicNodeInvocation const &invocation,
+    TensorInstanceBacking const &tensor_backing,
+    DeviceSpecificPtr<ManagedPerDeviceFFHandle> const &device_handle,
+    Realm::Event precondition);
+
+}
+
+#endif
diff --git a/lib/realm-execution/include/realm-execution/tasks/impl/nccl_task_args.dtg.toml b/lib/realm-execution/include/realm-execution/tasks/impl/nccl_task_args.dtg.toml
new file mode 100644
index 0000000000..b819b794df
--- /dev/null
+++ b/lib/realm-execution/include/realm-execution/tasks/impl/nccl_task_args.dtg.toml
@@ -0,0 +1,23 @@
+namespace = "FlexFlow"
+name = "NCCLTaskArgs"
+type = "struct"
+features = []
+
+includes = [
+  "realm-execution/device_specific_managed_per_device_ff_handle.h",
+  "realm-execution/device_specific_ptr.h",
+  "realm-execution/tensor_instance_backing.dtg.h",
+  "task-spec/dynamic_graph/dynamic_node_invocation.dtg.h",
+]
+
+[[fields]]
+name = "invocation"
+type = "::FlexFlow::DynamicNodeInvocation"
+
+[[fields]]
+name = "tensor_backing"
+type = "::FlexFlow::TensorInstanceBacking"
+
+[[fields]]
+name = "device_handle"
+type = "::FlexFlow::DeviceSpecificPtr<::FlexFlow::ManagedPerDeviceFFHandle>"
diff --git a/lib/realm-execution/include/realm-execution/tasks/impl/serializable_nccl_task_args.dtg.toml b/lib/realm-execution/include/realm-execution/tasks/impl/serializable_nccl_task_args.dtg.toml
new file mode 100644
index 0000000000..919913fd79
--- /dev/null
+++ b/lib/realm-execution/include/realm-execution/tasks/impl/serializable_nccl_task_args.dtg.toml
@@ -0,0 +1,24 @@
+namespace = "FlexFlow"
+name = "SerializableNcclTaskArgs"
+type = "struct"
+features = [
+  "json",
+]
+
+includes = [
+  "realm-execution/tasks/serializer/serializable_device_specific_ptr.dtg.h",
+  "realm-execution/tasks/serializer/serializable_tensor_instance_backing.dtg.h",
+  "task-spec/dynamic_graph/serializable_dynamic_node_invocation.dtg.h",
+]
+
+[[fields]]
+name = "invocation"
+type = "::FlexFlow::SerializableDynamicNodeInvocation"
+
+[[fields]]
+name = "tensor_backing"
+type = "::FlexFlow::SerializableTensorInstanceBacking"
+
+[[fields]]
+name = "device_handle"
+type = "::FlexFlow::SerializableDeviceSpecificPtr"
diff --git a/lib/realm-execution/include/realm-execution/tasks/impl/serializable_nccl_task_args.h b/lib/realm-execution/include/realm-execution/tasks/impl/serializable_nccl_task_args.h
new file mode 100644
index 0000000000..9e4eceef68
--- /dev/null
+++ b/lib/realm-execution/include/realm-execution/tasks/impl/serializable_nccl_task_args.h
@@ -0,0 +1,17 @@
+#ifndef _FLEXFLOW_LIB_REALM_EXECUTION_INCLUDE_REALM_EXECUTION_TASKS_IMPL_SERIALIZABLE_NCCL_TASK_ARGS_H
+#define _FLEXFLOW_LIB_REALM_EXECUTION_INCLUDE_REALM_EXECUTION_TASKS_IMPL_SERIALIZABLE_NCCL_TASK_ARGS_H
+
+#include "realm-execution/tasks/impl/nccl_task_args.dtg.h"
+#include "realm-execution/tasks/impl/serializable_nccl_task_args.dtg.h"
+
+namespace FlexFlow {
+
+SerializableNcclTaskArgs
+    nccl_task_args_to_serializable(NCCLTaskArgs const &);
+
+NCCLTaskArgs
+    nccl_task_args_from_serializable(SerializableNcclTaskArgs const &);
+
+} // namespace FlexFlow
+
+#endif
diff --git a/lib/realm-execution/include/realm-execution/tasks/task_id_t.dtg.toml b/lib/realm-execution/include/realm-execution/tasks/task_id_t.dtg.toml
index a7c2d54180..10961065da 100644
--- a/lib/realm-execution/include/realm-execution/tasks/task_id_t.dtg.toml
+++ b/lib/realm-execution/include/realm-execution/tasks/task_id_t.dtg.toml
@@ -270,6 +270,9 @@ name = "NCCL_GETUNIQUEID_TASK_ID"
 [[values]]
 name = "NCCL_INIT_COMMS_TASK_ID"
 
+[[values]]
+name = "NCCL_HELLO_WORLD_TASK_ID"
+
 [[values]]
 name = "STRATEGY_SEARCH_TASK_ID"
 
diff --git a/lib/realm-execution/src/realm-execution/tasks/impl/nccl_task.cc b/lib/realm-execution/src/realm-execution/tasks/impl/nccl_task.cc
new file mode 100644
index 0000000000..05d7a982f3
--- /dev/null
+++ b/lib/realm-execution/src/realm-execution/tasks/impl/nccl_task.cc
@@ -0,0 +1,110 @@
+#include "realm-execution/tasks/impl/nccl_task.h"
+#include "realm-execution/tasks/impl/nccl_task_args.dtg.h"
+#include "realm-execution/tasks/impl/serializable_nccl_task_args.h"
+#include "realm-execution/tasks/serializer/task_arg_serializer.h"
+#include "realm-execution/tasks/task_id_t.h"
+
+#include <cstdio>
+#include <nccl.h>
+
+namespace FlexFlow {
+
+ncclResult_t run_nccl_all_reduce(void const *send_buffer,
+                                void *receive_buffer,
+                                size_t count,
+                                ncclDataType_t data_type,
+                                ncclRedOp_t reduction_op,
+                                ncclComm_t communicator,
+                                ffStream_t stream) {
+  return ncclAllReduce(send_buffer,
+                       receive_buffer,
+                       count,
+                       data_type,
+                       reduction_op,
+                       communicator,
+                       stream);
+}
+
+ncclResult_t run_nccl_broadcast(void const *send_buffer,
+                                void *receive_buffer,
+                                size_t count,
+                                ncclDataType_t data_type,
+                                int root_rank,
+                                ncclComm_t communicator,
+                                ffStream_t stream) {
+  return ncclBroadcast(send_buffer,
+                       receive_buffer,
+                       count,
+                       data_type,
+                       root_rank,
+                       communicator,
+                       stream);
+}
+
+ncclResult_t run_nccl_reduce(void const *send_buffer,
+                             void *receive_buffer,
+                             size_t count,
+                             ncclDataType_t data_type,
+                             ncclRedOp_t reduction_op,
+                             int root_rank,
+                             ncclComm_t communicator,
+                             ffStream_t stream) {
+  return ncclReduce(send_buffer,
+                    receive_buffer,
+                    count,
+                    data_type,
+                    reduction_op,
+                    root_rank,
+                    communicator,
+                    stream);
+}
+
+void nccl_task_body(void const *args,
+                    size_t arglen,
+                    void const *userdata,
+                    size_t userdata_len,
+                    Realm::Processor proc) {
+  (void)userdata;
+  (void)userdata_len;
+  (void)proc;
+
+  NCCLTaskArgs task_args = nccl_task_args_from_serializable(
+      deserialize_task_args<SerializableNcclTaskArgs>(args, arglen));
+
+  int nccl_version = 0;
+  ncclResult_t result = ncclGetVersion(&nccl_version);
+
+  if (result != ncclSuccess) {
+    std::printf("NCCL error: %s\n", ncclGetErrorString(result));
+    return;
+  }
+
+  std::printf("NCCL version: %d\n", nccl_version);
+}
+
+Realm::Event spawn_nccl_task(
+    RealmContext &ctx,
+    Realm::Processor target_proc,
+    DynamicNodeInvocation const &invocation,
+    TensorInstanceBacking const &tensor_backing,
+    DeviceSpecificPtr<ManagedPerDeviceFFHandle> const &device_handle,
+    Realm::Event precondition) {
+  NCCLTaskArgs task_args = NCCLTaskArgs{
+      /*invocation=*/invocation,
+      /*tensor_backing=*/tensor_backing,
+      /*device_handle=*/device_handle,
+  };
+
+  std::string serialized_args =
+      serialize_task_args(nccl_task_args_to_serializable(task_args));
+
+  return ctx.spawn_task(
+      target_proc,
+      task_id_t::NCCL_HELLO_WORLD_TASK_ID,
+      serialized_args.data(),
+      serialized_args.size(),
+      Realm::ProfilingRequestSet{},
+      precondition);
+}
+
+} // namespace FlexFlow
diff --git a/lib/realm-execution/src/realm-execution/tasks/impl/serializable_nccl_task_args.cc b/lib/realm-execution/src/realm-execution/tasks/impl/serializable_nccl_task_args.cc
new file mode 100644
index 0000000000..c19cbc0c46
--- /dev/null
+++ b/lib/realm-execution/src/realm-execution/tasks/impl/serializable_nccl_task_args.cc
@@ -0,0 +1,33 @@
+#include "realm-execution/tasks/impl/serializable_nccl_task_args.h"
+#include "realm-execution/tasks/serializer/serializable_device_specific_ptr.h"
+#include "realm-execution/tasks/serializer/serializable_tensor_instance_backing.h"
+#include "task-spec/dynamic_graph/serializable_dynamic_node_invocation.h"
+
+namespace FlexFlow {
+
+SerializableNcclTaskArgs
+nccl_task_args_to_serializable(NCCLTaskArgs const &args) {
+  return SerializableNcclTaskArgs{
+      /*invocation=*/
+      dynamic_node_invocation_to_serializable(args.invocation),
+      /*tensor_backing=*/
+      tensor_instance_backing_to_serializable(args.tensor_backing),
+      /*device_handle=*/
+      device_specific_ptr_to_serializable(args.device_handle),
+  };
+}
+
+NCCLTaskArgs
+nccl_task_args_from_serializable(SerializableNcclTaskArgs const &args) {
+  return NCCLTaskArgs{
+      /*invocation=*/
+      dynamic_node_invocation_from_serializable(args.invocation),
+      /*tensor_backing=*/
+      tensor_instance_backing_from_serializable(args.tensor_backing),
+      /*device_handle=*/
+      device_specific_ptr_from_serializable<ManagedPerDeviceFFHandle>(
+          args.device_handle),
+  };
+}
+
+} // namespace FlexFlow
diff --git a/lib/realm-execution/src/realm-execution/tasks/realm_task_registry.cc b/lib/realm-execution/src/realm-execution/tasks/realm_task_registry.cc
index 406a2912a3..1f79036d5c 100644
--- a/lib/realm-execution/src/realm-execution/tasks/realm_task_registry.cc
+++ b/lib/realm-execution/src/realm-execution/tasks/realm_task_registry.cc
@@ -7,6 +7,7 @@
 #include "realm-execution/tasks/impl/per_device_op_state_init_task.h"
 #include "realm-execution/tasks/task_id_t.h"
 #include "utils/exception.h"
+#include "realm-execution/tasks/impl/nccl_task.h"
 
 namespace FlexFlow {
 
@@ -133,6 +134,16 @@ Realm::Event register_all_tasks() {
         register_task(Realm::Processor::TOC_PROC, task_id, op_task_body));
   }
 
+  pending_registrations.push_back(
+      register_task(Realm::Processor::LOC_PROC,
+                    task_id_t::NCCL_HELLO_WORLD_TASK_ID,
+                    nccl_task_body));
+
+  pending_registrations.push_back(
+      register_task(Realm::Processor::TOC_PROC,
+                    task_id_t::NCCL_HELLO_WORLD_TASK_ID,
+                    nccl_task_body));
+
   pending_registrations.push_back(register_task(Realm::Processor::LOC_PROC,
                                                 task_id_t::CONTROLLER_TASK_ID,
                                                 controller_task_body));
diff --git a/lib/realm-execution/test/src/realm-execution/nccl_task.cc b/lib/realm-execution/test/src/realm-execution/nccl_task.cc
new file mode 100644
index 0000000000..e4af71e716
--- /dev/null
+++ b/lib/realm-execution/test/src/realm-execution/nccl_task.cc
@@ -0,0 +1,173 @@
+#include "internal/realm_test_utils.h"
+#include "realm-execution/distributed_ff_handle.h"
+#include "op-attrs/ops/replicate_attrs.dtg.h"
+#include "op-attrs/pcg_operator_attrs.dtg.h"
+#include "realm-execution/realm_manager.h"
+#include "realm-execution/tasks/impl/nccl_task.h"
+#include "realm-execution/tensor_instance_backing.h"
+#include "task-spec/dynamic_graph/training_operation_attrs.dtg.h"
+
+#include <cuda_runtime.h>
+#include <doctest/doctest.h>
+#include <nccl.h>
+
+namespace test {
+
+using namespace ::FlexFlow;
+namespace Realm = ::FlexFlow::Realm;
+
+TEST_SUITE(FF_CUDA_TEST_SUITE) {
+
+TEST_CASE("NCCL task spawns successfully") {
+  std::vector<char *> fake_args =
+      make_fake_realm_args(/*num_cpus=*/1_p, /*num_gpus=*/1_n);
+
+  int fake_argc = fake_args.size();
+  char **fake_argv = fake_args.data();
+
+  RealmManager manager(&fake_argc, &fake_argv);
+
+  ControllerTaskResult result =
+      manager.start_controller([](RealmContext &ctx) {
+          Realm::Machine::ProcessorQuery processor_query(
+              Realm::Machine::get_machine());
+          processor_query.only_kind(Realm::Processor::TOC_PROC);
+
+        Realm::Processor gpu_proc = processor_query.first();
+
+        DistributedFfHandle distributed_handle =
+            create_distributed_ff_handle(
+                ctx,
+                /*workSpaceSize=*/1024 * 1024,
+                /*allowTensorOpMathConversion=*/true,
+                Realm::Event::NO_EVENT);
+        DynamicNodeInvocation invocation{
+            /*inputs=*/{},
+            /*node_attrs=*/
+            DynamicNodeAttrs{
+                /*task_type=*/std::nullopt,
+                /*device_coord=*/std::nullopt,
+                /*mapping=*/std::nullopt,
+                /*op_attrs=*/
+                TrainingOperationAttrs{
+                    PCGOperatorAttrs{
+                        ReplicateAttrs{
+                            /*replicate_degree=*/8_p,
+                        },
+                    },
+                },
+                /*layer_guid=*/
+                dynamic_layer_guid_t{
+                    parallel_layer_guid_t{
+                        Node{0},
+                    },
+                },
+                /*per_device_op_state=*/std::nullopt,
+            },
+            /*outputs=*/{},
+        };
+
+        TensorInstanceBacking tensor_backing =
+            make_empty_tensor_instance_backing();
+
+        Realm::Event event = spawn_nccl_task(
+            ctx,
+            gpu_proc,
+            invocation,
+            tensor_backing,
+            distributed_handle.at(gpu_proc),
+            Realm::Event::NO_EVENT);
+
+        event.wait();
+      });
+
+  result.wait();
+}
+
+TEST_CASE("NCCL broadcast and reduce helpers") {
+  constexpr size_t count = 8;
+  size_t const buffer_size = count * sizeof(int);
+
+  ncclUniqueId unique_id;
+  REQUIRE(ncclGetUniqueId(&unique_id) == ncclSuccess);
+
+  ncclComm_t communicator;
+  REQUIRE(ncclCommInitRank(
+              &communicator,
+              /*num_ranks=*/1,
+              unique_id,
+              /*rank=*/0) == ncclSuccess);
+
+  ffStream_t stream;
+  REQUIRE(cudaStreamCreate(&stream) == cudaSuccess);
+
+  int *send_buffer = nullptr;
+  int *receive_buffer = nullptr;
+
+  REQUIRE(cudaMalloc(&send_buffer, buffer_size) == cudaSuccess);
+  REQUIRE(cudaMalloc(&receive_buffer, buffer_size) == cudaSuccess);
+
+  std::vector<int> input = {1, 2, 3, 4, 5, 6, 7, 8};
+  std::vector<int> output(count, 0);
+
+  REQUIRE(cudaMemcpy(send_buffer,
+                     input.data(),
+                     buffer_size,
+                     cudaMemcpyHostToDevice) == cudaSuccess);
+
+  SUBCASE("broadcast") {
+    REQUIRE(cudaMemset(receive_buffer, 0, buffer_size) == cudaSuccess);
+
+    REQUIRE(run_nccl_broadcast(send_buffer,
+                               receive_buffer,
+                               count,
+                               ncclInt32,
+                               /*root_rank=*/0,
+                               communicator,
+                               stream) == ncclSuccess);
+
+    REQUIRE(cudaStreamSynchronize(stream) == cudaSuccess);
+
+    REQUIRE(cudaMemcpy(output.data(),
+                       receive_buffer,
+                       buffer_size,
+                       cudaMemcpyDeviceToHost) == cudaSuccess);
+
+    for (size_t i = 0; i < count; i++) {
+      CHECK(output[i] == input[i]);
+    }
+  }
+
+  SUBCASE("reduce") {
+    REQUIRE(cudaMemset(receive_buffer, 0, buffer_size) == cudaSuccess);
+
+    REQUIRE(run_nccl_reduce(send_buffer,
+                            receive_buffer,
+                            count,
+                            ncclInt32,
+                            ncclSum,
+                            /*root_rank=*/0,
+                            communicator,
+                            stream) == ncclSuccess);
+
+    REQUIRE(cudaStreamSynchronize(stream) == cudaSuccess);
+
+    REQUIRE(cudaMemcpy(output.data(),
+                       receive_buffer,
+                       buffer_size,
+                       cudaMemcpyDeviceToHost) == cudaSuccess);
+
+    for (size_t i = 0; i < count; i++) {
+      CHECK(output[i] == input[i]);
+    }
+  }
+
+  REQUIRE(cudaFree(send_buffer) == cudaSuccess);
+  REQUIRE(cudaFree(receive_buffer) == cudaSuccess);
+  REQUIRE(cudaStreamDestroy(stream) == cudaSuccess);
+  REQUIRE(ncclCommDestroy(communicator) == ncclSuccess);
+}
+
+} // TEST_SUITE
+
+} // namespace test
diff --git a/lib/utils/test/src/utils/containers/concat_vectors.cc b/lib/utils/test/src/utils/containers/concat_vectors.cc
new file mode 100644
index 0000000000..e22b8e8c6c
--- /dev/null
+++ b/lib/utils/test/src/utils/containers/concat_vectors.cc
@@ -0,0 +1,38 @@
+#include "utils/containers/concat_vectors.h"
+#include "test/utils/doctest/fmt/vector.h"
+#include <doctest/doctest.h>
+#include <vector>
+
+using namespace ::FlexFlow;
+
+TEST_SUITE(FF_TEST_SUITE) {
+  TEST_CASE("concat_vectors") {
+    SUBCASE("two vectors") {
+      std::vector<int> prefix = {1, 2};
+      std::vector<int> postfix = {3, 4};
+
+      std::vector<int> result = concat_vectors(prefix, postfix);
+      std::vector<int> correct = {1, 2, 3, 4};
+
+      CHECK(result == correct);
+    }
+
+    SUBCASE("vector of vectors") {
+      std::vector<std::vector<int>> vecs = {{1, 2}, {3}, {4, 5}};
+
+      std::vector<int> result = concat_vectors(vecs);
+      std::vector<int> correct = {1, 2, 3, 4, 5};
+
+      CHECK(result == correct);
+    }
+
+    SUBCASE("empty vector of vectors") {
+      std::vector<std::vector<int>> vecs = {};
+
+      std::vector<int> result = concat_vectors(vecs);
+      std::vector<int> correct = {};
+
+      CHECK(result == correct);
+    }
+  }
+}
diff --git a/lib/utils/test/src/utils/containers/extend_vector.cc b/lib/utils/test/src/utils/containers/extend_vector.cc
new file mode 100644
index 0000000000..a394cdf22f
--- /dev/null
+++ b/lib/utils/test/src/utils/containers/extend_vector.cc
@@ -0,0 +1,30 @@
+#include "utils/containers/extend_vector.h"
+#include "test/utils/doctest/fmt/vector.h"
+#include <doctest/doctest.h>
+#include <vector>
+
+using namespace ::FlexFlow;
+// checks rhs gets added to the end of lhs
+TEST_SUITE(FF_TEST_SUITE) {
+  TEST_CASE("extend_vector") {
+    SUBCASE("non-empty vectors") {
+      std::vector<int> lhs = {1, 2};
+      std::vector<int> rhs = {3, 4};
+
+      extend_vector(lhs, rhs);
+      std::vector<int> correct = {1, 2, 3, 4};
+
+      CHECK(lhs == correct);
+    }
+    // checks that lhs stays the same when the rhs is empty
+    SUBCASE("empty rhs") {
+      std::vector<int> lhs = {1, 2};
+      std::vector<int> rhs = {};
+
+      extend_vector(lhs, rhs);
+      std::vector<int> correct = {1, 2};
+
+      CHECK(lhs == correct);
+    }
+  }
+}
diff --git a/lib/utils/test/src/utils/containers/generate_vector.cc b/lib/utils/test/src/utils/containers/generate_vector.cc
new file mode 100644
index 0000000000..e560ba80bd
--- /dev/null
+++ b/lib/utils/test/src/utils/containers/generate_vector.cc
@@ -0,0 +1,31 @@
+#include "utils/containers/generate_vector.h"
+#include "test/utils/doctest/fmt/vector.h"
+#include "utils/exception.h"
+#include "utils/nonnegative_int/nonnegative_int.h"
+#include <doctest/doctest.h>
+#include <vector>
+
+using namespace ::FlexFlow;
+
+TEST_SUITE(FF_TEST_SUITE) {
+  TEST_CASE("generate_vector") {
+    SUBCASE("empty vector") {
+      std::vector<int> result = generate_vector(0_n, [](nonnegative_int) -> int {
+        PANIC("lambda should not be called");
+      });
+      std::vector<int> correct = {};
+
+      CHECK(result == correct);
+    }
+    SUBCASE("non-empty vector") {
+      std::vector<int> result = generate_vector(5_n, [](nonnegative_int idx) {
+        int i = idx.unwrap_nonnegative();
+        return i * i;
+
+      });
+      std::vector<int> correct = {0, 1, 4, 9, 16};
+
+      CHECK(result == correct);
+    }
+  }
+}