From 38b094499e862dda42a031ce8c2397c9692d01e4 Mon Sep 17 00:00:00 2001 From: Matthias Maier Date: Sat, 7 Mar 2020 14:19:59 -0600 Subject: [PATCH] Revert "remove asynchronous IO mechanism" This reverts commit 7980483d1654923c44f67491699b82e6b5ce4e51. --- examples/step-69/step-69.cc | 135 ++++++++++++++++++++++++++---------- 1 file changed, 99 insertions(+), 36 deletions(-) diff --git a/examples/step-69/step-69.cc b/examples/step-69/step-69.cc index 1e76eb80c4..74750c3444 100644 --- a/examples/step-69/step-69.cc +++ b/examples/step-69/step-69.cc @@ -504,6 +504,9 @@ namespace Step69 InitialValues initial_values; TimeStepping time_stepping; SchlierenPostprocessor schlieren_postprocessor; + + std::thread output_thread; + vector_type output_vector; }; // @sect3{Implementation} @@ -2675,6 +2678,12 @@ namespace Step69 } } + // We wait for any remaining background output thread to finish before + // printing a summary and exiting. + + if (output_thread.joinable()) + output_thread.join(); + computing_timer.print_summary(); pcout << timer_output.str() << std::endl; } @@ -2723,6 +2732,19 @@ namespace Step69 } // @sect5{Output and checkpointing} + // + // Writing out the final vtk files is a quite IO intensive task that can + // stall the main loop quite a bit. In order to avoid this we use an asynchronous + // IO strategy by creating a background thread that will perform IO + // while the main loop is allowed to continue. In order for this to work + // we have to be mindful of two things: + // - Before running the output_worker thread, we have to create + // a copy of the state vector U. We store it in the + // vector output_vector. + // - We have to avoid any MPI communication in the background thread, + // otherwise the program might deadlock. This implies that we have to + // run the postprocessing outside of the worker thread. template void MainLoop::output(const typename MainLoop::vector_type &U, @@ -2734,57 +2756,98 @@ namespace Step69 pcout << "MainLoop::output(t = " << t << ", checkpoint = " << checkpoint << ")" << std::endl; - TimerOutput::Scope scope(computing_timer, "main_loop - output"); + // We check whether the output thread is still running. If so, we have + // to wait to for it to finish because we would otherwise overwrite + // output_vector and rerun the + // schlieren_postprocessor before the output of the + // previous output cycle has been fully written back to disk. - if (checkpoint) + if (output_thread.joinable()) { - // We checkpoint the current state by doing the precise inverse - // operation to what we discussed for the resume - // logic: + TimerOutput::Scope timer(computing_timer, "main_loop - stalled output"); + output_thread.join(); + } - const unsigned int i = - discretization.triangulation.locally_owned_subdomain(); - std::string name = base_name + "-checkpoint-" + - Utilities::int_to_string(i, 4) + ".archive"; + constexpr auto problem_dimension = + ProblemDescription::problem_dimension; - std::ofstream file(name, std::ios::binary | std::ios::trunc); + // At this point we make a copy of the state vector and run the + // schlieren postprocessor. - boost::archive::binary_oarchive oa(file); - oa << t << cycle; - for (const auto &it1 : U) - for (const auto &it2 : it1) - oa << it2; + for (unsigned int i = 0; i < problem_dimension; ++i) + { + output_vector[i] = U[i]; + output_vector[i].update_ghost_values(); } - schlieren_postprocessor.compute_schlieren(U); + schlieren_postprocessor.compute_schlieren(output_vector); - // The actual output code is standard. We create a (local) DataOut - // instance, attach all data vectors we want to output and finally - // call to DataOut::write_vtu_with_pvtu_record + // Next we create a lambda function for the background thread. We capture + // the this pointer as well as most of the arguments of + // the output function by value so that we have access to them inside + // the lambda function. - DataOut data_out; - data_out.attach_dof_handler(offline_data.dof_handler); + const auto output_worker = [this, name, t, cycle, checkpoint]() { + if (checkpoint) + { + // We checkpoint the current state by doing the precise inverse + // operation to what we discussed for the resume + // logic: + + const unsigned int i = + discretization.triangulation.locally_owned_subdomain(); + std::string name = base_name + "-checkpoint-" + + Utilities::int_to_string(i, 4) + ".archive"; + + std::ofstream file(name, std::ios::binary | std::ios::trunc); + + boost::archive::binary_oarchive oa(file); + oa << t << cycle; + for (const auto &it1 : output_vector) + for (const auto &it2 : it1) + oa << it2; + } - constexpr auto problem_dimension = - ProblemDescription::problem_dimension; - const auto &component_names = ProblemDescription::component_names; + // The actual output code is standard. We create a (local) DataOut + // instance, attach all data vectors we want to output and finally + // call to DataOut::write_vtu_with_pvtu_record - for (unsigned int i = 0; i < problem_dimension; ++i) - data_out.add_data_vector(U[i], component_names[i]); + DataOut data_out; + data_out.attach_dof_handler(offline_data.dof_handler); - data_out.add_data_vector(schlieren_postprocessor.schlieren, - "schlieren_plot"); + constexpr auto problem_dimension = + ProblemDescription::problem_dimension; + const auto &component_names = ProblemDescription::component_names; - data_out.build_patches(discretization.mapping, - discretization.finite_element.degree - 1); + for (unsigned int i = 0; i < problem_dimension; ++i) + data_out.add_data_vector(output_vector[i], component_names[i]); + + data_out.add_data_vector(schlieren_postprocessor.schlieren, + "schlieren_plot"); + + data_out.build_patches(discretization.mapping, + discretization.finite_element.degree - 1); + + DataOutBase::VtkFlags flags(t, + cycle, + true, + DataOutBase::VtkFlags::best_speed); + data_out.set_flags(flags); + + data_out.write_vtu_with_pvtu_record("", name, cycle, mpi_communicator, 6); + }; - DataOutBase::VtkFlags flags(t, - cycle, - true, - DataOutBase::VtkFlags::best_speed); - data_out.set_flags(flags); + // We launch the thread by creating a + // std::thread + // object from the lambda function and moving it into the + // output_thread thread object. At this point we can + // return from the output() function and resume with the + // time stepping in the main loop - the thread will run in the + // background. - data_out.write_vtu_with_pvtu_record("", name, cycle, mpi_communicator, 6); + output_thread = std::move(std::thread(output_worker)); } } // namespace Step69 -- 2.39.5