<commit_before>/*
 * Copyright 2017 otavio.
 *
 * 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 <common/gru_alloc.h>

#include "contrib/options.h"
#include "scheduler.h"

gru_list_t *list;

static void *probe_scheduler_run(void *en) {
	logger_t logger = gru_logger_get();
	probe_entry_t *entry = (probe_entry_t *) en;
	gru_status_t status = gru_status_new();

	while (!entry->cancel) {
		logger(DEBUG, "Running probe %s", entry->name());
		entry->collect(&status);
		if (gru_status_error(&status)) {
			logger(ERROR, "Probe %s error: %s", entry->name(), status.message);

			break;
		}
	}

	logger(DEBUG, "Finished running probe %s", entry->name());
	return NULL;
}

static void probe_scheduler_launch_probe(const void *nodedata, void *payload) {
	probe_entry_t *entry = (probe_entry_t *) nodedata;
	const options_t *options = get_options_object();
	logger_t logger = gru_logger_get();

	gru_status_t status = gru_status_new();
	bool pret = entry->init(options, &status);
	if (!pret) {
		logger(ERROR, "Unable to initialize probe %s: %s", entry->name(), status.message);

		return;
	}

	int sys_ret = pthread_create(&entry->thread, NULL, probe_scheduler_run, entry);
	if (sys_ret != 0) {
		logger(ERROR, "Unable to create probe thread");

		return;
	}

	logger(INFO, "Probe %s created", entry->name());
}

static probe_entry_t *probe_scheduler_load_probe(const char *lib, const char *name) {
	void *handle = NULL;
	char *error = NULL;

	handle = dlopen(lib, RTLD_NOW);
	if (!handle) {
		fprintf(stderr, "Unable to open handle: %s\n", dlerror());

		return NULL;
	}

	dlerror();

	probe_entry_t *(*new_entry)(gru_status_t *);

	new_entry = (probe_entry_t * (*) (gru_status_t *) ) dlsym(handle, name);
	error = dlerror();
	if (error) {
		fprintf(stderr, "Unable to open handle: %s\n", error);

		return NULL;
	}

	gru_status_t status = gru_status_new();
	probe_entry_t *ret = (*new_entry)(&status);

	if (gru_status_error(&status)) {
		fprintf(stderr, "%s\n", status.message);
	}
	ret->handle = handle;

	return ret;
}

bool probe_scheduler_start(gru_status_t *status) {
	const options_t *options = get_options_object();
	logger_t logger = gru_logger_get();

	if (!options->probing) { 
		logger(INFO, "Disabling probes"); 
		return true;
	}

	if (!options->logdir) {
		logger(ERROR, "Log directory is mandatory for probes");
		return false;
	}

	list = gru_list_new(status);
	if (!list) {
		return false;
	}

	
	uint32_t num_mod = gru_list_count(options->probes);

	for (uint32_t i = 0; i < num_mod; i++) {
		const gru_node_t *node = gru_list_get(options->probes, i);
		char *mod_name = (char *) node->data;
		char lib[256] = {0};
		char ename[256] = {0};

		snprintf(lib, sizeof(lib) - 1, "libmpt-probe-%s.so", mod_name);
		snprintf(ename, sizeof(ename) - 1, "%s_entry", mod_name);

		logger(DEBUG, "Loading symbol %s@%s", ename, lib);

		probe_entry_t *net = probe_scheduler_load_probe(lib, ename);
		gru_list_append(list, net);
	}

	gru_list_for_each(list, probe_scheduler_launch_probe, NULL);

	return true;
}

static void probe_scheduler_stop_probe(const void *nodedata, void *payload) {
	probe_entry_t *entry = (probe_entry_t *) nodedata;

	entry->stop();
	entry->cancel = true;
	dlclose(entry->handle);
}

void probe_scheduler_stop() {
	const options_t *options = get_options_object();
	
	if (options->probing) { 
		gru_list_for_each(list, probe_scheduler_stop_probe, NULL);	
	}
}<commit_msg>FIX: handle the errors if unable to load a probe<commit_after>/*
 * Copyright 2017 otavio.
 *
 * 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 <common/gru_alloc.h>

#include "contrib/options.h"
#include "scheduler.h"

gru_list_t *list;

static void *probe_scheduler_run(void *en) {
	logger_t logger = gru_logger_get();
	probe_entry_t *entry = (probe_entry_t *) en;
	gru_status_t status = gru_status_new();

	while (!entry->cancel) {
		logger(DEBUG, "Running probe %s", entry->name());
		entry->collect(&status);
		if (gru_status_error(&status)) {
			logger(ERROR, "Probe %s error: %s", entry->name(), status.message);

			break;
		}
	}

	logger(DEBUG, "Finished running probe %s", entry->name());
	return NULL;
}

static void probe_scheduler_launch_probe(const void *nodedata, void *payload) {
	probe_entry_t *entry = (probe_entry_t *) nodedata;
	const options_t *options = get_options_object();
	logger_t logger = gru_logger_get();

	gru_status_t status = gru_status_new();
	bool pret = entry->init(options, &status);
	if (!pret) {
		logger(ERROR, "Unable to initialize probe %s: %s", entry->name(), status.message);

		return;
	}

	int sys_ret = pthread_create(&entry->thread, NULL, probe_scheduler_run, entry);
	if (sys_ret != 0) {
		logger(ERROR, "Unable to create probe thread");

		return;
	}

	logger(INFO, "Probe %s created", entry->name());
}

static probe_entry_t *probe_scheduler_load_probe(const char *lib, const char *name) {
	logger_t logger = gru_logger_get();
	void *handle = NULL;
	char *error = NULL;

	handle = dlopen(lib, RTLD_NOW);
	if (!handle) {
		logger(ERROR, "Unable to open handle: %s\n", dlerror());

		fflush(NULL);
		return NULL;
	}

	dlerror();

	probe_entry_t *(*new_entry)(gru_status_t *);

	new_entry = (probe_entry_t * (*) (gru_status_t *) ) dlsym(handle, name);
	error = dlerror();
	if (error) {
		logger(ERROR, "Unable to open handle: %s\n", error);

		return NULL;
	}

	gru_status_t status = gru_status_new();
	probe_entry_t *ret = (*new_entry)(&status);

	if (gru_status_error(&status)) {
		fprintf(stderr, "%s\n", status.message);
	}
	ret->handle = handle;

	return ret;
}

bool probe_scheduler_start(gru_status_t *status) {
	const options_t *options = get_options_object();
	logger_t logger = gru_logger_get();

	if (!options->probing) { 
		logger(INFO, "Disabling probes"); 
		return true;
	}

	if (!options->logdir) {
		logger(ERROR, "Log directory is mandatory for probes");
		return false;
	}

	list = gru_list_new(status);
	if (!list) {
		return false;
	}

	
	uint32_t num_mod = gru_list_count(options->probes);

	for (uint32_t i = 0; i < num_mod; i++) {
		const gru_node_t *node = gru_list_get(options->probes, i);
		char *mod_name = (char *) node->data;
		char lib[256] = {0};
		char ename[256] = {0};

		snprintf(lib, sizeof(lib) - 1, "libmpt-probe-%s.so", mod_name);
		snprintf(ename, sizeof(ename) - 1, "%s_entry", mod_name);

		logger(DEBUG, "Loading symbol %s@%s", ename, lib);

		probe_entry_t *probe = probe_scheduler_load_probe(lib, ename);
		if (probe) {
			gru_list_append(list, probe);
		}
		else {
			logger(ERROR, "Unable load probe: probe-%s", mod_name);
		}
	}

	gru_list_for_each(list, probe_scheduler_launch_probe, NULL);

	return true;
}

static void probe_scheduler_stop_probe(const void *nodedata, void *payload) {
	probe_entry_t *entry = (probe_entry_t *) nodedata;

	entry->stop();
	entry->cancel = true;
	dlclose(entry->handle);
}

void probe_scheduler_stop() {
	const options_t *options = get_options_object();
	
	if (options->probing) { 
		gru_list_for_each(list, probe_scheduler_stop_probe, NULL);	
	}
}
{'fraction_non_alphanumeric': 0.13947105788423153, 'fraction_numerical': 0.005988023952095809, 'mean_word_length': 4.645774647887324, 'pattern_counts': {'":': 0, '<': 7, '<?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': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '5874543', 'n_tokens_mistral': 2986, 'n_tokens_neox': 2802, 'n_words': 750}