<commit_before>/**
 *   Copyright 2017 <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.
 */
#include "worker_manager.h"

worker_ret_t worker_manager_clone(worker_t *worker,
								 worker_start_fn worker_start,
								 gru_status_t *status) {
	if (!worker_list_init(status)) {
		return false;
	}

	logger_t logger = gru_logger_get();

	logger(INFO, "Creating %d concurrent operations", worker->options->parallel_count);
	for (uint16_t i = 0; i < worker->options->parallel_count; i++) {
		pid_t child = fork();

		if (child == 0) {
			worker_snapshot_t snapshot = {0};

			const options_t *options = get_options_object();

			remap_log(options->logdir, worker->name, getppid(), getpid(), stderr, status);

			naming_info_t naming_info = {0};

			naming_info.source = worker->name;
			naming_info.location = options->logdir;
			naming_info.pid = getpid();
			naming_info.ppid = getppid();

			bool nmret = naming_initialize_writer(worker->writer,
												  worker->report_format,
												  worker->naming_options,
												  &naming_info,
												  status);
			if (!nmret) {
				logger(ERROR, "Error initializing performance report writer: %s",
					   status->message);
				return WORKER_FAILURE;
			}

			install_interrupt_handler();
			worker_ret_t wret = worker_start(worker, &snapshot, status);
			if (wret != WORKER_SUCCESS) {
				logger(ERROR, "Unable to execute worker: %s", status->message);
			}

			logger(INFO, "Test execution terminated");
			return wret | WORKER_CHILD;
		} else {
			worker_wait_setup();

			worker_info_t *worker_info = worker_info_new(worker, child, status);
			if (!worker_info) {
				logger(ERROR, "Unable to create worker info: %s", status->message);

				return WORKER_FAILURE;
			}

			if (!worker_list_append(worker_info, status)) {
				kill(worker_info->child, SIGKILL);
				worker_info_destroy(&worker_info);

				return WORKER_FAILURE;
			}
		}
	}

	return WORKER_SUCCESS;
}

static bool worker_manager_update_snapshot(worker_info_t *worker_info) {
	bool ret;

	ret = shr_buff_read(worker_info->shr, &worker_info->snapshot,
						sizeof(worker_snapshot_t));

	if (!ret) {
		logger_t logger = gru_logger_get();

		logger(WARNING,
			   "Unable to obtain performance snapshot from worker %d",
			   worker_info->child);
	}

	return ret;
}

static bool worker_manager_watchdog(worker_handler_t *handler, gru_status_t *status) {
	gru_node_t *node = NULL;
	logger_t logger = gru_logger_get();

	worker_list_lock();
	node = worker_list_root();

	while (node) {
		worker_info_t *worker_info = gru_node_get_data_ptr(worker_info_t, node);

		int wstatus = 0;
		pid_t pid = waitpid(worker_info->child, &wstatus, WNOHANG);

		// waitpid returns 0 if WNOHANG and there's no change of state for the process
		if (pid == 0) {
			worker_manager_update_snapshot(worker_info);

			if (handler->flags & WRK_HANDLE_PRINT) {
				handler->print(worker_info);
			}

			if (handler->flags & WRK_HANDLE_EVAL) {
				if (!handler->eval(worker_info, status)) {
					logger(DEBUG, "Worker handler eval failed: %s", status->message);

					worker_list_unlock();
					return false;
				}
			}

			node = node->next;
		} else {
			if (WIFEXITED(wstatus)) {
				logger(INFO,
					   "Child %d finished with status %d",
					   worker_info->child,
					   WEXITSTATUS(wstatus));
			} else if (WIFSIGNALED(wstatus)) {
				logger(INFO,
					   "Child %d received a signal %d",
					   worker_info->child,
					   WTERMSIG(wstatus));
			} else if (WIFSTOPPED(wstatus)) {
				logger(
					ERROR, "Child %d stopped %d", worker_info->child, WSTOPSIG(wstatus));
			}


			node = worker_list_remove_unlocked(node);
			worker_info_destroy(&worker_info);
		}
	}

	worker_list_unlock();
	return true;
}

void worker_manager_watchdog_loop(worker_handler_t *handler, gru_status_t *status) {
	const int wait_time = 250000;

	uint32_t count = worker_list_count();
	while (worker_list_is_running() && count > 0) {
		mpt_trace("There are still %d children running", count);
		if (worker_manager_watchdog(handler, status)) {
			usleep(wait_time);
			count = worker_list_count();
		} else {
			break;
		}
	}
}

bool worker_manager_stop() {
	gru_node_t *node = NULL;
	logger_t logger = gru_logger_get();

	worker_list_lock();
	node = worker_list_root();

	while (node) {
		worker_info_t *worker_info = gru_node_get_data_ptr(worker_info_t, node);<commit_msg>FIX: to ensure that child process errors are evaluated and reported<commit_after>/**
 *   Copyright 2017 <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.
 */
#include "worker_manager.h"

worker_ret_t worker_manager_clone(worker_t *worker,
								 worker_start_fn worker_start,
								 gru_status_t *status) {
	if (!worker_list_init(status)) {
		return false;
	}

	logger_t logger = gru_logger_get();

	logger(INFO, "Creating %d concurrent operations", worker->options->parallel_count);
	for (uint16_t i = 0; i < worker->options->parallel_count; i++) {
		pid_t child = fork();

		if (child == 0) {
			worker_snapshot_t snapshot = {0};

			const options_t *options = get_options_object();

			remap_log(options->logdir, worker->name, getppid(), getpid(), stderr, status);

			naming_info_t naming_info = {0};

			naming_info.source = worker->name;
			naming_info.location = options->logdir;
			naming_info.pid = getpid();
			naming_info.ppid = getppid();

			bool nmret = naming_initialize_writer(worker->writer,
												  worker->report_format,
												  worker->naming_options,
												  &naming_info,
												  status);
			if (!nmret) {
				logger(ERROR, "Error initializing performance report writer: %s",
					   status->message);
				return WORKER_FAILURE;
			}

			install_interrupt_handler();
			worker_ret_t wret = worker_start(worker, &snapshot, status);
			if (wret != WORKER_SUCCESS) {
				logger(ERROR, "Unable to execute worker: %s", status->message);
			}

			logger(INFO, "Test execution terminated");
			return wret | WORKER_CHILD;
		} else {
			worker_wait_setup();

			worker_info_t *worker_info = worker_info_new(worker, child, status);
			if (!worker_info) {
				logger(ERROR, "Unable to create worker info: %s", status->message);

				return WORKER_FAILURE;
			}

			if (!worker_list_append(worker_info, status)) {
				kill(worker_info->child, SIGKILL);
				worker_info_destroy(&worker_info);

				return WORKER_FAILURE;
			}
		}
	}

	return WORKER_SUCCESS;
}

static bool worker_manager_update_snapshot(worker_info_t *worker_info) {
	bool ret;

	ret = shr_buff_read(worker_info->shr, &worker_info->snapshot,
						sizeof(worker_snapshot_t));

	if (!ret) {
		logger_t logger = gru_logger_get();

		logger(WARNING,
			   "Unable to obtain performance snapshot from worker %d",
			   worker_info->child);
	}

	return ret;
}

static bool worker_manager_watchdog(worker_handler_t *handler, uint32_t *errors, gru_status_t *status) {
	gru_node_t *node = NULL;
	logger_t logger = gru_logger_get();

	worker_list_lock();
	node = worker_list_root();

	while (node) {
		worker_info_t *worker_info = gru_node_get_data_ptr(worker_info_t, node);

		int wstatus = 0;
		pid_t pid = waitpid(worker_info->child, &wstatus, WNOHANG);

		// waitpid returns 0 if WNOHANG and there's no change of state for the process
		if (pid == 0) {
			worker_manager_update_snapshot(worker_info);

			if (handler->flags & WRK_HANDLE_PRINT) {
				handler->print(worker_info);
			}

			if (handler->flags & WRK_HANDLE_EVAL) {
				if (!handler->eval(worker_info, status)) {
					logger(DEBUG, "Worker handler eval failed: %s", status->message);

					worker_list_unlock();
					return false;
				}
			}

			node = node->next;
		} else {
			if (WIFEXITED(wstatus)) {
				logger(INFO,
					   "Child %d finished with status %d",
					   worker_info->child,
					   WEXITSTATUS(wstatus));
			} else if (WIFSIGNALED(wstatus)) {
				(*errors)++;
				logger(INFO,
					   "Child %d received a signal %d",
					   worker_info->child,
					   WTERMSIG(wstatus));
			} else if (WIFSTOPPED(wstatus)) {
				int sig = WSTOPSIG(wstatus);

				if (sig != SIGTERM) {
					(*errors)++;

					logger(
						ERROR, "Child %d stopped %d", worker_info->child, sig);
				}
			}


			node = worker_list_remove_unlocked(node);
			worker_info_destroy(&worker_info);
		}
	}

	worker_list_unlock();
	return true;
}

void worker_manager_watchdog_loop(worker_handler_t *handler, gru_status_t *status) {
	const int wait_time = 250000;

	uint32_t count = worker_list_count();
	uint32_t errors;

	while (worker_list_is_running() && count > 0) {
		mpt_trace("There are still %d children running", count);
		if (worker_manager_watchdog(handler, &errors, status)) {
			usleep(wait_time);
			count = worker_list_count();
		} else {
			break;
		}
	}

	if (errors > 0) {
		gru_status_set(status,  GRU_FAILURE, "There were %"PRIu32" child process that ended with failure", errors);
	}
}

bool worker_manager_stop() {
	gru_node_t *node = NULL;
	logger_t logger = gru_logger_get();

	worker_list_lock();
	node = worker_list_root();

	while (node) {
		worker_info_t *worker_info = gru_node_get_data_ptr(worker_info_t, node);
{'fraction_non_alphanumeric': 0.11470384653416395, 'fraction_numerical': 0.005834074953030752, 'mean_word_length': 3.719552029864676, 'pattern_counts': {'":': 0, '<': 7, '<?xml version=': 0, '>': 64, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '746588', 'n_tokens_mistral': 3976, 'n_tokens_neox': 3326, 'n_words': 831}