text: /*
   * Copyright (c) 2021 Huawei Technologies Co.,Ltd.
   *
   * openGauss is licensed under Mulan PSL v2.
   * You can use this software according to the terms and conditions of the Mulan PSL v2.
   * You may obtain a copy of Mulan PSL v2 at:
   *
   *          http://license.coscl.org.cn/MulanPSL2
   *
   * THIS SOFTWARE IS PROVIDED ON AN "AS IS" BASIS, WITHOUT WARRANTIES OF ANY KIND,
   * EITHER EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO NON-INFRINGEMENT,
   * MERCHANTABILITY OR FIT FOR A PARTICULAR PURPOSE.
   * See the Mulan PSL v2 for more details.
   * -------------------------------------------------------------------------
   *
   * mec_agent.c
   *    mec process
   *
   * IDENTIFICATION
   *    src/network/mec/mec_agent.c
   *
   * -------------------------------------------------------------------------
   */
  
  #include "cm_log.h"
  #include "cm_file.h"
  #include "mec_reactor.h"
  #include "cm_memory.h"
  #include "mec_profile.h"
  #include "mec_func.h"
  
  #ifdef __cplusplus
  extern "C" {
  #endif
  
  status_t agent_create_pool(agent_pool_t *agent_pool, uint32 agent_num)
  {
      size_t size;
      uint32 loop;
      agent_t *agent = NULL;
      agent_pool->optimized_count = agent_num;
  
      size = sizeof(agent_t) * agent_pool->optimized_count;
      if (size == 0 || size / sizeof(agent_t) != agent_pool->optimized_count) {
          CM_THROW_ERROR(ERR_ALLOC_MEMORY, (uint64)size, "creating agent pool");
          return CM_ERROR;
      }
      agent_pool->agents = (agent_t *)malloc(size);
      if (agent_pool->agents == NULL) {
          CM_THROW_ERROR(ERR_ALLOC_MEMORY, (uint64)size, "creating agent pool");
          return CM_ERROR;
      }
      errno_t err = memset_s(agent_pool->agents, size, 0, size);
      if (err != EOK) {
          CM_FREE_PTR(agent_pool->agents);
          CM_THROW_ERROR(ERR_SYSTEM_CALL, err);
          return CM_ERROR;
      }
  
      agent_pool->lock_idle = 0;
      biqueue_init(&agent_pool->idle_agents);
  
      agent_pool->lock_new = 0;
      biqueue_init(&agent_pool->blank_agents);
      for (loop = 0; loop < agent_pool->optimized_count; ++loop) {
          agent = &agent_pool->agents[loop];
          agent->pool = agent_pool;
          biqueue_add_tail(&agent_pool->blank_agents, QUEUE_NODE_OF(agent));
      }
      agent_pool->blank_count = agent_pool->optimized_count;
  
      if (cm_event_init(&agent_pool->idle_evnt) != CM_SUCCESS) {
          CM_FREE_PTR(agent_pool->agents);
          CM_THROW_ERROR(ERR_CREATE_EVENT, cm_get_os_error());
          return CM_ERROR;
      }
  
      return CM_SUCCESS;
  }
  
  void agent_detach_and_set_oneshot(mec_pipe_t *pipe)
  {
      detach_agent(pipe, RECV_MODE);
  
      CM_MFENCE;
      if (reactor_set_oneshot(pipe) != CM_SUCCESS) {
          LOG_RUN_ERR("[MEC]set oneshot flag of socket failed, "
                      "channel %d, os error %d",
                      pipe->channel->id, cm_get_sock_error());
      }
  }
  
  static void try_process_multi_channels(agent_t *agent)
  {
      mec_pipe_t *pipe = NULL;
  
      for (;;) {
          // event will be set by reactor
          if (cm_event_timedwait(&agent->event, CM_SLEEP_50_FIXED) == CM_SUCCESS) {
              break;
          }
  
          if (agent->thread.closed) {
              return;
          }
      }
  
      if (agent->mode >= MODE_END || agent->pipe->attach[agent->mode].job == NULL) {
          LOG_DEBUG_ERR("[MEC]agent mode=%u or job=%p err.", agent->mode, agent->pipe->attach[agent->mode].job);
          return;
      }
  
      pipe = agent->pipe;
      pipe->attach[agent->mode].spid = cm_get_current_thread_id();
  
      LOG_DEBUG_INF("[MEC]begin to process job from inst id %u, channel id %u.",
                    MEC_INSTANCE_ID(pipe->channel->id),
                    MEC_CHANNEL_ID(pipe->channel->id));
      bool32 is_continue;
      while (!agent->thread.closed) {
          pipe->attach[agent->mode].job((void *)pipe, &is_continue);
          if (is_continue) {
              continue;
          }
          return;
      }
  }
  
  
  static inline void return_agent2blankqueue(agent_t *agent, agent_pool_t *agent_pool)
  {
      // when failed to start an agent, the agent has not be added to idle queue
      // so then pointer 'next' could be null
      if (agent->next != NULL) {
          // remove agent from idle queue
          cm_spin_lock(&agent_pool->lock_idle, NULL);
          if (agent->next != NULL) {          // re-check to protect change by reactor thread
              biqueue_del_node(QUEUE_NODE_OF(agent));
              agent_pool->idle_count--;
          }
          cm_spin_unlock(&agent_pool->lock_idle);
      }
  
      // add agent to blank queue
      cm_spin_lock(&agent_pool->lock_new, NULL);
      biqueue_add_tail(&agent_pool->blank_agents, QUEUE_NODE_OF(agent));
  
      --agent_pool->curr_count;
      agent_pool->blank_count++;
      // can not process agent member after agent back to blank queue, otherwise will core.
      cm_spin_unlock(&agent_pool->lock_new);
  }
  
  
  void agent_entry(thread_t *thread)
  {
      agent_t *agent = (agent_t *)thread->argument;
  
      (void)cm_set_thread_name("agent");
      LOG_RUN_INF("[MEC]agent thread started, tid:%lu, close:%u", thread->id, thread->closed);
      while (!thread->closed) {
          try_process_multi_channels(agent);
      }
      LOG_RUN_INF("[MEC]agent thread closed, tid:%lu, close:%u", thread->id, thread->closed);
      cm_event_destory(&agent->event);
      cm_release_thread(thread);
      return_agent2blankqueue(agent, agent->pool);
  }
  
  status_t start_agent(agent_t *agent, thread_entry_t entry)
  {
      return cm_create_thread(entry, 0, agent, &agent->thread);
  }
  
  
  void sync_agents_closed(agent_pool_t *agent_pool)
  {
      if (agent_pool->agents != NULL) {
          for (uint32 i = 0; i < agent_pool->optimized_count; i++) {
              agent_pool->agents[i].thread.closed = CM_TRUE;
          }
      }
  
      while (agent_pool->curr_count > 0) {
          cm_sleep(1);
      }
  }
  
  static void shutdown_agent_pool(agent_pool_t *agent_pool)
  {
      LOG_RUN_INF("[MEC]all agents' thread have been closed");
      sync_agents_closed(agent_pool);
      cm_event_destory(&agent_pool->idle_evnt);
      biqueue_init(&agent_pool->idle_agents);
      biqueue_init(&agent_pool->blank_agents);
      agent_pool->blank_count = 0;
      agent_pool->idle_count = 0;
      CM_FREE_PTR(agent_pool->agents);
  }
  
  void agent_destroy_pool(agent_pool_t *agent_pool)
  {
      LOG_RUN_INF("[MEC]begin to destroy agent pool");
      shutdown_agent_pool(agent_pool);
      LOG_RUN_INF("[MEC]destroy agent pool end");
  }
  
  
  static inline status_t create_agent(agent_t *agent)
  {
      if (cm_event_init(&agent->event) != CM_SUCCESS) {
          CM_THROW_ERROR(ERR_CREATE_EVENT, cm_get_os_error());
          return CM_ERROR;
      }
      if (start_agent(agent, agent_entry) != CM_SUCCESS) {
          LOG_RUN_ERR("[MEC]create agent thread failed, os error %d", cm_get_os_error());
          return CM_ERROR;
      }
  
      return CM_SUCCESS;
  }
  
  static inline void bind_channel_agent(mec_pipe_t *pipe, attach_mode_t mode, agent_t *agent)
  {
      agent->mode = mode;
      agent->pipe = pipe;
      pipe->attach[mode].agent = agent;
      pipe->attach[mode].status = ACTIVE;
  }
  
  
  static inline status_t try_create_agent(agent_pool_t *agent_pool, agent_t **agent)
  {
      biqueue_node_t *node = NULL;
      bool32 need_create;
  
      if (agent_pool->curr_count == agent_pool->optimized_count) {
          *agent = NULL;
          return CM_SUCCESS;
      }
  
      cm_spin_lock(&agent_pool->lock_new, NULL);
  
      need_create = agent_pool->curr_count < agent_pool->optimized_count;
      if (!need_create) {
          cm_spin_unlock(&agent_pool->lock_new);
          *agent = NULL;
          return CM_SUCCESS;
      }
      node = biqueue_del_head(&agent_pool->blank_agents);
      ++agent_pool->curr_count;
      agent_pool->blank_count--;
      cm_spin_unlock(&agent_pool->lock_new);
  
      *agent = OBJECT_OF(agent_t, node);
      if (create_agent(*agent) != CM_SUCCESS) {
          return_agent2blankqueue(*agent, agent_pool);
          *agent = NULL;
          return CM_ERROR;
      }
  
      return CM_SUCCESS;
  }
  
  status_t try_attach_agent(mec_pipe_t *pipe, agent_pool_t *agent_pool, attach_mode_t mode, agent_t **agent)
  {
      status_t status;
      biqueue_node_t *node = NULL;
  
      // if not empty , get agent from idle pool.
      if (!biqueue_empty(&agent_pool->idle_agents)) {
          cm_spin_lock(&agent_pool->lock_idle, NULL);
          node = biqueue_del_head(&agent_pool->idle_agents);
          if (node != NULL) {
              agent_pool->idle_count--;
          }
          cm_spin_unlock(&agent_pool->lock_idle);
  
          if (node != NULL) {
              *agent = OBJECT_OF(agent_t, node);
              bind_channel_agent(pipe, mode, *agent);
              return CM_SUCCESS;
          }
      }
  
      status = try_create_agent(agent_pool, agent);
      CM_RETURN_IFERR(status);
  
      if (*agent != NULL) {
          bind_channel_agent(pipe, mode, *agent);
      }
  
      return CM_SUCCESS;
  }
  
  status_t attach_agent(mec_pipe_t *pipe, agent_pool_t *agent_pool, attach_mode_t mode, agent_t **agent)
  {
      *agent = NULL;
      for (;;) {
          /* hit scenario: enter deadloop, after create agent failed */
          status_t status = try_attach_agent(pipe, agent_pool, mode, agent);
          CM_RETURN_IFERR(status);
  
          if (*agent != NULL) {
              return CM_SUCCESS;
          }
  
          cm_event_wait(&agent_pool->idle_evnt);
      }
  }
  
  void detach_agent(mec_pipe_t *pipe, attach_mode_t mode)
  {
      volatile agent_t *agent = pipe->attach[mode].agent;
      if (agent == NULL) {
          return;
      }
      agent_pool_t *agent_pool = agent->pool;
  
      CM_ASSERT(pipe == agent->pipe);
      agent->pipe = NULL;
      agent->mode = MODE_END;
      /* status might still be ACTIVE while being detached from agent, so need to reset */
      pipe->attach[mode].status = INACTIVE;
  
      cm_spin_lock(&agent_pool->lock_idle, NULL);
      biqueue_add_tail(&agent_pool->idle_agents, QUEUE_NODE_OF(agent));
      agent_pool->idle_count++;
      cm_spin_unlock(&agent_pool->lock_idle);
      cm_event_notify(&agent_pool->idle_evnt);
      CM_MFENCE;
      /* agent must be freed in the end */
      pipe->attach[mode].agent = NULL;
      LOG_DEBUG_INF("[MEC]detach channel %u from agent %lu success, idle agent num %u.",
                    pipe->channel->id, agent->thread.id, agent_pool->idle_count);
  }
  
  void show_agent_count(agent_pool_t *agent_pool, msg_priv_t priv)
  {
      LOG_DEBUG_INF("[MEC]priv[%u] agent_pool count info: "
                    "idle_count[%u] blank_count[%u] curr_count[%u] optimized_count[%u].",
                    priv,
                    agent_pool->idle_count,
                    agent_pool->blank_count,
                    agent_pool->curr_count,
                    agent_pool->optimized_count);
  }
  
  
  #ifdef __cplusplus
  }
  #endif
  