refactor: make aio decoupled from dsn_runtime (#778)

src/aio/disk_engine.cpp

@@ -29,19 +29,25 @@
 #include <dsn/utility/flags.h>
 
 #include "disk_engine.h"
-#include "sim_aio_provider.h"
 #include "runtime/service_engine.h"
+#include "native_linux_aio_provider.h"
 
 using namespace dsn::utils;
 
 namespace dsn {
-using namespace aio;
-
 DEFINE_TASK_CODE_AIO(LPC_AIO_BATCH_WRITE, TASK_PRIORITY_COMMON, THREAD_POOL_DEFAULT)
 
 const char *native_aio_provider = "dsn::tools::native_aio_provider";
 DSN_REGISTER_COMPONENT_PROVIDER(native_linux_aio_provider, native_aio_provider);
-DSN_REGISTER_COMPONENT_PROVIDER(sim_aio_provider, "dsn::tools::sim_aio_provider");
+
+struct disk_engine_initializer
+{
+    disk_engine_initializer() { disk_engine::instance(); }
+};
+
+// make disk_engine destructed after service_engine, which is inited in dsn_global_init,
+// because service_engine relies on the former to close files.
+static disk_engine_initializer disk_engine_init;
 
 DSN_DEFINE_string("core",
                   aio_factory_name,
@@ -151,10 +157,7 @@ aio_task *disk_file::on_write_completed(aio_task *wk, void *ctx, error_code err,
 disk_engine::disk_engine()
 {
     aio_provider *provider = utils::factory_store<aio_provider>::create(
-        FLAGS_aio_factory_name, dsn::PROVIDER_TYPE_MAIN, this);
-    if (nullptr == provider) {
-        dassert_f(false, "The config value of aio_factory_name is invalid");
-    }
+        native_aio_provider, dsn::PROVIDER_TYPE_MAIN, this);
     _provider.reset(provider);
 }
 

src/aio/native_linux_aio_provider.cpp

@@ -25,8 +25,10 @@
  */
 
 #include "native_linux_aio_provider.h"
+#include "runtime/service_engine.h"
 
 #include <dsn/tool-api/async_calls.h>
+#include <dsn/c/api_utilities.h>
 
 namespace dsn {
 
@@ -96,12 +98,17 @@ error_code native_linux_aio_provider::read(const aio_context &aio_ctx,
 
 void native_linux_aio_provider::submit_aio_task(aio_task *aio_tsk)
 {
+    // for the tests which use simulator need sync submit for aio
+    if (dsn_unlikely(service_engine::instance().is_simulator())) {
+        aio_internal(aio_tsk);
+        return;
+    }
+
     tasking::enqueue(
         aio_tsk->code(), aio_tsk->tracker(), [=]() { aio_internal(aio_tsk); }, aio_tsk->hash());
 }
 
-error_code native_linux_aio_provider::aio_internal(aio_task *aio_tsk,
-                                                   /*out*/ uint32_t *pbytes /*= nullptr*/)
+error_code native_linux_aio_provider::aio_internal(aio_task *aio_tsk)
 {
     aio_context *aio_ctx = aio_tsk->get_aio_context();
     error_code err = ERR_UNKNOWN;
@@ -117,10 +124,6 @@ error_code native_linux_aio_provider::aio_internal(aio_task *aio_tsk,
         return err;
     }
 
-    if (pbytes) {
-        *pbytes = processed_bytes;
-    }
-
     complete_io(aio_tsk, err, processed_bytes);
     return err;
 }

src/aio/native_linux_aio_provider.h

@@ -46,7 +46,7 @@ class native_linux_aio_provider : public aio_provider
     aio_context *prepare_aio_context(aio_task *tsk) override { return new aio_context; }
 
 protected:
-    error_code aio_internal(aio_task *aio, /*out*/ uint32_t *pbytes = nullptr);
+    error_code aio_internal(aio_task *aio);
 };
 
 } // namespace dsn

src/runtime/CMakeLists.txt

@@ -30,5 +30,5 @@ add_library(dsn_runtime STATIC
         tracer.cpp
         zlocks.cpp
         )
-target_link_libraries(dsn_runtime PRIVATE dsn_utils sasl2 gssapi_krb5 krb5 dsn_aio)
+target_link_libraries(dsn_runtime PRIVATE dsn_utils sasl2 gssapi_krb5 krb5)
 install(TARGETS dsn_runtime DESTINATION "lib")

src/runtime/service_api_c.cpp

@@ -24,7 +24,6 @@
  * THE SOFTWARE.
  */
 
-#include "aio/disk_engine.h"
 #include "service_engine.h"
 #include "utils/coredump.h"
 #include "runtime/rpc/rpc_engine.h"
@@ -292,11 +291,10 @@ extern void dsn_core_init();
 
 inline void dsn_global_init()
 {
-    // make perf_counters/disk_engine destructed after service_engine,
+    // make perf_counters destructed after service_engine,
     // because service_engine relies on the former to monitor
-    // task queues length and close files.
+    // task queues length.
     dsn::perf_counters::instance();
-    dsn::disk_engine::instance();
     dsn::service_engine::instance();
 }
 

src/runtime/service_engine.cpp

@@ -284,4 +284,8 @@ std::string service_engine::get_queue_info(const std::vector<std::string> &args)
     return ss.str();
 }
 
+bool service_engine::is_simulator() const { return _simulator; }
+
+void service_engine::set_simulator() { _simulator = true; }
+
 } // namespace dsn

src/runtime/service_engine.h

@@ -123,6 +123,8 @@ class service_engine : public utils::singleton<service_engine>
 
     void start_node(service_app_spec &app_spec);
     const service_nodes_by_app_id &get_all_nodes() const { return _nodes_by_app_id; }
+    bool is_simulator() const;
+    void set_simulator();
 
 private:
     service_spec _spec;
@@ -131,6 +133,8 @@ class service_engine : public utils::singleton<service_engine>
     dsn_handle_t _get_runtime_info_cmd;
     dsn_handle_t _get_queue_info_cmd;
 
+    bool _simulator;
+
     // <port, servicenode>
     typedef std::map<int, service_node *>
         node_engines_by_port; // multiple ports may share the same node

src/runtime/simulator.cpp

@@ -35,6 +35,7 @@
 
 #include <dsn/tool/simulator.h>
 #include "scheduler.h"
+#include "service_engine.h"
 
 #include "env.sim.h"
 #include "runtime/task/task_engine.sim.h"
@@ -114,6 +115,8 @@ void simulator::install(service_spec &spec)
 
     // the new sim_clock is taken over by unique_ptr in clock instance
     utils::clock::instance()->mock(new sim_clock());
+
+    service_engine::instance().set_simulator();
 }
 
 void simulator::on_system_exit(sys_exit_type st)