Line data Source code
1 : /* SPDX-License-Identifier: Apache-2.0 */
2 : /**
3 : * Copyright (c) 2023 Samsung Electronics Co., Ltd. All Rights Reserved.
4 : *
5 : * @file ml-api-service-extension.c
6 : * @date 1 September 2023
7 : * @brief ML service extension C-API.
8 : * @see https://github.com/nnstreamer/api
9 : * @author Jaeyun Jung <jy1210.jung@samsung.com>
10 : * @bug No known bugs except for NYI items
11 : */
12 :
13 : #include "ml-api-service-extension.h"
14 :
15 : /**
16 : * @brief The time to wait for new input data in message thread, in millisecond.
17 : */
18 : #define DEFAULT_TIMEOUT 200
19 :
20 : /**
21 : * @brief The max number of input data in message queue (0 for no limit).
22 : */
23 : #define DEFAULT_MAX_INPUT 5
24 :
25 : /**
26 : * @brief Internal enumeration for ml-service extension types.
27 : */
28 : typedef enum
29 : {
30 : ML_EXTENSION_TYPE_UNKNOWN = 0,
31 : ML_EXTENSION_TYPE_SINGLE = 1,
32 : ML_EXTENSION_TYPE_PIPELINE = 2,
33 :
34 : ML_EXTENSION_TYPE_MAX
35 : } ml_extension_type_e;
36 :
37 : /**
38 : * @brief Internal structure of the message in ml-service extension handle.
39 : */
40 : typedef struct
41 : {
42 : gchar *name;
43 : ml_tensors_data_h input;
44 : ml_tensors_data_h output;
45 : } ml_extension_msg_s;
46 :
47 : /**
48 : * @brief Internal structure for ml-service extension handle.
49 : */
50 : typedef struct
51 : {
52 : ml_extension_type_e type;
53 : gboolean running;
54 : guint timeout; /**< The time to wait for new input data in message thread, in millisecond (see DEFAULT_TIMEOUT). */
55 : guint max_input; /**< The max number of input data in message queue (see DEFAULT_MAX_INPUT). */
56 : GThread *msg_thread;
57 : GAsyncQueue *msg_queue;
58 :
59 : /**
60 : * Handles for each ml-service extension type.
61 : * - single : Default. Open model file and prepare invoke. The configuration should include model information.
62 : * - pipeline : Construct a pipeline from configuration. The configuration should include pipeline description.
63 : */
64 : ml_single_h single;
65 :
66 : ml_pipeline_h pipeline;
67 : GHashTable *node_table;
68 : } ml_extension_s;
69 :
70 : /**
71 : * @brief Internal function for message callback.
72 : */
73 : static void
74 0 : _ml_extension_message_cb (const char *type, const char *message, void *user_data)
75 : {
76 0 : ml_service_s *mls = (ml_service_s *) user_data;
77 :
78 0 : _ml_service_invoke_event_message (mls, type, message);
79 0 : }
80 :
81 : /**
82 : * @brief Internal function to handle the asynchronous invoke.
83 : */
84 : static int
85 0 : _ml_extension_async_cb (const ml_tensors_data_h data, void *user_data)
86 : {
87 0 : ml_service_s *mls = (ml_service_s *) user_data;
88 :
89 0 : return _ml_service_invoke_event_new_data (mls, NULL, data);
90 : }
91 :
92 : /**
93 : * @brief Internal function to create node info in pipeline.
94 : */
95 : static ml_service_node_info_s *
96 0 : _ml_extension_node_info_new (ml_service_s * mls, const gchar * name,
97 : ml_service_node_type_e type)
98 : {
99 0 : ml_extension_s *ext = (ml_extension_s *) mls->priv;
100 : ml_service_node_info_s *node_info;
101 :
102 0 : if (!STR_IS_VALID (name)) {
103 0 : _ml_error_report_return (NULL,
104 : "Cannot add new node info, invalid node name '%s'.", name);
105 : }
106 :
107 0 : if (g_hash_table_lookup (ext->node_table, name)) {
108 0 : _ml_error_report_return (NULL,
109 : "Cannot add duplicated node '%s' in ml-service pipeline.", name);
110 : }
111 :
112 0 : node_info = g_try_new0 (ml_service_node_info_s, 1);
113 0 : if (!node_info) {
114 0 : _ml_error_report_return (NULL,
115 : "Failed to allocate new memory for node info in ml-service pipeline. Out of memory?");
116 : }
117 :
118 0 : node_info->name = g_strdup (name);
119 0 : node_info->type = type;
120 0 : node_info->mls = mls;
121 :
122 0 : g_hash_table_insert (ext->node_table, g_strdup (name), node_info);
123 :
124 0 : return node_info;
125 : }
126 :
127 : /**
128 : * @brief Internal function to release pipeline node info.
129 : */
130 : static void
131 0 : _ml_extension_node_info_free (gpointer data)
132 : {
133 0 : ml_service_node_info_s *node_info = (ml_service_node_info_s *) data;
134 :
135 0 : if (!node_info)
136 0 : return;
137 :
138 0 : if (node_info->info)
139 0 : ml_tensors_info_destroy (node_info->info);
140 :
141 0 : g_clear_pointer (&node_info->name, g_free);
142 0 : g_free (node_info);
143 : }
144 :
145 : /**
146 : * @brief Internal function to get the node info in ml-service extension.
147 : */
148 : static ml_service_node_info_s *
149 0 : _ml_extension_node_info_get (ml_extension_s * ext, const gchar * name)
150 : {
151 0 : if (!STR_IS_VALID (name))
152 0 : return NULL;
153 :
154 0 : return g_hash_table_lookup (ext->node_table, name);
155 : }
156 :
157 : /**
158 : * @brief Internal function to release ml-service extension message.
159 : */
160 : static void
161 0 : _ml_extension_msg_free (gpointer data)
162 : {
163 0 : ml_extension_msg_s *msg = (ml_extension_msg_s *) data;
164 :
165 0 : if (!msg)
166 0 : return;
167 :
168 0 : if (msg->input)
169 0 : ml_tensors_data_destroy (msg->input);
170 0 : if (msg->output)
171 0 : ml_tensors_data_destroy (msg->output);
172 0 : g_clear_pointer (&msg->name, g_free);
173 :
174 0 : g_free (msg);
175 : }
176 :
177 : /**
178 : * @brief Internal function to process ml-service extension message.
179 : */
180 : static gpointer
181 0 : _ml_extension_msg_thread (gpointer data)
182 : {
183 0 : ml_service_s *mls = (ml_service_s *) data;
184 0 : ml_extension_s *ext = (ml_extension_s *) mls->priv;
185 : gchar *errmsg;
186 : int status;
187 :
188 0 : g_mutex_lock (&mls->lock);
189 0 : ext->running = TRUE;
190 0 : g_cond_signal (&mls->cond);
191 0 : g_mutex_unlock (&mls->lock);
192 :
193 0 : while (ext->running) {
194 : ml_extension_msg_s *msg;
195 :
196 0 : msg = g_async_queue_timeout_pop (ext->msg_queue,
197 0 : ext->timeout * G_TIME_SPAN_MILLISECOND);
198 :
199 0 : if (msg) {
200 0 : switch (ext->type) {
201 0 : case ML_EXTENSION_TYPE_SINGLE:
202 : {
203 0 : status = ml_single_invoke (ext->single, msg->input, &msg->output);
204 :
205 0 : if (status == ML_ERROR_NONE) {
206 0 : _ml_service_invoke_event_new_data (mls, NULL, msg->output);
207 : } else {
208 0 : errmsg = g_strdup ("Failed to invoke the model (single-shot) in ml-service extension thread.");
209 :
210 0 : _ml_error_report ("%s", errmsg);
211 0 : _ml_service_invoke_event_message (mls, "invoke-failure", errmsg);
212 0 : g_free (errmsg);
213 : }
214 0 : break;
215 : }
216 0 : case ML_EXTENSION_TYPE_PIPELINE:
217 : {
218 : ml_service_node_info_s *node_info;
219 :
220 0 : node_info = _ml_extension_node_info_get (ext, msg->name);
221 :
222 0 : if (node_info && node_info->type == ML_SERVICE_NODE_TYPE_INPUT) {
223 : /* The input data will be released in the pipeline. */
224 0 : status = ml_pipeline_src_input_data (node_info->handle, msg->input,
225 : ML_PIPELINE_BUF_POLICY_AUTO_FREE);
226 0 : msg->input = NULL;
227 :
228 0 : if (status != ML_ERROR_NONE) {
229 0 : errmsg = g_strdup_printf ("Failed to push input data into the input node '%s' in ml-service extension thread.", msg->name);
230 :
231 0 : _ml_error_report ("%s", errmsg);
232 0 : _ml_service_invoke_event_message (mls, "push-failure", errmsg);
233 0 : g_free (errmsg);
234 : }
235 : } else {
236 0 : errmsg = g_strdup_printf ("Failed to push input data into the pipeline, cannot find input node '%s'.", msg->name);
237 :
238 0 : _ml_error_report ("%s", errmsg);
239 0 : _ml_service_invoke_event_message (mls, "push-failure", errmsg);
240 0 : g_free (errmsg);
241 : }
242 0 : break;
243 : }
244 0 : default:
245 : /* Unknown ml-service extension type, skip this. */
246 0 : break;
247 : }
248 :
249 0 : _ml_extension_msg_free (msg);
250 : }
251 : }
252 :
253 0 : return NULL;
254 : }
255 :
256 : /**
257 : * @brief Wrapper to release tensors-info handle.
258 : */
259 : static void
260 0 : _ml_extension_destroy_tensors_info (void *data)
261 : {
262 0 : ml_tensors_info_h info = (ml_tensors_info_h) data;
263 :
264 0 : if (info)
265 0 : ml_tensors_info_destroy (info);
266 0 : }
267 :
268 : /**
269 : * @brief Internal function to parse common option from json.
270 : */
271 : static void
272 0 : _ml_extension_conf_parse_common (ml_service_s * mls, JsonObject * object)
273 : {
274 0 : const gchar *value = NULL;
275 :
276 0 : g_return_if_fail (object != NULL);
277 :
278 0 : if (json_object_has_member (object, "input_queue_size")) {
279 0 : value = json_object_get_string_member (object, "input_queue_size");
280 :
281 0 : if (STR_IS_VALID (value))
282 0 : _ml_service_extension_set_information (mls, "input_queue_size", value);
283 : }
284 :
285 0 : if (json_object_has_member (object, "max_input")) {
286 0 : value = json_object_get_string_member (object, "max_input");
287 :
288 0 : if (STR_IS_VALID (value))
289 0 : _ml_service_extension_set_information (mls, "max_input", value);
290 : }
291 :
292 0 : if (json_object_has_member (object, "timeout")) {
293 0 : value = json_object_get_string_member (object, "timeout");
294 :
295 0 : if (STR_IS_VALID (value))
296 0 : _ml_service_extension_set_information (mls, "timeout", value);
297 : }
298 : }
299 :
300 : /**
301 : * @brief Internal function to parse single-shot info from json.
302 : */
303 : static int
304 0 : _ml_extension_conf_parse_single (ml_service_s * mls, JsonObject * single)
305 : {
306 0 : ml_extension_s *ext = (ml_extension_s *) mls->priv;
307 : ml_option_h option;
308 : int status;
309 :
310 0 : status = ml_option_create (&option);
311 0 : if (status != ML_ERROR_NONE) {
312 0 : _ml_error_report_return (status,
313 : "Failed to parse configuration file, cannot create ml-option handle.");
314 : }
315 :
316 : /**
317 : * 1. "key" : load model info from ml-service agent.
318 : * 2. "model" : configuration file includes model path.
319 : */
320 0 : if (json_object_has_member (single, "key")) {
321 0 : const gchar *key = json_object_get_string_member (single, "key");
322 :
323 0 : if (STR_IS_VALID (key)) {
324 : ml_information_h model_info;
325 :
326 0 : status = ml_service_model_get_activated (key, &model_info);
327 0 : if (status == ML_ERROR_NONE) {
328 0 : gchar *paths = NULL;
329 :
330 : /** @todo parse desc and other information if necessary. */
331 0 : ml_information_get (model_info, "path", (void **) (&paths));
332 0 : ml_option_set (option, "models", g_strdup (paths), g_free);
333 :
334 0 : ml_information_destroy (model_info);
335 : } else {
336 0 : _ml_error_report
337 : ("Failed to parse configuration file, cannot get the model of '%s'.",
338 : key);
339 0 : goto error;
340 : }
341 : }
342 0 : } else if (json_object_has_member (single, "model")) {
343 0 : JsonNode *file_node = json_object_get_member (single, "model");
344 0 : gchar *paths = NULL;
345 :
346 0 : status = _ml_service_conf_parse_string (file_node, ",", &paths);
347 0 : if (status != ML_ERROR_NONE) {
348 0 : _ml_error_report
349 : ("Failed to parse configuration file, it should have valid model path.");
350 0 : goto error;
351 : }
352 :
353 0 : ml_option_set (option, "models", paths, g_free);
354 : } else {
355 0 : status = ML_ERROR_INVALID_PARAMETER;
356 0 : _ml_error_report
357 : ("Failed to parse configuration file, cannot get the model path.");
358 0 : goto error;
359 : }
360 :
361 0 : if (json_object_has_member (single, "framework")) {
362 0 : const gchar *fw = json_object_get_string_member (single, "framework");
363 :
364 0 : if (STR_IS_VALID (fw))
365 0 : ml_option_set (option, "framework_name", g_strdup (fw), g_free);
366 : }
367 :
368 0 : if (json_object_has_member (single, "input_info")) {
369 0 : JsonNode *info_node = json_object_get_member (single, "input_info");
370 : ml_tensors_info_h in_info;
371 :
372 0 : status = _ml_service_conf_parse_tensors_info (info_node, &in_info);
373 0 : if (status != ML_ERROR_NONE) {
374 0 : _ml_error_report
375 : ("Failed to parse configuration file, cannot parse input information.");
376 0 : goto error;
377 : }
378 :
379 0 : ml_option_set (option, "input_info", in_info,
380 : _ml_extension_destroy_tensors_info);
381 : }
382 :
383 0 : if (json_object_has_member (single, "output_info")) {
384 0 : JsonNode *info_node = json_object_get_member (single, "output_info");
385 : ml_tensors_info_h out_info;
386 :
387 0 : status = _ml_service_conf_parse_tensors_info (info_node, &out_info);
388 0 : if (status != ML_ERROR_NONE) {
389 0 : _ml_error_report
390 : ("Failed to parse configuration file, cannot parse output information.");
391 0 : goto error;
392 : }
393 :
394 0 : ml_option_set (option, "output_info", out_info,
395 : _ml_extension_destroy_tensors_info);
396 : }
397 :
398 : /* parse latency profiling option - "profile": "true" or "1" */
399 0 : if (json_object_has_member (single, "profile")) {
400 0 : const gchar *profile = json_object_get_string_member (single, "profile");
401 :
402 0 : if (STR_IS_VALID (profile))
403 0 : ml_option_set (option, "profile", g_strdup (profile), g_free);
404 : }
405 :
406 : /* parse latency profiling option - "latency": "true" or "1" */
407 0 : if (json_object_has_member (single, "latency")) {
408 0 : const gchar *latency = json_object_get_string_member (single, "latency");
409 :
410 0 : if (STR_IS_VALID (latency))
411 0 : ml_option_set (option, "profile", g_strdup (latency), g_free);
412 : }
413 :
414 0 : if (json_object_has_member (single, "custom")) {
415 0 : const gchar *custom = json_object_get_string_member (single, "custom");
416 :
417 0 : if (STR_IS_VALID (custom))
418 0 : ml_option_set (option, "custom", g_strdup (custom), g_free);
419 : }
420 :
421 0 : if (json_object_has_member (single, "invoke_dynamic")) {
422 : const gchar *invoke_dynamic =
423 0 : json_object_get_string_member (single, "invoke_dynamic");
424 :
425 0 : if (STR_IS_VALID (invoke_dynamic)) {
426 0 : ml_option_set (option, "invoke_dynamic", g_strdup (invoke_dynamic),
427 : g_free);
428 : }
429 : }
430 :
431 0 : if (json_object_has_member (single, "invoke_async")) {
432 : const gchar *invoke_async =
433 0 : json_object_get_string_member (single, "invoke_async");
434 :
435 0 : if (STR_IS_VALID (invoke_async)) {
436 0 : ml_option_set (option, "invoke_async", g_strdup (invoke_async), g_free);
437 :
438 0 : if (g_ascii_strcasecmp (invoke_async, "true") == 0) {
439 0 : ml_option_set (option, "async_callback", _ml_extension_async_cb, NULL);
440 0 : ml_option_set (option, "async_data", mls, NULL);
441 : }
442 : }
443 : }
444 :
445 0 : error:
446 0 : if (status == ML_ERROR_NONE)
447 0 : status = ml_single_open_with_option (&ext->single, option);
448 :
449 0 : ml_option_destroy (option);
450 0 : return status;
451 : }
452 :
453 : /**
454 : * @brief Internal function to parse the node info in pipeline.
455 : */
456 : static int
457 0 : _ml_extension_conf_parse_pipeline_node (ml_service_s * mls, JsonNode * node,
458 : ml_service_node_type_e type)
459 : {
460 0 : ml_extension_s *ext = (ml_extension_s *) mls->priv;
461 0 : JsonArray *array = NULL;
462 : JsonObject *object;
463 : guint i, n;
464 : int status;
465 :
466 0 : n = 1;
467 0 : if (JSON_NODE_HOLDS_ARRAY (node)) {
468 0 : array = json_node_get_array (node);
469 0 : n = json_array_get_length (array);
470 : }
471 :
472 0 : for (i = 0; i < n; i++) {
473 0 : const gchar *name = NULL;
474 : ml_service_node_info_s *node_info;
475 :
476 0 : if (array)
477 0 : object = json_array_get_object_element (array, i);
478 : else
479 0 : object = json_node_get_object (node);
480 :
481 0 : name = _ml_service_get_json_string_member (object, "name");
482 :
483 0 : node_info = _ml_extension_node_info_new (mls, name, type);
484 0 : if (!node_info) {
485 0 : _ml_error_report_return (ML_ERROR_INVALID_PARAMETER,
486 : "Failed to parse configuration file, cannot add new node information.");
487 : }
488 :
489 0 : if (json_object_has_member (object, "info")) {
490 0 : JsonNode *info_node = json_object_get_member (object, "info");
491 :
492 0 : status = _ml_service_conf_parse_tensors_info (info_node,
493 : &node_info->info);
494 0 : if (status != ML_ERROR_NONE) {
495 0 : _ml_error_report_return (status,
496 : "Failed to parse configuration file, cannot parse the information.");
497 : }
498 : } else {
499 0 : _ml_error_report_return (ML_ERROR_INVALID_PARAMETER,
500 : "Failed to parse configuration file, cannot find node information.");
501 : }
502 :
503 0 : switch (type) {
504 0 : case ML_SERVICE_NODE_TYPE_INPUT:
505 0 : status = ml_pipeline_src_get_handle (ext->pipeline, name,
506 0 : &node_info->handle);
507 0 : break;
508 0 : case ML_SERVICE_NODE_TYPE_OUTPUT:
509 0 : status = ml_pipeline_sink_register (ext->pipeline, name,
510 0 : _ml_service_pipeline_sink_cb, node_info, &node_info->handle);
511 0 : break;
512 0 : default:
513 0 : status = ML_ERROR_INVALID_PARAMETER;
514 0 : break;
515 : }
516 :
517 0 : if (status != ML_ERROR_NONE) {
518 0 : _ml_error_report_return (status,
519 : "Failed to parse configuration file, cannot get the handle for pipeline node.");
520 : }
521 : }
522 :
523 0 : return ML_ERROR_NONE;
524 : }
525 :
526 : /**
527 : * @brief Internal function to parse pipeline info from json.
528 : */
529 : static int
530 0 : _ml_extension_conf_parse_pipeline (ml_service_s * mls, JsonObject * pipe)
531 : {
532 0 : ml_pipeline_preset preset = { 0, };
533 0 : ml_extension_s *ext = (ml_extension_s *) mls->priv;
534 0 : g_autofree gchar *desc = NULL;
535 : int status;
536 :
537 : /**
538 : * 1. "key" : load pipeline from ml-service agent.
539 : * 2. "description" : configuration file includes pipeline description.
540 : */
541 0 : if (json_object_has_member (pipe, "key")) {
542 0 : const gchar *key = json_object_get_string_member (pipe, "key");
543 :
544 0 : if (STR_IS_VALID (key)) {
545 0 : status = ml_service_pipeline_get (key, &desc);
546 0 : if (status != ML_ERROR_NONE) {
547 0 : _ml_error_report_return (status,
548 : "Failed to parse configuration file, cannot get the pipeline of '%s'.",
549 : key);
550 : }
551 : }
552 0 : } else if (json_object_has_member (pipe, "description")) {
553 0 : desc = g_strdup (json_object_get_string_member (pipe, "description"));
554 : } else {
555 0 : _ml_error_report_return (ML_ERROR_INVALID_PARAMETER,
556 : "Failed to parse configuration file, cannot get the pipeline description.");
557 : }
558 :
559 0 : preset.description = desc;
560 0 : preset.message_cb.cb = _ml_extension_message_cb;
561 0 : preset.message_cb.user_data = mls;
562 0 : preset.is_internal = FALSE;
563 :
564 0 : status = _ml_pipeline_construct_custom (&preset, &ext->pipeline);
565 0 : if (status != ML_ERROR_NONE) {
566 0 : _ml_error_report_return (status,
567 : "Failed to parse configuration file, cannot construct the pipeline.");
568 : }
569 :
570 0 : if (json_object_has_member (pipe, "input_node")) {
571 0 : JsonNode *node = json_object_get_member (pipe, "input_node");
572 :
573 0 : status = _ml_extension_conf_parse_pipeline_node (mls, node,
574 : ML_SERVICE_NODE_TYPE_INPUT);
575 0 : if (status != ML_ERROR_NONE) {
576 0 : _ml_error_report_return (status,
577 : "Failed to parse configuration file, cannot get the input node.");
578 : }
579 : } else {
580 0 : _ml_logw
581 : ("No input node is defined in the pipeline. Might Non-appsrc be used?");
582 : }
583 :
584 0 : if (json_object_has_member (pipe, "output_node")) {
585 0 : JsonNode *node = json_object_get_member (pipe, "output_node");
586 :
587 0 : status = _ml_extension_conf_parse_pipeline_node (mls, node,
588 : ML_SERVICE_NODE_TYPE_OUTPUT);
589 0 : if (status != ML_ERROR_NONE) {
590 0 : _ml_error_report_return (status,
591 : "Failed to parse configuration file, cannot get the output node.");
592 : }
593 : } else {
594 0 : _ml_logw ("No output node is defined in the pipeline.");
595 : }
596 :
597 : /* Start pipeline when creating ml-service handle to check pipeline description. */
598 0 : status = ml_pipeline_start (ext->pipeline);
599 0 : if (status != ML_ERROR_NONE) {
600 0 : _ml_error_report_return (status,
601 : "Failed to parse configuration file, cannot start the pipeline.");
602 : }
603 :
604 0 : return ML_ERROR_NONE;
605 : }
606 :
607 : /**
608 : * @brief Internal function to parse configuration file.
609 : */
610 : static int
611 0 : _ml_extension_conf_parse_json (ml_service_s * mls, JsonObject * object)
612 : {
613 0 : ml_extension_s *ext = (ml_extension_s *) mls->priv;
614 0 : JsonObject *sub = NULL;
615 : int status;
616 :
617 0 : if (json_object_has_member (object, "single")) {
618 0 : sub = json_object_get_object_member (object, "single");
619 :
620 0 : status = _ml_extension_conf_parse_single (mls, sub);
621 0 : if (status != ML_ERROR_NONE)
622 0 : return status;
623 :
624 0 : ext->type = ML_EXTENSION_TYPE_SINGLE;
625 0 : } else if (json_object_has_member (object, "pipeline")) {
626 0 : sub = json_object_get_object_member (object, "pipeline");
627 :
628 0 : status = _ml_extension_conf_parse_pipeline (mls, sub);
629 0 : if (status != ML_ERROR_NONE)
630 0 : return status;
631 :
632 0 : ext->type = ML_EXTENSION_TYPE_PIPELINE;
633 : } else {
634 0 : _ml_error_report_return (ML_ERROR_INVALID_PARAMETER,
635 : "Failed to parse configuration file, cannot get the valid type from configuration.");
636 : }
637 :
638 0 : _ml_extension_conf_parse_common (mls, sub);
639 0 : return ML_ERROR_NONE;
640 : }
641 :
642 : /**
643 : * @brief Internal function to create ml-service extension.
644 : */
645 : int
646 0 : _ml_service_extension_create (ml_service_s * mls, JsonObject * object)
647 : {
648 : ml_extension_s *ext;
649 0 : g_autofree gchar *thread_name = g_strdup_printf ("ml-ext-msg-%d", getpid ());
650 : int status;
651 :
652 0 : mls->priv = ext = g_try_new0 (ml_extension_s, 1);
653 0 : if (ext == NULL) {
654 0 : _ml_error_report_return (ML_ERROR_OUT_OF_MEMORY,
655 : "Failed to allocate memory for ml-service extension. Out of memory?");
656 : }
657 :
658 0 : ext->type = ML_EXTENSION_TYPE_UNKNOWN;
659 0 : ext->running = FALSE;
660 0 : ext->timeout = DEFAULT_TIMEOUT;
661 0 : ext->max_input = DEFAULT_MAX_INPUT;
662 0 : ext->node_table = g_hash_table_new_full (g_str_hash, g_str_equal, g_free,
663 : _ml_extension_node_info_free);
664 :
665 0 : status = _ml_extension_conf_parse_json (mls, object);
666 0 : if (status != ML_ERROR_NONE) {
667 0 : _ml_error_report_return (status,
668 : "Failed to parse the ml-service extension configuration.");
669 : }
670 :
671 0 : g_mutex_lock (&mls->lock);
672 :
673 0 : ext->msg_queue = g_async_queue_new_full (_ml_extension_msg_free);
674 0 : ext->msg_thread = g_thread_new (thread_name, _ml_extension_msg_thread, mls);
675 :
676 : /* Wait until the message thread has been initialized. */
677 0 : g_cond_wait (&mls->cond, &mls->lock);
678 0 : g_mutex_unlock (&mls->lock);
679 :
680 0 : return ML_ERROR_NONE;
681 : }
682 :
683 : /**
684 : * @brief Internal function to release ml-service extension.
685 : */
686 : int
687 0 : _ml_service_extension_destroy (ml_service_s * mls)
688 : {
689 0 : ml_extension_s *ext = (ml_extension_s *) mls->priv;
690 :
691 : /* Supposed internal function call to release handle. */
692 0 : if (!ext)
693 0 : return ML_ERROR_NONE;
694 :
695 : /**
696 : * Close message thread.
697 : * If model inference is running, it may wait for the result in message thread.
698 : * This takes time, so do not call join with extension lock.
699 : */
700 0 : ext->running = FALSE;
701 0 : if (ext->msg_thread) {
702 0 : g_thread_join (ext->msg_thread);
703 0 : ext->msg_thread = NULL;
704 : }
705 :
706 0 : if (ext->msg_queue) {
707 0 : g_async_queue_unref (ext->msg_queue);
708 0 : ext->msg_queue = NULL;
709 : }
710 :
711 0 : if (ext->single) {
712 0 : ml_single_close (ext->single);
713 0 : ext->single = NULL;
714 : }
715 :
716 0 : if (ext->pipeline) {
717 0 : ml_pipeline_stop (ext->pipeline);
718 0 : ml_pipeline_destroy (ext->pipeline);
719 0 : ext->pipeline = NULL;
720 : }
721 :
722 0 : if (ext->node_table) {
723 0 : g_hash_table_destroy (ext->node_table);
724 0 : ext->node_table = NULL;
725 : }
726 :
727 0 : g_free (ext);
728 0 : mls->priv = NULL;
729 :
730 0 : return ML_ERROR_NONE;
731 : }
732 :
733 : /**
734 : * @brief Internal function to start ml-service extension.
735 : */
736 : int
737 0 : _ml_service_extension_start (ml_service_s * mls)
738 : {
739 0 : ml_extension_s *ext = (ml_extension_s *) mls->priv;
740 0 : int status = ML_ERROR_NONE;
741 :
742 0 : switch (ext->type) {
743 0 : case ML_EXTENSION_TYPE_PIPELINE:
744 0 : status = ml_pipeline_start (ext->pipeline);
745 0 : break;
746 0 : case ML_EXTENSION_TYPE_SINGLE:
747 : /* Do nothing. */
748 0 : break;
749 0 : default:
750 0 : status = ML_ERROR_NOT_SUPPORTED;
751 0 : break;
752 : }
753 :
754 0 : return status;
755 : }
756 :
757 : /**
758 : * @brief Internal function to stop ml-service extension.
759 : */
760 : int
761 0 : _ml_service_extension_stop (ml_service_s * mls)
762 : {
763 0 : ml_extension_s *ext = (ml_extension_s *) mls->priv;
764 0 : int status = ML_ERROR_NONE;
765 :
766 0 : switch (ext->type) {
767 0 : case ML_EXTENSION_TYPE_PIPELINE:
768 0 : status = ml_pipeline_stop (ext->pipeline);
769 0 : break;
770 0 : case ML_EXTENSION_TYPE_SINGLE:
771 : /* Do nothing. */
772 0 : break;
773 0 : default:
774 0 : status = ML_ERROR_NOT_SUPPORTED;
775 0 : break;
776 : }
777 :
778 0 : return status;
779 : }
780 :
781 : /**
782 : * @brief Internal function to get the information of required input data.
783 : */
784 : int
785 0 : _ml_service_extension_get_input_information (ml_service_s * mls,
786 : const char *name, ml_tensors_info_h * info)
787 : {
788 0 : ml_extension_s *ext = (ml_extension_s *) mls->priv;
789 : int status;
790 :
791 0 : switch (ext->type) {
792 0 : case ML_EXTENSION_TYPE_SINGLE:
793 0 : status = ml_single_get_input_info (ext->single, info);
794 0 : break;
795 0 : case ML_EXTENSION_TYPE_PIPELINE:
796 : {
797 : ml_service_node_info_s *node_info;
798 :
799 0 : node_info = _ml_extension_node_info_get (ext, name);
800 :
801 0 : if (node_info && node_info->type == ML_SERVICE_NODE_TYPE_INPUT) {
802 0 : status = _ml_tensors_info_create_from (node_info->info, info);
803 : } else {
804 0 : status = ML_ERROR_INVALID_PARAMETER;
805 : }
806 0 : break;
807 : }
808 0 : default:
809 0 : status = ML_ERROR_NOT_SUPPORTED;
810 0 : break;
811 : }
812 :
813 0 : return status;
814 : }
815 :
816 : /**
817 : * @brief Internal function to get the information of output data.
818 : */
819 : int
820 0 : _ml_service_extension_get_output_information (ml_service_s * mls,
821 : const char *name, ml_tensors_info_h * info)
822 : {
823 0 : ml_extension_s *ext = (ml_extension_s *) mls->priv;
824 : int status;
825 :
826 0 : switch (ext->type) {
827 0 : case ML_EXTENSION_TYPE_SINGLE:
828 0 : status = ml_single_get_output_info (ext->single, info);
829 0 : break;
830 0 : case ML_EXTENSION_TYPE_PIPELINE:
831 : {
832 : ml_service_node_info_s *node_info;
833 :
834 0 : node_info = _ml_extension_node_info_get (ext, name);
835 :
836 0 : if (node_info && node_info->type == ML_SERVICE_NODE_TYPE_OUTPUT) {
837 0 : status = _ml_tensors_info_create_from (node_info->info, info);
838 : } else {
839 0 : status = ML_ERROR_INVALID_PARAMETER;
840 : }
841 0 : break;
842 : }
843 0 : default:
844 0 : status = ML_ERROR_NOT_SUPPORTED;
845 0 : break;
846 : }
847 :
848 0 : if (status != ML_ERROR_NONE) {
849 0 : if (*info) {
850 0 : ml_tensors_info_destroy (*info);
851 0 : *info = NULL;
852 : }
853 : }
854 :
855 0 : return status;
856 : }
857 :
858 : /**
859 : * @brief Internal function to set the information for ml-service extension.
860 : */
861 : int
862 0 : _ml_service_extension_set_information (ml_service_s * mls, const char *name,
863 : const char *value)
864 : {
865 0 : ml_extension_s *ext = (ml_extension_s *) mls->priv;
866 :
867 : /**
868 : * Check limitation of message queue and other options.
869 : * When adding new value, you should fix _ml_extension_conf_parse_common() also.
870 : */
871 0 : if (g_ascii_strcasecmp (name, "input_queue_size") == 0 ||
872 0 : g_ascii_strcasecmp (name, "max_input") == 0) {
873 0 : ext->max_input = (guint) g_ascii_strtoull (value, NULL, 10);
874 0 : } else if (g_ascii_strcasecmp (name, "timeout") == 0) {
875 0 : ext->timeout = (guint) g_ascii_strtoull (value, NULL, 10);
876 : }
877 :
878 0 : return ML_ERROR_NONE;
879 : }
880 :
881 : /**
882 : * @brief Internal function to add an input data to process the model in ml-service extension handle.
883 : */
884 : int
885 0 : _ml_service_extension_request (ml_service_s * mls, const char *name,
886 : const ml_tensors_data_h data)
887 : {
888 0 : ml_extension_s *ext = (ml_extension_s *) mls->priv;
889 : ml_extension_msg_s *msg;
890 : int status, len;
891 :
892 0 : if (ext->type == ML_EXTENSION_TYPE_PIPELINE) {
893 : ml_service_node_info_s *node_info;
894 :
895 0 : if (!STR_IS_VALID (name)) {
896 0 : _ml_error_report_return (ML_ERROR_INVALID_PARAMETER,
897 : "The parameter, name '%s', is invalid.", name);
898 : }
899 :
900 0 : node_info = _ml_extension_node_info_get (ext, name);
901 :
902 0 : if (!node_info || node_info->type != ML_SERVICE_NODE_TYPE_INPUT) {
903 0 : _ml_error_report_return (ML_ERROR_INVALID_PARAMETER,
904 : "The parameter, name '%s', is invalid, cannot find the input node from pipeline.",
905 : name);
906 : }
907 : }
908 :
909 0 : len = g_async_queue_length (ext->msg_queue);
910 :
911 0 : if (ext->max_input > 0 && len > 0 && ext->max_input <= len) {
912 0 : _ml_error_report_return (ML_ERROR_STREAMS_PIPE,
913 : "Failed to push input data into the queue, the max number of input is %u.",
914 : ext->max_input);
915 : }
916 :
917 0 : msg = g_try_new0 (ml_extension_msg_s, 1);
918 0 : if (!msg) {
919 0 : _ml_error_report_return (ML_ERROR_OUT_OF_MEMORY,
920 : "Failed to allocate the ml-service extension message. Out of memory?");
921 : }
922 :
923 0 : msg->name = g_strdup (name);
924 0 : status = ml_tensors_data_clone (data, &msg->input);
925 :
926 0 : if (status != ML_ERROR_NONE) {
927 0 : _ml_extension_msg_free (msg);
928 0 : _ml_error_report_return (status, "Failed to clone input data.");
929 : }
930 :
931 0 : g_async_queue_push (ext->msg_queue, msg);
932 :
933 0 : return ML_ERROR_NONE;
934 : }
|
