<commit_before>/*
 * Copyright 2016 <NAME>
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

#define NDEBUG

#include <ac_comp_mgr.h>

#include <ac_assert.h>
#include <ac_dispatcher.h>
#include <ac_memmgr.h>
#include <ac_msg.h>
#include <ac_printf.h>
#include <ac_debug_printf.h>
#include <ac_receptor.h>
#include <ac_string.h>
#include <ac_thread.h>

#if AC_PLATFORM == pc_x86_64
extern void remove_zombies(void);
#endif

typedef struct DispatchThreadParams DispatchThreadParams;

/**
 * A opaque component info for an AcComp
 */
typedef struct AcCompInfo {
  AcCompMgr* mgr;
  AcComp* comp;
  AcDispatchableComp* dc;
  ac_u32 comp_idx;
  DispatchThreadParams* dtp;
} AcCompInfo;

/**
 * Parameters for each thread to manage its components
 */
typedef struct DispatchThreadParams {
  ac_bool thread_started;
  ac_thread_hdl_t thread_hdl;
  AcDispatcher* d;
  ac_u32 max_comps;
  AcCompInfo** cis;
  AcReceptor* done;
  AcReceptor* ready;
  AcReceptor* waiting;
  ac_bool stop_processing_msgs;
} DispatchThreadParams;

typedef struct AcCompMgr {
  AcCompInfo* comp_infos;     // Array of AcCompInfo objects being managed
                              // across all of the threads
  ac_u32 comp_infos_max_count;// Number of elements in comp_infos array
  DispatchThreadParams* dtps; // Array of DispathThreadParams, one for each thread
  ac_u32 max_dtps;            // Number of threads in the dtps array
  ac_u32 next_dtps;           // Next thread
} AcCompMgr;

/**
 * A thread which dispatches message to its components.
 */
static void* dispatch_thread(void *param) {
  DispatchThreadParams* params = (DispatchThreadParams*)(param);

  ac_debug_printf("dispatch_thread:+starting params=%p\n", params);

  // Get a dispatcher and add a queue and message processor
  params->d = AcDispatcher_get(params->max_comps);
  if (params->d == AC_NULL) {
    ac_debug_printf("dispatch_thread: no AcDispatcher params=%p\n", params);
    goto done;
  }

  // Create the waiting receptor and init our not stopped flag
  params->waiting = AcReceptor_get();
  ac_assert(params->waiting != AC_NULL);
  __atomic_store_n(&params->stop_processing_msgs, AC_FALSE, __ATOMIC_RELEASE);

  // Signal dispatch_thread is ready
  AcReceptor_signal(params->ready);

  // Continuously dispatch messages until we're told to stop
  while (__atomic_load_n(&params->stop_processing_msgs, __ATOMIC_ACQUIRE) == AC_FALSE) {
    if (!AcDispatcher_dispatch(params->d)) {
      ac_debug_printf("dispatch_thread: waiting\n");<commit_msg>Fix comment in dispatch_thread<commit_after>/*
 * Copyright 2016 <NAME>
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

#define NDEBUG

#include <ac_comp_mgr.h>

#include <ac_assert.h>
#include <ac_dispatcher.h>
#include <ac_memmgr.h>
#include <ac_msg.h>
#include <ac_printf.h>
#include <ac_debug_printf.h>
#include <ac_receptor.h>
#include <ac_string.h>
#include <ac_thread.h>

#if AC_PLATFORM == pc_x86_64
extern void remove_zombies(void);
#endif

typedef struct DispatchThreadParams DispatchThreadParams;

/**
 * A opaque component info for an AcComp
 */
typedef struct AcCompInfo {
  AcCompMgr* mgr;
  AcComp* comp;
  AcDispatchableComp* dc;
  ac_u32 comp_idx;
  DispatchThreadParams* dtp;
} AcCompInfo;

/**
 * Parameters for each thread to manage its components
 */
typedef struct DispatchThreadParams {
  ac_bool thread_started;
  ac_thread_hdl_t thread_hdl;
  AcDispatcher* d;
  ac_u32 max_comps;
  AcCompInfo** cis;
  AcReceptor* done;
  AcReceptor* ready;
  AcReceptor* waiting;
  ac_bool stop_processing_msgs;
} DispatchThreadParams;

typedef struct AcCompMgr {
  AcCompInfo* comp_infos;     // Array of AcCompInfo objects being managed
                              // across all of the threads
  ac_u32 comp_infos_max_count;// Number of elements in comp_infos array
  DispatchThreadParams* dtps; // Array of DispathThreadParams, one for each thread
  ac_u32 max_dtps;            // Number of threads in the dtps array
  ac_u32 next_dtps;           // Next thread
} AcCompMgr;

/**
 * A thread which dispatches message to its components.
 */
static void* dispatch_thread(void *param) {
  DispatchThreadParams* params = (DispatchThreadParams*)(param);

  ac_debug_printf("dispatch_thread:+starting params=%p\n", params);

  // Get a dispatcher and add a queue and message processor
  params->d = AcDispatcher_get(params->max_comps);
  if (params->d == AC_NULL) {
    ac_debug_printf("dispatch_thread: no AcDispatcher params=%p\n", params);
    goto done;
  }

  // Create the waiting receptor and init our stop_processing_msgs flag
  params->waiting = AcReceptor_get();
  ac_assert(params->waiting != AC_NULL);
  __atomic_store_n(&params->stop_processing_msgs, AC_FALSE, __ATOMIC_RELEASE);

  // Signal dispatch_thread is ready
  AcReceptor_signal(params->ready);

  // Continuously dispatch messages until we're told to stop
  while (__atomic_load_n(&params->stop_processing_msgs, __ATOMIC_ACQUIRE) == AC_FALSE) {
    if (!AcDispatcher_dispatch(params->d)) {
      ac_debug_printf("dispatch_thread: waiting\n");
{'fraction_non_alphanumeric': 0.09223140495867768, 'fraction_numerical': 0.007272727272727273, 'mean_word_length': 4.202923473774721, 'pattern_counts': {'":': 0, '<': 25, '<?xml version=': 0, '>': 43, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 2, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '5545791', 'n_tokens_mistral': 1964, 'n_tokens_neox': 1883, 'n_words': 583}