libqb/lib/log_thread.c

/*
 * Copyright (C) 2011 Red Hat, Inc.
 *
 * All rights reserved.
 *
 * Author: Angus Salkeld <asalkeld@redhat.com>
 *
 * libqb is free software: you can redistribute it and/or modify
 * it under the terms of the GNU Lesser General Public License as published by
 * the Free Software Foundation, either version 2.1 of the License, or
 * (at your option) any later version.
 *
 * libqb 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 for more details.
 *
 * You should have received a copy of the GNU Lesser General Public License
 * along with libqb.  If not, see <http://www.gnu.org/licenses/>.
 */
#include "os_base.h"

#include <pthread.h>
#include <semaphore.h>

#include <qb/qbdefs.h>
#include <qb/qblist.h>
#include <qb/qbutil.h>
#include "log_int.h"

static int wthread_active = QB_FALSE;

static int wthread_should_exit = QB_FALSE;

static qb_thread_lock_t *logt_wthread_lock = NULL;

static QB_LIST_DECLARE(logt_print_finished_records);

static int logt_memory_used = 0;

static int logt_dropped_messages = 0;

static sem_t logt_thread_start;

static sem_t logt_print_finished;

static int logt_sched_param_queued = QB_FALSE;

static int logt_sched_policy;

#if defined(HAVE_PTHREAD_SETSCHEDPARAM)
static struct sched_param logt_sched_param;
#endif /* HAVE_PTHREAD_SETSCHEDPARAM */

static pthread_t logt_thread_id = 0;

static void *qb_logt_worker_thread(void *data) __attribute__ ((noreturn));
static void *
qb_logt_worker_thread(void *data)
{
	struct qb_log_record *rec;
	int dropped = 0;
	int res;

	/* Signal qb_log_thread_start that the initialization may continue */
	sem_post(&logt_thread_start);
	for (;;) {
retry_sem_wait:
		res = sem_wait(&logt_print_finished);
		if (res == -1 && errno == EINTR) {
			goto retry_sem_wait;
		} else if (res == -1) {
			/* This case shouldn't happen */
			pthread_exit(NULL);
		}

		(void)qb_thread_lock(logt_wthread_lock);
		if (wthread_should_exit) {
			int value = -1;

			(void)sem_getvalue(&logt_print_finished, &value);
			if (value == 0) {
				(void)qb_thread_unlock(logt_wthread_lock);
				pthread_exit(NULL);
			}
		}

		rec =
		    qb_list_first_entry(&logt_print_finished_records,
				  struct qb_log_record, list);
		qb_list_del(&rec->list);
		logt_memory_used = logt_memory_used - strlen(rec->buffer) -
		    sizeof(struct qb_log_record) - 1;
		dropped = logt_dropped_messages;
		logt_dropped_messages = 0;
		if (dropped) {
			printf("%d messages lost\n", dropped);
		}

		qb_log_thread_log_write(rec->cs, &rec->timestamp, rec->buffer);

		(void)qb_thread_unlock(logt_wthread_lock);
		free(rec->buffer);
		free(rec);
	}
}

int32_t
qb_log_thread_priority_set(int32_t policy, int32_t priority)
{
	int res = 0;

#if defined(HAVE_PTHREAD_SETSCHEDPARAM)

	logt_sched_policy = policy;

	if (policy == SCHED_OTHER
#ifdef SCHED_IDLE
	    || policy == SCHED_IDLE
#endif
#if defined(SCHED_BATCH) && !defined(QB_DARWIN)
	    || policy == SCHED_BATCH
#endif
	    ) {
		logt_sched_param.sched_priority = 0;
	} else {
		logt_sched_param.sched_priority = priority;
	}
	if (wthread_active == QB_FALSE) {
		logt_sched_param_queued = QB_TRUE;
	} else {
		res = pthread_setschedparam(logt_thread_id, policy,
					    &logt_sched_param);
		if (res != 0) {
			res = -res;
		}
	}
#endif
	return res;
}

int32_t
qb_log_thread_start(void)
{
	int res;
	qb_thread_lock_t *wthread_lock;

	if (wthread_active) {
		return 0;
	}

	wthread_active = QB_TRUE;
	sem_init(&logt_thread_start, 0, 0);
	sem_init(&logt_print_finished, 0, 0);
	errno = 0;
	logt_wthread_lock = qb_thread_lock_create(QB_THREAD_LOCK_SHORT);
	if (logt_wthread_lock == NULL) {
		return errno ? -errno : -1;
	}
	res = pthread_create(&logt_thread_id, NULL,
			     qb_logt_worker_thread, NULL);
	if (res != 0) {
		wthread_active = QB_FALSE;
		(void)qb_thread_lock_destroy(logt_wthread_lock);
		return -res;
	}
	sem_wait(&logt_thread_start);

	if (logt_sched_param_queued) {
		res = qb_log_thread_priority_set(logt_sched_policy,
#if defined(HAVE_PTHREAD_SETSCHEDPARAM)
		                                 logt_sched_param.sched_priority);
#else
		                                 0);
#endif
		if (res != 0) {
			goto cleanup_pthread;
		}
		logt_sched_param_queued = QB_FALSE;
	}

	return 0;

cleanup_pthread:
	wthread_should_exit = QB_TRUE;
	sem_post(&logt_print_finished);
	pthread_join(logt_thread_id, NULL);

	wthread_active = QB_FALSE;
	wthread_lock = logt_wthread_lock;
	logt_wthread_lock = NULL;
	(void)qb_thread_lock_destroy(wthread_lock);

	sem_destroy(&logt_print_finished);
	sem_destroy(&logt_thread_start);

	return res;
}


void
qb_log_thread_pause(struct qb_log_target *t)
{
	if (t->threaded) {
		(void)qb_thread_lock(logt_wthread_lock);
	}
}

void
qb_log_thread_resume(struct qb_log_target *t)
{
	if (t->threaded) {
		(void)qb_thread_unlock(logt_wthread_lock);
	}
}

void
qb_log_thread_log_post(struct qb_log_callsite *cs,
		       struct timespec *timestamp, const char *buffer)
{
	struct qb_log_record *rec;
	size_t buf_size;
	size_t total_size;

	rec = malloc(sizeof(struct qb_log_record));
	if (rec == NULL) {
		return;
	}

	buf_size = strlen(buffer) + 1;
	total_size = sizeof(struct qb_log_record) + buf_size;

	rec->cs = cs;
	rec->buffer = malloc(buf_size);
	if (rec->buffer == NULL) {
		goto free_record;
	}
	memcpy(rec->buffer, buffer, buf_size);

	memcpy(&rec->timestamp, timestamp, sizeof(struct timespec));

	qb_list_init(&rec->list);
	(void)qb_thread_lock(logt_wthread_lock);
	logt_memory_used += total_size;
	if (logt_memory_used > 512000) {
		free(rec->buffer);
		free(rec);
		logt_memory_used = logt_memory_used - total_size;
		logt_dropped_messages += 1;
		(void)qb_thread_unlock(logt_wthread_lock);
		return;

	} else {
		qb_list_add_tail(&rec->list, &logt_print_finished_records);
	}
	(void)qb_thread_unlock(logt_wthread_lock);

	sem_post(&logt_print_finished);
	return;

free_record:
	free(rec);
}

void
qb_log_thread_stop(void)
{
	int res;
	int value;
	struct qb_log_record *rec;

	if (wthread_active == QB_FALSE && logt_wthread_lock == NULL) {
		return;
	}
	if (wthread_active == QB_FALSE) {
		for (;;) {
			res = sem_getvalue(&logt_print_finished, &value);
			if (res != 0 || value == 0) {
				break;
			}
			sem_wait(&logt_print_finished);

			(void)qb_thread_lock(logt_wthread_lock);

			rec = qb_list_first_entry(&logt_print_finished_records,
					    struct qb_log_record, list);
			qb_list_del(&rec->list);
			logt_memory_used = logt_memory_used -
					   strlen(rec->buffer) -
					   sizeof(struct qb_log_record) - 1;

			qb_log_thread_log_write(rec->cs, &rec->timestamp,
						rec->buffer);
			(void)qb_thread_unlock(logt_wthread_lock);
			free(rec->buffer);
			free(rec);
		}
	} else {
		(void)qb_thread_lock(logt_wthread_lock);
		wthread_should_exit = QB_TRUE;
		(void)qb_thread_unlock(logt_wthread_lock);
		sem_post(&logt_print_finished);
		pthread_join(logt_thread_id, NULL);
	}
	(void)qb_thread_lock_destroy(logt_wthread_lock);
	sem_destroy(&logt_print_finished);
	sem_destroy(&logt_thread_start);
}