dmeventd.c

Linux Device Mapper Source Code

/*
 * Copyright (C) 2005-2007 Red Hat, Inc. All rights reserved.
 *
 * This file is part of the device-mapper userspace tools.
 *
 * This copyrighted material is made available to anyone wishing to use,
 * modify, copy, or redistribute it subject to the terms and conditions
 * of the GNU Lesser General Public License v.2.1.
 *
 * You should have received a copy of the GNU Lesser General Public License
 * along with this program; if not, write to the Free Software Foundation,
 * Inc., 59 Temple Place, Suite 330, Boston, MA  02111-1307  USA
 */

/*
 * dmeventd - dm event daemon to monitor active mapped devices
 */

#define _GNU_SOURCE
#define _FILE_OFFSET_BITS 64

#include "configure.h"
#include "libdevmapper.h"
#include "libdevmapper-event.h"
#include "list.h"
#include "dmeventd.h"
//#include "libmultilog.h"
#include "log.h"

#include <dlfcn.h>
#include <errno.h>
#include <pthread.h>
#include <sys/file.h>
#include <sys/stat.h>
#include <sys/wait.h>
#include <sys/time.h>
#include <sys/resource.h>
#include <unistd.h>
#include <arpa/inet.h>		/* for htonl, ntohl */

#ifdef linux
#  include <malloc.h>

#  define OOM_ADJ_FILE "/proc/self/oom_adj"

/* From linux/oom.h */
#  define OOM_DISABLE (-17)
#  define OOM_ADJUST_MIN (-16)

#endif

/* FIXME We use syslog for now, because multilog is not yet implemented */
#include <syslog.h>

static volatile sig_atomic_t _exit_now = 0;	/* set to '1' when signal is given to exit */
static volatile sig_atomic_t _thread_registries_empty = 1;	/* registries are empty initially */
static int _debug = 0;

/* List (un)link macros. */
#define	LINK(x, head)		list_add(head, &(x)->list)
#define	LINK_DSO(dso)		LINK(dso, &_dso_registry)
#define	LINK_THREAD(thread)	LINK(thread, &_thread_registry)

#define	UNLINK(x)		list_del(&(x)->list)
#define	UNLINK_DSO(x)		UNLINK(x)
#define	UNLINK_THREAD(x)	UNLINK(x)

#define DAEMON_NAME "dmeventd"

/*
  Global mutex for thread list access. Has to be held when:
  - iterating thread list
  - adding or removing elements from thread list
  - changing or reading thread_status's fields:
    processing, status, events
  Use _lock_mutex() and _unlock_mutex() to hold/release it
*/
static pthread_mutex_t _global_mutex;

/*
  There are three states a thread can attain (see struct
  thread_status, field int status):

  - DM_THREAD_RUNNING: thread has started up and is either working or
  waiting for events... transitions to either SHUTDOWN or DONE
  - DM_THREAD_SHUTDOWN: thread is still doing something, but it is
  supposed to terminate (and transition to DONE) as soon as it
  finishes whatever it was doing at the point of flipping state to
  SHUTDOWN... the thread is still on the thread list
  - DM_THREAD_DONE: thread has terminated and has been moved over to
  unused thread list, cleanup pending
 */
#define DM_THREAD_RUNNING  0
#define DM_THREAD_SHUTDOWN 1
#define DM_THREAD_DONE     2

#define THREAD_STACK_SIZE (300*1024)

#define DEBUGLOG(fmt, args...) _debuglog(fmt, ## args)

/* Data kept about a DSO. */
struct dso_data {
	struct list list;

	char *dso_name;		/* DSO name (eg, "evms", "dmraid", "lvm2"). */

	void *dso_handle;	/* Opaque handle as returned from dlopen(). */
	unsigned int ref_count;	/* Library reference count. */

	/*
	 * Event processing.
	 *
	 * The DSO can do whatever appropriate steps if an event
	 * happens such as changing the mapping in case a mirror
	 * fails, update the application metadata etc.
	 *
	 * This function gets a dm_task that is a result of
	 * DM_DEVICE_WAITEVENT ioctl (results equivalent to
	 * DM_DEVICE_STATUS). It should not destroy it.
	 * The caller must dispose of the task.
	 */
	void (*process_event)(struct dm_task *dmt, enum dm_event_mask event, void **user);

	/*
	 * Device registration.
	 *
	 * When an application registers a device for an event, the DSO
	 * can carry out appropriate steps so that a later call to
	 * the process_event() function is sane (eg, read metadata
	 * and activate a mapping).
	 */
	int (*register_device)(const char *device, const char *uuid, int major,
			       int minor, void **user);

	/*
	 * Device unregistration.
	 *
	 * In case all devices of a mapping (eg, RAID10) are unregistered
	 * for events, the DSO can recognize this and carry out appropriate
	 * steps (eg, deactivate mapping, metadata update).
	 */
	int (*unregister_device)(const char *device, const char *uuid,
				 int major, int minor, void **user);
};
static LIST_INIT(_dso_registry);

/* Structure to keep parsed register variables from client message. */
struct message_data {
	char *id;
	char *dso_name;		/* Name of DSO. */
	char *device_uuid;	/* Mapped device path. */
	union {
		char *str;	/* Events string as fetched from message. */
		enum dm_event_mask field;	/* Events bitfield. */
	} events;
	union {
		char *str;
		uint32_t secs;
	} timeout;
	struct dm_event_daemon_message *msg;	/* Pointer to message buffer. */
};

/*
 * Housekeeping of thread+device states.
 *
 * One thread per mapped device which can block on it until an event
 * occurs and the event processing function of the DSO gets called.
 */
struct thread_status {
	struct list list;

	pthread_t thread;

	struct dso_data *dso_data;	/* DSO this thread accesses. */

	struct {
		char *uuid;
		char *name;
		int major, minor;
	} device;
	uint32_t event_nr;	/* event number */
	int processing;		/* Set when event is being processed */

	int status;		/* see DM_THREAD_{RUNNING,SHUTDOWN,DONE}
				   constants above */
	enum dm_event_mask events;	/* bitfield for event filter. */
	enum dm_event_mask current_events;	/* bitfield for occured events. */
	struct dm_task *current_task;
	time_t next_time;
	uint32_t timeout;
	struct list timeout_list;
	void *dso_private; /* dso per-thread status variable */
};
static LIST_INIT(_thread_registry);
static LIST_INIT(_thread_registry_unused);

static int _timeout_running;
static LIST_INIT(_timeout_registry);
static pthread_mutex_t _timeout_mutex = PTHREAD_MUTEX_INITIALIZER;
static pthread_cond_t _timeout_cond = PTHREAD_COND_INITIALIZER;

static void _debuglog(const char *fmt, ...)
{
        time_t P;
        va_list ap;
 
        if (!_debug)
                return;
 
        va_start(ap,fmt);

        time(&P);
        fprintf(stderr, "dmeventd[%p]: %.15s ", (void *) pthread_self(), ctime(&P)+4 );
        vfprintf(stderr, fmt, ap);
	fprintf(stderr, "\n");

        va_end(ap);
}

/* Allocate/free the status structure for a monitoring thread. */
static struct thread_status *_alloc_thread_status(struct message_data *data,
						  struct dso_data *dso_data)
{
	struct thread_status *ret = (typeof(ret)) dm_malloc(sizeof(*ret));

	if (!ret)
		return NULL;

	memset(ret, 0, sizeof(*ret));
	if (!(ret->device.uuid = dm_strdup(data->device_uuid))) {
		dm_free(ret);
		return NULL;
	}

	ret->current_task = NULL;
	ret->device.name = NULL;
	ret->device.major = ret->device.minor = 0;
	ret->dso_data = dso_data;
	ret->events = data->events.field;
	ret->timeout = data->timeout.secs;
	list_init(&ret->timeout_list);

	return ret;
}

static void _free_thread_status(struct thread_status *thread)
{
	if (thread->current_task)
		dm_task_destroy(thread->current_task);
	dm_free(thread->device.uuid);
	dm_free(thread->device.name);
	dm_free(thread);
}

/* Allocate/free DSO data. */
static struct dso_data *_alloc_dso_data(struct message_data *data)
{
	struct dso_data *ret = (typeof(ret)) dm_malloc(sizeof(*ret));

	if (!ret)
		return NULL;

	memset(ret, 0, sizeof(*ret));
	if (!(ret->dso_name = dm_strdup(data->dso_name))) {
		dm_free(ret);
		return NULL;
	}

	return ret;
}

/* Create a device monitoring thread. */
static int _pthread_create_smallstack(pthread_t *t, void *(*fun)(void *), void *arg)
{
	pthread_attr_t attr;
	pthread_attr_init(&attr);
	/*
	 * We use a smaller stack since it gets preallocated in its entirety
	 */
	pthread_attr_setstacksize(&attr, THREAD_STACK_SIZE);
	return pthread_create(t, &attr, fun, arg);
}

static void _free_dso_data(struct dso_data *data)
{
	dm_free(data->dso_name);
	dm_free(data);
}

/*
 * Fetch a string off src and duplicate it into *ptr.
 * Pay attention to zero-length strings.
 */
/* FIXME? move to libdevmapper to share with the client lib (need to
   make delimiter a parameter then) */
static int _fetch_string(char **ptr, char **src, const int delimiter)
{
	int ret = 0;
	char *p;
	size_t len;

	if ((p = strchr(*src, delimiter)))
		*p = 0;

	if ((*ptr = dm_strdup(*src))) {
		if ((len = strlen(*ptr)))
			*src += len;
		else {
			dm_free(*ptr);
			*ptr = NULL;
		}

		(*src)++;
		ret = 1;
	}

	if (p)
		*p = delimiter;

	return ret;
}

/* Free message memory. */
static void _free_message(struct message_data *message_data)
{
	if (message_data->id)
		dm_free(message_data->id);
	if (message_data->dso_name)
		dm_free(message_data->dso_name);

	if (message_data->device_uuid)
		dm_free(message_data->device_uuid);

}

/* Parse a register message from the client. */
static int _parse_message(struct message_data *message_data)
{
	int ret = 0;
	char *p = message_data->msg->data;
	struct dm_event_daemon_message *msg = message_data->msg;

	if (!msg->data)
		return 0;

	/*
	 * Retrieve application identifier, mapped device
	 * path and events # string from message.
	 */
	if (_fetch_string(&message_data->id, &p, ' ') &&
	    _fetch_string(&message_data->dso_name, &p, ' ') &&
	    _fetch_string(&message_data->device_uuid, &p, ' ') &&
	    _fetch_string(&message_data->events.str, &p, ' ') &&
	    _fetch_string(&message_data->timeout.str, &p, ' ')) {
		if (message_data->events.str) {
			enum dm_event_mask i = atoi(message_data->events.str);

			/*
			 * Free string representaion of events.
			 * Not needed an more.
			 */
			dm_free(message_data->events.str);
			message_data->events.field = i;
		}
		if (message_data->timeout.str) {
			uint32_t secs = atoi(message_data->timeout.str);
			dm_free(message_data->timeout.str);
			message_data->timeout.secs = secs ? secs :
			    DM_EVENT_DEFAULT_TIMEOUT;
		}

		ret = 1;
	}

	dm_free(msg->data);
	msg->data = NULL;
	msg->size = 0;
	return ret;
};

/* Global mutex to lock access to lists et al. See _global_mutex
   above. */
static int _lock_mutex(void)
{
	return pthread_mutex_lock(&_global_mutex);
}

static int _unlock_mutex(void)
{
	return pthread_mutex_unlock(&_global_mutex);
}

/* Store pid in pidfile. */
static int _storepid(int lf)
{
	int len;
	char pid[8];

	if ((len = snprintf(pid, sizeof(pid), "%u\n", getpid())) < 0)
		return 0;

	if (len > (int) sizeof(pid))
		len = (int) sizeof(pid);

	if (write(lf, pid, (size_t) len) != len)
		return 0;

	fsync(lf);

	return 1;
}

/* Check, if a device exists. */
static int _fill_device_data(struct thread_status *ts)
{
	struct dm_task *dmt;
	struct dm_info dmi;

	if (!ts->device.uuid)
		return 0;

	ts->device.name = NULL;
	ts->device.major = ts->device.minor = 0;

	dmt = dm_task_create(DM_DEVICE_INFO);
	if (!dmt)
		return 0;

	dm_task_set_uuid(dmt, ts->device.uuid);
	if (!dm_task_run(dmt))
		goto fail;

	ts->device.name = dm_strdup(dm_task_get_name(dmt));
	if (!ts->device.name)
		goto fail;

	if (!dm_task_get_info(dmt, &dmi))
		goto fail;

	ts->device.major = dmi.major;
	ts->device.minor = dmi.minor;

	dm_task_destroy(dmt);
	return 1;

      fail:
	dm_task_destroy(dmt);
	dm_free(ts->device.name);
	return 0;
}

/*
 * Find an existing thread for a device.
 *
 * Mutex must be held when calling this.
 */
static struct thread_status *_lookup_thread_status(struct message_data *data)
{
	struct thread_status *thread;

	list_iterate_items(thread, &_thread_registry)
	    if (!strcmp(data->device_uuid, thread->device.uuid))
		return thread;

	return NULL;
}

/* Cleanup at exit. */
static void _exit_dm_lib(void)
{
	dm_lib_release();
	dm_lib_exit();
}

static void _exit_timeout(void *unused __attribute((unused)))
{
	_timeout_running = 0;
	pthread_mutex_unlock(&_timeout_mutex);
}

/* Wake up monitor threads every so often. */
static void *_timeout_thread(void *unused __attribute((unused)))
{
	struct timespec timeout;
	time_t curr_time;

	timeout.tv_nsec = 0;
	pthread_cleanup_push(_exit_timeout, NULL);
	pthread_mutex_lock(&_timeout_mutex);

	while (!list_empty(&_timeout_registry)) {
		struct thread_status *thread;

		timeout.tv_sec = 0;
		curr_time = time(NULL);

		list_iterate_items_gen(thread, &_timeout_registry, timeout_list) {
			if (thread->next_time <= curr_time) {
				thread->next_time = curr_time + thread->timeout;
				pthread_kill(thread->thread, SIGALRM);
			}

			if (thread->next_time < timeout.tv_sec || !timeout.tv_sec)
				timeout.tv_sec = thread->next_time;
		}

		pthread_cond_timedwait(&_timeout_cond, &_timeout_mutex,
				       &timeout);
	}

	pthread_cleanup_pop(1);

	return NULL;
}

static int _register_for_timeout(struct thread_status *thread)
{
	int ret = 0;

	pthread_mutex_lock(&_timeout_mutex);

	thread->next_time = time(NULL) + thread->timeout;

	if (list_empty(&thread->timeout_list)) {
		list_add(&_timeout_registry, &thread->timeout_list);
		if (_timeout_running)
			pthread_cond_signal(&_timeout_cond);
	}

	if (!_timeout_running) {
		pthread_t timeout_id;

		if (!(ret = -_pthread_create_smallstack(&timeout_id, _timeout_thread, NULL)))
			_timeout_running = 1;
	}

	pthread_mutex_unlock(&_timeout_mutex);

	return ret;
}

static void _unregister_for_timeout(struct thread_status *thread)
{
	pthread_mutex_lock(&_timeout_mutex);
	if (!list_empty(&thread->timeout_list)) {
		list_del(&thread->timeout_list);
		list_init(&thread->timeout_list);
	}
	pthread_mutex_unlock(&_timeout_mutex);
}

static void _no_intr_log(int level, const char *file, int line,
			const char *f, ...)
{
	va_list ap;

	if (errno == EINTR)
		return;
	if (level > _LOG_WARN)
		return;

	va_start(ap, f);

	if (level < _LOG_WARN)
		vfprintf(stderr, f, ap);
	else
		vprintf(f, ap);

	va_end(ap);

	if (level < _LOG_WARN)
		fprintf(stderr, "\n");
	else
		fprintf(stdout, "\n");
}

static sigset_t _unblock_sigalrm(void)
{
	sigset_t set, old;

	sigemptyset(&set);
	sigaddset(&set, SIGALRM);
	pthread_sigmask(SIG_UNBLOCK, &set, &old);
	return old;
}

#define DM_WAIT_RETRY 0
#define DM_WAIT_INTR 1