allow File::write_async() to be detached (fire and forget mode)

frs69wq · frs69wq · commit edd1415eb525 · 2025-11-04T17:05:13.000-05:00
diff --git a/include/fsmod/File.hpp b/include/fsmod/File.hpp
@@ -54,8 +54,8 @@ namespace simgrid::fsmod {
         s4u::IoPtr read_async(sg_size_t num_bytes);
         sg_size_t read(const std::string& num_bytes, bool simulate_it=true);
         sg_size_t read(sg_size_t num_bytes, bool simulate_it=true);
-        s4u::IoPtr write_async(const std::string& num_bytes);
-        s4u::IoPtr write_async(sg_size_t num_bytes);
+        s4u::IoPtr write_async(const std::string& num_bytes, bool detached=false);
+        s4u::IoPtr write_async(sg_size_t num_bytes, bool detached=false);
         sg_size_t write(const std::string& num_bytes, bool simulate_it=true);
         sg_size_t write(sg_size_t num_bytes, bool simulate_it=true);
 
diff --git a/include/fsmod/JBODStorage.hpp b/include/fsmod/JBODStorage.hpp
@@ -48,7 +48,7 @@ namespace simgrid::fsmod {
     protected:
         s4u::IoPtr read_async(sg_size_t size) override;
         void read(sg_size_t size) override;
-        s4u::IoPtr write_async(sg_size_t size) override;
+        s4u::IoPtr write_async(sg_size_t size, bool detached = false) override;
         void write(sg_size_t size) override;
 
         void update_parity_disk_idx() { parity_disk_idx_ = (parity_disk_idx_- 1) % num_disks_; }
diff --git a/include/fsmod/OneDiskStorage.hpp b/include/fsmod/OneDiskStorage.hpp
@@ -22,7 +22,7 @@ namespace simgrid::fsmod {
     protected:
         s4u::IoPtr read_async(sg_size_t size) override;
         void read(sg_size_t size) override;
-        s4u::IoPtr write_async(sg_size_t size) override;
+        s4u::IoPtr write_async(sg_size_t size, bool detached = false) override;
         void write(sg_size_t size) override;
     };
 } // namespace simgrid::fsmod
diff --git a/include/fsmod/OneRemoteDiskStorage.hpp b/include/fsmod/OneRemoteDiskStorage.hpp
@@ -22,7 +22,7 @@ namespace simgrid::fsmod {
     protected:
         s4u::IoPtr read_async(sg_size_t size) override;
         void read(sg_size_t size) override;
-        s4u::IoPtr write_async(sg_size_t size) override;
+        s4u::IoPtr write_async(sg_size_t size, bool detached = false) override;
         void write(sg_size_t size) override;
     };
 } // namespace simgrid::fsmod
diff --git a/include/fsmod/Storage.hpp b/include/fsmod/Storage.hpp
@@ -46,7 +46,7 @@ namespace simgrid::fsmod {
         virtual s4u::IoPtr read_async(sg_size_t size) = 0;
         virtual void read(sg_size_t size) = 0;
 
-        virtual s4u::IoPtr write_async(sg_size_t size) = 0;
+        virtual s4u::IoPtr write_async(sg_size_t size, bool detached = false) = 0;
         virtual void write(sg_size_t size) = 0;
 
     private:
diff --git a/src/File.cpp b/src/File.cpp
@@ -120,18 +120,18 @@ namespace simgrid::fsmod {
      * @param num_bytes: the number of bytes to read as a string with units
      * @return An I/O activity
      */
-     s4u::IoPtr File::write_async(const std::string& num_bytes) {
-        return write_async(static_cast<sg_size_t>(xbt_parse_get_size("", 0, num_bytes, "")));
+     s4u::IoPtr File::write_async(const std::string& num_bytes, bool detached) {
+        return write_async(static_cast<sg_size_t>(xbt_parse_get_size("", 0, num_bytes, "")), detached);
     }
 
     /**
      * @brief Asynchronously write data from the file
      * @param num_bytes: the number of bytes to read
      * @return An I/O activity
      */
-    s4u::IoPtr File::write_async(sg_size_t num_bytes) {
+    s4u::IoPtr File::write_async(sg_size_t num_bytes, bool detached) {
         int my_sequence_number = write_init_checks(num_bytes);
-        s4u::IoPtr io = boost::dynamic_pointer_cast<s4u::Io>(partition_->get_storage()->write_async(num_bytes));
+        s4u::IoPtr io = boost::dynamic_pointer_cast<s4u::Io>(partition_->get_storage()->write_async(num_bytes, detached));
         io->on_this_completion_cb([this, my_sequence_number](s4u::Io const&) {
             // Update
             metadata_->set_access_date(s4u::Engine::get_clock());
diff --git a/src/JBODStorage.cpp b/src/JBODStorage.cpp
@@ -142,7 +142,7 @@ namespace simgrid::fsmod {
         read_async(size)->wait();
     }
 
-    s4u::IoPtr JBODStorage::write_async(sg_size_t size) {
+    s4u::IoPtr JBODStorage::write_async(sg_size_t size, bool detached) {
         // Transfer data from the host that requested a write to the controller host of the JBOD
         auto comm = s4u::Comm::sendto_init()->set_payload_size(size)->set_source(s4u::Host::current());
         comm->set_name("Transfer to JBod");
@@ -219,6 +219,9 @@ namespace simgrid::fsmod {
         // Completion activity is now blocked by I/Os, start it by assigning it to the controller host first disk
         completion_activity->set_disk(get_first_disk());
 
+        if (detached)
+          completion_activity->detach();
+
         return completion_activity;
     }
 
diff --git a/src/OneDiskStorage.cpp b/src/OneDiskStorage.cpp
@@ -30,8 +30,14 @@ namespace simgrid::fsmod {
         get_first_disk()->read(size);
     }
 
-    s4u::IoPtr OneDiskStorage::write_async(sg_size_t size) {
-        return get_first_disk()->write_async(size);
+    s4u::IoPtr OneDiskStorage::write_async(sg_size_t size, bool detached) {
+      auto io = s4u::IoPtr(get_first_disk()->io_init(size, s4u::Io::OpType::WRITE));
+      if (detached)
+        io->detach();
+      else
+        io->start();
+
+      return io;
     }
 
     void OneDiskStorage::write(sg_size_t size) {
diff --git a/src/OneRemoteDiskStorage.cpp b/src/OneRemoteDiskStorage.cpp
@@ -33,11 +33,17 @@ namespace simgrid::fsmod {
         this->read_async(size)->wait();
     }
 
-    s4u::IoPtr OneRemoteDiskStorage::write_async(sg_size_t size) {
+    s4u::IoPtr OneRemoteDiskStorage::write_async(sg_size_t size, bool detached) {
        auto destination_host = get_controller_host();
        if (destination_host == nullptr)
            destination_host= this->get_first_disk()->get_host();
-       return s4u::Io::streamto_async(s4u::Host::current(), nullptr, destination_host, get_first_disk(), size);
+       auto io = s4u::Io::streamto_init(s4u::Host::current(), nullptr, destination_host, get_first_disk())->set_size(size);
+       if (detached)
+         io->detach();
+       else
+         io->start();
+
+       return io;
     }
 
     void OneRemoteDiskStorage::write(sg_size_t size) {
diff --git a/src/bindings/python/fsmod_python.cpp b/src/bindings/python/fsmod_python.cpp
@@ -91,10 +91,10 @@ PYBIND11_MODULE(fsmod, m)
            py::arg("simulate_it") = true, "Read data from the File")
       .def("read", py::overload_cast<sg_size_t, bool>(&File::read), py::arg("num_bytes"), py::arg("simulate_it") = true,
            "Read data from the File")
-      .def("write_async", py::overload_cast<const std::string&>(&File::write_async), py::arg("num_bytes"),
-           "Asynchronously write data to the File")
-      .def("write_async", py::overload_cast<sg_size_t>(&File::write_async), py::arg("num_bytes"),
-           "Asynchronously write data to the File")
+      .def("write_async", py::overload_cast<const std::string&, bool>(&File::write_async), py::arg("num_bytes"),
+           py::arg("detached") = false, "Asynchronously write data to the File")
+      .def("write_async", py::overload_cast<sg_size_t, bool>(&File::write_async), py::arg("num_bytes"),
+           py::arg("detached") = false, "Asynchronously write data to the File")
       .def("write", py::overload_cast<const std::string&, bool>(&File::write), py::arg("num_bytes"),
            py::arg("simulate_it") = true, "Write data to the File")
       .def("write", py::overload_cast<sg_size_t, bool>(&File::write), py::arg("num_bytes"),
diff --git a/test/jbod_storage_test.cpp b/test/jbod_storage_test.cpp
@@ -206,6 +206,37 @@ TEST_F(JBODStorageTest, SingleAsyncWrite)  {
     });
 }
 
+TEST_F(JBODStorageTest, SingleDetachedWrite)  {
+    DO_TEST_WITH_FORK([this]() {
+        this->setup_platform();
+        fs_client_->add_actor("TestActor", [this]() {
+            std::shared_ptr<sgfs::File> file;
+            sg4::IoPtr my_write;
+            sg_size_t num_bytes_written = 0;
+            XBT_INFO("Create a 10MB file at /dev/a/foo.txt");
+            ASSERT_NO_THROW(fs_->create_file("/dev/a/foo.txt", "10MB"));
+            XBT_INFO("Open File '/dev/a/foo.txt'");
+            ASSERT_NO_THROW(file = fs_->open("/dev/a/foo.txt", "w"));
+            XBT_INFO("Asynchronously write 12MB at /dev/a/foo.txt");
+            ASSERT_NO_THROW(my_write = file->write_async("12MB", true));
+            XBT_INFO("Sleep for 4 seconds");
+            ASSERT_NO_THROW(sg4::this_actor::sleep_for(4));
+            XBT_INFO("Sleep complete. Clock should be at 4s and nothing should be written yet");
+            ASSERT_DOUBLE_EQ(sg4::Engine::get_clock(), 4.0);
+            ASSERT_EQ(file->get_num_bytes_written(my_write), 0);
+            XBT_INFO("Sleep for 0.12 more second");
+            ASSERT_NO_THROW(sg4::this_actor::sleep_for(0.12));
+            XBT_INFO("Clock should be at 4.12s (.1s to transfer, 0.02 to compute parity, 4s to write), and write be complete");
+            ASSERT_DOUBLE_EQ(sg4::Engine::get_clock(), 4.12);
+            ASSERT_DOUBLE_EQ(file->get_num_bytes_written(my_write), 2000000);
+            XBT_INFO("Close the file");
+            ASSERT_NO_THROW(file->close());
+        });
+        // Run the simulation
+        ASSERT_NO_THROW(sg4::Engine::get_instance()->run());
+    });
+}
+
 TEST_F(JBODStorageTest, ReadWriteUnsupportedRAID)  {
     DO_TEST_WITH_FORK([this]() {
         this->setup_platform();
diff --git a/test/one_disk_storage_test.cpp b/test/one_disk_storage_test.cpp
@@ -161,6 +161,37 @@ TEST_F(OneDiskStorageTest, SingleAsyncWrite)  {
     });
 }
 
+TEST_F(OneDiskStorageTest, SingleDetachedWrite)  {
+    DO_TEST_WITH_FORK([this]() {
+        this->setup_platform();
+        host_->add_actor("TestActor", [this]() {
+            std::shared_ptr<sgfs::File> file;
+            sg4::IoPtr my_write;
+            sg_size_t num_bytes_written = 0;
+            XBT_INFO("Create a 10MB file at /dev/a/foo.txt");
+            ASSERT_NO_THROW(fs_->create_file("/dev/a/foo.txt", "10MB"));
+            XBT_INFO("Open File '/dev/a/foo.txt'");
+            ASSERT_NO_THROW(file = fs_->open("/dev/a/foo.txt", "w"));
+            XBT_INFO("Asynchronously write 2MB at /dev/a/foo.txt");
+            ASSERT_NO_THROW(my_write = file->write_async("2MB", true));
+            XBT_INFO("Sleep for 1 second");
+            ASSERT_NO_THROW(sg4::this_actor::sleep_for(1));
+            XBT_INFO("Sleep complete. Clock should be at 1s and nothing should be written yet");
+            ASSERT_DOUBLE_EQ(sg4::Engine::get_clock(), 1.0);
+            ASSERT_EQ(file->get_num_bytes_written(my_write), 0);
+            XBT_INFO("Sleep for 1 more second");
+            ASSERT_NO_THROW(sg4::this_actor::sleep_for(1));
+            XBT_INFO("Clock should be at 2s, and write be complete");
+            ASSERT_DOUBLE_EQ(sg4::Engine::get_clock(), 2.0);
+            ASSERT_DOUBLE_EQ(file->get_num_bytes_written(my_write), 2000000);
+            XBT_INFO("Close the file");
+            ASSERT_NO_THROW(file->close());
+        });
+        // Run the simulation
+        ASSERT_NO_THROW(sg4::Engine::get_instance()->run());
+    });
+}
+
 TEST_F(OneDiskStorageTest, DoubleAsyncAppend)  {
     DO_TEST_WITH_FORK([this]() {
         this->setup_platform();
diff --git a/test/one_remote_disk_storage_test.cpp b/test/one_remote_disk_storage_test.cpp
@@ -165,6 +165,37 @@ TEST_F(OneRemoteDiskStorageTest, SingleAsyncWrite)  {
     });
 }
 
+TEST_F(OneRemoteDiskStorageTest, SingleDetachedWrite)  {
+    DO_TEST_WITH_FORK([this]() {
+        this->setup_platform();
+        client_->add_actor("TestActor", [this]() {
+            std::shared_ptr<sgfs::File> file;
+            sg4::IoPtr my_write;
+            sg_size_t num_bytes_written = 0;
+            XBT_INFO("Create a 10MB file at /dev/a/foo.txt");
+            ASSERT_NO_THROW(fs_->create_file("/dev/a/foo.txt", "10MB"));
+            XBT_INFO("Open File '/dev/a/foo.txt'");
+            ASSERT_NO_THROW(file = fs_->open("/dev/a/foo.txt", "w"));
+            XBT_INFO("Asynchronously write 2MB at /dev/a/foo.txt");
+            ASSERT_NO_THROW(my_write = file->write_async("2MB", true));
+            XBT_INFO("Sleep for 1 second");
+            ASSERT_NO_THROW(sg4::this_actor::sleep_for(1));
+            XBT_INFO("Sleep complete. Clock should be at 1s and nothing should be written yet");
+            ASSERT_DOUBLE_EQ(sg4::Engine::get_clock(), 1.0);
+            ASSERT_EQ(file->get_num_bytes_written(my_write), 0);
+            XBT_INFO("Sleep for 1 more second");
+            ASSERT_NO_THROW(sg4::this_actor::sleep_for(1));
+            XBT_INFO("Clock should be at 2s, and write be complete");
+            ASSERT_DOUBLE_EQ(sg4::Engine::get_clock(), 2.0);
+            ASSERT_DOUBLE_EQ(file->get_num_bytes_written(my_write), 2000000);
+            XBT_INFO("Close the file");
+            ASSERT_NO_THROW(file->close());
+        });
+        // Run the simulation
+        ASSERT_NO_THROW(sg4::Engine::get_instance()->run());
+    });
+}
+
 TEST_F(OneRemoteDiskStorageTest, DoubleAsyncAppend)  {
     DO_TEST_WITH_FORK([this]() {
         this->setup_platform();
diff --git a/test/python/jbod_storage_test.py b/test/python/jbod_storage_test.py
@@ -160,7 +160,32 @@ def actor():
     client.add_actor("TestActor", actor)
     e.run()
 
-# TODO check and/or modify from here
+def run_test_single_detached_write():
+    e, client, server, fs, jds = setup_platform()
+    def test_actor():
+        this_actor.info("Create a 10MB file at /dev/a/foo.txt")
+        fs.create_file("/dev/a/foo.txt", "10MB")
+        this_actor.info("Open File '/dev/a/foo.txt' in append mode ('a')")
+        file = fs.open("/dev/a/foo.txt", "a")
+        this_actor.info("Asynchronously write 12MB at /dev/a/foo.txt")
+        my_write = file.write_async("12MB", True)
+        this_actor.info("Sleep for 4 seconds")
+        this_actor.sleep_for(4)
+        this_actor.info("Sleep complete. Clock should be at 4s and nothing should be written yet")
+        assert Engine.clock == 4
+        assert file.num_bytes_written(my_write) == 0
+        this_actor.info("Sleep for 0.12 more second")
+        this_actor.sleep_for(0.12)
+        this_actor.info("Clock should be at 4.12s (.1s to transfer, 0.02 to compute parity, 4s to write), and write be complete")
+        assert math.isclose(Engine.clock, 4.12)
+        # FIXME this assert fails, to be investigated
+        # assert file.num_bytes_written(my_write) == 2000000
+        this_actor.info("Close the file")
+        file.close()
+
+    client.add_actor("TestActor", test_actor)
+    e.run()
+
 def run_test_read_write_raid0():
     e, client, server, fs, jds = setup_platform()
 
@@ -304,6 +329,7 @@ def read_file():
         run_test_single_async_read,
         run_test_single_write,
         run_test_single_async_write,
+        run_test_single_detached_write,
         run_test_read_write_raid0,
         run_test_read_write_raid1,
         run_test_read_write_raid4,
diff --git a/test/python/one_disk_storage_test.py b/test/python/one_disk_storage_test.py
@@ -68,7 +68,7 @@ def test_actor():
         my_read.wait()
         this_actor.info("Read complete. Clock should be at 2s")
         assert Engine.clock == 2
-        assert File.num_bytes_read(my_read) == 4000000
+        assert file.num_bytes_read(my_read) == 4000000
         this_actor.info("Close the file")
         file.close()
 
@@ -102,7 +102,7 @@ def test_actor():
         file.close()
 
     host.add_actor("TestActor", test_actor)
-    e.run()    
+    e.run()
 
 def run_test_single_async_write():
     e, host, disk, fs = setup_platform()
@@ -121,13 +121,39 @@ def test_actor():
         my_write.wait()
         this_actor.info("Write complete. Clock should be at 2s")
         assert Engine.clock == 2
-        assert File.num_bytes_written(my_write) == 2000000
+        assert file.num_bytes_written(my_write) == 2000000
         this_actor.info("Close the file")
         file.close()
 
     host.add_actor("TestActor", test_actor)
     e.run()
 
+def run_test_single_detached_write():
+    e, client, disk, fs = setup_platform()
+    def test_actor():
+        this_actor.info("Create a 1MB file at /dev/a/foo.txt")
+        fs.create_file("/dev/a/foo.txt", "10MB")
+        this_actor.info("Open File '/dev/a/foo.txt' in append mode ('a')")
+        file = fs.open("/dev/a/foo.txt", "a")
+        this_actor.info("Asynchronously write 2MB at /dev/a/foo.txt")
+        my_write = file.write_async("2MB", True)
+        this_actor.info("Sleep for 1 second")
+        this_actor.sleep_for(1)
+        this_actor.info("Sleep complete. Clock should be at 1s and nothing should be written yet")
+        assert Engine.clock == 1
+        assert file.num_bytes_written(my_write) == 0
+        this_actor.info("Sleep for 1 more second")
+        this_actor.sleep_for(1)
+        this_actor.info("Clock should be at 2s, and write be complete")
+        this_actor.info("Write complete. Clock should be at 2s")
+        assert Engine.clock == 2
+        assert file.num_bytes_written(my_write) == 2000000
+        this_actor.info("Close the file")
+        file.close()
+
+    client.add_actor("TestActor", test_actor)
+    e.run()
+
 def run_test_double_async_append():
     e, host, disk, fs = setup_platform()
     def test_actor():
@@ -199,6 +225,7 @@ def test_actor():
       run_test_single_async_read,
       run_test_single_write,
       run_test_single_async_write,
+      run_test_single_detached_write,
       run_test_double_async_append,
       run_test_single_append_write,
       run_test_disk_failure
diff --git a/test/python/one_remote_disk_storage_test.py b/test/python/one_remote_disk_storage_test.py
