<commit_before>/*
 * ivykis, an event handling library
 * Copyright (C) 2010 <NAME>
 * Dedicated to <NAME>.
 *
 * This library is free software; you can redistribute it and/or modify
 * it under the terms of the GNU Lesser General Public License version
 * 2.1 as published by the Free Software Foundation.
 *
 * This library is distributed in the hope that it will be useful,
 * but WITHOUT ANY WARRANTY; without even the implied warranty of
 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 * GNU Lesser General Public License version 2.1 for more details.
 *
 * You should have received a copy of the GNU Lesser General Public
 * License version 2.1 along with this library; if not, write to the
 * Free Software Foundation, Inc., 51 Franklin Street - Fifth Floor,
 * Boston, MA 02110-1301, USA.
 */

#include <stdio.h>
#include <stdlib.h>
#include <iv.h>
#include <iv_list.h>
#include <iv_thread.h>
#include <iv_tls.h>
#include <iv_work.h>
#include <pthread.h>
#include <syslog.h>

/* data structures **********************************************************/
struct work_pool_priv {
	pthread_mutex_t		lock;
	struct iv_event		ev;
	int			shutting_down;
	int			started_threads;
	struct iv_list_head	idle_threads;
	void			*cookie;
	void			(*thread_start)(void *cookie);
	void			(*thread_stop)(void *cookie);
	struct iv_list_head	work_items;
	struct iv_list_head	work_done;
};

struct work_pool_thread {
	struct work_pool_priv	*pool;
	struct iv_list_head	list;
	int			kicked;
	struct iv_event		kick;
	struct iv_timer		idle_timer;
};


/* worker thread ************************************************************/
static void __iv_work_thread_die(struct work_pool_thread *thr)
{
	struct work_pool_priv *pool = thr->pool;

	if (thr->kicked) {
		syslog(LOG_CRIT, "__iv_work_thread_die: called on "
				 "kicked thread");
		abort();
	}

	iv_event_unregister(&thr->kick);
	free(thr);

	pool->started_threads--;

	if (pool->thread_stop != NULL)
		pool->thread_stop(pool->cookie);

	if (pool->shutting_down && !pool->started_threads)
		iv_event_post(&pool->ev);
}

static void iv_work_thread_got_event(void *_thr)
{
	struct work_pool_thread *thr = _thr;
	struct work_pool_priv *pool = thr->pool;

	pthread_mutex_lock(&pool->lock);

	thr->kicked = 0;

	iv_list_del(&thr->list);
	iv_timer_unregister(&thr->idle_timer);

	while (1) {
		struct iv_work_item *work;

		if (iv_list_empty(&pool->work_items))
			break;

		work = iv_container_of(pool->work_items.next,
				       struct iv_work_item, list);
		iv_list_del(&work->list);

		pthread_mutex_unlock(&pool->lock);
		work->work(work->cookie);
		iv_invalidate_now();
		pthread_mutex_lock(&pool->lock);

		iv_list_add_tail(&work->list, &pool->work_done);
		iv_event_post(&pool->ev);
	}

	if (!pool->shutting_down) {
		iv_list_add(&thr->list, &pool->idle_threads);
		iv_validate_now();
		thr->idle_timer.expires = iv_now;
		thr->idle_timer.expires.tv_sec += 10;
		iv_timer_register(&thr->idle_timer);
	} else {
		__iv_work_thread_die(thr);
	}

	pthread_mutex_unlock(&pool->lock);
}

static void iv_work_thread_idle_timeout(void *_thr)
{
	struct work_pool_thread *thr = _thr;
	struct work_pool_priv *pool = thr->pool;

	pthread_mutex_lock(&pool->lock);
	
	if (thr->kicked) {
		thr->idle_timer.expires = iv_now;
		thr->idle_timer.expires.tv_sec += 10;
		iv_timer_register(&thr->idle_timer);

		pthread_mutex_unlock(&pool->lock);

		return;
	}

	iv_list_del(&thr->list);
	__iv_work_thread_die(thr);

	pthread_mutex_unlock(&pool->lock);
}

static void iv_work_thread(void *_thr)
{
	struct work_pool_thread *thr = _thr;
	struct work_pool_priv *pool = thr->pool;

	iv_init();

	thr->kicked = 0;

	IV_EVENT_INIT(&thr->kick);
	thr->kick.cookie = thr;
	thr->kick.handler = iv_work_thread_got_event;
	iv_event_register(&thr->kick);

	IV_TIMER_INIT(&thr->idle_timer);
	iv_validate_now();
	thr->idle_timer.expires = iv_now;
	thr->idle_timer.expires.tv_sec += 10;
	thr->idle_timer.cookie = thr;
	thr->idle_timer.handler = iv_work_thread_idle_timeout;
	iv_timer_register(&thr->idle_timer);

	pthread_mutex_lock(&pool->lock);
	iv_list_add_tail(&thr->list, &pool->idle_threads);
	pthread_mutex_unlock(&pool->lock);

	if (pool->thread_start != NULL)
		pool->thread_start(pool->cookie);

	iv_event_post(&thr->kick);

	iv_main();

	iv_deinit();
}


/* main thread **************************************************************/
static void iv_work_event(void *_pool)
{
	struct work_pool_priv *pool = _pool;

	pthread_mutex_lock(&pool->lock);

	while (!iv_list_empty(&pool->work_done)) {
		struct iv_work_item *work;

		work = iv_container_of(pool->work_done.next,
				       struct iv_work_item, list);

		iv_list_del(&work->list);

		pthread_mutex_unlock(&pool->lock);
		work->completion(work->cookie);
		pthread_mutex_lock(&pool->lock);
	}

	if (pool->shutting_down && !pool->started_threads) {
		pthread_mutex_unlock(&pool->lock);
		pthread_mutex_destroy(&pool->lock);
		iv_event_unregister(&pool->ev);
		free(pool);
		return;
	}

	pthread_mutex_unlock(&pool->lock);
}

int iv_work_pool_create(struct iv_work_pool *this)
{
	struct work_pool_priv *pool;
	int ret;

	pool = malloc(sizeof(*pool));
	if (pool == NULL)
		return -1;

	ret = pthread_mutex_init(&pool->lock, NULL);
	if (ret) {
		free(pool);
		return -1;
	}

	IV_EVENT_INIT(&pool->ev);
	pool->ev.cookie = pool;
	pool->ev.handler = iv_work_event;
	iv_event_register(&pool->ev);

	pool->shutting_down = 0;
	pool->started_threads = 0;
	INIT_IV_LIST_HEAD(&pool->idle_threads);
	pool->cookie = this->cookie;
	pool->thread_start = this->thread_start;
	pool->thread_stop = this->thread_stop;
	INIT_IV_LIST_HEAD(&pool->work_items);
	INIT_IV_LIST_HEAD(&pool->work_done);

	this->priv = pool;

	return 0;
}

void iv_work_pool_put(struct iv_work_pool *this)
{
	struct work_pool_priv *pool = this->priv;
	struct iv_list_head *ilh;

	pthread_mutex_lock(&pool->lock);

	this->priv = NULL;
	pool->shutting_down = 1;

	if (!pool->started_threads) {
		pthread_mutex_unlock(&pool->lock);
		iv_event_post(&pool->ev);
		return;
	}

	iv_list_for_each (ilh, &pool->idle_threads) {
		struct work_pool_thread *thr;

		thr = iv_container_of(ilh, struct work_pool_thread, list);
		iv_event_post(&thr->kick);
	}

	pthread_mutex_unlock(&pool->lock);
}

static int iv_work_start_thread(struct work_pool_priv *pool)
{
	struct work_pool_thread *thr;
	char name[512];
	int ret;

	thr = malloc(sizeof(*thr));
	if (thr == NULL)
		return -1;

	thr->pool = pool;

	snprintf(name, sizeof(name), "iv_work pool %p thread %p", pool, thr);

	ret = iv_thread_create(name, iv_work_thread, thr);
	if (ret < 0) {
		free(thr);
		return -1;
	}

	pool->started_threads++;

	return 0;
}

static void
iv_work_submit_pool(struct iv_work_pool *this, struct iv_work_item *work)
{
	struct work_pool_priv *pool = this->priv;

	pthread_mutex_lock(&pool->lock);

	iv_list_add_tail(&work->list, &pool->work_items);

	if (!iv_list_empty(&pool->idle_threads)) {
		struct work_pool_thread *thr;

		thr = iv_container_of(pool->idle_threads.next,
				      struct work_pool_thread, list);
		thr->kicked = 1;
		iv_event_post(&thr->kick);
	} else if (pool->started_threads < this->max_threads) {
		iv_work_start_thread(pool);
	}

	pthread_mutex_unlock(&pool->lock);
}

struct iv_work_thr_info {
	struct iv_task		task;
	struct iv_list_head	work_items;
};

static void iv_work_handle_local(void *_tinfo);

static void iv_work_tls_init_thread(void *_tinfo)
{
	struct iv_work_thr_info *tinfo = _tinfo;

	IV_TASK_INIT(&tinfo->task);
	tinfo->task.cookie = tinfo;
	tinfo->task.handler = iv_work_handle_local;

	INIT_IV_LIST_HEAD(&tinfo->work_items);
}

static struct iv_tls_user iv_work_tls_user = {
	.sizeof_state	= sizeof(struct iv_work_thr_info),
	.init_thread	= iv_work_tls_init_thread,
};

static void iv_work_tls_init(void) __attribute__((constructor));
static void iv_work_tls_init(void)
{
	iv_tls_user_register(&iv_work_tls_user);
}

static void iv_work_handle_local(void *_tinfo)
{
	struct iv_work_thr_info *tinfo = _tinfo;

	while (!iv_list_empty(&tinfo->work_items)) {
		struct iv_work_item *work;

		work = iv_container_of(tinfo->work_items.next,
				       struct iv_work_item, list);

		iv_list_del(&work->list);

		work->work(work->cookie);
		work->completion(work->cookie);
	}
}

static void iv_work_submit_local(struct iv_work_item *work)
{
	struct iv_work_thr_info *tinfo = iv_tls_user_ptr(&iv_work_tls_user);
	int was_empty;

	was_empty = iv_list_empty(&tinfo->work_items);

	iv_list_add_tail(&work->list, &tinfo->work_items);
	if (was_empty)
		iv_task_register(&tinfo->task);
}

void
iv_work_pool_submit_work(struct iv_work_pool *this, struct iv_work_item *work)
{
	if (this != NULL)
		iv_work_submit_pool(this, work);
	else
		iv_work_submit_local(work);
}
<commit_msg>iv_work: Don't re-kick callback event from worker thread if still pending.<commit_after>/*
 * ivykis, an event handling library
 * Copyright (C) 2010 <NAME>
 * Dedicated to <NAME>.
 *
 * This library is free software; you can redistribute it and/or modify
 * it under the terms of the GNU Lesser General Public License version
 * 2.1 as published by the Free Software Foundation.
 *
 * This library is distributed in the hope that it will be useful,
 * but WITHOUT ANY WARRANTY; without even the implied warranty of
 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 * GNU Lesser General Public License version 2.1 for more details.
 *
 * You should have received a copy of the GNU Lesser General Public
 * License version 2.1 along with this library; if not, write to the
 * Free Software Foundation, Inc., 51 Franklin Street - Fifth Floor,
 * Boston, MA 02110-1301, USA.
 */

#include <stdio.h>
#include <stdlib.h>
#include <iv.h>
#include <iv_list.h>
#include <iv_thread.h>
#include <iv_tls.h>
#include <iv_work.h>
#include <pthread.h>
#include <syslog.h>

/* data structures **********************************************************/
struct work_pool_priv {
	pthread_mutex_t		lock;
	struct iv_event		ev;
	int			shutting_down;
	int			started_threads;
	struct iv_list_head	idle_threads;
	void			*cookie;
	void			(*thread_start)(void *cookie);
	void			(*thread_stop)(void *cookie);
	struct iv_list_head	work_items;
	struct iv_list_head	work_done;
};

struct work_pool_thread {
	struct work_pool_priv	*pool;
	struct iv_list_head	list;
	int			kicked;
	struct iv_event		kick;
	struct iv_timer		idle_timer;
};


/* worker thread ************************************************************/
static void __iv_work_thread_die(struct work_pool_thread *thr)
{
	struct work_pool_priv *pool = thr->pool;

	if (thr->kicked) {
		syslog(LOG_CRIT, "__iv_work_thread_die: called on "
				 "kicked thread");
		abort();
	}

	iv_event_unregister(&thr->kick);
	free(thr);

	pool->started_threads--;

	if (pool->thread_stop != NULL)
		pool->thread_stop(pool->cookie);

	if (pool->shutting_down && !pool->started_threads)
		iv_event_post(&pool->ev);
}

static void iv_work_thread_got_event(void *_thr)
{
	struct work_pool_thread *thr = _thr;
	struct work_pool_priv *pool = thr->pool;

	pthread_mutex_lock(&pool->lock);

	thr->kicked = 0;

	iv_list_del(&thr->list);
	iv_timer_unregister(&thr->idle_timer);

	while (1) {
		struct iv_work_item *work;

		if (iv_list_empty(&pool->work_items))
			break;

		work = iv_container_of(pool->work_items.next,
				       struct iv_work_item, list);
		iv_list_del(&work->list);

		pthread_mutex_unlock(&pool->lock);
		work->work(work->cookie);
		iv_invalidate_now();
		pthread_mutex_lock(&pool->lock);

		if (iv_list_empty(&pool->work_done))
			iv_event_post(&pool->ev);
		iv_list_add_tail(&work->list, &pool->work_done);
	}

	if (!pool->shutting_down) {
		iv_list_add(&thr->list, &pool->idle_threads);
		iv_validate_now();
		thr->idle_timer.expires = iv_now;
		thr->idle_timer.expires.tv_sec += 10;
		iv_timer_register(&thr->idle_timer);
	} else {
		__iv_work_thread_die(thr);
	}

	pthread_mutex_unlock(&pool->lock);
}

static void iv_work_thread_idle_timeout(void *_thr)
{
	struct work_pool_thread *thr = _thr;
	struct work_pool_priv *pool = thr->pool;

	pthread_mutex_lock(&pool->lock);
	
	if (thr->kicked) {
		thr->idle_timer.expires = iv_now;
		thr->idle_timer.expires.tv_sec += 10;
		iv_timer_register(&thr->idle_timer);

		pthread_mutex_unlock(&pool->lock);

		return;
	}

	iv_list_del(&thr->list);
	__iv_work_thread_die(thr);

	pthread_mutex_unlock(&pool->lock);
}

static void iv_work_thread(void *_thr)
{
	struct work_pool_thread *thr = _thr;
	struct work_pool_priv *pool = thr->pool;

	iv_init();

	thr->kicked = 0;

	IV_EVENT_INIT(&thr->kick);
	thr->kick.cookie = thr;
	thr->kick.handler = iv_work_thread_got_event;
	iv_event_register(&thr->kick);

	IV_TIMER_INIT(&thr->idle_timer);
	iv_validate_now();
	thr->idle_timer.expires = iv_now;
	thr->idle_timer.expires.tv_sec += 10;
	thr->idle_timer.cookie = thr;
	thr->idle_timer.handler = iv_work_thread_idle_timeout;
	iv_timer_register(&thr->idle_timer);

	pthread_mutex_lock(&pool->lock);
	iv_list_add_tail(&thr->list, &pool->idle_threads);
	pthread_mutex_unlock(&pool->lock);

	if (pool->thread_start != NULL)
		pool->thread_start(pool->cookie);

	iv_event_post(&thr->kick);

	iv_main();

	iv_deinit();
}


/* main thread **************************************************************/
static void iv_work_event(void *_pool)
{
	struct work_pool_priv *pool = _pool;

	pthread_mutex_lock(&pool->lock);

	while (!iv_list_empty(&pool->work_done)) {
		struct iv_work_item *work;

		work = iv_container_of(pool->work_done.next,
				       struct iv_work_item, list);

		iv_list_del(&work->list);

		pthread_mutex_unlock(&pool->lock);
		work->completion(work->cookie);
		pthread_mutex_lock(&pool->lock);
	}

	if (pool->shutting_down && !pool->started_threads) {
		pthread_mutex_unlock(&pool->lock);
		pthread_mutex_destroy(&pool->lock);
		iv_event_unregister(&pool->ev);
		free(pool);
		return;
	}

	pthread_mutex_unlock(&pool->lock);
}

int iv_work_pool_create(struct iv_work_pool *this)
{
	struct work_pool_priv *pool;
	int ret;

	pool = malloc(sizeof(*pool));
	if (pool == NULL)
		return -1;

	ret = pthread_mutex_init(&pool->lock, NULL);
	if (ret) {
		free(pool);
		return -1;
	}

	IV_EVENT_INIT(&pool->ev);
	pool->ev.cookie = pool;
	pool->ev.handler = iv_work_event;
	iv_event_register(&pool->ev);

	pool->shutting_down = 0;
	pool->started_threads = 0;
	INIT_IV_LIST_HEAD(&pool->idle_threads);
	pool->cookie = this->cookie;
	pool->thread_start = this->thread_start;
	pool->thread_stop = this->thread_stop;
	INIT_IV_LIST_HEAD(&pool->work_items);
	INIT_IV_LIST_HEAD(&pool->work_done);

	this->priv = pool;

	return 0;
}

void iv_work_pool_put(struct iv_work_pool *this)
{
	struct work_pool_priv *pool = this->priv;
	struct iv_list_head *ilh;

	pthread_mutex_lock(&pool->lock);

	this->priv = NULL;
	pool->shutting_down = 1;

	if (!pool->started_threads) {
		pthread_mutex_unlock(&pool->lock);
		iv_event_post(&pool->ev);
		return;
	}

	iv_list_for_each (ilh, &pool->idle_threads) {
		struct work_pool_thread *thr;

		thr = iv_container_of(ilh, struct work_pool_thread, list);
		iv_event_post(&thr->kick);
	}

	pthread_mutex_unlock(&pool->lock);
}

static int iv_work_start_thread(struct work_pool_priv *pool)
{
	struct work_pool_thread *thr;
	char name[512];
	int ret;

	thr = malloc(sizeof(*thr));
	if (thr == NULL)
		return -1;

	thr->pool = pool;

	snprintf(name, sizeof(name), "iv_work pool %p thread %p", pool, thr);

	ret = iv_thread_create(name, iv_work_thread, thr);
	if (ret < 0) {
		free(thr);
		return -1;
	}

	pool->started_threads++;

	return 0;
}

static void
iv_work_submit_pool(struct iv_work_pool *this, struct iv_work_item *work)
{
	struct work_pool_priv *pool = this->priv;

	pthread_mutex_lock(&pool->lock);

	iv_list_add_tail(&work->list, &pool->work_items);

	if (!iv_list_empty(&pool->idle_threads)) {
		struct work_pool_thread *thr;

		thr = iv_container_of(pool->idle_threads.next,
				      struct work_pool_thread, list);
		thr->kicked = 1;
		iv_event_post(&thr->kick);
	} else if (pool->started_threads < this->max_threads) {
		iv_work_start_thread(pool);
	}

	pthread_mutex_unlock(&pool->lock);
}

struct iv_work_thr_info {
	struct iv_task		task;
	struct iv_list_head	work_items;
};

static void iv_work_handle_local(void *_tinfo);

static void iv_work_tls_init_thread(void *_tinfo)
{
	struct iv_work_thr_info *tinfo = _tinfo;

	IV_TASK_INIT(&tinfo->task);
	tinfo->task.cookie = tinfo;
	tinfo->task.handler = iv_work_handle_local;

	INIT_IV_LIST_HEAD(&tinfo->work_items);
}

static struct iv_tls_user iv_work_tls_user = {
	.sizeof_state	= sizeof(struct iv_work_thr_info),
	.init_thread	= iv_work_tls_init_thread,
};

static void iv_work_tls_init(void) __attribute__((constructor));
static void iv_work_tls_init(void)
{
	iv_tls_user_register(&iv_work_tls_user);
}

static void iv_work_handle_local(void *_tinfo)
{
	struct iv_work_thr_info *tinfo = _tinfo;

	while (!iv_list_empty(&tinfo->work_items)) {
		struct iv_work_item *work;

		work = iv_container_of(tinfo->work_items.next,
				       struct iv_work_item, list);

		iv_list_del(&work->list);

		work->work(work->cookie);
		work->completion(work->cookie);
	}
}

static void iv_work_submit_local(struct iv_work_item *work)
{
	struct iv_work_thr_info *tinfo = iv_tls_user_ptr(&iv_work_tls_user);
	int was_empty;

	was_empty = iv_list_empty(&tinfo->work_items);

	iv_list_add_tail(&work->list, &tinfo->work_items);
	if (was_empty)
		iv_task_register(&tinfo->task);
}

void
iv_work_pool_submit_work(struct iv_work_pool *this, struct iv_work_item *work)
{
	if (this != NULL)
		iv_work_submit_pool(this, work);
	else
		iv_work_submit_local(work);
}

{'fraction_non_alphanumeric': 0.1508411747932706, 'fraction_numerical': 0.0050185343598517255, 'mean_word_length': 5.189904694669961, 'pattern_counts': {'":': 0, '<': 29, '<?xml version=': 0, '>': 286, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, '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': '2834289', 'n_tokens_mistral': 7063, 'n_tokens_neox': 6646, 'n_words': 1260}