diff --git a/c/src/ml-api-service-training-offloading.c b/c/src/ml-api-service-training-offloading.c index c8a8c1bb..ec856941 100644 --- a/c/src/ml-api-service-training-offloading.c +++ b/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); diff --git a/tests/capi/unittest_capi_service_training_offloading.cc b/tests/capi/unittest_capi_service_training_offloading.cc index db6968cc..2d2aa2dd 100644 --- a/tests/capi/unittest_capi_service_training_offloading.cc +++ b/tests/capi/unittest_capi_service_training_offloading.cc @@ -8,10 +8,15 @@ */ #include +#include +#ifdef __GLIBC__ +#include +#endif #include #include #include #include +#include #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. */