dmeventd.c

Linux Device Mapper Source Code

{
	struct thread_status *thread;
	struct dm_event_daemon_message *msg = message_data->msg;

	if (msg->data)
		dm_free(msg->data);

	_lock_mutex();
	if ((thread = _lookup_thread_status(message_data))) {
		msg->size =
		    dm_asprintf(&(msg->data), "%s %" PRIu32, message_data->id,
				thread->timeout);
	} else {
		msg->data = NULL;
		msg->size = 0;
	}
	_unlock_mutex();

	return thread ? 0 : -ENODEV;
}

/* Initialize a fifos structure with path names. */
static void _init_fifos(struct dm_event_fifos *fifos)
{
	memset(fifos, 0, sizeof(*fifos));

	fifos->client_path = DM_EVENT_FIFO_CLIENT;
	fifos->server_path = DM_EVENT_FIFO_SERVER;
}

/* Open fifos used for client communication. */
static int _open_fifos(struct dm_event_fifos *fifos)
{
	/* Create fifos */
	if (((mkfifo(fifos->client_path, 0600) == -1) && errno != EEXIST) ||
	    ((mkfifo(fifos->server_path, 0600) == -1) && errno != EEXIST)) {
		syslog(LOG_ERR, "%s: Failed to create a fifo.\n", __func__);
		stack;
		return -errno;
	}

	struct stat st;

	/* Warn about wrong permissions if applicable */
	if ((!stat(fifos->client_path, &st)) && (st.st_mode & 0777) != 0600)
		syslog(LOG_WARNING, "Fixing wrong permissions on %s",
		       fifos->client_path);

	if ((!stat(fifos->server_path, &st)) && (st.st_mode & 0777) != 0600)
		syslog(LOG_WARNING, "Fixing wrong permissions on %s",
		       fifos->server_path);

	/* If they were already there, make sure permissions are ok. */
	if (chmod(fifos->client_path, 0600)) {
		syslog(LOG_ERR, "Unable to set correct file permissions on %s",
		       fifos->client_path);
		return -errno;
	}

	if (chmod(fifos->server_path, 0600)) {
		syslog(LOG_ERR, "Unable to set correct file permissions on %s",
		       fifos->server_path);
		return -errno;
	}

	/* Need to open read+write or we will block or fail */
	if ((fifos->server = open(fifos->server_path, O_RDWR)) < 0) {
		stack;
		return -errno;
	}

	/* Need to open read+write for select() to work. */
	if ((fifos->client = open(fifos->client_path, O_RDWR)) < 0) {
		stack;
		close(fifos->server);
		return -errno;
	}

	return 0;
}

/*
 * Read message from client making sure that data is available
 * and a complete message is read.  Must not block indefinitely.
 */
static int _client_read(struct dm_event_fifos *fifos,
		       struct dm_event_daemon_message *msg)
{
	struct timeval t;
	unsigned bytes = 0;
	int ret = 0;
	fd_set fds;
	int header = 1;
	size_t size = 2 * sizeof(uint32_t);	/* status + size */
	char *buf = alloca(size);

	msg->data = NULL;

	errno = 0;
	while (bytes < size && errno != EOF) {
		/* Watch client read FIFO for input. */
		FD_ZERO(&fds);
		FD_SET(fifos->client, &fds);
		t.tv_sec = 1;
		t.tv_usec = 0;
		ret = select(fifos->client + 1, &fds, NULL, NULL, &t);

		if (!ret && !bytes)	/* nothing to read */
			return 0;

		if (!ret)	/* trying to finish read */
			continue;

		if (ret < 0)	/* error */
			return 0;

		ret = read(fifos->client, buf + bytes, size - bytes);
		bytes += ret > 0 ? ret : 0;
		if (bytes == 2 * sizeof(uint32_t) && header) {
			msg->cmd = ntohl(*((uint32_t *) buf));
			msg->size = ntohl(*((uint32_t *) buf + 1));
			buf = msg->data = dm_malloc(msg->size);
			size = msg->size;
			bytes = 0;
			header = 0;
		}
	}

	if (bytes != size) {
		if (msg->data)
			dm_free(msg->data);
		msg->data = NULL;
		msg->size = 0;
	}

	return bytes == size;
}

/*
 * Write a message to the client making sure that it is ready to write.
 */
static int _client_write(struct dm_event_fifos *fifos,
			struct dm_event_daemon_message *msg)
{
	unsigned bytes = 0;
	int ret = 0;
	fd_set fds;

	size_t size = 2 * sizeof(uint32_t) + msg->size;
	char *buf = alloca(size);

	*((uint32_t *)buf) = htonl(msg->cmd);
	*((uint32_t *)buf + 1) = htonl(msg->size);
	if (msg->data)
		memcpy(buf + 2 * sizeof(uint32_t), msg->data, msg->size);

	errno = 0;
	while (bytes < size && errno != EIO) {
		do {
			/* Watch client write FIFO to be ready for output. */
			FD_ZERO(&fds);
			FD_SET(fifos->server, &fds);
		} while (select(fifos->server + 1, NULL, &fds, NULL, NULL) !=
			 1);

		ret = write(fifos->server, buf + bytes, size - bytes);
		bytes += ret > 0 ? ret : 0;
	}

	return bytes == size;
}

/*
 * Handle a client request.
 *
 * We put the request handling functions into
 * a list because of the growing number.
 */
static int _handle_request(struct dm_event_daemon_message *msg,
			  struct message_data *message_data)
{
	static struct {
		unsigned int cmd;
		int (*f)(struct message_data *);
	} requests[] = {
		{ DM_EVENT_CMD_REGISTER_FOR_EVENT, _register_for_event},
		{ DM_EVENT_CMD_UNREGISTER_FOR_EVENT, _unregister_for_event},
		{ DM_EVENT_CMD_GET_REGISTERED_DEVICE, _get_registered_device},
		{ DM_EVENT_CMD_GET_NEXT_REGISTERED_DEVICE,
			_get_next_registered_device},
		{ DM_EVENT_CMD_SET_TIMEOUT, _set_timeout},
		{ DM_EVENT_CMD_GET_TIMEOUT, _get_timeout},
		{ DM_EVENT_CMD_ACTIVE, _active},
	}, *req;

	for (req = requests; req < requests + sizeof(requests); req++)
		if (req->cmd == msg->cmd)
			return req->f(message_data);

	return -EINVAL;
}

/* Process a request passed from the communication thread. */
static int _do_process_request(struct dm_event_daemon_message *msg)
{
	int ret;
	char *answer;
	static struct message_data message_data;

	/* Parse the message. */
	memset(&message_data, 0, sizeof(message_data));
	message_data.msg = msg;
	if (msg->cmd == DM_EVENT_CMD_HELLO)  {
		ret = 0;
		answer = msg->data;
		if (answer) {
			msg->size = dm_asprintf(&(msg->data), "%s HELLO", answer);
			dm_free(answer);
		} else {
			msg->size = 0;
			msg->data = NULL;
		}
	} else if (msg->cmd != DM_EVENT_CMD_ACTIVE && !_parse_message(&message_data)) {
		stack;
		ret = -EINVAL;
	} else
		ret = _handle_request(msg, &message_data);

	msg->cmd = ret;
	if (!msg->data)
		msg->size = dm_asprintf(&(msg->data), "%s %s", message_data.id, strerror(-ret));

	_free_message(&message_data);

	return ret;
}

/* Only one caller at a time. */
static void _process_request(struct dm_event_fifos *fifos)
{
	struct dm_event_daemon_message msg;

	memset(&msg, 0, sizeof(msg));

	/*
	 * Read the request from the client (client_read, client_write
	 * give true on success and false on failure).
	 */
	if (!_client_read(fifos, &msg))
		return;

	/* _do_process_request fills in msg (if memory allows for
	   data, otherwise just cmd and size = 0) */
	_do_process_request(&msg);

	if (!_client_write(fifos, &msg))
		stack;

	if (msg.data)
		dm_free(msg.data);
}

static void _cleanup_unused_threads(void)
{
	int ret;
	struct list *l;
	struct thread_status *thread;

	_lock_mutex();
	while ((l = list_first(&_thread_registry_unused))) {
		thread = list_item(l, struct thread_status);
		if (thread->processing)
			break;	/* cleanup on the next round */

		if (thread->status == DM_THREAD_RUNNING) {
			thread->status = DM_THREAD_SHUTDOWN;
			break;
		}

		if (thread->status == DM_THREAD_SHUTDOWN) {
			if (!thread->events) {
				/* turn codes negative -- should we be returning this? */
				ret = _terminate_thread(thread);

				if (ret == ESRCH) {
					thread->status = DM_THREAD_DONE;
				} else if (ret) {
					syslog(LOG_ERR,
					       "Unable to terminate thread: %s\n",
					       strerror(-ret));
					stack;
				}
				break;
			}

			list_del(l);
			syslog(LOG_ERR,
			       "thread can't be on unused list unless !thread->events");
			thread->status = DM_THREAD_RUNNING;
			LINK_THREAD(thread);

			continue;
		}

		if (thread->status == DM_THREAD_DONE) {
			list_del(l);
			pthread_join(thread->thread, NULL);
			_lib_put(thread->dso_data);
			_free_thread_status(thread);
		}
	}

	_unlock_mutex();
}

static void _sig_alarm(int signum __attribute((unused)))
{
	pthread_testcancel();
}

/* Init thread signal handling. */
static void _init_thread_signals(void)
{
	sigset_t my_sigset;
	struct sigaction act;

	memset(&act, 0, sizeof(act));
	act.sa_handler = _sig_alarm;
	sigaction(SIGALRM, &act, NULL);
	sigfillset(&my_sigset);

	/* These are used for exiting */
	sigdelset(&my_sigset, SIGTERM);
	sigdelset(&my_sigset, SIGINT);
	sigdelset(&my_sigset, SIGHUP);
	sigdelset(&my_sigset, SIGQUIT);

	pthread_sigmask(SIG_BLOCK, &my_sigset, NULL);
}

/*
 * exit_handler
 * @sig
 *
 * Set the global variable which the process should
 * be watching to determine when to exit.
 */
static void _exit_handler(int sig __attribute((unused)))
{
	/*
	 * We exit when '_exit_now' is set.
	 * That is, when a signal has been received.
	 *
	 * We can not simply set '_exit_now' unless all
	 * threads are done processing.
	 */
	if (!_thread_registries_empty) {
		syslog(LOG_ERR, "There are still devices being monitored.");
		syslog(LOG_ERR, "Refusing to exit.");
	} else
		_exit_now = 1;

}

static int _lock_pidfile(void)
{
	int lf;
	char pidfile[] = DMEVENTD_PIDFILE;

	if ((lf = open(pidfile, O_CREAT | O_RDWR, 0644)) < 0)
		exit(EXIT_OPEN_PID_FAILURE);

	if (flock(lf, LOCK_EX | LOCK_NB) < 0)
		exit(EXIT_LOCKFILE_INUSE);

	if (!_storepid(lf))
		exit(EXIT_FAILURE);

	return 0;
}

#ifdef linux
/*
 * Protection against OOM killer if kernel supports it
 */
static int _set_oom_adj(int val)
{
	FILE *fp;

	struct stat st;

	if (stat(OOM_ADJ_FILE, &st) == -1) {
		if (errno == ENOENT)
			DEBUGLOG(OOM_ADJ_FILE " not found");
		else
			perror(OOM_ADJ_FILE ": stat failed");
		return 1;
	}

	if (!(fp = fopen(OOM_ADJ_FILE, "w"))) {
		perror(OOM_ADJ_FILE ": fopen failed");
		return 0;
	}

	fprintf(fp, "%i", val);
	if (dm_fclose(fp))
		perror(OOM_ADJ_FILE ": fclose failed");

	return 1;
}
#endif

static void _daemonize(void)
{
	int child_status;
	int fd;
	pid_t pid;
	struct rlimit rlim;
	struct timeval tval;
	sigset_t my_sigset;

	sigemptyset(&my_sigset);
	if (sigprocmask(SIG_SETMASK, &my_sigset, NULL) < 0) {
		fprintf(stderr, "Unable to restore signals.\n");
		exit(EXIT_FAILURE);
	}
	signal(SIGTERM, &_exit_handler);

	switch (pid = fork()) {
	case -1:
		perror("fork failed:");
		exit(EXIT_FAILURE);

	case 0:		/* Child */
		break;

	default:
		/* Wait for response from child */
		while (!waitpid(pid, &child_status, WNOHANG) && !_exit_now) {
			tval.tv_sec = 0;
			tval.tv_usec = 250000;	/* .25 sec */
			select(0, NULL, NULL, NULL, &tval);
		}

		if (_exit_now)	/* Child has signaled it is ok - we can exit now */
			exit(EXIT_SUCCESS);

		/* Problem with child.  Determine what it is by exit code */
		switch (WEXITSTATUS(child_status)) {
		case EXIT_LOCKFILE_INUSE:
			fprintf(stderr, "Another dmeventd daemon is already running\n");
			break;
		case EXIT_DESC_CLOSE_FAILURE:
		case EXIT_DESC_OPEN_FAILURE:
		case EXIT_OPEN_PID_FAILURE:
		case EXIT_FIFO_FAILURE:
		case EXIT_CHDIR_FAILURE:
		default:
			fprintf(stderr, "Child exited with code %d\n", WEXITSTATUS(child_status));
			break;
		}

		exit(WEXITSTATUS(child_status));
	}

	if (chdir("/"))
		exit(EXIT_CHDIR_FAILURE);

	if (getrlimit(RLIMIT_NOFILE, &rlim) < 0)
		fd = 256;	/* just have to guess */
	else
		fd = rlim.rlim_cur;

	for (--fd; fd >= 0; fd--)
		close(fd);

	if ((open("/dev/null", O_RDONLY) < 0) ||
	    (open("/dev/null", O_WRONLY) < 0) ||
	    (open("/dev/null", O_WRONLY) < 0))
		exit(EXIT_DESC_OPEN_FAILURE);

	setsid();
}

static void usage(char *prog, FILE *file)
{
	fprintf(file, "Usage:\n");
	fprintf(file, "%s [Vhd]\n", prog);
	fprintf(file, "\n");
	fprintf(file, "   -V       Show version of dmeventd\n");
	fprintf(file, "   -h       Show this help information\n");
	fprintf(file, "   -d       Don't fork, run in the foreground\n");
	fprintf(file, "\n");
}

int main(int argc, char *argv[])
{
	int ret;
	signed char opt;
	struct dm_event_fifos fifos;
	//struct sys_log logdata = {DAEMON_NAME, LOG_DAEMON};

	opterr = 0;
	optind = 0;

	while ((opt = getopt(argc, argv, "?hVd")) != EOF) {
		switch (opt) {
		case 'h':
			usage(argv[0], stdout);
			exit(0);
		case '?':
			usage(argv[0], stderr);
			exit(0);
		case 'd':
			_debug++;
			break;
		case 'V':
			printf("dmeventd version: %s\n", DM_LIB_VERSION);
			exit(1);
			break;
		}
	}

	if (!_debug)
		_daemonize();

	openlog("dmeventd", LOG_PID, LOG_DAEMON);

	_lock_pidfile();		/* exits if failure */

	/* Set the rest of the signals to cause '_exit_now' to be set */
	signal(SIGINT, &_exit_handler);
	signal(SIGHUP, &_exit_handler);
	signal(SIGQUIT, &_exit_handler);

#ifdef linux
	if (!_set_oom_adj(OOM_DISABLE) && !_set_oom_adj(OOM_ADJUST_MIN))
		syslog(LOG_ERR, "Failed to set oom_adj to protect against OOM killer");
#endif

	_init_thread_signals();

	//multilog_clear_logging();
	//multilog_add_type(std_syslog, &logdata);
	//multilog_init_verbose(std_syslog, _LOG_DEBUG);
	//multilog_async(1);

	_init_fifos(&fifos);

	pthread_mutex_init(&_global_mutex, NULL);

#ifdef MCL_CURRENT
	if (mlockall(MCL_CURRENT | MCL_FUTURE) == -1)
		exit(EXIT_FAILURE);
#endif

	if ((ret = _open_fifos(&fifos)))
		exit(EXIT_FIFO_FAILURE);

	/* Signal parent, letting them know we are ready to go. */
	kill(getppid(), SIGTERM);
	syslog(LOG_NOTICE, "dmeventd ready for processing.");

	while (!_exit_now) {
		_process_request(&fifos);
		_cleanup_unused_threads();
		if (!list_empty(&_thread_registry)
		    || !list_empty(&_thread_registry_unused))
			_thread_registries_empty = 0;
		else
			_thread_registries_empty = 1;
	}

	_exit_dm_lib();

#ifdef MCL_CURRENT
	munlockall();
#endif
	pthread_mutex_destroy(&_global_mutex);

	syslog(LOG_NOTICE, "dmeventd shutting down.");
	closelog();

	exit(EXIT_SUCCESS);
}