/*
 *
 *   (C) Copyright IBM Corp. 2001, 2003
 *
 *   This program is free software;  you can redistribute it and/or modify
 *   it under the terms of the GNU General Public License as published by
 *   the Free Software Foundation; either version 2 of the License, or
 *   (at your option) any later version.
 *
 *   This program 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 General Public License for more details.
 *
 *   You should have received a copy of the GNU 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
 *
 * Module: engine.c
 *
 */

#include <stdlib.h>
#include <stdio.h>
#include <stdarg.h>
#include <string.h>
#include <ctype.h>
#include <errno.h>
#include <fcntl.h>
#include <dirent.h>
#include <glob.h>
#include <pthread.h>
#include <signal.h>
#include <time.h>
#include <wait.h>
#include <unistd.h>
#include <sys/ioctl.h>
#include <sys/mount.h>
#include <sys/stat.h>
#include <sys/time.h>

#include "fullengine.h"
#include "engine.h"
#include "lists.h"
#include "dload.h"
#include "faulthdlr.h"
#include "handlemgr.h"
#include "common.h"
#include "internalAPI.h"
#include "memman.h"
#include "commit.h"
#include "discover.h"
#include "message.h"
#include "volume.h"
#include "config.h"
#include "dm.h"
#include "cluster.h"
#include "remote.h"
#include "daemon.h"
#include "dmonmsg.h"


/*----------------------------------------------------------------------------+
+                                                                             +
+                                  GLOBAL DATA                                +
+                                                                             +
+-----------------------------------------------------------------------------*/

evms_version_t engine_version =
{
	MAJOR_VERSION,
	MINOR_VERSION,
	PATCH_LEVEL
};

evms_version_t engine_api_version =
{
	ENGINE_API_MAJOR_VERSION,
	ENGINE_API_MINOR_VERSION,
	ENGINE_API_PATCH_LEVEL
};

evms_version_t engine_services_api_version =
{
	ENGINE_SERVICES_API_MAJOR_VERION,
	ENGINE_SERVICES_API_MINOR_VERION,
	ENGINE_SERVICES_API_PATCH_LEVEL
};

evms_version_t engine_plugin_api_version =
{
	ENGINE_PLUGIN_API_MAJOR_VERION,
	ENGINE_PLUGIN_API_MINOR_VERION,
	ENGINE_PLUGIN_API_PATCH_LEVEL
};

evms_version_t engine_fsim_api_version =
{
	ENGINE_FSIM_API_MAJOR_VERION,
	ENGINE_FSIM_API_MINOR_VERION,
	ENGINE_FSIM_API_PATCH_LEVEL
};

evms_version_t engine_container_api_version =
{
	ENGINE_CONTAINER_API_MAJOR_VERION,
	ENGINE_CONTAINER_API_MINOR_VERION,
	ENGINE_CONTAINER_API_PATCH_LEVEL
};

evms_version_t engine_cluster_api_version =
{
	ENGINE_CLUSTER_API_MAJOR_VERION,
	ENGINE_CLUSTER_API_MINOR_VERION,
	ENGINE_CLUSTER_API_PATCH_LEVEL
};

evms_version_t engine_daemon_msg_version =
{
	EVMS_DMONMSG_MAJOR_VERSION,
	EVMS_DMONMSG_MINOR_VERSION,
	EVMS_DMONMSG_PATCH_LEVEL
};


ece_nodeid_t   no_nodeid = ECE_NO_NODE;
ece_nodeid_t * current_nodeid = NULL;
char         * current_node_name;
boolean        local_focus = TRUE;

STATIC_LIST_DECL(plugins_list);
STATIC_LIST_DECL(disks_list);
STATIC_LIST_DECL(segments_list);
STATIC_LIST_DECL(containers_list);
STATIC_LIST_DECL(regions_list);
STATIC_LIST_DECL(EVMS_objects_list);
STATIC_LIST_DECL(volumes_list);
STATIC_LIST_DECL(kill_sectors_list);
STATIC_LIST_DECL(volume_delete_list);
STATIC_LIST_DECL(deactivate_list);
STATIC_LIST_DECL(rename_volume_list);

plugin_record_t * local_disk_manager = NULL;
plugin_record_t * replace_plugin = NULL;
plugin_record_t * cluster_manager = NULL;
cluster_functions_t * ece_funcs = NULL;
plugin_record_t * cluster_segment_manager = NULL;

int dm_control_fd = 0;

engine_mode_t engine_mode = ENGINE_CLOSED;

debug_level_t debug_level = DEFAULT;

char  * log_file_name = DEFAULT_LOG_FILE;
int     log_file_fd = 0;
boolean log_usec = FALSE;
boolean log_pid = FALSE;
u_char log_buf[LOG_BUF_SIZE];
pthread_mutex_t log_mutex = PTHREAD_MUTEX_INITIALIZER;

#define ENGINE_LOCK_FILE_DIR	"/var/lock"
#define ENGINE_LOCK_FILE	ENGINE_LOCK_FILE_DIR "/evms-engine"
int           lock_file_fd = 0;
size_t        lock_file_size = 0;
lock_file_t * lock_file;

sem_t   shutdown_sem;
#define SHUTDOWN_TIMEOUT 30	/* seconds */

boolean is_2_4_kernel;

pthread_attr_t pthread_attr_detached;

char metadata_db_name[PATH_MAX];
int metadata_db_fd = 0;


/*
 * The following functions are used to prevent potential deadlock in a child
 * process after a fork().  If the log_mutex is held by another thread when
 * a thread forks, the child preocess will inherit a copy of the locked mutex.
 * The child process will never see the mutex unlocked, and so will deadlock
 * once it tries to lock the mutex.
 */
static void fork_prepare(void) {
	pthread_mutex_lock(&log_mutex);
}

static void fork_parent(void) {
	pthread_mutex_unlock(&log_mutex);
}

static void fork_child(void) {
	pthread_mutex_unlock(&log_mutex);
}


/*
 * my_init() is run when the library is first loaded.
 */
void __attribute__ ((constructor)) my_init(void) {
	pthread_atfork(fork_prepare, fork_parent, fork_child);
}


/*-------------------------------------------------------------------------------------+
+                                                                                      +
+                                PRIVATE SUBROUTINES                                   +
+                                                                                      +
+-------------------------------------------------------------------------------------*/

static char * build_archive_log_name(char * log_name, int index) {

	char * archive_log_file_name;
	char * pch1;
	char * pch2;

	/*
	 * Get memory to build the archive file names.  Get enough for the name,
	 * three bytes for the ".nn" that will be inserted/appended, plus one for
	 * the null terminator.
	 */
	archive_log_file_name = engine_alloc(strlen(log_name) + 4);

	if (archive_log_file_name != NULL) {

		strcpy(archive_log_file_name, log_name);

		/*
		 * The ".nn" will be inserted ahead of the last '.' in the file name.
		 * If the file name doesn't have a '.' in it, the .nn will be
		 * appended at the end.
		 */
		pch1 = strrchr(archive_log_file_name, '.');
		if (pch1 == NULL) {
			pch1 = archive_log_file_name + strlen(archive_log_file_name);
		}

		/* Add/insert the ".nn". */
		*(pch1++) = '.';
		sprintf(pch1, "%d", index);

		/* Copy the remainder of the file name, if it exists. */
		pch2 = strrchr(log_name, '.');
		if (pch2 != NULL) {
			strcat(pch1, pch2);
		}
	}

	return archive_log_file_name;
}


#define MAX_LOG_ARCHIVES 9

static int start_logging(char * filename) {

	int rc = 0;
	int i;
	char * old_archive_log_name;
	char * new_archive_log_name;

	if (log_file_fd == 0) {

		/* Roll back the archive log files. */
		old_archive_log_name = build_archive_log_name(filename, MAX_LOG_ARCHIVES);
		unlink(old_archive_log_name);

		for (i = MAX_LOG_ARCHIVES; i > 1; i--) {
			new_archive_log_name = build_archive_log_name(filename, i - 1);

			rename(new_archive_log_name, old_archive_log_name);

			engine_free(old_archive_log_name);
			old_archive_log_name = new_archive_log_name;
		}

		rename(filename, old_archive_log_name);
		engine_free(old_archive_log_name);

		log_file_fd = open(filename, O_CREAT | O_TRUNC | O_WRONLY, 0664);

		if (log_file_fd < 0) {
			rc = errno;
			log_file_fd = 0;
		} else {
			/* Close this file for all exec'd processes. */
			fcntl(log_file_fd, F_SETFD, FD_CLOEXEC);
		}

	} else {
		rc = EEXIST;
	}

	return rc;
}


static int stop_logging() {

	int rc = 0;

	if (log_file_fd > 0) {
		close(log_file_fd);
		log_file_fd = 0;
	} else {
		rc = ENOENT;
	}

	return rc;
}


/*
 * Put a time stamp in the buffer.
 * This function assumes the buffer has enough room for the timestamp.
 */
void timestamp(char * buf, size_t len, debug_level_t level) {

	time_t t;
	size_t stamp_len;

	time(&t);

	strftime(buf, len, "%b %d %H:%M:%S", localtime(&t));

	if (log_usec) {
		struct timezone tz;
		struct timeval  utime;

		gettimeofday(&utime, &tz);
		sprintf(buf + strlen(buf), ".%06ld ", utime.tv_usec);
	} else {
		strcat(buf, " ");
	}

	stamp_len = strlen(buf);

	gethostname(buf + stamp_len, len - stamp_len);

	sprintf(buf + strlen(buf), " _%d_ ", level);

	if (log_pid) {
		sprintf(log_buf + strlen(log_buf), "%lx ", pthread_self());
	}
}


int engine_write_log_entry(debug_level_t level,
			   char        * fmt,
			   ...) {
	int rc = 0;
	int len;
	va_list args;

	if (dm_device_suspended) {
		return rc;
	}

	if (level <= debug_level) {
		if (log_file_fd > 0) {
			pthread_mutex_lock(&log_mutex);
			timestamp(log_buf, LOG_BUF_SIZE, level);

			if (engine_mode & ENGINE_DAEMON) {
				strcat(log_buf, "Daemon: ");
			} else {
				strcat(log_buf, "Engine: ");
			}
			len = strlen(log_buf);

			va_start(args, fmt);
			len += vsprintf(log_buf + strlen(log_buf), fmt, args);
			va_end(args);

			if (write(log_file_fd, log_buf, len) < 0) {
				rc = errno;
			}
			pthread_mutex_unlock(&log_mutex);

		} else {
			rc = ENOENT;
		}
	}

	return rc;
}


/*
 * Scan a line for device info.
 * This is a simple, job specific function.  Rather than try to use sscanf() or
 * other complex parsers, we do a quick and dirty job here.
 * The input buffer should have the form:
 * | +[0-9]+ +name *[\n]|
 * In English, start with one or more blanks, then a number, one or more blanks,
 * then a name, followed by blanks and/or a newline.
 * Return the number and the name via parameters.
 * The function returns TRUE if it successfully found a number and a name,
 * else FALSE.
 */
static boolean scan_dev_info(char * buf, int * dev_num, char * dev_name) {

	int result = FALSE;
	char * p = buf;

	/* Skip over white space. */
	while ((*p != '\0') &&
	       ((*p == ' ') || (*p == '\t') || (*p == '\n'))) {
		p++;
	}

	if ((*p != '\0') &&
	    (*p >= '0') && (*p <= '9')) {
		*dev_num = atoi(p);

		/* Skip over the number. */
		while ((*p != '\0') &&
		       (*p != ' ') && (*p != '\t') && (*p != '\n')) {
			p++;
		}

		/* Skip over white space. */
		while ((*p != '\0') &&
		       ((*p == ' ') || (*p == '\t') || (*p == '\n'))) {
			p++;
		}

		if (*p != '\0') {
			/* Copy string up to white space. */
			while ((*p != '\0') &&
			       (*p != ' ') && (*p != '\t') && (*p != '\n')) {
				*dev_name++ = *p++;
			}

			result = TRUE;
		}
	}

	/*
	 * Terminate the name. If a name was not found, this will set the
	 * returned device name to "".
	 */
	*dev_name = '\0';

	return result;
}


/*
 * Check if /proc and /sys (on 2.6) are mounted. If they aren't, mount them.
 */

static int mounted_procfs = FALSE;

static int check_for_procfs(void) {

	struct stat st;
	int rc;

	LOG_PROC_ENTRY();

	rc = stat("/proc/filesystems", &st);
	if (rc) {
		LOG_WARNING("The /proc filesystem is not mounted. Attempting to mount now.\n");

		rc = stat("/proc", &st);
		if (rc) {
			rc = make_directory("/proc", (S_IFDIR | S_IRWXU |
						      S_IRGRP | S_IXGRP |
						      S_IROTH | S_IXOTH));
		}

		if (!rc) {
			rc = mount("none", "/proc", "proc", 0, NULL);
			if (!rc) {
				mounted_procfs = TRUE;
			}
		}

		if (rc) {
			LOG_ERROR("Unable to mount /proc.\n");
		}
	}

	LOG_PROC_EXIT_INT(rc);
	return rc;
}


static int mounted_sysfs = FALSE;

static int check_for_sysfs(void) {

	FILE *fp;
	char string[50];
	struct stat st;
	int sysfs = FALSE;
	int rc = 0;

	LOG_PROC_ENTRY();

	/* See if sysfs is supported on this kernel. */
	fp = fopen("/proc/filesystems", "r");
	if (fp) {
		while (fscanf(fp, "%s", string) != EOF) {
			if (!strncmp(string, "sysfs", 50)) {
				sysfs = TRUE;
			}
		}
		fclose(fp);
	}

	if (sysfs) {
		rc = stat("/sys/block", &st);
		if (rc) {
			LOG_WARNING("The /sys filesystem is not mounted. Attempting to mount now.\n");

			rc = stat("/sys", &st);
			if (rc) {
				rc = make_directory("/sys", (S_IFDIR | S_IRWXU |
							     S_IRGRP | S_IXGRP |
							     S_IROTH | S_IXOTH));
			}

			if (!rc) {
				rc = mount("none", "/sys", "sysfs", 0, NULL);
				if (!rc) {
					mounted_sysfs = TRUE;
				}
			}

			if (rc) {
				LOG_ERROR("Unable to mount /sys.\n");
			}
		}
	}

	LOG_PROC_EXIT_INT(rc);
	return rc;
}


static int check_for_filesystems(void) {

	int rc;

	LOG_PROC_ENTRY();

	rc = check_for_procfs();
	if (!rc) {
		rc = check_for_sysfs();
	}

	LOG_PROC_EXIT_INT(rc);
	return rc;
}


static void unmount_filesystems(void) {

	LOG_PROC_ENTRY();

	if (mounted_procfs) {
		umount("/proc");
	}

	if (mounted_sysfs) {
		umount("/sys");
	}

	LOG_PROC_EXIT_VOID();
}


static int load_dm_module() {

	int rc = 0;
	pid_t pid;
	int   status;
	char * argv[3] = {"modprobe", "dm-mod", NULL};

	LOG_PROC_ENTRY();

        pid = fork();

	switch (pid) {
	case -1:
		rc = errno;
		LOG_WARNING("fork() to run \"%s %s\" returned error %d: %s\n", argv[0], argv[1], rc, strerror(rc));
		break;

	case 0:
		/* Child process */

		execvp(argv[0], argv);

		/* Should not get here unless execvp() fails. */
		rc = errno;
		LOG_WARNING("execvp() to run \"%s %s\" returned error %d: %s\n", argv[0], argv[1], rc, strerror(rc));

		/* exit() kills the GUI.  Use _exit() instead. */
		_exit(rc);
		break;

	default:
		/* Parent process */
		waitpid(pid, &status, 0);
	}

	if (rc == 0) {
		if (WIFSIGNALED(status)) {
			LOG_WARNING("\"%s %s\" was terminated by signal %s\n", argv[0], argv[1], sys_siglist[WTERMSIG(status)]);
			rc = EINTR;

		} else {
			rc = WEXITSTATUS(status);
			LOG_DEBUG("\"%s %s\" exited with error code %d: %s\n", argv[0], argv[1], rc, strerror(rc));
		}
	}

	LOG_PROC_EXIT_INT(rc);
	return rc;
}


/*
 * Setup a dev node for the Device Mapper control node.
 * The Device Mapper control node is a character device.
 * Its major number is the major number of "misc", found in /proc/devices.
 * Its Minor number is listed in /proc/misc as "device"mapper".
 */
static int open_dm_control_node(void) {

	int rc = 0;
	FILE * file;
	char work_buf[256];
	char dev_name[64];
	int dev_num;
	int dm_control_major = 0;
	int dm_control_minor = 0;
	boolean found_major = FALSE;
	boolean found_minor = FALSE;

	LOG_PROC_ENTRY();

	file = fopen("/proc/devices", "r");

	if (file != NULL) {
		while (fgets(work_buf,sizeof(work_buf),file) != NULL) {
			if (scan_dev_info(work_buf, &dev_num, dev_name)) {
				if (strcmp(dev_name, "misc") == 0) {
					dm_control_major = dev_num;
					found_major = TRUE;
					break;
				}
			}
		}

		fclose(file);

		if (found_major) {
			boolean retry = TRUE;

retry:
			file = fopen("/proc/misc", "r");

			if (file != NULL) {
				while (fgets(work_buf,sizeof(work_buf),file) != NULL) {
					if (scan_dev_info(work_buf, &dev_num, dev_name)) {
						if (strcmp(dev_name, "device-mapper") == 0) {
							dm_control_minor = dev_num;
							found_minor = TRUE;
							break;
						}
					}
				}

				fclose(file);

				if (found_minor) {
					dev_t devt = makedev(dm_control_major, dm_control_minor);

					make_directory(EVMS_DM_CONTROL_DIR, (S_IFDIR | S_IRWXU |
									     S_IRGRP | S_IXGRP |
									     S_IROTH | S_IXOTH));
					unlink(EVMS_DM_CONTROL);

					rc = mknod(EVMS_DM_CONTROL, (S_IFCHR | S_IRUSR | S_IWUSR | S_IRGRP | S_IWGRP), devt);

					if (rc == 0) {
						rc = open(EVMS_DM_CONTROL, O_RDWR);

						if (rc > 0) {
							dm_control_fd = rc;

							rc = dm_check_version();
							if (rc != 0) {
								LOG_WARNING("Checking Device-Mapper interface version failed with error code %d: %s\n", rc, strerror(rc));
								close(dm_control_fd);
								dm_control_fd = 0;
							} else {
								/* Close this file for all exec'd processes. */
								fcntl(dm_control_fd, F_SETFD, FD_CLOEXEC);
							}

						} else {
							rc = errno;
							LOG_WARNING("Open of " EVMS_DM_CONTROL " failed with error code %d: %s\n", rc, strerror(rc));
						}

					} else {
						rc = errno;
						LOG_WARNING("mknod of " EVMS_DM_CONTROL " (%d:%d) failed with error code %d: %s\n", dm_control_major, dm_control_minor, rc, strerror(rc));
					}

				} else {
					if (retry) {
						rc = load_dm_module();

						/* Whether success or error, don't try again. */
						retry = FALSE;

						if (rc == 0) {
							goto retry;
						}

					} else {
						LOG_WARNING("Could not find an entry for \"device-mapper\" in /proc/misc.\n");
						rc = ENOENT;
					}
				}

			} else {
				rc = errno;
				LOG_WARNING("Open of /proc/misc failed with error code %d: %s\n", rc, strerror(rc));
			}

		} else {
			LOG_WARNING("Could not find an entry for \"misc\" in /proc/devices.\n");
			rc = ENOENT;
		}

	} else {
		rc = errno;
		LOG_WARNING("Open of /proc/devices failed with error code %d: %s\n", rc, strerror(rc));
	}

	LOG_PROC_EXIT_INT(rc);
	return rc;
}


void send_shutdown(pid_t pid) {

	int rc = 0;
	int seconds_to_wait = SHUTDOWN_TIMEOUT + 5;

	LOG_PROC_ENTRY();

	/* The Engine uses the SIGUSR1 signal to indicate a shutdown. */
	LOG_DEBUG("Send SIGUSR1 to pid %d.\n", pid);
        kill(pid, SIGUSR1);

	/*
	 * Wait for the process to die by sending it a harmless signal.
	 * When the kill() fails we can know that the process has gone
	 * away.
	 */
	while ((rc == 0) && (seconds_to_wait > 0)) {
		usleep(1000000);
		LOG_DEBUG("Send SIGCONT to pid %d to check if is gone.\n", pid);
		rc = kill(pid, SIGCONT);
		seconds_to_wait--;
	}

	if (rc == 0) {
		/*
		 * The process did not respond to the SIGUSR1 signal.
		 * Shut it down.
		 */

		/* Try to kill it nicely. */
		LOG_DEBUG("Send SIGQUIT to pid %d to tell it to quit.\n", pid);
		kill(pid, SIGQUIT);

		usleep(3000000);

		/* Just kill it. */
		LOG_DEBUG("Send SIGKILL to pid %d to kill it.\n", pid);
		kill(pid, SIGKILL);
	}

	LOG_PROC_EXIT_VOID();
}


static char lock_msg[MAX_LOCK_MESSAGE_LEN];

int lock_engine(engine_mode_t mode, const char * * msg) {

	int     rc = 0;
	int     status;
	struct  flock lockinfo = {0};
	boolean retry = FALSE;
	char  * buf1 = NULL;
	char  * buf2;
	struct stat statbuf = {0};

	LOG_PROC_ENTRY();

	lock_msg[0] = '\0';

	status = stat(ENGINE_LOCK_FILE, &statbuf);
	
	/*
	 * Get memory for the lock file contents.
	 * If the lock file doesn't exist, the stat() above failed and
	 * statbuf.st_size will be left at its initial value of 0.
	 */
	if (lock_file == NULL) {
		lock_file_size = max(statbuf.st_size, sizeof(lock_file_t));
		lock_file = engine_alloc(lock_file_size);
		if (lock_file == NULL) {
			LOG_CRITICAL("Error getting memory for the lock file.\n");
			LOG_PROC_EXIT_INT(ENOMEM);
			return ENOMEM;
		}

	} else {
		/* Make sure the memory for the lock file is big enough. */
		if (lock_file_size < statbuf.st_size) {
			lock_file = engine_realloc(lock_file, statbuf.st_size);
			if (lock_file != NULL) {
				lock_file_size = statbuf.st_size;

			} else {
				LOG_CRITICAL("Error reallocating memory for the lock file.\n");
				LOG_PROC_EXIT_INT(ENOMEM);
				return ENOMEM;
			}
		}
	}

	if (status != 0) {

		/*
		 * The lock file does not exist.  Make sure the lock file
		 * directory exists so that we can create the lock file.
		 */
		char dir_name[] = ENGINE_LOCK_FILE_DIR;

		rc = make_directory(dir_name, (S_IFDIR | S_IRWXU |
					       S_IRGRP | S_IXGRP |
					       S_IROTH | S_IXOTH));

		if (rc != 0) {
			LOG_SERIOUS("Unable to create the directory for the lock file.\n");
		}
	}

start:
	if (rc == 0) {
		lock_file_fd = open(ENGINE_LOCK_FILE, O_CREAT | O_RDWR, 0660);
		if (lock_file_fd < 0) {
			rc = errno;
		}
	}

	if (rc == 0) {

		/* Close this file for all exec'd processes. */
		fcntl(lock_file_fd, F_SETFD, FD_CLOEXEC);

		lseek(lock_file_fd, offsetof(lock_file_t, pid), SEEK_SET);
		read(lock_file_fd, &lock_file->pid, lock_file_size - offsetof(lock_file_t, pid));

		/*
		 * Lock our access to the Engine.  Figure out whether to lock for
		 * reading or for writing based on the mode.
		 */
		if (mode & (ENGINE_WRITE | ENGINE_DAEMON)) {
			lockinfo.l_type = F_WRLCK;
		} else {
			lockinfo.l_type = F_RDLCK;
		}

		lockinfo.l_whence = SEEK_SET;
		if (mode & ENGINE_DAEMON) {
			lockinfo.l_start = offsetof(lock_file_t, daemon_lock);
			lockinfo.l_len = sizeof(lock_file->daemon_lock);

		} else {
			lockinfo.l_start = offsetof(lock_file_t, engine_lock);
			lockinfo.l_len = sizeof(lock_file->engine_lock);
		}

		rc = fcntl(lock_file_fd, F_SETLK, &lockinfo);

		if (rc == 0) {
			/* Put our information in the lock file. */
			if (mode & ENGINE_DAEMON) {
				lock_file->daemon_lock = 1;
				lseek(lock_file_fd, offsetof(lock_file_t, daemon_lock), SEEK_SET);
				write(lock_file_fd, &lock_file->daemon_lock, sizeof(lock_file->daemon_lock));

			} else {
				lock_file->engine_lock = 1;
				lock_file->pid = getpid();
				lock_file->mode = mode;
				/*
				 * Leave lock_file->node as an empty string.  If
				 * the Engine is being opened on behalf of
				 * another node, the daemon will fill in the
				 * node field.
				 */
				lseek(lock_file_fd, offsetof(lock_file_t, engine_lock), SEEK_SET);
				write(lock_file_fd, &lock_file->engine_lock, lock_file_size - offsetof(lock_file_t, engine_lock));
			}

		} else {
			/*
			 * We could not lock the Engine.  It must be in use by some
			 * other process.  Try to find out what process it is and tell
			 * the user.
			 */
			rc = fcntl(lock_file_fd, F_GETLK, &lockinfo);

			if (rc == 0) {
				int bytes;

				/*
				 * Check if the process is alive by sending it a harmless
				 * signal.
				 */
				rc = kill(lockinfo.l_pid, SIGCONT);
				if (rc != 0) {
					if ((errno == ESRCH) && !retry) {
						/*
						 * The locking process is gone.
						 * Try deleting the lock file and starting again.
						 */
						close(lock_file_fd);
						lock_file_fd = 0;
						rc = unlink(ENGINE_LOCK_FILE);
						if (rc == 0) {
							retry = TRUE;
							goto start;
						}
					}
				}

				if ((mode & ENGINE_CRITICAL) &&
				    (!(lock_file->mode & ENGINE_CRITICAL))) {
					send_shutdown(lockinfo.l_pid);
					close(lock_file_fd);
					goto start;
				}

				buf1 = engine_alloc(PATH_MAX + 1);
				if (buf1 == NULL) {
					LOG_CRITICAL("Error getting memory for buffer 1.\n");
					LOG_PROC_EXIT_INT(ENOMEM);
					return ENOMEM;
				}

				buf2 = engine_alloc(PATH_MAX + 1);
				if (buf2 == NULL) {
					LOG_CRITICAL("Error getting memory for buffer 2.\n");
					engine_free(buf1);
					LOG_PROC_EXIT_INT(ENOMEM);
					return ENOMEM;
				}
				
				/*
				 * The file name for the pid is symlinked at
				 * /proc/{pid}/exe
				 */
				sprintf(buf1, "/proc/%d/exe", lockinfo.l_pid);

				bytes = readlink(buf1, buf2, PATH_MAX + 1);

				if (bytes > 0) {
					/*
					 * readlink() does not null terminate the returned
					 * file name.
					 */
					buf2[bytes] = '\0';

					if (mode & ENGINE_DAEMON) {
						sprintf(lock_msg, "The EVMS Daemon is already running in process %d (%s).\n",
							lockinfo.l_pid, buf2);
					} else {
						sprintf(lock_msg, "The EVMS Engine is currently in use by process %d (%s).\n",
							lockinfo.l_pid, buf2);
					}

				} else {
					/*
					 * Couldn't get the name for the process ID.
					 * Log a message with just the PID.
					 */
					if (mode & ENGINE_DAEMON) {
						sprintf(lock_msg, "The EVMS Daemon is already running in process %d.\n",
							lockinfo.l_pid);
					} else {
						sprintf(lock_msg, "The EVMS Engine is currently in use by process %d.\n",
							lockinfo.l_pid);
					}
				}

				if ((!(mode & ENGINE_DAEMON)) &&
				    (lock_file->node[0] != '\0')) {
					sprintf(lock_msg + strlen(lock_msg), "The process has locked the Engine on behalf of node %s.\n", lock_file->node);
				}

				engine_free(buf2);

			} else {
				/*
				 * Couldn't get the lockinfo of the locking process.
				 * Log a generic message.
				 */
				if (mode & ENGINE_DAEMON) {
					sprintf(lock_msg, _("The EVMS Daemon is already running.\n"));
				} else {
					sprintf(lock_msg, _("The EVMS Engine is currently in use by another process.\n"));
				}
			}

			engine_free(buf1);

			if (!((engine_mode & ENGINE_DAEMON) &&
			      (!(mode & ENGINE_DAEMON))) ) {
				engine_user_message(NULL, NULL, lock_msg);
			}

			close(lock_file_fd);
			lock_file_fd = 0;
			rc = EACCES;
		}

	} else {
		sprintf(lock_msg, "Unable to open the Engine lock file %s: %s.  The Engine is not protected against other instances of the Engine being opened at the same time.\n", ENGINE_LOCK_FILE, strerror(rc));
                rc = 0;
		engine_user_message(NULL, NULL, lock_msg);
		lock_file_fd = 0;
	}

	if (msg != NULL) {
		*msg = lock_msg;
	}

	LOG_PROC_EXIT_INT(rc);
	return rc;
}


int unlock_engine(engine_mode_t mode) {

	int rc = 0;

	LOG_PROC_ENTRY();

	if (lock_file_fd != 0) {
		struct flock lockinfo = {0};

		/* Clear the contents of the lock file. */
		if (mode & ENGINE_DAEMON) {
			lock_file->daemon_lock = 0;
			lseek(lock_file_fd, offsetof(lock_file_t, daemon_lock), SEEK_SET);
			write(lock_file_fd, &lock_file->daemon_lock, sizeof(lock_file->daemon_lock));

		} else {
			lock_file->engine_lock = 0;
			lock_file->pid = 0;
			lock_file->mode = ENGINE_CLOSED;
			memset(lock_file->node, '\0', lock_file_size - offsetof(lock_file_t, node));
			lseek(lock_file_fd, offsetof(lock_file_t, engine_lock), SEEK_SET);
			write(lock_file_fd, &lock_file->engine_lock, lock_file_size - offsetof(lock_file_t, engine_lock));
		}

		lockinfo.l_type = F_UNLCK;
		lockinfo.l_whence = SEEK_SET;
		if (mode & ENGINE_DAEMON) {
			lockinfo.l_start = offsetof(lock_file_t, daemon_lock);
			lockinfo.l_len = sizeof(lock_file->daemon_lock);

		} else {
			lockinfo.l_start = offsetof(lock_file_t, engine_lock);
			lockinfo.l_len = sizeof(lock_file->engine_lock);
		}

		rc = fcntl(lock_file_fd, F_SETLK, &lockinfo);
		if (rc == 0) {
			close(lock_file_fd);
			lock_file_fd = 0;
		} else {
			rc = errno;
		}

		engine_free(lock_file);
		lock_file = NULL;
	}

	LOG_PROC_EXIT_INT(rc);
	return rc;
}


static void free_list_items(list_anchor_t list,
			    object_type_t type) {

	list_element_t iter1;
	list_element_t iter2;
	void * thing;

	LIST_FOR_EACH_SAFE(list,iter1, iter2, thing) {
		
		switch (type) {
		case DISK:
			{
				storage_object_t * obj = (storage_object_t *) thing;

				/*
				 * Mark the object inactive so that
				 * engine_free_logical_disk() doesn't put the
				 * object on the deactivate_list.
				 */
				obj->flags &= ~SOFLAG_ACTIVE;
				engine_free_logical_disk(obj);
			}
			break;

		case SEGMENT:
			{
				storage_object_t * obj = (storage_object_t *) thing;

				/*
				 * Mark the object inactive so that
				 * engine_free_segment() doesn't put the
				 * object on the deactivate_list.
				 */
				obj->flags &= ~SOFLAG_ACTIVE;
				engine_free_segment(obj);
			}
			break;

		case REGION:
			{
				storage_object_t * obj = (storage_object_t *) thing;

				/*
				 * Mark the object inactive so that
				 * engine_free_region() doesn't put the
				 * object on the deactivate_list.
				 */
				obj->flags &= ~SOFLAG_ACTIVE;
				engine_free_region(obj);
			}
			break;

		case EVMS_OBJECT:
			{
				storage_object_t * obj = (storage_object_t *) thing;

				/*
				 * Mark the object inactive so that
				 * engine_free_evms_object() doesn't put the
				 * object on the deactivate_list.
				 */
				obj->flags &= ~SOFLAG_ACTIVE;
				engine_free_evms_object(obj);
			}
			break;

		case CONTAINER:
			{
				storage_container_t * con = (storage_container_t *) thing;

				engine_free_container(con);
			}
			break;

		case VOLUME:
			{
				logical_volume_t * vol = (logical_volume_t *) thing;

				destroy_handle(vol->app_handle);
				engine_unregister_name(vol->name);

				/*
				 * The volume may also be on the the
				 * volume_delete_list.  If we are processing
				 * the volumes_list, remove the volume from the
				 * volume_delete_list.
				 */
				if (list == &volumes_list) {
					remove_thing(&volume_delete_list, thing);
				}
			}

			/*
			 * Fall through to remove the logical_volume from
			 * the list and free its memory.
			 */

		default:
			/*
			 * Don't know what the object is.  Delete the item from the
			 * list and free its memory
			 */
			delete_element(iter1);
			break;
		}
	}
}


static void cleanup_evms_lists(void) {

	LOG_PROC_ENTRY();

	delete_all_elements(&plugins_list);

	free_list_items(&disks_list, DISK);
	delete_all_elements(&disks_list);

	free_list_items(&segments_list, SEGMENT);
	delete_all_elements(&segments_list);

	free_list_items(&containers_list, CONTAINER);
	delete_all_elements(&containers_list);

	free_list_items(&regions_list, REGION);
	delete_all_elements(&regions_list);

	free_list_items(&EVMS_objects_list, EVMS_OBJECT);
	delete_all_elements(&EVMS_objects_list);

	free_list_items(&volumes_list, VOLUME);
	delete_all_elements(&volumes_list);

	free_list_items(&kill_sectors_list, 0);
	delete_all_elements(&kill_sectors_list);

	free_list_items(&volume_delete_list, VOLUME);
	delete_all_elements(&volume_delete_list);

	free_list_items(&deactivate_list, 0);
	delete_all_elements(&deactivate_list);

	delete_all_elements(&rename_volume_list);

	LOG_PROC_EXIT_VOID();
}


int evms_close_engine(void) {

	int rc = 0;

	LOG_PROC_ENTRY();

	if (engine_mode != ENGINE_CLOSED) {
		close(dm_control_fd);
		dm_control_fd = 0;
		evms_free_config();
		unload_plugins();
		unmount_filesystems();
		clear_name_registry();
		destroy_all_handles();
		cleanup_evms_lists();
		remove_signal_handlers();
		unlock_engine(engine_mode);
		engine_mode = ENGINE_CLOSED;

	} else {
		LOG_DEBUG("The Engine is already closed.\n");
		rc = EPERM;
	}

	LOG_PROC_EXIT_INT(rc);

	stop_logging();

	return rc;
}


static void * thread_engine_user_message(void * arg) {

	char * msg = (char *) arg;

	engine_user_message(NULL, NULL, msg);

	return NULL;
}


static char msg_buf[256];

static void shutdown_engine() {

	int seconds_until_death = SHUTDOWN_TIMEOUT;
	pthread_t msg_tid;

	LOG_PROC_ENTRY();

	if (engine_mode & ENGINE_WORKER) {
		/*
		 * Hey, I'm not the one running the show.  Send the shutdown
		 * on to the node that started me.
		 */
		remote_shutdown();

		LOG_PROC_EXIT_VOID();
		return;

	}
	
	while (seconds_until_death > 0) {
		if (engine_mode & ENGINE_WRITE) {
			sprintf(msg_buf,
				"Another process urgently needs the Engine.  "
				"Please save your changes or quit now.  "
				"This process will self destruct in %d seconds.\n",
				seconds_until_death);

		} else {
			sprintf(msg_buf,
				"Another process urgently needs the Engine.  "
				"Please quit now.  "
				"This process will self destruct in %d seconds.\n",
				seconds_until_death);
		}

		pthread_create(&msg_tid,
			       &pthread_attr_detached,
			       thread_engine_user_message,
			       msg_buf);

		/* Warn every ten seconds and on the last five seconds. */

		if (seconds_until_death >= 10) {
			usleep(10000000);
			seconds_until_death -= 10;
		} else {
			usleep(5000000);
			seconds_until_death -= 5;
		}
	}

	if (seconds_until_death <= 0) {
		pthread_create(&msg_tid,
			       &pthread_attr_detached,
			       thread_engine_user_message,
			       _("Self destruct sequence initiated.\n"));


	}

	/* Don't self destruct in the middle of a commit. */
	while (commit_in_progress) {
		usleep(1000000);
	}

	/* Close the Engine nicely if the user/UI didn't do it. */
	evms_close_engine();

	/* Try to exit nicely. */
	raise(SIGQUIT);

	usleep(3000000);

	/* Die, die, die. */
	raise(SIGKILL);

	/* We should never get here.  But if we do, we'll see it in the log. */
	LOG_PROC_EXIT_VOID();
}


/*
 * Check if the old EVMS kernel is running on this system.
 */
static int check_for_evms_kernel(void) {

	int rc = 0;
	int status;
	struct stat statbuf;
	dev_t evms_devt = makedev(EVMS_MAJOR, 0);
	int fd;
	char temp_node_name[65];

	LOG_PROC_ENTRY();

	/*
	 * Simple check first. Is there a "/proc/evms" directory?
	 * If the EVMS kernel is running it will exist.
	 */
	status = stat("/proc/evms", &statbuf);
	if (status == 0) {
		if (S_ISDIR(statbuf.st_mode)) {

			/*
			 * Make a dev node for EVMS_MAJOR:0 and see if it can
			 * be opened.
			 */

			/* Find a random file name. */
			do {
				sprintf(temp_node_name, "%x", rand());

				status = stat(temp_node_name, &statbuf);
			} while (status == 0);

			status = mknod(temp_node_name, (S_IFBLK | S_IRUSR | S_IWUSR | S_IRGRP | S_IWGRP), evms_devt);

			if (status == 0) {
				fd = open(temp_node_name, O_RDWR|O_NONBLOCK);
				if (fd > 0) {
					/* Open succeeded.  EVMS kernel is running. */
					close(fd);
					rc = EEXIST;
				}

				unlink(temp_node_name);
			}
		}
	}

	LOG_PROC_EXIT_INT(rc);
	return rc;
}


static void check_for_2_4_kernel(void)
{
	int fd;
	char needle[] = "version";
	char haystack[256];
	char *ver;
	int major, minor, patch;

	LOG_PROC_ENTRY();

	/* Assume the worst. */
	is_2_4_kernel = TRUE;

	fd = open("/proc/version", O_RDONLY);
	if (fd >= 0) {
		read(fd, haystack, 256);
		close(fd);
		
		ver = strstr(haystack, needle);
		if (ver) {
			sscanf(ver, "%*s %d.%d.%d", &major, &minor, &patch);

			LOG_DETAILS("Kernel version is: %d.%d.%d\n", major, minor, patch);

			if ((major != 2) || (minor != 4)) {
				is_2_4_kernel = FALSE;
			}

		} else {
			LOG_WARNING("Could not find \"version\" in the version string in /proc/version.");
			LOG_WARNING("Assuming kernel is version 2.4.\n");
		}

	} else {
		LOG_WARNING("Open of /proc/version failed with error code %d: %s\n", errno, strerror(errno));
		LOG_WARNING("Assuming kernel is version 2.4.\n");
	}

	LOG_PROC_EXIT_VOID()
	return;
}


/*
 * Check if the Engine is currently allowing read access.  The Engine must be
 * open and a commit must not be in progress.
 */
int check_engine_read_access() {

	int rc = 0;

	LOG_PROC_ENTRY();

	if ((engine_mode == ENGINE_CLOSED) ||
	    commit_in_progress) {
		if (engine_mode == ENGINE_CLOSED) {
			LOG_ERROR("The Engine is not open.\n");
		}
		if (commit_in_progress) {
			LOG_ERROR("The Engine is currently committing changes.\n");
		}
		rc = EACCES;
	}

	LOG_PROC_EXIT_INT(rc);
	return rc;
}


/*
 * Check if the Engine is allowing write access, i.e., data structures, etc.
 * can be updated.  The Engine must be opened in a mode that allows writing
 * and a commit must not be in progress.
 */
int check_engine_write_access() {

	int rc = 0;

	LOG_PROC_ENTRY();

	if ((engine_mode == ENGINE_CLOSED) ||
	    (!(engine_mode & ENGINE_WRITE)) ||
	    commit_in_progress) {
		if (engine_mode == ENGINE_CLOSED) {
			LOG_ERROR("The Engine is not open.\n");
		} else {
			if (engine_mode & ENGINE_WRITE) {
				LOG_ERROR("The Engine is not open for writing.\n");
			}
		}
		if (commit_in_progress) {
			LOG_ERROR("The Engine is currently committing changes.\n");
		}
		rc = EACCES;
	}

	LOG_PROC_EXIT_INT(rc);
	return rc;
}


static void get_config_mode(char          * key,
			    engine_mode_t * open_mode) {

	int rc;
	const char * config_string;

	rc = evms_get_config_string(key, &config_string);
	if (rc == 0) {
		if (strcasecmp(config_string, "readonly") == 0) {
			*open_mode = ENGINE_READONLY;
		} else if (strcasecmp(config_string, "readwrite") == 0) {
			*open_mode = ENGINE_READWRITE;
		} else if (strcasecmp(config_string, "daemon") == 0) {
			*open_mode = ENGINE_DAEMON;
		}
	}
}


static void get_config_debug_level(char          * key,
				   debug_level_t * debug_level) {

	int rc;
	const char * config_string;

	rc = evms_get_config_string(key, &config_string);
	if (rc == 0) {
		if (strcasecmp(config_string, "CRITICAL") == 0) {
			*debug_level = CRITICAL;
		} else if (strcasecmp(config_string, "SERIOUS") == 0) {
			*debug_level = SERIOUS;
		} else if (strcasecmp(config_string, "ERROR") == 0) {
			*debug_level = ERROR;
		} else if (strcasecmp(config_string, "WARNING") == 0) {
			*debug_level = WARNING;
		} else if (strcasecmp(config_string, "DEFAULT") == 0) {
			*debug_level = DEFAULT;
		} else if (strcasecmp(config_string, "DETAILS") == 0) {
			*debug_level = DETAILS;
		} else if (strcasecmp(config_string, "DEBUG") == 0) {
			*debug_level = DEBUG;
		} else if (strcasecmp(config_string, "EXTRA") == 0) {
			*debug_level = EXTRA;
		} else if (strcasecmp(config_string, "ENTRY_EXIT") == 0) {
			*debug_level = ENTRY_EXIT;
		} else if (strcasecmp(config_string, "EVERYTHING") == 0) {
			*debug_level = EVERYTHING;
		}
	}
}


static void sigusr1_handler(int sig_no) {

	LOG_CRITICAL("***\n");
	LOG_CRITICAL("*** Received shutdown signal: %s.\n", sys_siglist[sig_no]);
	LOG_CRITICAL("***\n");

	sem_post(&shutdown_sem);
}


static void * shutdown_thread(void * arg) {

	sem_t * shutdown_sem = (sem_t *) arg;
	int rc;

	LOG_PROC_ENTRY();

	LOG_DEBUG("Register SIGUSR1 handler.\n");
	signal(SIGUSR1, sigusr1_handler);

	sem_init(shutdown_sem, 0, 0);

	do {
		LOG_DEBUG("Wait for SIGUSR1 shutdown signal.\n");
		rc = sem_wait(shutdown_sem);
		LOG_DEBUG("sem_wait returned %d.\n", rc);

		if (rc != 0) {
			LOG_DEBUG("errno is %d: %s\n", errno, strerror(errno));
		}

	} while ((rc != 0) && (errno == EINTR));
					
	LOG_DEBUG("Shutdown signal received.\n");

	if (rc == 0) {
		shutdown_engine();

	} else {
		LOG_CRITICAL("sem_wait() failed with error code %d: %s\n", rc, strerror(rc));
		LOG_CRITICAL("The shutdown thread is disabled.\n");
	}

	LOG_PROC_EXIT_VOID();

	return NULL;
}


#define VERSION_FIELDS(ver) ver.major, ver.minor, ver.patchlevel

int evms_get_api_version(evms_version_t * version) {

	*version = engine_api_version;
	return 0;
}


int evms_open_engine(char           * node_name,
		     engine_mode_t    mode,
		     ui_callbacks_t * callbacks,
		     debug_level_t    level,
		     char           * log_name) {
	int rc = 0;
	engine_mode_t open_mode;
	pthread_t shutdown_tid;

	if (geteuid() != 0) {
		if (callbacks != NULL) {
			if (callbacks->user_message != NULL) {
				callbacks->user_message(_("You must have root privilege to open the EVMS Engine."), NULL, NULL);
			}
		}

		return EACCES;
	}

	/* Setup defaults. */
	current_node_name = NULL;
	open_mode = ENGINE_READWRITE;
	debug_level = DEFAULT;
	log_usec = FALSE;
	log_pid = FALSE;

	if ((mode != -1) && (mode & ENGINE_DAEMON)) {
		log_file_name = DEFAULT_DAEMON_LOG_FILE;
	} else {
		log_file_name = DEFAULT_LOG_FILE;
	}

	ui_callbacks = callbacks;

	/*
	 * Free any configuration that may have been read in and abandoned by
	 * an application outside of the Engine being open.
	 */
	evms_free_config();

	/*
	 * Get settings from the evms.conf.  They may override the default
	 * settings above.
	 */
	if (evms_get_config(NULL) == 0) {
		if ((mode != -1) && (mode & ENGINE_DAEMON)) {
			get_config_debug_level("daemon.debug_level", &debug_level);
			evms_get_config_string("daemon.log_file", (const char * *) &log_file_name);
			evms_get_config_bool("daemon.log_usec", &log_usec);
			evms_get_config_bool("daemon.log_pid", &log_pid);

		} else {
			get_config_mode("engine.mode", &open_mode);
			get_config_debug_level("engine.debug_level", &debug_level);
			evms_get_config_string("engine.node", (const char * *) &current_node_name);
			evms_get_config_string("engine.log_file", (const char * *) &log_file_name);
			evms_get_config_bool("engine.log_usec", &log_usec);
			evms_get_config_bool("engine.log_pid", &log_pid);
		}
	}

	/*
	 * Get user supplied settings, which override both the defaults and the
	 * settings in evms.conf.
	 */
	if (node_name != NULL) {
		current_node_name = node_name;
	}

	if (mode != -1) {
		open_mode = mode;
	}

	if (level != -1) {
		debug_level = level;
	}

	if (log_name != NULL) {
		log_file_name = log_name;
	}

	start_logging(log_file_name);

	LOG_PROC_ENTRY();

	LOG_DEFAULT("Engine version:          %2d.%d.%d\n", VERSION_FIELDS(engine_version));
	LOG_DEFAULT("External API version:    %2d.%d.%d\n", VERSION_FIELDS(engine_api_version));
	LOG_DEFAULT("Engine services version: %2d.%d.%d\n", VERSION_FIELDS(engine_services_api_version));
	LOG_DEFAULT("Plug-in API version:     %2d.%d.%d\n", VERSION_FIELDS(engine_plugin_api_version));
	LOG_DEFAULT("Container API version:   %2d.%d.%d\n", VERSION_FIELDS(engine_container_api_version));
	LOG_DEFAULT("FSIM API version:        %2d.%d.%d\n", VERSION_FIELDS(engine_fsim_api_version));
	LOG_DEFAULT("Cluster API version:     %2d.%d.%d\n", VERSION_FIELDS(engine_cluster_api_version));
	LOG_DEFAULT("Daemon protocol version: %2d.%d.%d\n", VERSION_FIELDS(engine_daemon_msg_version));

	sprintf(message_buffer, "Requested open mode is %#x.", mode);
	if (mode == -1) {
		strcat(message_buffer, " (default)\n");

	} else {
		if (mode & ENGINE_READ) {
			strcat(message_buffer, " ENGINE_READ");
		}
		if (mode & ENGINE_WRITE) {
			strcat(message_buffer, " ENGINE_WRITE");
		}
		if (mode & ENGINE_CRITICAL) {
			strcat(message_buffer, " ENGINE_CRITICAL");
		}
		if (mode & ENGINE_DAEMON) {
			strcat(message_buffer, " ENGINE_DAEMON");
		}
		if (mode & ENGINE_WORKER) {
			strcat(message_buffer, " ENGINE_WORKER");
		}
		strcat(message_buffer, "\n");
	}
	LOG_DEFAULT("%s", message_buffer);

	if (!(open_mode & (ENGINE_READ | ENGINE_DAEMON))) {
		LOG_ERROR("Open mode of %d is not valid.\n", open_mode);

		evms_free_config();
		LOG_PROC_EXIT_INT(EINVAL);
		return EINVAL;
	}

	if (engine_mode != ENGINE_CLOSED) {
		/* This Engine is already open. */
		LOG_ERROR("The Engine is already opened.\n");
		evms_free_config();
		LOG_PROC_EXIT_INT(EINVAL);
		return EINVAL;
	}

	/* Check for /proc and /sys. */
	rc = check_for_filesystems();
	if (rc != 0) {
		evms_free_config();
		LOG_PROC_EXIT_INT(EINVAL);
		return EINVAL;
	}

	/* Check if we can access the EVMS kernel code. */
	rc = check_for_evms_kernel();
	if (rc != 0) {
		engine_user_message(NULL, NULL,
				    _("This system is running the EVMS kernel.  "
				      "EVMS Engine versions 1.9.0 and above do not run with the EVMS kernel.  "
				      "Use EVMS Engine version 1.2 to configure with the EVMS kernel.\n"));
		evms_free_config();
		LOG_PROC_EXIT_INT(rc);
		return rc;
	}

	engine_mode = open_mode;
	sprintf(message_buffer, "Open mode is %#x.", engine_mode);
	if (engine_mode & ENGINE_READ) {
		strcat(message_buffer, " ENGINE_READ");
	}
	if (engine_mode & ENGINE_WRITE) {
		strcat(message_buffer, " ENGINE_WRITE");
	}
	if (engine_mode & ENGINE_CRITICAL) {
		strcat(message_buffer, " ENGINE_CRITICAL");
	}
	if (engine_mode & ENGINE_DAEMON) {
		strcat(message_buffer, " ENGINE_DAEMON");
	}
	if (engine_mode & ENGINE_WORKER) {
		strcat(message_buffer, " ENGINE_WORKER");
	}
	strcat(message_buffer, "\n");
	LOG_DEFAULT("%s", message_buffer);

	check_for_2_4_kernel();

	/* Initialize the random number generator. */
	srand(time(NULL) + getpid());

	/* Install signal handlers. */
	install_signal_handlers();

	if (open_dm_control_node() != 0) {
		engine_user_message(NULL, NULL,
				    _("Unable to open the control node for Device-Mapper.  "
				      "The Engine will run without Device-Mapper support.\n"));
	}

	/* Setup the pthread attribute for starting threads detached. */
	/* That is, no other thread will wait for the created thread to exit. */
	pthread_attr_init(&pthread_attr_detached);
	pthread_attr_setdetachstate(&pthread_attr_detached, PTHREAD_CREATE_DETACHED);

	rc = lock_engine(engine_mode, NULL);
	if (rc == 0) {

		rc = load_plugins();
		if (rc == 0) {

			if (engine_mode & ENGINE_DAEMON) {

				/* Daemon can't run without a cluster manager. */
				if (cluster_manager == NULL) {
					engine_user_message(NULL, NULL,
							    _("There is no cluster manager plug-in loaded on this system.\n"));
					rc = ENODEV;
				}

			} else {
				if (cluster_manager != NULL) {
					if (!(engine_mode & ENGINE_WORKER)) {
						/*
						 * Open the Engine on the other nodes
						 * in the cluster.
						 */
						rc = remote_open_engine(NULL, engine_mode, NULL, debug_level, log_file_name);
						if (rc != 0) {
							/*
							 * At least one of the Engine on
							 * another node failed to start.
							 * Release any Engines we may
							 * have already started on other
							 * nodes.
							 */
							remote_close_engine();
							disconnect_from_ece();
							local_focus = TRUE;

							engine_user_message(NULL, NULL,
									    _("There was an error when starting EVMS on the other nodes in the cluster.  "
									      "The error code was %d: %s.  "
									      "EVMS will only manage local devices on this system.\n"),
									    rc, evms_strerror(rc));
							rc = 0;
						}
					}

				} else {
					if (current_node_name != NULL) {
						engine_user_message(NULL, NULL,
								    _("There is no cluster manager plug-in loaded on this system.  "
								      "The node_name parameter \"%s\" is ignored.\n"),
								    current_node_name);
					}
				}

				pthread_create(&shutdown_tid,
					       &pthread_attr_detached,
					       shutdown_thread,
					       &shutdown_sem);

				/* Initialize the Handle Manager. */
				if (initialize_handle_manager()) {

					rc = initial_discovery();
					if (rc != 0) {
						destroy_all_handles();
					}

				} else {
					LOG_CRITICAL("Handle Manager failed to initialize.\n");
					rc = ENOMEM;
				}
			}

			if (rc != 0) {
				unload_plugins();
			}
		}

		if (rc != 0) {
			unlock_engine(engine_mode);
		}
	}

	if (rc != 0) {
		cleanup_evms_lists();
	}

	if (rc != 0) {
		close(dm_control_fd);
		evms_free_config();
		engine_mode = ENGINE_CLOSED;
	}

	LOG_PROC_EXIT_INT(rc);
	return rc;
}


void free_changes_pending_record_array_contents(void * thing) {

	change_record_array_t * changes = (change_record_array_t *) thing;
	int i;

	for (i = 0; i < changes->count; i++) {
		engine_free(changes->changes_pending[i].name);
	}
}


int evms_changes_pending(boolean                 * result,
			 change_record_array_t * * p_changes) {

	int rc = 0;

	LOG_PROC_ENTRY();

	rc = check_engine_read_access();
	if (rc != 0) {
		LOG_PROC_EXIT_INT(rc);
		return rc;
	}

	if (!local_focus) {
		rc = remote_changes_pending(result, p_changes);
		LOG_PROC_EXIT_INT(rc);
		return rc;
	}

	if (p_changes == NULL) {
		*result = are_changes_pending(NULL);

	} else {
		change_record_array_t * changes = NULL;
		change_record_array_t * user_changes = NULL;

		*result = are_changes_pending(&changes);

		if (changes != NULL) {
			user_changes = alloc_app_struct(sizeof(change_record_array_t) + sizeof(change_record_t) * changes->count,
							free_changes_pending_record_array_contents);

			if (user_changes != NULL) {
				int i;

				user_changes->count = changes->count;

				for (i = 0; i < changes->count; i++) {
					user_changes->changes_pending[i].name = engine_strdup(changes->changes_pending[i].name);
					user_changes->changes_pending[i].type = changes->changes_pending[i].type;
					user_changes->changes_pending[i].changes = changes->changes_pending[i].changes;
				}

			} else {
				LOG_CRITICAL("Error allocating memory for a change_record_array_t.\n");
			}

			engine_free(changes);
		}

		*p_changes = user_changes;
	}

	LOG_PROC_EXIT_INT(rc);
	return rc;
}


/*
 * Get the Engine's current setting of the debug level.
 */
int evms_get_debug_level(debug_level_t * level) {

	int rc = 0;

	LOG_PROC_ENTRY();

	rc = check_engine_read_access();

	/*
	 * Debug level is the same across all nodes in the cluster,
	 * so we can just get the level from the local node.
	 */
	if (rc == 0) {
		*level = debug_level;
	}

	LOG_PROC_EXIT_INT(0)
	return 0;
}


/*
 * Set the Engine's debug level.
 */
int evms_set_debug_level(debug_level_t level) {

	int rc = 0;
	int new_debug_level = debug_level;


	/*
	 * Always log the activity for setting the debug level.  If the debug
	 * level changes from below ENTRY_EXIT to ENTRY_EXIT or above the log
	 * will not contain a line for entering this function.  Conversely,
	 * if the debug level changes from ENTRY_EXIT or above to below
	 * ENTRY_EXIT the log will not contain the line for exiting this
	 * function.  Changing the log level in the middle of
	 * evms_set_debug_level() does funky things to the log.
	 */
	if (debug_level < DEBUG) {
		debug_level = DEBUG;
        }

	LOG_PROC_ENTRY();

	LOG_DEBUG("Request to set debug level to %d.\n", level);

	/*
	 * People that opened the Engine for read-only are allowed to set their
	 * debug level.
	 */
	rc = check_engine_read_access();
	if (rc != 0) {
		LOG_PROC_EXIT_INT(rc)
		return(rc);
	}

	if ((level >= CRITICAL) &&
	    (level <= EVERYTHING)) {
		/*
		 * If we are running in a cluster and this is not a
		 * worker, set the debug level on the other nodes in the
		 * cluster.  (A worker must not forward the
		 * set_debug_level() on to other nodes.  It would cause
		 * recursion as the worker(s) would then receive another
		 * set_debug_level() and attmpt to forward that one
		 * too...)
		 */
		if ((current_nodeid != NULL) && !(engine_mode & ENGINE_WORKER)) {
			rc = remote_set_debug_level(level);
		}

		new_debug_level = level;

		LOG_DEBUG("Debug level is set to %d.\n", new_debug_level);

	} else {
		LOG_ERROR("Debug level is out of range.  It must be between %d and %d, inclusive.\n", CRITICAL, EVERYTHING);
		rc = EINVAL;
	}

	LOG_PROC_EXIT_INT(rc)

	debug_level = new_debug_level;
	return(rc);
}


/*-----------------------------------------------------------------------------+
+                                                                              +
+                              Logging Functions                               +
+                                                                              +
+-----------------------------------------------------------------------------*/

int evms_write_log_entry(debug_level_t level,
			 char        * module_name,
			 char        * fmt,
			 ...) {
	int rc = 0;
	int len;
	va_list args;

	if (dm_device_suspended) {
		return rc;
	}

	if (level <= debug_level) {
		if (log_file_fd > 0) {
			pthread_mutex_lock(&log_mutex);
			timestamp(log_buf, LOG_BUF_SIZE, level);
			strcat(log_buf, module_name);
			strcat(log_buf, ": ");
			len = strlen(log_buf);

			va_start(args, fmt);
			len += vsprintf(log_buf + strlen(log_buf), fmt, args);
			va_end(args);

			if (write(log_file_fd, log_buf, len) < 0) {
				rc = errno;
			}
			pthread_mutex_unlock(&log_mutex);

		} else {
			rc = ENOENT;
		}
	}

	return rc;
}


/*-----------------------------------------------------------------------------+
+                                                                              +
+                             Some List Handlers                               +
+                                                                              +
+-----------------------------------------------------------------------------*/

/*
 * Make sure this object has an app_handle.
 */
int ensure_app_handle(void * thing) {

	int rc = 0;
	/*
	 * All exportable Engine structures start with the common header.
	 */
	common_header_t * hdr = (common_header_t *) thing;

	LOG_PROC_ENTRY();

	if (hdr->app_handle == 0) {
		LOG_DEBUG("Create a handle for a thing of type %d.\n", hdr->type);
		rc = create_handle(thing,
				   hdr->type,
				   &hdr->app_handle);
		if (rc == 0) {
			LOG_DEBUG("Handle is %d.\n", hdr->app_handle);
		}
	}

	LOG_PROC_EXIT_INT(rc);
	return rc;
}


/*
 * Add the app_handle for a given object to a handle_array_t.
 */
int make_handle_entry(void            * thing,
		      handle_array_t  * ha) {
	int rc = 0;
	/*
	 * All exportable Engine structures start with the common header.
	 */
	common_header_t * hdr = (common_header_t *) thing;

	LOG_PROC_ENTRY();

	rc = ensure_app_handle(thing);
	if (rc == 0) {
		ha->handle[ha->count] = hdr->app_handle;
		ha->count++;
	}

	LOG_PROC_EXIT_INT(rc);
	return rc;
}


/*
 * Make a user (it has a Engine memory block header) array of handles
 * (handle_array_t) for the objects in a list_anchor_t.
 */
int make_user_handle_array(list_anchor_t      list,
			   handle_array_t * * ha) {
	int rc = 0;
	uint count;
	uint size;

	LOG_PROC_ENTRY();

	count = list_count(list);

	size = sizeof(handle_array_t) + (count * sizeof(object_handle_t));

	*ha = alloc_app_struct(size, NULL);
	if (*ha != NULL) {
		list_element_t iter;
		void * thing;

		LIST_FOR_EACH(list, iter, thing) {
			make_handle_entry(thing, *ha);
		}

	} else {
		rc = ENOMEM;
	}

	LOG_PROC_EXIT_INT(rc);
	return rc;
}


/*
 * Get a list of handles for plug-ins, optionally filtering by plug-in type.
 */
int evms_get_plugin_list(evms_plugin_code_t    type,
			 plugin_search_flags_t flags,
			 handle_array_t    * * plugin_handle_list) {
	int rc = 0;
	list_anchor_t plugins;

	LOG_PROC_ENTRY();

	rc = check_engine_read_access();
	if (rc != 0) {
		LOG_PROC_EXIT_INT(rc);
		return rc;
	}

	if (!local_focus) {
		rc = remote_get_plugin_list(type, flags, plugin_handle_list);
		LOG_PROC_EXIT_INT(rc);
		return rc;
	}

	if (plugin_handle_list != NULL) {
		rc = engine_get_plugin_list(type, flags, &plugins);

		if (rc == 0) {
			rc = make_user_handle_array(plugins, plugin_handle_list);

			destroy_list(plugins);
		}

	} else {
		LOG_DEBUG("User specified NULL pointer for plugin_handle_list. There is nowhere to put the results.\n");
		rc = EINVAL;
	}

	LOG_PROC_EXIT_INT(rc);
	return rc;
}


/*
 * Get the handle for a plug-in with a given ID.
 */
int evms_get_plugin_by_ID(plugin_id_t       plugin_ID,
			  plugin_handle_t * plugin_handle) {
	int rc = 0;
	plugin_record_t * pPlugRec;

	LOG_PROC_ENTRY();

	rc = check_engine_read_access();
	if (rc != 0) {
		LOG_PROC_EXIT_INT(rc);
		return rc;
	}

	if (!local_focus) {
		rc = remote_get_plugin_by_ID(plugin_ID, plugin_handle);
		LOG_PROC_EXIT_INT(rc);
		return rc;
	}

	rc = engine_get_plugin_by_ID(plugin_ID, &pPlugRec);

	if (rc == 0) {
		rc = ensure_app_handle(pPlugRec);

		if (rc == 0) {
			*plugin_handle = pPlugRec->app_handle;
		}
	}

	LOG_PROC_EXIT_INT(rc);
	return rc;
}


/*
 * Get the handle for a plug-in with a given short name.
 */
int evms_get_plugin_by_name(char            * plugin_name,
			    plugin_handle_t * plugin_handle) {
	int rc = 0;

	plugin_record_t * pPlugRec;

	LOG_PROC_ENTRY();

	rc = check_engine_read_access();
	if (rc != 0) {
		LOG_PROC_EXIT_INT(rc);
		return rc;
	}

	if (!local_focus) {
		rc = remote_get_plugin_by_name(plugin_name, plugin_handle);
		LOG_PROC_EXIT_INT(rc);
		return rc;
	}

	rc = engine_get_plugin_by_name(plugin_name, &pPlugRec);

	if (rc == 0) {
		rc = ensure_app_handle(pPlugRec);

		if (rc == 0) {
			*plugin_handle = pPlugRec->app_handle;
		}
	}

	LOG_PROC_EXIT_INT(rc);
	return rc;
}


/********************************************/
/* Functions for updating the dev node tree */
/********************************************/

/*
 * Make a directory in the EVMS dev node tree.
 * This function is a utility function for make_evms_dev_node() below.
 */
int make_directory(char * dir_name, mode_t mode) {

	int rc = 0;
	int status;
	char name_buf[EVMS_VOLUME_NAME_SIZE + 1];
	char * tmp_ptr = name_buf;
	struct stat statbuf;

	LOG_PROC_ENTRY();

	/* See if this directory exists */
	status = stat(dir_name, &statbuf);
	if (status != 0) {
		if (errno == ENOENT) {
			/* This directory doesn't exist. */
			strcpy(name_buf, dir_name);

			/* Strip of any trailing '/'. */
			if (name_buf[strlen(name_buf) - 1] == '/') {
				name_buf[strlen(name_buf) - 1] = '\0';
			}

			/* Find the parent directory of this one. */
			tmp_ptr = strrchr(name_buf, '/');

			if ((tmp_ptr != NULL) && (tmp_ptr != name_buf)) {
				/*
				 * This directory has a parent directory.
				 * Peel off the directory name and call myself
				 * recursively to make the parent directory.
				 */
				*tmp_ptr = 0;

				rc = make_directory(name_buf, mode);
			}

			if (rc == 0) {
				LOG_DEBUG("Make directory \"%s\".\n", dir_name);
				status = mkdir(dir_name, mode);

				if (status != 0) {
					if (errno == EEXIST) {
						rc = 0;

					} else {
						rc = errno;
						LOG_WARNING("mkdir(%s) failed with error code %d.\n", dir_name, rc);
					}
				}
			}

		} else {
			rc = errno;
			LOG_WARNING("stat(%s) failed with error code %d.\n", dir_name, rc);
		}

	} else {
		/* Make sure this is a directory */
		if (!(statbuf.st_mode & S_IFDIR)) {
			LOG_ERROR("%s is not a directory.\n", dir_name);
			rc = EINVAL;
		} else {
			LOG_DEBUG("Directory %s already exists.\n", dir_name);
		}
	}

	LOG_PROC_EXIT_INT(rc);
	return rc;
}


int evms_update_evms_dev_tree(void) {

	int rc = 0;

	LOG_PROC_ENTRY();

	if (!local_focus) {
		LOG_PROC_EXIT_INT(ENOSYS);
		return ENOSYS;
	}

	/* Opening the Engine updates the /dev/evms tree. */
	rc = evms_open_engine(NULL, ENGINE_READONLY, NULL, DEFAULT, NULL);

	if (rc == 0) {
		evms_close_engine();
	}

	LOG_PROC_EXIT_INT(rc);
	return rc;
}


const char * evms_strerror(int err_num) {

	unsigned int err = abs(err_num);

	if (err <= EMEDIUMTYPE) {
		return strerror(err_num);
	}

	if (IS_HANDLE_MANAGER_ERROR(err)) {
		return handlemgr_strerror(err);
	}

	if (err == E_CANCELED) {
		return "Operation canceled";
	}

	if (err == E_NOLOAD) {
		return "Plug-in did not want to load";
	}

	return "Unknown error code";
}


static int backup_fh(char             * parent_name,
		     storage_object_t * child) {

	int rc = 0;

	LOG_PROC_ENTRY();

	feature_header_cpu_to_disk(child->feature_header);
	child->feature_header->crc = 0;
	child->feature_header->crc = CPU_TO_DISK32(evms_calculate_crc(EVMS_INITIAL_CRC,
								       child->feature_header,
								       sizeof(evms_feature_header_t)));

	rc = engine_save_metadata(parent_name,
				  child->name,
				  child->size - (EVMS_FEATURE_HEADER_SECTORS * 2),
				  EVMS_FEATURE_HEADER_SECTORS,
				  child->feature_header);
	if (rc == 0) {
		rc = engine_save_metadata(parent_name,
					  child->name,
					  child->size - EVMS_FEATURE_HEADER_SECTORS,
					  EVMS_FEATURE_HEADER_SECTORS,
					  child->feature_header);
	}

	if (rc != 0) {
		LOG_SERIOUS("Error backing up feature header for %s.  Error code is %d: %s\n",
			    parent_name, rc, evms_strerror(rc));
	}

	/* Convert the feature header back to CPU endian format. */
	feature_header_disk_to_cpu(child->feature_header);

	LOG_PROC_EXIT_INT(rc);
	return rc;
}


/*
 * Backup the feature headers for an object.  The feature headers reside on the
 * object's child objects.
 */
static int backup_feature_headers(storage_object_t * obj,
				  char           * * ancestors,
				  int                depth) {

	int rc = 0;
	list_element_t iter;
	storage_object_t * child;
	int i;

	LOG_PROC_ENTRY();

	LIST_FOR_EACH(obj->child_objects, iter, child) {
		if (child->feature_header != NULL) {

			/* This child may have feature headers to build it. */
			ancestors[depth] = obj->name;
			rc = backup_feature_headers(child, ancestors, depth + 1);

			if (rc == 0) {
				if (obj->volume != NULL) {
					rc = backup_fh(obj->volume->name, child);
				}
			}

			for (i = 0; (rc == 0) && (i < depth); i++) {
				rc = backup_fh(ancestors[i], child);
			}

			if (rc == 0) {
				rc = backup_fh(obj->name, child);
			}

			if (rc != 0) {
				break;
			}
		}
	}

	LOG_PROC_EXIT_INT(rc);
	return rc;
}


int evms_metadata_backup(const char * directory) {

	int rc = 0;
	int i;
	change_record_array_t * changes;
	time_t time_now;
	struct tm * tm_now;

	list_anchor_t object_lists[] = {&disks_list,
					&segments_list,
					&regions_list,
					&EVMS_objects_list,
					NULL};
	list_element_t iter;
	storage_object_t * obj;
	storage_container_t * con;
	logical_volume_t * vol;
	list_anchor_t top_objects;

	LOG_PROC_ENTRY();

	rc = check_engine_read_access();
	if (rc != 0) {
		LOG_PROC_EXIT_INT(rc);
		return rc;
	}

	if (!local_focus) {
		rc = remote_metadata_backup(directory);
		LOG_PROC_EXIT_INT(rc);
		return rc;
	}

	/*
	 * Make sure nothing is dirty.  We can't do a backup of a system
	 * that is currently being configured.  All changes must be on disk.
	 * Needing activation/deactivation is OK since it doesn't involve
	 * writing metadata.
	 */
	if (are_changes_pending(&changes)) {

		int i;
		boolean nothing_dirty = TRUE;

		for (i = 0; nothing_dirty && (i < changes->count); i++) {
			if (changes->changes_pending[i].changes & ~(CHANGE_ACTIVATE |
								    CHANGE_REACTIVATE |
								    CHANGE_DEACTIVATE)) {
				nothing_dirty = FALSE;
			}
		}

		engine_free(changes);

		if (!nothing_dirty) {
			engine_user_message(NULL,NULL,
					    _("EVMS cannot backup metadata while changes are pending.\n"));

			LOG_PROC_EXIT_INT(EPERM);
			return EPERM;
		}
	}

	/* Build the metadata database file name. */
	if (directory == NULL) {
		directory = DEFAULT_METADATA_DIR;
		evms_get_config_string("engine.metadata_backup_dir", &directory);
	}

	strcpy(metadata_db_name, directory);
	if (metadata_db_name[strlen(metadata_db_name) -1] != '/') {
		strcat(metadata_db_name, "/");
	}

	rc = make_directory(metadata_db_name, (S_IFDIR | S_IRWXU |
					       S_IRGRP | S_IXGRP |
					       S_IROTH | S_IXOTH));
	if (rc != 0) {
		engine_user_message(NULL, NULL,
				    "Error creating metadata backup directory %s.  Error code is %d: %s\n",
				    metadata_db_name, rc, evms_strerror(rc));
		
		LOG_PROC_EXIT_INT(rc);
		return rc;
	}

	strcat(metadata_db_name, EVMS_METADATA_FILE_NAME_PREFIX);

	time_now = time(NULL);
	tm_now = localtime(&time_now);

	sprintf(metadata_db_name + strlen(metadata_db_name), "%04u-%02u-%02u-%02u.%02u.%02u",
		tm_now->tm_year + 1900,
		tm_now->tm_mon + 1,
		tm_now->tm_mday,
		tm_now->tm_hour,
		tm_now->tm_min,
		tm_now->tm_sec);

	LOG_DEBUG("Metadata database file name is %s.\n", metadata_db_name);
	metadata_db_fd = open(metadata_db_name, O_CREAT | O_TRUNC | O_WRONLY, 0664);
	if (metadata_db_fd < 0) {

		rc = errno;
		
		engine_user_message(NULL, NULL,
				    "Error opening metadata backup file %s.  open() returned error %d: %s\n",
				    metadata_db_name, rc, strerror(rc));
		
		LOG_PROC_EXIT_INT(rc);
		return rc;
	}

	/* Call the plug-ins for all objects and have them save their metadata. */
	for (i = 0; object_lists[i] != NULL; i++) {
		LIST_FOR_EACH(object_lists[i], iter, obj) {
			rc = obj->plugin->functions.plugin->backup_metadata(obj);
			if (rc != 0) {
				if (rc == ENOSYS) {
					engine_user_message(NULL, NULL,
							    "Warning: The %s plug-in does not support backing up metadata for object %s.  "
							    "The metadata backup may be incomplete.\n",
							    obj->plugin->short_name, obj->name);
					rc = 0;

				} else {
					engine_user_message(NULL, NULL,
							    _("The %s plug-in returned error %d (%s) when asked to backup the metadata for object %s.\n"),
							    obj->plugin->short_name, rc, evms_strerror(rc), obj->name);
					goto error_out;
				}
			}
		}
	}

	/*
	 * Call the plug-ins for all containers and have them save their
	 * metadata.
	 */
	LIST_FOR_EACH(&containers_list, iter, con) {
		rc = con->plugin->container_functions->backup_container_metadata(con);
		if (rc != 0) {
			if (rc == ENOSYS) {
				LOG_WARNING("The %s plug-in does not support backing up metadata for container %s.\n",
					    con->plugin->short_name, con->name);
				rc = 0;

			} else {
				engine_user_message(NULL, NULL,
						    _("The %s plug-in returned error %d (%s) when asked to backup the metadata for container %s.\n"),
						    con->plugin->short_name, rc, evms_strerror(rc), con->name);
				goto error_out;
			}
		}
	}

	/*
	 * Save dummy entries for the parents of top objects.
	 * They need an entry in the metadata database in order to save
	 * the stop data that is put on the object.
	 */
	rc = engine_get_object_list(DISK | SEGMENT | REGION | EVMS_OBJECT,
					DATA_TYPE,
					NULL,
					NULL,
					TOPMOST,
					&top_objects);

	if (rc != 0) {
		LOG_SERIOUS("Error getting a list of top objects.  Return code is %d: %s\n",
			    rc, evms_strerror(rc));
		goto error_out;
	}

	if (stop_data.crc == 0) {
		feature_header_cpu_to_disk(&stop_data);
		stop_data.crc = CPU_TO_DISK32(evms_calculate_crc(EVMS_INITIAL_CRC, &stop_data, sizeof(evms_feature_header_t)));
	}

	LIST_FOR_EACH(top_objects, iter, obj) {
		if (obj->object_type == EVMS_OBJECT) {
			char * ancestors[8];

			rc = backup_feature_headers(obj, ancestors, 0);
			if (rc != 0) {
				goto error_out;
			}

		} else {
			rc = engine_save_metadata(NULL,
						  obj->name,
						  obj->size - (EVMS_FEATURE_HEADER_SECTORS * 2),
						  EVMS_FEATURE_HEADER_SECTORS,
						  &stop_data);
			if (rc != 0) {
				goto error_out;
			}

			rc = engine_save_metadata(NULL,
						  obj->name,
						  obj->size - EVMS_FEATURE_HEADER_SECTORS,
						  EVMS_FEATURE_HEADER_SECTORS,
						  &stop_data);
			if (rc != 0) {
				goto error_out;
			}
		}
	}

	destroy_list(top_objects);
	top_objects = NULL;

	LIST_FOR_EACH(&volumes_list, iter, vol) {

		if (vol->flags & VOLFLAG_HAS_OWN_DEVICE) {
			/*
			 * Entries for EVMS volumes that have their own
			 * device contain the feature headers that go on
			 * the object.
			 */
			rc = backup_fh(vol->name, get_working_top_object(vol->object));

		} else {
			rc = engine_save_metadata(vol->name,
						  vol->object->name,
						  0,
						  0,
						  NULL);

			if ((rc == 0) && !(vol->flags & VOLFLAG_COMPATIBILITY)){
				char * ancestors[8];

				rc = backup_feature_headers(vol->object, ancestors, 0);
			}
		}

		if (rc != 0) {
			break;
		}

		/*
		 * Save a dummy entry for a "parent" of the volume.
		 * The dummy parent is used for terminating a parent walk up the
		 * stack.  It is also used as a starting point for a recursive
		 * "restore all children" walk down the stack.
		 */
		rc = engine_save_metadata(NULL,
					  vol->name,
					  0,
					  0,
					  NULL);
		if (rc != 0){
			break;
		}
	}

error_out:
	close(metadata_db_fd);
	metadata_db_fd = 0;
	if (rc != 0) {
		unlink(metadata_db_name);
	}

	LOG_PROC_EXIT_INT(rc);
	return rc;
}


int evms_engine_read(object_handle_t handle,
		     lsn_t           lsn,
		     sector_count_t  length,
		     void          * buffer) {

	int rc = 0;
	object_type_t type;
	storage_object_t * obj;

	LOG_PROC_ENTRY();

	rc = check_engine_read_access();
	if (rc != 0) {
		LOG_PROC_EXIT_INT(rc);
		return rc;
	}

	if (!local_focus) {
		rc = remote_engine_read(handle, lsn, length, buffer);
		LOG_PROC_EXIT_INT(rc);
		return rc;
	}

	rc = translate_handle(handle, &obj, &type);
	if (rc == 0) {
		switch (type) {
			case DISK:
			case SEGMENT:
			case REGION:
			case EVMS_OBJECT:
				rc = obj->plugin->functions.plugin->read(obj, lsn, length, buffer);
				break;

			case CONTAINER:
				LOG_ERROR("Cannot read from container %s.\n", ((storage_container_t *) obj)->name);
				rc = EINVAL;
				break;

			case VOLUME:
				LOG_ERROR("Cannot read from container %s.\n", ((logical_volume_t *) obj)->name);
				rc = EINVAL;
				break;

			default:
				LOG_ERROR("I don't know how to read from a thing of type %d (%#x).\n", type, type);
				rc = EINVAL;
				break;
		}
	}

	LOG_PROC_EXIT_INT(rc);
	return rc;
}


int evms_engine_write(object_handle_t handle,
		      lsn_t           lsn,
		      sector_count_t  length,
                      void          * buffer) {

	int rc = 0;
	object_type_t type;
	storage_object_t * obj;

	LOG_PROC_ENTRY();

	rc = check_engine_write_access();
	if (rc != 0) {
		LOG_PROC_EXIT_INT(rc);
		return rc;
	}

	if (!local_focus) {
		rc = remote_engine_write(handle, lsn, length, buffer);
		LOG_PROC_EXIT_INT(rc);
		return rc;
	}

	rc = translate_handle(handle, &obj, &type);
	if (rc == 0) {
		switch (type) {
			case DISK:
			case SEGMENT:
			case REGION:
			case EVMS_OBJECT:
				rc = obj->plugin->functions.plugin->write(obj, lsn, length, buffer);
				break;

			case CONTAINER:
				LOG_ERROR("Cannot write to container %s.\n", ((storage_container_t *) obj)->name);
				rc = EINVAL;
				break;

			case VOLUME:
				LOG_ERROR("Cannot write to container %s.\n", ((logical_volume_t *) obj)->name);
				rc = EINVAL;
				break;

			default:
				LOG_ERROR("I don't know how to write to a thing of type %d (%#x).\n", type, type);
				rc = EINVAL;
				break;
		}
	}

	LOG_PROC_EXIT_INT(rc);
	return rc;
}