[Service] fix use-after-free on training offloading destroy

c/src/ml-api-service-training-offloading.c

@@ -911,12 +911,12 @@ _ml_service_training_offloading_destroy (ml_service_s * mls)
     training_s->transfer_data_table = NULL;
   }
 
-  if (training_s->node_table) {
-    g_hash_table_destroy (training_s->node_table);
-    training_s->node_table = NULL;
-  }
-
+  /* Stop the pipeline before releasing the node info the sink callback uses. */
   if (training_s->pipeline_h) {
+    if (ml_pipeline_stop (training_s->pipeline_h) != ML_ERROR_NONE) {
+      _ml_error_report ("Failed to stop ml pipeline, destroy it anyway.");
+    }
+
     ret = ml_pipeline_destroy (training_s->pipeline_h);
     if (ret != ML_ERROR_NONE) {
       _ml_error_report ("Failed to destroy ml pipeline, clear handle anyway.");
@@ -925,6 +925,11 @@ _ml_service_training_offloading_destroy (ml_service_s * mls)
     training_s->pipeline_h = NULL;
   }
 
+  if (training_s->node_table) {
+    g_hash_table_destroy (training_s->node_table);
+    training_s->node_table = NULL;
+  }
+
   g_clear_pointer (&training_s->path, g_free);
   g_clear_pointer (&training_s->trained_model_path, g_free);
   g_clear_pointer (&training_s->receiver_pipe_json_str, g_free);

tests/capi/unittest_capi_service_training_offloading.cc

@@ -8,10 +8,15 @@
  */
 
 #include <gtest/gtest.h>
+#include <glib/gstdio.h>
+#ifdef __GLIBC__
+#include <malloc.h>
+#endif
 #include <ml-api-inference-pipeline-internal.h>
 #include <ml-api-internal.h>
 #include <ml-api-service-private.h>
 #include <ml-api-service.h>
+#include <nnstreamer-edge.h>
 
 #include "ml-api-service-offloading.h"
 #include "ml-api-service-training-offloading.h"
@@ -332,6 +337,187 @@ TEST_F (MLServiceTrainingOffloading, create_p)
   EXPECT_EQ (g_remove (receiver_config), 0);
 }
 
+/**
+ * @brief Pipeline the receiver would normally get from the remote sender.
+ * It is self-contained on purpose: the teardown order, not the training
+ * framework, is under test here.
+ */
+static const gchar *receiver_pipe_json
+    = R"JSON({"pipeline":{"description":"videotestsrc is-live=true ! videoconvert ! video/x-raw,format=RGB,width=16,height=16,framerate=30/1 ! tensor_converter ! tensor_sink name=training_result async=false","output_node":[{"name":"training_result"}]}})JSON";
+
+/**
+ * @brief Time the sink callback stays in the pipeline, in microseconds.
+ */
+#define SINK_CB_HOLD_TIME (300000)
+
+/**
+ * @brief Handshake between the sink callback and the thread tearing the
+ * service down.
+ */
+typedef struct {
+  GMutex lock;
+  GCond cond;
+  gboolean entered;
+} sink_hold_s;
+
+/**
+ * @brief Callback holding the streaming thread while the service is destroyed.
+ *
+ * The 'name' of the event data is the name of the node info owned by the
+ * training offloading handle, handed over without a copy. Holding the callback
+ * makes the teardown overlap with it, so reading the name afterwards fails if
+ * the node info was released while the pipeline was still running.
+ */
+static void
+_hold_new_data_cb (ml_service_event_e event, ml_information_h event_data, void *user_data)
+{
+  sink_hold_s *hold = (sink_hold_s *) user_data;
+  char *node_name = NULL;
+
+  if (event != ML_SERVICE_EVENT_NEW_DATA)
+    return;
+
+  g_mutex_lock (&hold->lock);
+  hold->entered = TRUE;
+  g_cond_broadcast (&hold->cond);
+  g_mutex_unlock (&hold->lock);
+
+  g_usleep (SINK_CB_HOLD_TIME);
+
+  EXPECT_EQ (ml_information_get (event_data, "name", (void **) &node_name), ML_ERROR_NONE);
+  EXPECT_STREQ (node_name, "training_result");
+}
+
+/**
+ * @brief Bring a receiver service up to the point where the sink callback has
+ * entered and is holding the streaming thread.
+ */
+static void
+_start_receiver_pipeline (ml_service_h receiver_h, const gchar *path, sink_hold_s *hold)
+{
+  ml_service_s *mls = (ml_service_s *) receiver_h;
+  nns_edge_data_h data_h = NULL;
+  gint64 deadline;
+  gboolean entered;
+  int status;
+
+  status = _ml_service_training_offloading_set_path (mls, path);
+  ASSERT_EQ (status, ML_ERROR_NONE);
+
+  status = ml_service_set_event_cb (receiver_h, _hold_new_data_cb, hold);
+  ASSERT_EQ (status, ML_ERROR_NONE);
+
+  ASSERT_EQ (nns_edge_data_create (&data_h), NNS_EDGE_ERROR_NONE);
+  status = _ml_service_training_offloading_process_received_data (mls, data_h,
+      path, receiver_pipe_json, ML_SERVICE_OFFLOADING_TYPE_PIPELINE_RAW);
+  nns_edge_data_destroy (data_h);
+  ASSERT_EQ (status, ML_ERROR_NONE);
+
+  status = _ml_service_training_offloading_start (mls);
+  ASSERT_EQ (status, ML_ERROR_NONE);
+
+  deadline = g_get_monotonic_time () + 10 * G_TIME_SPAN_SECOND;
+
+  g_mutex_lock (&hold->lock);
+  while (!hold->entered) {
+    if (!g_cond_wait_until (&hold->cond, &hold->lock, deadline))
+      break;
+  }
+  entered = hold->entered;
+  g_mutex_unlock (&hold->lock);
+
+  ASSERT_TRUE (entered);
+}
+
+/**
+ * @brief Destroying a running service must not release the node info the sink
+ * callback is still using.
+ */
+TEST_F (MLServiceTrainingOffloading, destroyWhileRunning_p)
+{
+  int status;
+  sink_hold_s hold = {};
+  ml_service_h receiver_h = NULL;
+  g_autofree gchar *path = g_dir_make_tmp ("ml-training-offloading-XXXXXX", NULL);
+
+  ASSERT_NE (nullptr, path);
+
+  guint avail_port = get_available_port ();
+  g_autofree gchar *receiver_config
+      = prepare_test_config ("training_offloading_receiver.conf", avail_port);
+
+  g_mutex_init (&hold.lock);
+  g_cond_init (&hold.cond);
+
+  status = ml_service_new (receiver_config, &receiver_h);
+  ASSERT_EQ (status, ML_ERROR_NONE);
+
+  _start_receiver_pipeline (receiver_h, path, &hold);
+
+#ifdef __GLIBC__
+  /**
+   * Scrub memory as it is released, so that a node name read after the node
+   * table is gone is caught rather than read back intact. glibc skips this for
+   * a chunk that fits the tcache, which the node name normally does; there the
+   * detection instead comes from tcache_put() writing its own link fields over
+   * the first 16 bytes of the chunk. Neither is a property of the code under
+   * test, so a sanitizer build remains the only airtight net for this.
+   */
+  mallopt (M_PERTURB, 0xAA);
+#endif
+
+  /* Destroy without stopping first. */
+  status = ml_service_destroy (receiver_h);
+  EXPECT_EQ (ML_ERROR_NONE, status);
+
+#ifdef __GLIBC__
+  mallopt (M_PERTURB, 0);
+#endif
+
+  g_mutex_clear (&hold.lock);
+  g_cond_clear (&hold.cond);
+
+  EXPECT_EQ (g_remove (receiver_config), 0);
+  EXPECT_EQ (g_rmdir (path), 0);
+}
+
+/**
+ * @brief Stopping the service before destroying it keeps working.
+ */
+TEST_F (MLServiceTrainingOffloading, destroyAfterStop_p)
+{
+  int status;
+  sink_hold_s hold = {};
+  ml_service_h receiver_h = NULL;
+  g_autofree gchar *path = g_dir_make_tmp ("ml-training-offloading-XXXXXX", NULL);
+
+  ASSERT_NE (nullptr, path);
+
+  guint avail_port = get_available_port ();
+  g_autofree gchar *receiver_config
+      = prepare_test_config ("training_offloading_receiver.conf", avail_port);
+
+  g_mutex_init (&hold.lock);
+  g_cond_init (&hold.cond);
+
+  status = ml_service_new (receiver_config, &receiver_h);
+  ASSERT_EQ (status, ML_ERROR_NONE);
+
+  _start_receiver_pipeline (receiver_h, path, &hold);
+
+  status = ml_service_stop (receiver_h);
+  EXPECT_EQ (ML_ERROR_NONE, status);
+
+  status = ml_service_destroy (receiver_h);
+  EXPECT_EQ (ML_ERROR_NONE, status);
+
+  g_mutex_clear (&hold.lock);
+  g_cond_clear (&hold.cond);
+
+  EXPECT_EQ (g_remove (receiver_config), 0);
+  EXPECT_EQ (g_rmdir (path), 0);
+}
+
 /**
  * @brief Test _ml_service_training_offloading_destroy.
  */
@@ -343,6 +529,48 @@ TEST_F (MLServiceTrainingOffloading, destroyInvalidParam1_n)
   EXPECT_EQ (ML_ERROR_INVALID_PARAMETER, status);
 }
 
+/**
+ * @brief Test _ml_service_training_offloading_destroy with a service that is
+ * not in training mode.
+ */
+TEST_F (MLServiceTrainingOffloading, destroyInvalidParam2_n)
+{
+  int status;
+  ml_service_s *mls;
+
+  g_autoptr (JsonParser) parser = NULL;
+  g_autofree gchar *json_string = NULL;
+  JsonNode *root;
+  JsonObject *object;
+
+  guint avail_port = get_available_port ();
+  g_autofree gchar *receiver_config
+      = prepare_test_config ("service_offloading_receiver.conf", avail_port);
+
+  ASSERT_TRUE (g_file_get_contents (receiver_config, &json_string, NULL, NULL));
+  parser = json_parser_new ();
+  ASSERT_TRUE (json_parser_load_from_data (parser, json_string, -1, NULL));
+  root = json_parser_get_root (parser);
+  ASSERT_NE (nullptr, root);
+  object = json_node_get_object (root);
+  ASSERT_NE (nullptr, object);
+
+  mls = _ml_service_create_internal (ML_SERVICE_TYPE_OFFLOADING);
+  ASSERT_NE (nullptr, mls);
+
+  /* The configuration has no 'training' member, so the mode stays NONE. */
+  status = _ml_service_offloading_create (mls, object);
+  EXPECT_EQ (ML_ERROR_NONE, status);
+
+  status = _ml_service_training_offloading_destroy (mls);
+  EXPECT_EQ (ML_ERROR_INVALID_PARAMETER, status);
+
+  status = _ml_service_destroy_internal (mls);
+  EXPECT_EQ (ML_ERROR_NONE, status);
+
+  EXPECT_EQ (g_remove (receiver_config), 0);
+}
+
 /**
  * @brief Test _ml_service_training_offloading_set_path.
  */