move commit end log before teardown submission

project-tsurugi/tsurugi-issues#801

src/jogasaki/commit_common.cpp

@@ -0,0 +1,57 @@
+/*
+ * Copyright 2018-2023 Project Tsurugi.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+#include "commit_common.h"
+
+#include <atomic>
+#include <memory>
+
+#include <jogasaki/commit_response.h>
+#include <jogasaki/external_log/event_logging.h>
+#include <jogasaki/logging.h>
+#include <jogasaki/logging_helper.h>
+#include <jogasaki/request_context.h>
+#include <jogasaki/utils/external_log_utils.h>
+#include <jogasaki/utils/hex.h>
+
+namespace jogasaki {
+
+void log_commit_end(request_context& rctx) {
+    auto txid = rctx.transaction()->transaction_id();
+    auto jobid = rctx.job()->id();
+    VLOG(log_debug_timing_event) << "/:jogasaki:timing:committed "
+        << txid
+        << " job_id:"
+        << utils::hex(jobid);
+    VLOG(log_debug_timing_event) << "/:jogasaki:timing:transaction:finished "
+        << txid
+        << " status:"
+        << (rctx.status_code() == status::ok ? "committed" : "aborted");
+    rctx.transaction()->profile()->set_commit_job_completed();
+    auto tx_type = utils::tx_type_from(*rctx.transaction());
+    auto result = utils::result_from(rctx.status_code());
+    rctx.transaction()->end_time(transaction_context::clock::now());
+    external_log::tx_end(
+        rctx.req_info(),
+        "",
+        txid,
+        tx_type,
+        result,
+        rctx.transaction()->duration<std::chrono::nanoseconds>().count(),
+        rctx.transaction()->label()
+    );
+}
+
+}  // namespace jogasaki

src/jogasaki/commit_common.h

@@ -0,0 +1,32 @@
+/*
+ * Copyright 2018-2023 Project Tsurugi.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+#pragma once
+
+#include <atomic>
+#include <memory>
+
+#include <jogasaki/request_context.h>
+#include <jogasaki/commit_response.h>
+
+namespace jogasaki {
+
+/**
+ * @brief log commit end event
+ * @param rctx the request context
+ */
+void log_commit_end(request_context& rctx);
+
+}  // namespace jogasaki

src/jogasaki/durability_common.cpp

@@ -27,6 +27,7 @@
 
 #include <jogasaki/api/impl/database.h>
 #include <jogasaki/api/impl/request_context_factory.h>
+#include <jogasaki/commit_common.h>
 #include <jogasaki/commit_profile.h>
 #include <jogasaki/configuration.h>
 #include <jogasaki/durability_manager.h>
@@ -53,6 +54,7 @@ void submit_commit_response(
     auto& ts = *rctx->scheduler();
     ts.schedule_task(
         scheduler::create_custom_task(rctx.get(), [rctx, kind, teardown_try_on_suspended_worker, is_error]() {
+            log_commit_end(*rctx);
             if(is_error) {
                 rctx->commit_ctx()->on_error()(kind, rctx->status_code(), rctx->error_info());
             } else {

src/jogasaki/executor/executor.cpp

@@ -41,6 +41,7 @@
 #include <jogasaki/api/parameter_set.h>
 #include <jogasaki/api/statement_handle.h>
 #include <jogasaki/api/transaction_handle.h>
+#include <jogasaki/commit_common.h>
 #include <jogasaki/commit_profile.h>
 #include <jogasaki/commit_response.h>
 #include <jogasaki/configuration.h>
@@ -732,6 +733,9 @@ void process_commit_callback(
         auto& ts = *rctx->scheduler();
         ts.schedule_task(
             scheduler::create_custom_task(rctx.get(), [rctx, last_less_equals_accepted]() {
+                if(last_less_equals_accepted) {
+                    log_commit_end(*rctx);
+                }
                 rctx->commit_ctx()->on_response()(commit_response_kind::accepted);
                 if(last_less_equals_accepted) {
                     // if the last response is accepted or requested, we can finish job and clean up resource here
@@ -811,28 +815,9 @@ scheduler::job_context::job_id_type commit_async(
             });
         return model::task_result::complete;
     }, true);
-    rctx->job()->callback([rctx, jobid, txid, req_info](){  // callback is copy-based
-        VLOG(log_debug_timing_event) << "/:jogasaki:timing:committed "
-            << txid
-            << " job_id:"
-            << utils::hex(jobid);
-        VLOG(log_debug_timing_event) << "/:jogasaki:timing:transaction:finished "
-            << txid
-            << " status:"
-            << (rctx->status_code() == status::ok ? "committed" : "aborted");
-        rctx->transaction()->profile()->set_commit_job_completed();
-        auto tx_type = utils::tx_type_from(*rctx->transaction());
-        auto result = utils::result_from(rctx->status_code());
-        rctx->transaction()->end_time(transaction_context::clock::now());
-        external_log::tx_end(
-            req_info,
-            "",
-            txid,
-            tx_type,
-            result,
-            rctx->transaction()->duration<std::chrono::nanoseconds>().count(),
-            rctx->transaction()->label()
-        );
+    rctx->job()->callback([rctx, txid, jobid](){
+        // no-op just log and keep rctx
+        VLOG_LP(log_trace) << "commit job end job_id:" << utils::hex(jobid) << " " << txid;
     });
     std::weak_ptr wrctx{rctx};
     rctx->job()->completion_readiness([wrctx=std::move(wrctx)]() {