src/framework/engine-controller/interface.c

   1 /*
   2  * sync-agent
   3  * Copyright (c) 2012 Samsung Electronics Co., Ltd.
   4  *
   5  * Licensed under the Apache License, Version 2.0 (the License);
   6  * you may not use this file except in compliance with the License.
   7  * You may obtain a copy of the License at
   8  *
   9  *     http://www.apache.org/licenses/LICENSE-2.0
  10  *
  11  * Unless required by applicable law or agreed to in writing, software
  12  * distributed under the License is distributed on an "AS IS" BASIS,
  13  * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  14  * See the License for the specific language governing permissions and
  15  * limitations under the License.
  16  */
  17
  18 #include <stdbool.h>
  19 #include <assert.h>
  20 #include <stdio.h>
  21 #include <string.h>
  22 #include <stdlib.h>
  23 #include <signal.h>
  24 #include "utility/fw_assert.h"
  25 #include "utility/sync_util.h"
  26 #include "utility/fw_async_queue_internal.h"
  27 #include "engine-controller/interface.h"
  28 #include "engine-controller/interface_internal.h"
  29 #include "engine-controller/internal.h"
  30 #include "engine-controller/task_message.h"
  31 #include "engine-controller/task_spec.h"
  32 #include "engine-controller/task_pool.h"
  33 #include "engine-controller/task_spec_internal.h"
  34 #include "engine-controller/task_error_internal.h"
  35
  36 #ifndef EXPORT_API
  37 #define EXPORT_API __attribute__ ((visibility("default")))
  38 #endif
  39
  40 #ifndef SYNC_AGENT_LOG
  41 #undef LOG_TAG
  42 #define LOG_TAG "AF_EC"
  43 #endif
  44
  45 GPrivate *ec_cancel_task = NULL;
  46
  47 typedef void *(*ec_engine_controller_logic_func_cb) (ec_engine_controller_t *);
  48
  49 typedef void *(*ec_thread_start_routine_cb) (void *);
  50
  51 static ec_engine_controller_t *global_engine_controller = NULL;
  52 static ec_engine_controller_receiver_t *global_engine_controller_receiver = NULL;
  53
  54 typedef struct ec_sync_task_output_s ec_sync_task_output_t;
  55 struct ec_sync_task_output_s {
  56         pthread_mutex_t mutex;
  57         pthread_cond_t output_set_done_cond;
  58         sync_agent_ec_boolean output_set_done;
  59         sync_agent_ec_task_error_e task_error;
  60         sync_agent_ec_uint out_param_cnt;
  61         sync_agent_ec_param_param_s **out_param_array;
  62 };
  63
  64 static ec_sync_task_output_t *_alloc_and_init_sync_task_output(void)
  65 {
  66         _INNER_FUNC_ENTER;
  67
  68         ec_sync_task_output_t *output = (ec_sync_task_output_t *) calloc(1, sizeof(ec_sync_task_output_t));
  69         if (output == NULL) {
  70                 goto error_part;
  71         }
  72
  73         output->output_set_done = false;
  74         pthread_mutex_init(&(output->mutex), NULL);
  75         pthread_cond_init(&(output->output_set_done_cond), NULL);
  76
  77         _INNER_FUNC_EXIT;
  78
  79         return output;
  80
  81  error_part:
  82         return NULL;
  83 }
  84
  85 static void _destory_sync_task_output(ec_sync_task_output_t * output)
  86 {
  87         _INNER_FUNC_ENTER;
  88
  89         if (output != NULL) {
  90                 pthread_cond_destroy(&(output->output_set_done_cond));
  91                 pthread_mutex_destroy(&(output->mutex));
  92
  93                 free(output);
  94         }
  95
  96         _INNER_FUNC_EXIT;
  97 }
  98
  99 static void _static_sync_task_finish_callback(sync_agent_ec_task_error_e task_error, sync_agent_ec_uint out_param_cnt, sync_agent_ec_param_param_s ** out_param_array, sync_agent_ec_pointer usr_data)
 100 {
 101         _INNER_FUNC_ENTER;
 102
 103         retm_if(usr_data == NULL, "ec_sync_task_output_t is NULL !!");
 104
 105         ec_sync_task_output_t *output = (ec_sync_task_output_t *) usr_data;
 106
 107         pthread_mutex_lock(&(output->mutex));
 108         /* set output */
 109         output->task_error = task_error;
 110         output->out_param_cnt = out_param_cnt;
 111         output->out_param_array = out_param_array;
 112
 113         g_atomic_int_set(&(output->output_set_done), true);
 114         _DEBUG_TRACE("before conditional signal\n");
 115         /* wake up receiver */
 116         pthread_cond_signal(&(output->output_set_done_cond));
 117         pthread_mutex_unlock(&(output->mutex));
 118         _DEBUG_TRACE("after conditional signal\n");
 119
 120         _DEBUG_TRACE("finished\n");
 121
 122         _INNER_FUNC_EXIT;
 123 }
 124
 125 static void _send_parent_EC_TASK_DONE_msg_to_itseft(ec_task_t * parent_task)
 126 {
 127         _INNER_FUNC_ENTER;
 128
 129         _DEBUG_TRACE("called. task_name = %s\n", parent_task->task_spec->task_name);
 130         ec_task_message_t *new_task_msg = ec_task_message_create(EC_TASK_DONE, parent_task, 0);
 131         if (new_task_msg == NULL) {
 132                 _DEBUG_ERROR("error. out of memory. we can not create parent container done message.");
 133                 assert(false);
 134                 /* TODO : error handling */
 135         }
 136         ec_send_msg_to_engine_controller_with_compare_priority(new_task_msg, ec_compare_priority_of_task_message_append_way, NULL);
 137
 138         _INNER_FUNC_EXIT;
 139 }
 140
 141 static void _send_new_runnerable_child_task_msgs_to_itself(GSList * runnable_child_task_list, ec_task_message_t * received_task_msg)
 142 {
 143         _INNER_FUNC_ENTER;
 144
 145         retm_if(runnable_child_task_list == NULL, "GSList is NULL !!");
 146         retm_if(received_task_msg == NULL, "ec_task_message_t is NULL !!");
 147
 148         ec_task_message_t *new_task_msg = NULL;
 149         ec_task_t *child_task = NULL;
 150         GSList *iter = NULL;
 151         for (iter = runnable_child_task_list; iter != NULL; iter = g_slist_next(iter)) {
 152                 child_task = (ec_task_t *) (iter->data);
 153                 new_task_msg = ec_task_message_create(EC_TASK_START, child_task, 0);
 154                 if (received_task_msg->task_state == EC_TASK_START) {
 155                         /* note that we have to prepend task message in front of other messages with same priority */
 156                         /* otherwise, child task in container task always has penalty then new task input */
 157                         ec_send_msg_to_engine_controller_with_compare_priority(new_task_msg, ec_compare_priority_of_task_message_append_way, NULL);
 158                 } else if (received_task_msg->task_state == EC_TASK_DONE) {
 159                         ec_send_msg_to_engine_controller_with_compare_priority(new_task_msg, ec_compare_priority_of_task_message_append_way, NULL);
 160                 } else {
 161                         _DEBUG_ERROR("task msg has invalid task state");
 162                         assert(false);
 163                 }
 164         }
 165
 166         _INNER_FUNC_EXIT;
 167 }
 168
 169 static void *_engine_controller_logic(ec_engine_controller_t * engine_controller)
 170 {
 171         _INNER_FUNC_ENTER;
 172
 173         retvm_if(engine_controller == NULL, NULL, "ec_engine_controller_t is NULL !!");
 174
 175         sync_agent_util_async_queue_s *queue = engine_controller->queue;
 176         ec_thread_pool_t *pool = engine_controller->thread_pool;
 177         ec_task_pool_t *task_pool = engine_controller->task_pool;
 178         ec_task_t *task = NULL;
 179         ec_task_t *parent_task = NULL;
 180         sync_agent_ec_int child_task_index_in_parent_task = -1;
 181         ec_task_message_t *received_task_msg = NULL;
 182         GSList *runnable_child_task_list = NULL;
 183
 184         while (true) {
 185                 /* receive msg */
 186                 received_task_msg = sync_agent_receive_msg_async_queue(queue);
 187                 if (received_task_msg == NULL) {
 188                         _DEBUG_ERROR("null returned from receive_msg_fw_async_queue\n");
 189                         continue;
 190                 }
 191
 192                 if (received_task_msg->task_state == EC_TASK_ROOT_START) {
 193                         task = ec_task_ref_task(received_task_msg->u.task);
 194                         ec_request_msg_t *request_msg = (ec_request_msg_t *) sync_agent_get_original_object(task->request_msg);
 195                         _DEBUG_TRACE("\n\n======= [CONTROLLER : EC_TASK_ROOT_START] task(msg id : %d, task name : %s, task spec id : %d) ========= \n\n", request_msg->msg_head.msg_id, task->task_spec->task_name, request_msg->task_spec_id);
 196
 197                         assert(ec_task_is_root_task(task));
 198                         ec_task_pool_add_task(task_pool, request_msg->msg_head.msg_id, task);
 199
 200                         ec_task_message_t *new_task_msg = ec_task_message_create(EC_TASK_START, task, 0);
 201                         if (new_task_msg == NULL) {
 202                                 /* TODO */
 203                                 assert(false);
 204                         }
 205                         ec_send_msg_to_engine_controller_with_compare_priority(new_task_msg, ec_compare_priority_of_task_message_append_way, NULL);
 206
 207                         ec_task_unref_task(task);
 208
 209                 } else if (received_task_msg->task_state == EC_TASK_START) {
 210                         task = ec_task_ref_task(received_task_msg->u.task);
 211                         ec_request_msg_t *request_msg = (ec_request_msg_t *) sync_agent_get_original_object(task->request_msg);
 212                         _DEBUG_TRACE("\n\n======= [CONTROLLER : EC_TASK_START] =========\ntask(task name : %s) related to root_task(msg id : %d, task spec id : %d)\n\n", task->task_spec->task_name, request_msg->msg_head.msg_id, request_msg->task_spec_id);
 213
 214                         /* queuing if task applied to queuing rule */
 215                         sync_agent_ec_boolean is_pending_task = ec_task_add_pending_task_list_of_first_progress_blocking_realized_entity(task);
 216                         if (is_pending_task) {
 217                                 _DEBUG_TRACE("task name : %s pending by queuing rule\n", task->task_spec->task_name);
 218                                 ec_task_unref_task(task);
 219                                 goto last;
 220                         }
 221
 222                         if (ec_task_is_simple_task(task)) {
 223                                 ec_thread_pool_push_task(pool, task);
 224                         }
 225
 226                         if (ec_task_is_container_task(task)) {
 227                                 /* update progress blocking realized entity when dynamic task case */
 228                                 /* this can be done since other dynamic case conditions can be ignored now */
 229                                 if (ec_task_spec_is_dynamic_container(task->task_spec)) {
 230                                         ec_task_update_progress_blocking_realized_entity_and_pop_if_possible(task);
 231                                 }
 232
 233                                 /* check runnable child task */
 234                                 runnable_child_task_list = ec_collect_firstly_runnable_child_tasks(task);
 235                                 if (runnable_child_task_list == NULL) {
 236                                         _DEBUG_ERROR("no runnable child task exist.\n");
 237                                         assert(false);
 238                                 }
 239
 240                                 /* send task message to engine controller itself */
 241                                 _send_new_runnerable_child_task_msgs_to_itself(runnable_child_task_list, received_task_msg);
 242                         }
 243
 244                         ec_task_unref_task(task);
 245
 246                 } else if (received_task_msg->task_state == EC_TASK_ROOT_DONE) {
 247                         task = ec_task_ref_task(received_task_msg->u.task);
 248                         ec_request_msg_t *request_msg = (ec_request_msg_t *) sync_agent_get_original_object(task->request_msg);
 249
 250                         _DEBUG_TRACE("\n\n======= [CONTROLLER : EC_TASK_ROOT_DONE] task(result : %s), msg id : %d, task name : %s, task spec id : %d) ========= \n\n", ec_task_error_string(task->task_error), request_msg->msg_head.msg_id,
 251                                      task->task_spec->task_name, request_msg->task_spec_id);
 252
 253                         ec_task_pool_remove_task(task_pool, request_msg->msg_head.msg_id);
 254                         ec_task_call_task_finish_callback(task);
 255                         _DEBUG_TRACE("root task ec_task_call_task_finish_callback done\n");
 256                         _DEBUG_TRACE("root task done\n");
 257                 } else if (received_task_msg->task_state == EC_TASK_DONE) {
 258                         task = ec_task_ref_task(received_task_msg->u.task);
 259                         ec_request_msg_t *request_msg = (ec_request_msg_t *) sync_agent_get_original_object(task->request_msg);
 260
 261                         /* loggiing */
 262                         _DEBUG_TRACE("\n\n======= [CONTROLLER : EC_TASK_DONE : %s] =========\ntask(task name : %s) related to root_task(msg id : %d, task spec id : %d)\n\n", ec_task_error_string(task->task_error), task->task_spec->task_name,
 263                                      request_msg->msg_head.msg_id, request_msg->task_spec_id);
 264
 265                         ec_task_update_progress_blocking_realized_entity_and_pop_if_possible(task);
 266
 267                         if (ec_task_is_root_task(task)) {
 268
 269                                 ec_task_message_t *new_task_msg = ec_task_message_create(EC_TASK_ROOT_DONE, task, 0);
 270                                 if (new_task_msg == NULL) {
 271                                         /* TODO */
 272                                         assert(false);
 273                                 }
 274                                 ec_send_msg_to_engine_controller_with_compare_priority(new_task_msg, ec_compare_priority_of_task_message_append_way, NULL);
 275                         } else {
 276                                 switch (task->task_error) {
 277                                 case SYNC_AGENT_EC_TASK_ERROR_RUN_SUCCESS:
 278                                         {
 279                                                 parent_task = ec_task_get_parent_task(task);
 280                                                 child_task_index_in_parent_task = ec_task_get_child_index_in_parent_task(task);
 281
 282                                                 ec_task_decrease_left_child_to_run(parent_task);
 283
 284                                                 if (ec_task_is_parent_EC_TASK_DONE(parent_task)) {
 285                                                         ec_task_set_task_error(parent_task, SYNC_AGENT_EC_TASK_ERROR_RUN_SUCCESS);
 286
 287                                                         /* get parent task output from child tasks */
 288                                                         ec_task_collect_parent_task_output_parameter(parent_task);
 289
 290                                                         /* send parent task done message to engine controller itself */
 291                                                         _send_parent_EC_TASK_DONE_msg_to_itseft(parent_task);
 292
 293                                                         /* TODO : update */
 294                                                 } else {
 295                                                         /* remove dependencies related to done operation */
 296                                                         /* if some operations has no dependency, then push in thread_pool */
 297                                                         runnable_child_task_list = ec_collect_become_runnable_child_tasks_by_remove_control_flow(parent_task, child_task_index_in_parent_task);
 298
 299                                                         /* send task message to engine controller itself */
 300                                                         _send_new_runnerable_child_task_msgs_to_itself(runnable_child_task_list, received_task_msg);
 301                                                 }
 302                                         }
 303                                         break;
 304                                 case SYNC_AGENT_EC_TASK_ERROR_RUN_FAILED:
 305                                 case SYNC_AGENT_EC_TASK_ERROR_CANCELED:
 306                                         parent_task = ec_task_get_parent_task(task);
 307
 308                                         if (ec_task_is_not_yet_run(parent_task)) {
 309                                                 /* FIXME : do i have to wait other child task finished? */
 310                                                 ec_task_set_task_error(parent_task, task->task_error);
 311
 312                                                 /* send parent task done message to engine controller itself */
 313                                                 _send_parent_EC_TASK_DONE_msg_to_itseft(parent_task);
 314                                         }
 315                                         break;
 316                                 default:        /* TODO : other task error must be handled */
 317                                         break;
 318                                 }
 319                         }
 320
 321                         ec_task_unref_task(task);
 322
 323                 } else if (received_task_msg->task_state == EC_TASK_CANCEL) {
 324                         task = ec_task_pool_fetch_task(task_pool, received_task_msg->u.request_msg_id);
 325                         if (task != NULL) {
 326                                 ec_request_msg_t *request_msg = (ec_request_msg_t *) sync_agent_get_original_object(task->request_msg);
 327                                 _DEBUG_TRACE("\n\n======= [CONTROLLER : EC_TASK_CANCEL] =========\ncancel task(request message id : %d, task name : %s) related to root_task(msg id : %d, task spec id : %d)\n\n", received_task_msg->u.request_msg_id,
 328                                              task->task_spec->task_name, request_msg->msg_head.msg_id, request_msg->task_spec_id);
 329
 330                                 ec_task_do_cancellation(task);
 331                                 ec_task_unref_task(task);
 332                         } else {
 333                                 /* TODO : task request may be not handled yet. becuase cancel request has high priority */
 334                                 _DEBUG_TRACE("\n\n======= [CONTROLLER : EC_TASK_CANCEL] =========\ncancel task(request message id : %d) received, " "but  no such task exist. it could be already executed or you never requested\n\n",
 335                                              received_task_msg->u.request_msg_id);
 336                         }
 337                 } else if (received_task_msg->task_state == EC_TASK_CANCEL_ALL) {
 338                         _DEBUG_TRACE("\n\n======= [CONTROLLER : EC_TASK_CANCEL_ALL] =========\n\n\n");
 339
 340                         GList *all_tasks_list = ec_task_pool_fetch_all_tasks(task_pool);
 341                         GList *iter = NULL;
 342                         ec_task_t *task = NULL;
 343                         sync_agent_ec_uint i = 0;
 344                         ec_request_msg_t *request_msg = NULL;
 345                         for (i = 0, iter = all_tasks_list; iter != NULL; i++, iter = g_list_next(iter)) {
 346                                 task = iter->data;
 347                                 request_msg = (ec_request_msg_t *) sync_agent_get_original_object(task->request_msg);
 348                                 _DEBUG_TRACE("\n\n======= [CONTROLLER : EC_TASK_CANCEL_ALL_DETAIL (%d)] =========\ncancel task(request message id : %d, task name : %s) related to root_task(msg id : %d, task spec id : %d)\n\n", i,
 349                                              received_task_msg->u.request_msg_id, task->task_spec->task_name, request_msg->msg_head.msg_id, request_msg->task_spec_id);
 350
 351                                 ec_task_do_cancellation(task);
 352                                 ec_task_unref_task(task);
 353                         }
 354
 355                         if (all_tasks_list != NULL)
 356                                 g_list_free(all_tasks_list);
 357                 } else {
 358                         _DEBUG_ERROR("task msg has invalid task state\n");
 359                         assert(false);
 360                 }
 361
 362  last:
 363                 ec_task_message_free(received_task_msg);
 364         }
 365
 366         _INNER_FUNC_EXIT;
 367
 368         return NULL;            /* TODO */
 369 }
 370
 371 static void *_engine_controller_receiver_logic(ec_engine_controller_receiver_t * engine_controller_receiver)
 372 {
 373         _INNER_FUNC_ENTER;
 374
 375         retvm_if(engine_controller_receiver == NULL, NULL, "ec_engine_controller_receiver_t is NULL !!");
 376
 377         sync_agent_util_async_queue_s *queue = engine_controller_receiver->queue;
 378         ec_task_info_pool_t *task_info_pool = engine_controller_receiver->task_info_pool;
 379         ec_queuing_rule_spec_pool_t *queuing_rule_spec_pool = engine_controller_receiver->queuing_rule_spec_pool;
 380         sync_agent_ec_error_e ec_error = SYNC_AGENT_EC_OK;
 381
 382         ec_task_message_t *task_msg = NULL;
 383         SYNC_AGENT_UTIL_ASSERT_CONDITION(queue != NULL, "engine controller receiver has no queue\n");
 384
 385         while (true) {
 386                 void *msg = sync_agent_receive_msg_async_queue(queue);
 387                 if (ec_msg_is_register_msg(msg)) {
 388                         _DEBUG_TRACE("\n\n======= [RECEIVER : REGISTER TASK SPEC MSG] =========\n\n");
 389                         ec_register_msg_t *register_msg = (ec_register_msg_t *) msg;
 390
 391                         /* create task info and add to task info pool */
 392                         ec_task_info_t *task_info = ec_task_info_new(register_msg->task_spec_id, register_msg->task_spec);
 393
 394                         ec_task_info_pool_add_task_info(task_info_pool, task_info, false);
 395                         /* TODO */
 396
 397                         ec_msg_free_register_msg(register_msg);
 398                 } else if (ec_msg_is_unregister_msg(msg)) {
 399                         _DEBUG_TRACE("\n\n======= [RECEIVER : UNREGISTER TASK SPEC MSG] =========\n\n");
 400                         /* TODO */
 401                 } else if (ec_msg_is_register_queuing_rule_spec_msg(msg)) {
 402                         _DEBUG_TRACE("\n\n======= [RECEIVER : REGISTER QUEUING RULE SPEC MSG] =========\n\n");
 403                         sync_agent_ec_uint queuing_rule_id = 0;
 404                         ec_register_queuing_rule_spec_msg_t *queuing_rule_register_msg = (ec_register_queuing_rule_spec_msg_t *) msg;
 405
 406                         ec_error = ec_queuing_rule_spec_pool_add_queuing_rule_spec(queuing_rule_spec_pool, queuing_rule_register_msg->spec, &queuing_rule_id);
 407
 408                         if (queuing_rule_register_msg->register_finish_callback != NULL) {
 409                                 queuing_rule_register_msg->register_finish_callback(ec_error, queuing_rule_id, queuing_rule_register_msg->usr_data);
 410                         }
 411                         /* TODO */
 412                 } else if (ec_msg_is_cancel_msg(msg)) {
 413                         _DEBUG_TRACE("\n\n======= [RECEIVER : CANCEL MSG] =========\n\n");
 414                         ec_cancel_msg_t *cancel_msg = ((ec_cancel_msg_t *) msg);
 415
 416                         task_msg = ec_task_message_create(EC_TASK_CANCEL, NULL, cancel_msg->request_msg_id_to_cancel);
 417                         if (task_msg == NULL) {
 418                                 _DEBUG_ERROR("out of memory during ec_task_message_create\n");
 419                                 assert(false);
 420                                 continue;
 421                         }
 422
 423                         ec_send_msg_to_engine_controller_with_compare_priority(task_msg, ec_compare_priority_of_task_message_append_way, NULL);
 424
 425                 } else if (ec_msg_is_cancel_all_msg(msg)) {
 426                         _DEBUG_TRACE("\n\n======= [RECEIVER : CANCEL ALL MSG] =========\n\n");
 427
 428                         task_msg = ec_task_message_create(EC_TASK_CANCEL_ALL, NULL, 0);
 429                         if (task_msg == NULL) {
 430                                 _DEBUG_ERROR("out of memory during ec_task_message_create\n");
 431                                 assert(false);
 432                                 continue;
 433                         }
 434
 435                         ec_send_msg_to_engine_controller_with_compare_priority(task_msg, ec_compare_priority_of_task_message_append_way, NULL);
 436
 437                 } else if (ec_msg_is_request_msg(msg)) {
 438
 439                         _DEBUG_TRACE("\n\n======= [RECEIVER : REQUEST MSG] =========\n\n");
 440                         ec_request_msg_t *request_msg = ((ec_request_msg_t *) msg);
 441
 442                         ec_task_info_t *task_info = ec_task_info_pool_search_task_info(task_info_pool, request_msg->task_spec_id);
 443                         sync_agent_ec_task_spec_s *task_spec = NULL;
 444                         ec_task_t *task = NULL;
 445
 446                         if (task_info == NULL) {
 447                                 _DEBUG_TRACE("no msg spec found\n");
 448                         } else {
 449                                 /* create task */
 450                                 task_spec = task_info->task_spec;
 451                                 task = ec_task_alloc_root_task(task_spec, request_msg, task_info_pool);
 452                                 if (task == NULL) {
 453                                         _DEBUG_ERROR("out of memory during ec_task_alloc_root_task\n");
 454                                         assert(false);
 455                                         continue;
 456                                 }
 457
 458                                 task_msg = ec_task_message_create(EC_TASK_ROOT_START, task, 0);
 459                                 if (task_msg == NULL) {
 460                                         _DEBUG_ERROR("out of memory during ec_task_message_create\n");
 461                                         assert(false);
 462                                         continue;
 463                                 }
 464
 465                                 ec_send_msg_to_engine_controller_with_compare_priority(task_msg, ec_compare_priority_of_task_message_append_way, NULL);
 466                                 ec_task_unref_task(task);
 467                         }
 468                 } else {
 469                         _DEBUG_TRACE("\n\n======= [RECEIVER : UNKNOWN MSG] =========\n\n");
 470                         /* unknown msg type */
 471                         _DEBUG_ERROR("unknown msg type\n");
 472                 }
 473         }
 474
 475         _INNER_FUNC_EXIT;
 476
 477         return NULL;            /* TODO */
 478 }
 479
 480 ec_engine_controller_t *ec_alloc_engine_controller(sync_agent_ec_uint max_thread_count)
 481 {
 482         _EXTERN_FUNC_ENTER;
 483
 484         ec_engine_controller_t *engine_controller = (ec_engine_controller_t *) calloc(1, sizeof(ec_engine_controller_t));
 485         if (engine_controller == NULL) {
 486                 goto return_part;
 487         }
 488
 489         engine_controller->queue = sync_agent_alloc_async_queue();
 490         if (engine_controller->queue == NULL) {
 491                 goto error_part;
 492         }
 493
 494         engine_controller->task_pool = ec_task_pool_create_task_pool();
 495         if (engine_controller->task_pool == NULL) {
 496                 goto error_part;
 497         }
 498
 499         engine_controller->queuing_rule_spec_pool = NULL;
 500
 501         engine_controller->thread_pool = ec_thread_pool_alloc_and_init(max_thread_count);
 502         if (engine_controller->thread_pool == NULL) {
 503                 goto error_part;
 504         }
 505 #if !GLIB_CHECK_VERSION (2, 32, 0)
 506         ec_cancel_task = ec_thread_pool_get_cancel_task(engine_controller->thread_pool);
 507 #else
 508         ec_cancel_task = ec_thread_pool_get_cancel_task();
 509 #endif
 510         if (ec_cancel_task == NULL) {
 511                 goto error_part;
 512         }
 513
 514  return_part:
 515         _EXTERN_FUNC_EXIT;
 516         return engine_controller;
 517
 518  error_part:
 519         ec_free_engine_controller(engine_controller);
 520         return NULL;
 521 }
 522
 523 void ec_free_engine_controller(ec_engine_controller_t * engine_controller)
 524 {
 525         _EXTERN_FUNC_ENTER;
 526
 527         retm_if(engine_controller == NULL, "ec_engine_controller_t is NULL !!");
 528
 529         if (engine_controller != NULL) {
 530                 if (engine_controller->thread_pool) {
 531                         ec_thread_pool_destroy(engine_controller->thread_pool);
 532                         engine_controller->thread_pool = NULL;
 533                 }
 534                 if (engine_controller->queuing_rule_spec_pool) {
 535                         ec_queuing_rule_spec_pool_free(engine_controller->queuing_rule_spec_pool);
 536                         engine_controller->queuing_rule_spec_pool = NULL;
 537                 }
 538                 if (engine_controller->task_pool) {
 539                         ec_task_pool_free_task_pool(engine_controller->task_pool);
 540                         engine_controller->task_pool = NULL;
 541                 }
 542                 if (engine_controller->queue) {
 543                         util_destroy_async_queue(engine_controller->queue);
 544                         engine_controller->queue = NULL;
 545                 }
 546
 547                 free(engine_controller);
 548         }
 549
 550         _EXTERN_FUNC_EXIT;
 551 }
 552
 553 /* TODO error handling */
 554 void ec_run_engine_controller(ec_engine_controller_t * engine_controller)
 555 {
 556         _EXTERN_FUNC_ENTER;
 557
 558         retm_if(engine_controller == NULL, "ec_engine_controller_t is NULL !!");
 559
 560         int status_thread_create = 0;
 561         status_thread_create = pthread_create(&(engine_controller->thread_id), NULL, (ec_thread_start_routine_cb) _engine_controller_logic, (void *)engine_controller);
 562         if(status_thread_create != 0) {
 563                 _DEBUG_WARNING("pthread_create [%d]", status_thread_create );
 564                 _EXTERN_FUNC_EXIT;
 565                 return;
 566         }
 567
 568         pthread_detach(engine_controller->thread_id);
 569
 570         _EXTERN_FUNC_EXIT;
 571 }
 572
 573 void ec_stop_engine_controller(ec_engine_controller_t * engine_controller)
 574 {
 575         _EXTERN_FUNC_ENTER;
 576
 577         retm_if(engine_controller == NULL, "ec_engine_controller_t is NULL !!");
 578
 579         pthread_cancel(engine_controller->thread_id);
 580
 581         _EXTERN_FUNC_EXIT;
 582 }
 583
 584 ec_engine_controller_receiver_t *ec_alloc_engine_controller_receiver(ec_engine_controller_t * engine_controller)
 585 {
 586         _EXTERN_FUNC_ENTER;
 587
 588         retvm_if(engine_controller == NULL, NULL, "ec_engine_controller_t is NULL !!");
 589
 590         ec_engine_controller_receiver_t *receiver = NULL;
 591
 592         receiver = (ec_engine_controller_receiver_t *) calloc(1, sizeof(ec_engine_controller_receiver_t));
 593         if (receiver == NULL) {
 594                 goto return_part;
 595         }
 596
 597         receiver->next_msg_id = 0;
 598
 599         receiver->queue = sync_agent_alloc_async_queue();
 600         if (receiver->queue == NULL) {
 601                 goto error_part;
 602         }
 603
 604         receiver->task_info_pool = ec_task_info_pool_alloc();
 605         if (receiver->task_info_pool == NULL) {
 606                 goto error_part;
 607         }
 608
 609         /* TODO : remove below 1024 hard coding */
 610         receiver->queuing_rule_spec_pool = ec_queuing_rule_spec_pool_alloc(1024, receiver->task_info_pool);
 611         if (receiver->queuing_rule_spec_pool == NULL) {
 612                 goto error_part;
 613         }
 614
 615         receiver->engine_controller = engine_controller;
 616
 617         /* TODO */
 618         engine_controller->queuing_rule_spec_pool = receiver->queuing_rule_spec_pool;
 619
 620  return_part:
 621         _EXTERN_FUNC_EXIT;
 622         return receiver;
 623
 624  error_part:
 625         ec_free_engine_controller_receiver(receiver);
 626         return NULL;
 627 }
 628
 629 void ec_free_engine_controller_receiver(ec_engine_controller_receiver_t * engine_controller_receiver)
 630 {
 631         _EXTERN_FUNC_ENTER;
 632
 633         retm_if(engine_controller_receiver == NULL, "ec_engine_controller_receiver_t is NULL !!");
 634
 635         if (engine_controller_receiver != NULL) {
 636                 if (engine_controller_receiver->queuing_rule_spec_pool) {
 637                         ec_queuing_rule_spec_pool_free(engine_controller_receiver->queuing_rule_spec_pool);
 638                         engine_controller_receiver->queuing_rule_spec_pool = NULL;
 639                 }
 640                 if (engine_controller_receiver->task_info_pool) {
 641                         ec_task_info_pool_free(engine_controller_receiver->task_info_pool);
 642                         engine_controller_receiver->task_info_pool = NULL;
 643                 }
 644                 if (engine_controller_receiver->queue) {
 645                         util_destroy_async_queue(engine_controller_receiver->queue);
 646                         engine_controller_receiver->queue = NULL;
 647                 }
 648
 649                 free(engine_controller_receiver);
 650         }
 651
 652         _EXTERN_FUNC_EXIT;
 653 }
 654
 655 /* TODO error handling */
 656 void ec_run_engine_controller_receiver(ec_engine_controller_receiver_t * engine_controller_receiver)
 657 {
 658         _EXTERN_FUNC_ENTER;
 659
 660         retm_if(engine_controller_receiver == NULL, "ec_engine_controller_receiver_t is NULL !!");
 661
 662         /* TODO */
 663         int status_thread_create = 0;
 664         status_thread_create = pthread_create(&(engine_controller_receiver->thread_id), NULL, (ec_thread_start_routine_cb) _engine_controller_receiver_logic, (void *)engine_controller_receiver);
 665         if(status_thread_create != 0) {
 666                 _DEBUG_WARNING("pthread_create [%d]", status_thread_create );
 667                 _EXTERN_FUNC_EXIT;
 668                 return;
 669         }
 670
 671         pthread_detach(engine_controller_receiver->thread_id);
 672
 673         _EXTERN_FUNC_EXIT;
 674 }
 675
 676 void ec_stop_engine_controller_receiver(ec_engine_controller_receiver_t * engine_controller_receiver)
 677 {
 678         _EXTERN_FUNC_ENTER;
 679
 680         retm_if(engine_controller_receiver == NULL, "ec_engine_controller_receiver_t is NULL !!");
 681
 682         pthread_cancel(engine_controller_receiver->thread_id);
 683
 684         _EXTERN_FUNC_EXIT;
 685 }
 686
 687 void ec_send_msg_to_engine_controller_receiver(ec_msg_head_t * msg)
 688 {
 689         _EXTERN_FUNC_ENTER;
 690
 691         retm_if(msg == NULL, "ec_msg_head_t is NULL !!");
 692
 693         /* msg id management */
 694
 695 //      g_atomic_int_exchange_and_add() to be deprecated
 696 //      msg->msg_id = g_atomic_int_exchange_and_add(&(global_engine_controller_receiver->next_msg_id), 1);
 697         msg->msg_id = global_engine_controller_receiver->next_msg_id;
 698         g_atomic_int_add(&(global_engine_controller_receiver->next_msg_id), 1);
 699
 700         _DEBUG_INFO("============= msg id = %d =============\n", msg->msg_id);
 701
 702         sync_agent_send_msg_async_queue(global_engine_controller_receiver->queue, (void *)msg);
 703
 704         _EXTERN_FUNC_EXIT;
 705 }
 706
 707 void ec_send_msg_to_engine_controller(ec_task_message_t * task_msg)
 708 {
 709         _EXTERN_FUNC_ENTER;
 710
 711         retm_if(task_msg == NULL, "ec_task_message_t is NULL !!");
 712
 713         sync_agent_send_msg_async_queue(global_engine_controller->queue, (void *)task_msg);
 714
 715         _EXTERN_FUNC_EXIT;
 716 }
 717
 718 void ec_send_msg_to_engine_controller_with_compare_priority(ec_task_message_t * task_msg, ec_compare_task_msg_priority_func_cb ctmp_func, sync_agent_ec_pointer user_data)
 719 {
 720         _EXTERN_FUNC_ENTER;
 721
 722         retm_if(task_msg == NULL, "ec_task_message_t is NULL !!");
 723
 724         util_send_msg_async_queue_with_compare_priority(global_engine_controller->queue, (void *)task_msg, (sync_agent_compare_priority_cb) ctmp_func, (void *)user_data);
 725
 726         _EXTERN_FUNC_EXIT;
 727 }
 728
 729 /* external interfaces */
 730 bool ec_init_engine_controller(unsigned int max_thread_count)
 731 {
 732         _EXTERN_FUNC_ENTER;
 733
 734         bool success = true;
 735         ec_engine_controller_t *engine_controller = NULL;
 736         ec_engine_controller_receiver_t *engine_controller_receiver = NULL;
 737
 738         /* alloc engine controller */
 739         engine_controller = ec_alloc_engine_controller(max_thread_count);
 740         if (engine_controller == NULL) {
 741                 success = false;
 742                 goto return_part;
 743         }
 744
 745         /* alloc engine controller receiver */
 746         engine_controller_receiver = ec_alloc_engine_controller_receiver(engine_controller);
 747         if (engine_controller_receiver == NULL) {
 748                 success = false;
 749                 goto return_part;
 750         }
 751
 752         /* run engine controller */
 753         ec_run_engine_controller(engine_controller);
 754         ec_run_engine_controller_receiver(engine_controller_receiver);
 755
 756  return_part:
 757         if (success) {
 758                 global_engine_controller = engine_controller;
 759                 global_engine_controller_receiver = engine_controller_receiver;
 760         } else {
 761                 /* TODO : stop engine controller & receiver & free */
 762         }
 763
 764         _EXTERN_FUNC_EXIT;
 765
 766         return success;
 767 }
 768
 769 void ec_deinit_engine_controller()
 770 {
 771         _EXTERN_FUNC_ENTER;
 772
 773         ec_stop_engine_controller_receiver(global_engine_controller_receiver);
 774         ec_stop_engine_controller(global_engine_controller);
 775
 776         ec_free_engine_controller_receiver(global_engine_controller_receiver);
 777         global_engine_controller_receiver = NULL;
 778         ec_free_engine_controller(global_engine_controller);
 779         global_engine_controller = NULL;
 780
 781         _EXTERN_FUNC_EXIT;
 782 }
 783
 784 /* TODO : error handling */
 785 EXPORT_API void sync_agent_register_task_spec(sync_agent_ec_uint task_spec_id, sync_agent_ec_char * task_spec_name, sync_agent_ec_task_spec_s * task_spec, sync_agent_calculate_identifier_cb cal_func)
 786 {
 787         _EXTERN_FUNC_ENTER;
 788
 789         retm_if(task_spec == NULL, "sync_agent_ec_task_spec_s is NULL !!");
 790
 791         ec_register_msg_t *register_msg = ec_msg_create_register_msg(task_spec_id, task_spec, cal_func);
 792         ec_send_msg_to_engine_controller_receiver((ec_msg_head_t *) register_msg);
 793
 794         _EXTERN_FUNC_EXIT;
 795 }
 796
 797 EXPORT_API void sync_agent_register_async_queuing_rule_spec(sync_agent_ec_queuing_rule_spec_s * spec, sync_agent_register_finish_cb register_finish_callback, sync_agent_ec_pointer usr_data)
 798 {
 799         _EXTERN_FUNC_ENTER;
 800
 801         ec_register_queuing_rule_spec_msg_t *msg = ec_msg_create_register_queuing_rule_spec_msg(spec, register_finish_callback, usr_data);
 802
 803         ec_send_msg_to_engine_controller_receiver((ec_msg_head_t *) msg);
 804
 805         _EXTERN_FUNC_EXIT;
 806 }
 807
 808 void ec_register_sync_queuing_rule_spec(sync_agent_ec_queuing_rule_spec_s * spec, sync_agent_ec_error_e * ec_error, sync_agent_ec_uint * registered_id)
 809 {
 810         _EXTERN_FUNC_ENTER;
 811
 812         /* TODO */
 813
 814         _EXTERN_FUNC_EXIT;
 815 }
 816
 817 /* TODO : error handling */
 818 EXPORT_API void sync_agent_request_async_task(sync_agent_ec_uint task_spec_id, sync_agent_ec_uint identifier, sync_agent_ec_int cnt_in_param, sync_agent_ec_int * in_param_index_array, sync_agent_ec_value_type_e * in_param_value_type_array,
 819                                               sync_agent_ec_pointer * in_param_value_array, sync_agent_task_finish_cb task_finish_callback, sync_agent_ec_pointer simple_task_finish_callback_usr_data, sync_agent_ec_int * request_id)
 820 {
 821         _EXTERN_FUNC_ENTER;
 822
 823         ec_request_msg_t *msg = ec_msg_create_request_msg(task_spec_id, identifier, cnt_in_param,
 824                                                           in_param_index_array, in_param_value_type_array, in_param_value_array,
 825                                                           task_finish_callback, simple_task_finish_callback_usr_data);
 826
 827         retm_if(msg == NULL, "ec_request_msg_t is NULL !!");
 828
 829         ec_send_msg_to_engine_controller_receiver((ec_msg_head_t *) msg);
 830         *request_id = msg->msg_head.msg_id;
 831
 832         _EXTERN_FUNC_EXIT;
 833 }
 834
 835 EXPORT_API void sync_agent_request_sync_task(sync_agent_ec_uint task_spec_id, sync_agent_ec_uint identifier, sync_agent_ec_int cnt_in_param, sync_agent_ec_int * in_param_index_array, sync_agent_ec_value_type_e * in_param_value_type_array,
 836                                              sync_agent_ec_pointer * in_param_value_array, sync_agent_ec_int * request_id, sync_agent_ec_task_error_e * task_error, sync_agent_ec_uint * out_param_cnt, sync_agent_ec_param_param_s *** out_param_array)
 837 {
 838         _EXTERN_FUNC_ENTER;
 839
 840         sync_agent_task_finish_cb sync_task_finish_callback = _static_sync_task_finish_callback;
 841         ec_sync_task_output_t *sync_task_output = _alloc_and_init_sync_task_output();
 842         if (sync_task_output == NULL) {
 843                 return;         /* error */
 844         }
 845
 846         ec_request_msg_t *msg = ec_msg_create_request_msg(task_spec_id, identifier, cnt_in_param,
 847                                                           in_param_index_array, in_param_value_type_array, in_param_value_array,
 848                                                           sync_task_finish_callback, sync_task_output);
 849
 850         retm_if(msg == NULL, "ec_request_msg_t is NULL !!");
 851
 852         pthread_mutex_lock(&(sync_task_output->mutex));
 853         ec_send_msg_to_engine_controller_receiver((ec_msg_head_t *) msg);
 854
 855         while (!g_atomic_int_get(&(sync_task_output->output_set_done))) {
 856                 pthread_cond_wait(&(sync_task_output->output_set_done_cond), &(sync_task_output->mutex));
 857         }
 858
 859         pthread_mutex_unlock(&(sync_task_output->mutex));
 860
 861         *request_id = msg->msg_head.msg_id;
 862         *task_error = sync_task_output->task_error;
 863         *out_param_cnt = sync_task_output->out_param_cnt;
 864         *out_param_array = sync_task_output->out_param_array;
 865
 866         _destory_sync_task_output(sync_task_output);
 867
 868         _EXTERN_FUNC_EXIT;
 869 }
 870
 871 EXPORT_API void sync_agent_cancel_task(sync_agent_ec_int request_id_to_cancel)
 872 {
 873         _EXTERN_FUNC_ENTER;
 874
 875         ec_cancel_msg_t *cancel_msg = ec_msg_create_cancel_msg(request_id_to_cancel);
 876         ec_send_msg_to_engine_controller_receiver((ec_msg_head_t *) cancel_msg);
 877
 878         _EXTERN_FUNC_EXIT;
 879 }
 880
 881 EXPORT_API void sync_agent_cancel_all_tasks()
 882 {
 883         _EXTERN_FUNC_ENTER;
 884
 885         ec_cancel_all_msg_t *cancel_all_msg = ec_msg_create_cancel_all_msg();
 886         ec_send_msg_to_engine_controller_receiver((ec_msg_head_t *) cancel_all_msg);
 887
 888         _EXTERN_FUNC_EXIT;
 889 }