/*
 *
 *   (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: discover.c
 *
 */

#include <stdlib.h>
#include <stdio.h>
#include <string.h>
#include <errno.h>
#include <dirent.h>
#include <unistd.h>
#include <sys/stat.h>
#include <sys/time.h>

#include "fullengine.h"
#include "discover.h"
#include "engine.h"
#include "lists.h"
#include "internalAPI.h"
#include "crc.h"
#include "volume.h"
#include "commit.h"
#include "common.h"
#include "message.h"
#include "dm.h"
#include "memman.h"
#include "cluster.h"
#include "dmonmsg.h"
#include <ldm_funcs.h>
#include "activation.h"


boolean discover_in_progress = FALSE;

static int validate_feature_header(evms_feature_header_t * fh);

/******************************************************************************
 *
 * General purpose utility functions
 *
 ******************************************************************************/


/*
 * This function scans a list, removing any storage_object_t that has its
 * SOFLAG_CORRUPT set.
 */
void remove_corrupt_objects(list_anchor_t list) {

	list_element_t iter1;
	list_element_t iter2;
	storage_object_t * obj;

	LOG_PROC_ENTRY();

	LIST_FOR_EACH_SAFE(list, iter1, iter2, obj) {
		if (obj->flags & SOFLAG_CORRUPT) {
			delete_element(iter1);
		}
	}

	LOG_PROC_EXIT_VOID();
	return;
}


static void remove_unclaimed_objects(list_anchor_t list,
				     list_anchor_t unclaimed_object_list) {

	list_element_t iter1;
	list_element_t iter2;
	storage_object_t * obj;

	LOG_PROC_ENTRY();

	LIST_FOR_EACH_SAFE(list, iter1, iter2, obj) {
		LOG_DEBUG("Examining object %s.\n", obj->name);

		if (obj->flags & SOFLAG_NOT_CLAIMED) {
			list_element_t el;

			obj->flags &= ~SOFLAG_NOT_CLAIMED;

			/* Put the unclaimed object on the unclaimed_object_list. */
			LOG_DEBUG("Put object %s on the unclaimed list.\n", obj->name);
			el = insert_thing(unclaimed_object_list, obj, INSERT_AFTER, NULL);

			if (el == NULL) {
				LOG_CRITICAL("Error when putting object %s on the unclaimed object list.\n", obj->name);
			}

			delete_element(iter1);

		} else {
			LOG_DEBUG("Object %s is not marked not claimed.\n", obj->name);
		}
	}

	LOG_PROC_EXIT_VOID();
	return;
}


/*
 * Check if a dev node exists for a volume and that is has the correct minor
 * number.
 * Returns:
 * 0        dev node exists with the correct major:minor
 * ENOENT   dev node does not exist
 * EEXIST   dev node exists but has the wrong major:minor
 * other    some unexpected error code from stat()
 */
int hasa_dev_node(char * name, u_int32_t major, u_int32_t minor) {

	int rc = 0;
	struct stat statbuf;

	LOG_PROC_ENTRY();

	rc = stat(name, &statbuf);

	if (rc == 0) {

		/* dev node exists. Check if it has the right minor number. */
		dev_t devt = makedev(major, minor);

		if (statbuf.st_rdev != devt) {

			/* dev node exists but it has the wrong minor number. */
			rc = EEXIST;
		}

	} else {
		/* stat() failed.  Return the error code. */
		rc = errno;
	}

	LOG_PROC_EXIT_INT(rc);
	return rc;
}


int ensure_dev_node(char * name, u_int32_t major, u_int32_t minor) {

	int rc = 0;
	dev_t devt = makedev(major, minor);
	char dir_name[EVMS_OBJECT_NODE_PATH_LEN + EVMS_NAME_SIZE + 1];

	LOG_PROC_ENTRY();

	/* Make sure major:minor is valid. */
	if (major != 0) {
		switch (hasa_dev_node(name, major, minor)) {
			case 0:
				/* Dev node exists and has the correct major:minor. */
				break;

			case EEXIST:
				/* Dev node exists but has the wrong major:minor.  Fix it. */
				LOG_DEBUG("The device node \"%s\" has the wrong major:minor.  I'm fixing it to %d:%d.\n", name, major, minor);

				rc = unlink(name);
				if (rc == 0) {
					LOG_DEBUG("Make dev node for \"%s\".\n", name);
					rc = mknod(name, S_IFBLK | S_IRUSR | S_IWUSR | S_IRGRP | S_IWGRP, devt);

					if (rc != 0) {
						LOG_WARNING("mknod(%s) failed with error code %d.\n", name, rc);
					}
				} else {
					rc = errno;
					LOG_WARNING("unlink(\"%s\") failed with error code %d: %s\n", name, rc, strerror(rc));
				}
				break;

			case ENOENT:
				LOG_DEBUG("Device node \"%s\" does not exist.", name);

				strcpy(dir_name, name);
				*(strrchr(dir_name, '/')) = '\0';

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

				if (rc == 0) {
					LOG_DEBUG("Make dev node for \"%s\".\n", name);
					rc = mknod(name, S_IFBLK | S_IRUSR | S_IWUSR | S_IRGRP | S_IWGRP, devt);

					if (rc != 0) {
						LOG_WARNING("mknod(%s) failed with error code %d.\n", name, rc);
					}


				} else {
					LOG_WARNING("Error code %d when making directory %s.\n", rc, dir_name);
				}
				break;

			default:
				LOG_WARNING("hasa_dev_node() for \"%s\", major %d, minor %d failed with unexpected error %d: %s.\n", name, major, minor, rc, strerror(rc));
		}

	} else {
		LOG_DEBUG("Major:minor %d:%d is not valid.\n", major, minor);
		rc = EINVAL;
	}

	if (rc == 0) {
		LOG_DEBUG("Device node %s is for major %d, minor %d.\n", name, major, minor);
	}

	LOG_PROC_EXIT_INT(rc);
	return rc;
}


int make_object_dev_node(storage_object_t * obj) {

	int rc = 0;

	LOG_PROC_ENTRY();

	if (obj->flags & SOFLAG_ACTIVE) {
		char dev_name[EVMS_OBJECT_NODE_PATH_LEN + EVMS_NAME_SIZE + 1];

		strcpy(dev_name, EVMS_OBJECT_NODE_PATH);

		if (current_nodeid != NULL) {
			if ((memcmp(current_nodeid, &no_nodeid, sizeof(ece_nodeid_t)) != 0) &&
			    (current_nodeid != my_nodeid)) {
				/*
				 * This is a remote object on shared storage which
				 * we will access locally.  Add the node's name to the
				 * device name.
				 */
				strcat(dev_name, nodeid_to_string(current_nodeid));
				strcat(dev_name, "/");
			}
		}
		strcat(dev_name, obj->name);
		rc = ensure_dev_node(dev_name, obj->dev_major, obj->dev_minor);
	}

	LOG_PROC_EXIT_INT(rc);
	return rc;
}


static void propigate_cluster_info_to_container(storage_container_t * con) {

	storage_object_t * child_obj = first_thing(con->objects_consumed, NULL);

	if (child_obj->disk_group != NULL) {
		con->disk_group = child_obj->disk_group;

		if (child_obj->flags & SOFLAG_CLUSTER_PRIVATE) {
			con->flags |= SCFLAG_CLUSTER_PRIVATE;
		}
		if (child_obj->flags & SOFLAG_CLUSTER_SHARED) {
			con->flags |= SCFLAG_CLUSTER_SHARED;
		}
		if (child_obj->flags & SOFLAG_CLUSTER_DEPORTED) {
			con->flags |= SCFLAG_CLUSTER_DEPORTED;
		}
	}
}


void propigate_cluster_info(list_anchor_t objects) {

	list_element_t iter;
	storage_object_t * obj;

	LIST_FOR_EACH(objects, iter, obj) {

		/* Disks don't have cluster info. */
		if (obj->object_type != DISK) {

			if (obj->producing_container != NULL) {
				storage_container_t * con = obj->producing_container;

				propigate_cluster_info_to_container(obj->producing_container);

				obj->disk_group = con->disk_group;

				if (con->flags & SCFLAG_CLUSTER_PRIVATE) {
					obj->flags |= SOFLAG_CLUSTER_PRIVATE;
				}
				if (con->flags & SCFLAG_CLUSTER_SHARED) {
					obj->flags |= SOFLAG_CLUSTER_SHARED;
				}
				if (con->flags & SCFLAG_CLUSTER_DEPORTED) {
					obj->flags |= SOFLAG_CLUSTER_DEPORTED;
				}

			} else {
				storage_object_t * child_obj = first_thing(obj->child_objects, NULL);

				if (child_obj != NULL) {
					obj->disk_group = child_obj->disk_group;
					obj->flags |= (child_obj->flags & (SOFLAG_CLUSTER_PRIVATE | SOFLAG_CLUSTER_SHARED | SOFLAG_CLUSTER_DEPORTED));
				}
			}
		}
	}
}


/*
 * Return a string with a number of periods based on *status_count.
 * Increment *status_count.
 */
char * status_dots(int * status_count) {

	char * ptr;

	switch (*status_count % 4) {
		case 0:
			ptr = "...";
			break;
		case 1:
			ptr = "   ";
			break;
		case 2:
			ptr = ".  ";
			break;
		case 3:
			ptr = ".. ";
			break;
		default:
			ptr = "";
	}

	(*status_count)++;

	return ptr;
}


static int discover_replace_objects(list_anchor_t object_list) {

	int rc = 0;
	STATIC_LIST_DECL(result_object_list);

	LOG_PROC_ENTRY();

	if (replace_plugin != NULL) {
		rc = replace_plugin->functions.plugin->discover(object_list, &result_object_list, FALSE);
		delete_all_elements(object_list);
		merge_lists(object_list, &result_object_list, NULL, NULL);
	}

	LOG_PROC_EXIT_INT(rc);
	return rc;
}


/******************************************************************************
 *
 * Functions for discovering logical disks
 *
 ******************************************************************************/

static int discover_logical_disks(list_anchor_t object_list, boolean report_status) {

	int rc = 0;
	int status_count = 0;
	list_element_t iter;
	plugin_record_t * pPlugRec = NULL;
	STATIC_LIST_DECL(result_object_list);

	LOG_PROC_ENTRY();

	/*
	 * Call all of the device manager plug-ins to have them discover their
	 * disks.
	 */
	LIST_FOR_EACH(&plugins_list, iter, pPlugRec) {

		if (GetPluginType(pPlugRec->id) == EVMS_DEVICE_MANAGER) {
			list_element_t iter;
			storage_object_t * obj;

			if (report_status) {
				status_message(_("Discovering disks%s\n"), status_dots(&status_count));
			}

			LOG_DEBUG("Calling discover() in %s.\n", pPlugRec->short_name);
			rc = pPlugRec->functions.plugin->discover(object_list, &result_object_list, FALSE);
			LOG_DEBUG("Return code from discover() is %d: %s.\n", rc, evms_strerror(rc));

			/* Remove any corrupt objects from the list. */
			remove_corrupt_objects(&result_object_list);

			/* Make sure we have a dev node for each object. */
			LIST_FOR_EACH(&result_object_list, iter, obj) {
				if (obj->flags & SOFLAG_ACTIVE) {
					make_object_dev_node(obj);
				}
			}

			/* Remove any remaining objects from the source list. */
			delete_all_elements(object_list);

			/*
			 * Put the objects from the result list onto the source
			 * list.
			 */
			merge_lists(object_list, &result_object_list, NULL, NULL);

			discover_replace_objects(object_list);
		}
	}

	LOG_PROC_EXIT_INT(rc);
	return rc;
}


/******************************************************************************
 *
 * Functions for discovering disk segments
 *
 ******************************************************************************/

static int discover_segments(list_anchor_t object_list, boolean report_status) {

	int rc = 0;
	int status_count = 0;
	plugin_record_t * pPlugRec = NULL;
	STATIC_LIST_DECL(result_object_list);
	STATIC_LIST_DECL(unclaimed_object_list);
	boolean finished = FALSE;
	boolean last_pass = FALSE;

	LOG_PROC_ENTRY();

	while ((rc == 0) && (!finished || last_pass)) {
		list_element_t iter1;
		storage_object_t * obj;

		/*
		 * Mark all the objects in the list as unclaimed so that they
		 * can be removed at the end of the loop if no plug-in claims
		 * them.
		 */
		LIST_FOR_EACH(object_list, iter1, obj) {
			obj->flags |= SOFLAG_NOT_CLAIMED;
		}

		/* Assume we will finish discovery on this loop. */
		finished = TRUE;

		LIST_FOR_EACH(&plugins_list, iter1, pPlugRec) {

			if (GetPluginType(pPlugRec->id) == EVMS_SEGMENT_MANAGER) {
				list_element_t iter2;

				if (report_status) {
					status_message(_("Discovering segments%s\n"), status_dots(&status_count));
				}

				LOG_DEBUG("Calling discover() in %s.  last_pass is %s.\n", pPlugRec->short_name, (last_pass) ? "TRUE" : "FALSE");
				rc = pPlugRec->functions.plugin->discover(object_list, &result_object_list, last_pass);
				LOG_DEBUG("discover() returned %d objects.\n", rc);

				/* Remove any corrupt objects from the list. */
				remove_corrupt_objects(&result_object_list);

				/* Make sure we have a dev node for each object. */
				LIST_FOR_EACH(&result_object_list, iter2, obj) {
					if (obj->flags & SOFLAG_ACTIVE) {
						make_object_dev_node(obj);
					}
				}

				propigate_cluster_info(&result_object_list);

				/*
				 * Segment managers will return a positive number (the
				 * number of objects discovered) if they find any
				 * segments.  If rc is positive, turn off the finished
				 * flag so that we make another pass at discovery
				 * through the segment managers.  Segment managers
				 * return a 0 if they discovered nothing and had no
				 * errors.  They return a negative error code if
				 * something went wrong.  We ignore error codes to let
				 * the discovery continue.
				 */
				if (!last_pass) {
					if (rc > 0) {
						finished = FALSE;
					}
				}

				rc = 0;

				/*
				 * Remove any remaining objects from the source list.
				 */
				delete_all_elements(object_list);

				/*
				 * Put the objects from the result list onto the source
				 * list.
				 */
				merge_lists(object_list, &result_object_list, NULL, NULL);

				discover_replace_objects(object_list);
			}
		}

		/*
		 * Remove any objects that have not been claimed and put them on the
		 * unclaimed_object_list.  These objects will never be discovered in
		 * subsequent passes of discovery.  Therefore we set them aside so
		 * that all the plug-ins don't bother reexamining them on every pass
		 * of discovery.
		 */
		remove_unclaimed_objects(object_list, &unclaimed_object_list);

		if (finished && !last_pass) {
			last_pass = TRUE;
		} else {
			last_pass = FALSE;
		}
	}

	/*
	 * Put the unclaimed objects back on the object_list so that they
	 * can be examined by the next layer of discovery.
	 */
	merge_lists(object_list, &unclaimed_object_list, NULL, NULL);

	LOG_PROC_EXIT_INT(rc);
	return rc;
}


/******************************************************************************
 *
 * Functions for discovering regions
 *
 ******************************************************************************/

static int discover_regions(list_anchor_t object_list, boolean report_status) {

	int rc = 0;
	int status_count = 0;
	plugin_record_t * pPlugRec = NULL;
	STATIC_LIST_DECL(result_object_list);
	STATIC_LIST_DECL(unclaimed_object_list);
	boolean finished = FALSE;
	boolean last_pass = FALSE;

	LOG_PROC_ENTRY();

	while ((rc == 0) && (!finished || last_pass)) {
		list_element_t iter1;
		storage_object_t * obj;

		/*
		 * Mark all the objects in the list as unclaimed so that they
		 * can be removed at the end of the loop if no plug-in claims
		 * them.
		 */
		LIST_FOR_EACH(object_list, iter1, obj) {
			obj->flags |= SOFLAG_NOT_CLAIMED;
		}

		/* Assume we will finish discovery on this loop. */
		finished = TRUE;

		LIST_FOR_EACH(&plugins_list, iter1, pPlugRec) {

			if (GetPluginType(pPlugRec->id) == EVMS_REGION_MANAGER) {
				list_element_t iter2;

				if (report_status) {
					status_message(_("Discovering regions%s\n"), status_dots(&status_count));
				}

				LOG_DEBUG("Calling discover() in %s.  last_pass is %s.\n", pPlugRec->short_name, (last_pass) ? "TRUE" : "FALSE");
				rc = pPlugRec->functions.plugin->discover(object_list, &result_object_list, last_pass);
				LOG_DEBUG("discover() returned %d objects.\n", rc);

				/* Remove any corrupt objects from the list. */
				remove_corrupt_objects(&result_object_list);

				/* Make sure we have a dev node for each object. */
				LIST_FOR_EACH(&result_object_list, iter2, obj) {
					if (obj->flags & SOFLAG_ACTIVE) {
						make_object_dev_node(obj);
					}
				}

				propigate_cluster_info(&result_object_list);

				/*
				 * Region managers will return a positive number (the
				 * number of objects discovered) if they find any
				 * regions.  If rc is positive, turn off the finished
				 * flag so that we make another pass at discovery
				 * through the region managers.  Region managers return
				 * a 0 if they discovered nothing and had no errors.
				 * They return a negative error code if something went
				 * wrong.  We ignore error codes to let the discovery
				 * continue.
				 */
				if (!last_pass) {
					if (rc > 0) {
						finished = FALSE;
					}
				}

				rc = 0;

				/*
				 * Remove any remaining objects from the source list.
				 */
				delete_all_elements(object_list);

				/*
				 * Put the objects from the result list onto the source
				 * list.  AppendList deletes them from the
				 * result_object_list.
				 */
				merge_lists(object_list, &result_object_list, NULL, NULL);

				discover_replace_objects(object_list);
			}
		}

		/*
		 * Remove any objects that have not been claimed and put them on the
		 * unclaimed_object_list.  These objects will never be discovered in
		 * subsequent passes of discovery.  Therefore we set them aside so
		 * that all the plug-ins don't bother reexamining them on every pass
		 * of discovery.
		 */
		remove_unclaimed_objects(object_list, &unclaimed_object_list);

		if (finished && !last_pass) {
			last_pass = TRUE;
		} else {
			last_pass = FALSE;
		}
	}

	/*
	 * Put the unclaimed objects back on the object_list so that they
	 * can be examined by the next layer of discovery.
	 */
	merge_lists(object_list, &unclaimed_object_list, NULL, NULL);

	LOG_PROC_EXIT_INT(rc);
	return rc;
}


/******************************************************************************
 *
 * Support functions for making volumes
 *
 ******************************************************************************/

/*
 * Set the volume pointer in an object.  Set the volume pointer in its child
 * objects.
 */
void set_volume_in_object(storage_object_t * obj,
			  logical_volume_t  * vol) {

	LOG_PROC_ENTRY();

	/*
	 * This function can be called for a child of any object.
	 */
	switch (obj->object_type) {
		case DISK:
		case SEGMENT:
		case REGION:
		case EVMS_OBJECT:
			{
				LOG_DEBUG("Set volume %s in object %s.\n", (vol != NULL) ? vol->name : "(none)", obj->name);

				if (!discover_in_progress) {
					/* If the volume for this object is changing and it has a
					 * feature header, mark the feature header dirty so that
					 * it will be rewritten with the new volume data.
					 */
					if (obj->volume != vol) {
						if (obj->feature_header != NULL) {
							obj->flags |= SOFLAG_FEATURE_HEADER_DIRTY;
						}
					}
				}

				obj->volume = vol;

				/*
				 * Clear the volume pointer in a disk object if it has multiple
				 * parents or its single parent is a segment manager (which
				 * could potentially make more parents).
				 */
				if (obj->object_type == DISK) {
					uint parent_count = list_count(obj->parent_objects);

					if (parent_count > 1) {
						obj->volume = NULL;

					} else {
						if (parent_count == 1) {
							storage_object_t * parent_object;

							parent_object = first_thing(obj->parent_objects, NULL);

							if (GetPluginType(parent_object->plugin->id) == EVMS_SEGMENT_MANAGER) {
								obj->volume = NULL;
							}
						}
					}
				}

				/*
				 * If this is not discovery time (this function can be called on
				 * a create or revert of a volume) notify the feature of the
				 * object's change in volume status.
				 */
				if (!discover_in_progress) {
					obj->plugin->functions.plugin->set_volume(obj, (vol != NULL));
				}

				/*
				 * If the object was not produced by a container, then call
				 * this function recursively to set the volume pointers in
				 * the child objects.  If the object was produced by a
				 * container then its child objects may not belong exclusively
				 * to this volume, so we can't set the volume pointer in the
				 * child objects; the recursion stops.
				 */
				if (obj->producing_container == NULL) {
					list_element_t iter;
					storage_object_t * child;

					LIST_FOR_EACH(obj->child_objects, iter, child) {
						set_volume_in_object(child, vol);
					}
				}
			}
			break;

		default:
			/*Ignore any other kind of objects. */
			break;
	}

	LOG_PROC_EXIT_VOID();
	return;
}

/*
 * If a volume has an FSIM, get the file system size and limits.
 * Else set default values.
 */
int get_volume_sizes_and_limits(logical_volume_t * vol) {

	int rc = 0;
	plugin_record_t * fsim = vol->file_system_manager;

	LOG_PROC_ENTRY();

	if (fsim == NULL) {
		/* Volume has no FSIM.  Set default sizes and limits. */
		vol->fs_size      = round_down_to_hard_sector(vol->vol_size, vol->object);
		vol->min_fs_size  = 0;
		vol->max_fs_size  = round_down_to_hard_sector(-1, vol->object);
		vol->max_vol_size = round_down_to_hard_sector(-1, vol->object);

	} else {
		/*
		 * If the FSIM has changed, don't call the new FSIM for the sizes and
		 * limits.
		 */
		if (fsim == vol->original_fsim) {
			int rc2;

			rc = fsim->functions.fsim->get_fs_size(vol, &vol->fs_size);

			if (rc != 0) {
				LOG_WARNING("FSIM %s returned error code %d from call to get_fs_size() for volume %s.\n", fsim->short_name, rc, vol->name);
			}

			rc2 = fsim->functions.fsim->get_fs_limits(vol, &vol->min_fs_size,
								  &vol->max_fs_size,
								  &vol->max_vol_size);

			if (rc2 != 0) {
				LOG_WARNING("FSIM %s returned error code %d from call to get_fs_limits() for volume %s.\n", fsim->short_name, rc, vol->name);
			}

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

			/* Make sure sizes are multiples of the hard sector size. */
			vol->fs_size      = round_down_to_hard_sector(vol->fs_size, vol->object);
			vol->min_fs_size  = round_up_to_hard_sector(vol->min_fs_size, vol->object);
			vol->max_fs_size  = round_down_to_hard_sector(vol->max_fs_size, vol->object);
			vol->max_vol_size = round_down_to_hard_sector(vol->max_vol_size, vol->object);

		} else {
			if (vol->original_fsim == NULL) {
				LOG_DEBUG("Cannot get sizes and limits for volume %s because it does not have a file system on it right now.\n", vol->name);
			} else {
				LOG_DEBUG("Cannot get sizes and limits for volume %s because it is being converted from %s to %s.\n", vol->name, vol->original_fsim->short_name, vol->file_system_manager->short_name);
			}
		}
	}

	LOG_PROC_EXIT_INT(rc);
	return rc;
}


/*
 * Given a volume, see if there is an FSIM that will claim that it manages the
 * volume.
 */
void find_fsim_for_volume(logical_volume_t * volume) {

	list_element_t iter;
	plugin_record_t * pPlugRec = NULL;

	LOG_PROC_ENTRY();

	volume->file_system_manager = NULL;

	LIST_FOR_EACH(&plugins_list, iter, pPlugRec) {
		if (GetPluginType(pPlugRec->id) == EVMS_FILESYSTEM_INTERFACE_MODULE) {

			if (pPlugRec->functions.fsim->probe(volume) == 0) {
				LOG_DEBUG("Volume %s belongs to %s.\n", volume->name, pPlugRec->short_name);
				volume->file_system_manager = pPlugRec;
				volume->original_fsim = volume->file_system_manager;
				volume->original_fsim_private_data = volume->private_data;
				break;
			}
		}
	}

	get_volume_sizes_and_limits(volume);

	LOG_PROC_EXIT_VOID();
}


int make_volume_dev_node(logical_volume_t * vol) {

	int rc = 0;

	LOG_PROC_ENTRY();

	/*
	 * Make sure the volume has a valid major:minor before we
	 * make a dev node for it.
	 */
	if (vol->dev_major != 0) {
		rc = ensure_dev_node(vol->name, vol->dev_major, vol->dev_minor);

		if (rc == 0) {
			memcpy(vol->dev_node, vol->name, sizeof(vol->name));

		} else {
			memset(vol->dev_node, '\0', sizeof(vol->dev_node));
		}

	} else {
		rc = EINVAL;
	}

	LOG_PROC_EXIT_INT(rc);
	return rc;
}


static int get_volume_dev_node_and_fsim(logical_volume_t * vol) {

	int rc = 0;

	LOG_PROC_ENTRY();

	/* Safety check */
	if (vol == NULL) {
		LOG_WARNING("Volume pointer is NULL.\n");
		LOG_PROC_EXIT_INT(EINVAL);
		return EINVAL;
	}

	if (vol->dev_major != 0) {
		if (make_volume_dev_node(vol) == 0) {

			if (vol->object->flags & SOFLAG_ACTIVE) {
				vol->flags |= VOLFLAG_ACTIVE;

				find_fsim_for_volume(vol);

				if (is_volume_mounted(vol)) {
					LOG_DEBUG("Volume \"%s\" is mounted on %s.\n", vol->name, vol->mount_point);
				}
			}
		}
	}

	LOG_PROC_EXIT_INT(rc);
	return rc;
}


/******************************************************************************
 *
 * Functions for discovering EVMS objects
 *
 ******************************************************************************/

/*
 * A structure for grouping EVMS objects by volume.
 */
typedef struct volume_objects {
	u_int64_t               volume_serial;
	list_anchor_t           objects;
	struct volume_objects * next;
	char                    volume_name[EVMS_VOLUME_NAME_SIZE];
	u_int32_t               flags;
} volume_objects_t;


/*
 * For each object in list, if the object is managed by the plug-in specified
 * in the parameters, then remove it from the list and put it into the
 * plugin_object_list.
 */
static int extract_plugin_objects(list_anchor_t list,
				  plugin_id_t   plugin_id,
				  list_anchor_t plugin_object_list) {

	int rc = 0;
	list_element_t iter1;
	list_element_t iter2;
	storage_object_t * obj;

	LOG_PROC_ENTRY();

	LIST_FOR_EACH_SAFE(list, iter1, iter2, obj) {

		/*
		 * If this object has the same plug-in ID as the one we're
		 * looking for, then add it to the list specified in the parms
		 * and remove it from the main list that is being traversed.
		 */
		if (obj->feature_header->feature_id == plugin_id) {
			list_element_t el;

			LOG_DEBUG("Extract object %s.\n", obj->name);
			remove_element(iter1);
			el = insert_element(plugin_object_list,
					    iter1,
					    INSERT_AFTER,
					    NULL);

			if (el == NULL) {
				LOG_CRITICAL("Error inserting object %s into the plugin object list.\n", obj->name);
				rc = ENOMEM;
			}
		}
	}

	LOG_PROC_EXIT_INT(rc);
	return rc;
}


/*
 * Convert a feature header in disk endian (little endian) format to the CPU
 * endian format.
 */
void inline feature_header_disk_to_cpu(evms_feature_header_t * fh) {

	fh->signature                 = DISK_TO_CPU32(fh->signature);
	fh->version.major             = DISK_TO_CPU32(fh->version.major);
	fh->version.minor             = DISK_TO_CPU32(fh->version.minor);
	fh->version.patchlevel        = DISK_TO_CPU32(fh->version.patchlevel);
	fh->engine_version.major      = DISK_TO_CPU32(fh->engine_version.major);
	fh->engine_version.minor      = DISK_TO_CPU32(fh->engine_version.minor);
	fh->engine_version.patchlevel = DISK_TO_CPU32(fh->engine_version.patchlevel);
	fh->flags                     = DISK_TO_CPU32(fh->flags);
	fh->feature_id                = DISK_TO_CPU32(fh->feature_id);
	fh->sequence_number           = DISK_TO_CPU64(fh->sequence_number);
	fh->alignment_padding         = DISK_TO_CPU64(fh->alignment_padding);
	fh->feature_data1_start_lsn   = DISK_TO_CPU64(fh->feature_data1_start_lsn);
	fh->feature_data1_size        = DISK_TO_CPU64(fh->feature_data1_size);
	fh->feature_data2_start_lsn   = DISK_TO_CPU64(fh->feature_data2_start_lsn);
	fh->feature_data2_size        = DISK_TO_CPU64(fh->feature_data2_size);
	fh->volume_serial_number      = DISK_TO_CPU64(fh->volume_serial_number);
	fh->volume_system_id          = DISK_TO_CPU32(fh->volume_system_id);
	fh->object_depth              = DISK_TO_CPU32(fh->object_depth);
}


/*
 * Check for a valid feature header -- has a signature and the CRC is correct.
 *
 * Returns:
 * ENOENT    if no signature
 * EINVAL    if has signature but CRC is bad
 *           or version is bad
 */
static int validate_feature_header(evms_feature_header_t * fh) {

	int rc = 0;

	LOG_PROC_ENTRY();

	if (DISK_TO_CPU32(fh->signature) == EVMS_FEATURE_HEADER_SIGNATURE) {
		u_int32_t old_crc;
		u_int32_t new_crc;

		old_crc = DISK_TO_CPU32(fh->crc);
		fh->crc = 0;
		new_crc = evms_calculate_crc(EVMS_INITIAL_CRC, fh, sizeof(evms_feature_header_t));

		if ((new_crc == old_crc) || (old_crc == EVMS_MAGIC_CRC)) {

			/* Convert the feature header to CPU endian format. */
			feature_header_disk_to_cpu(fh);

			/*
			 * Make sure the version of the feature header is one that
			 * we can understand.
			 */
			if (fh->version.major == EVMS_FEATURE_HEADER_MAJOR) {
				if (fh->version.minor <= EVMS_FEATURE_HEADER_MINOR) {
					if (fh->version.minor == EVMS_FEATURE_HEADER_MINOR) {
						if (fh->version.patchlevel <= EVMS_FEATURE_HEADER_PATCHLEVEL) {
							/* Feature header looks OK. */

						} else {
							LOG_WARNING("Feature header is version %d.%d.%d.  The Engine handles version %d.%d.%d or compatible.\n", fh->version.major, fh->version.minor, fh->version.patchlevel, EVMS_FEATURE_HEADER_MAJOR, EVMS_FEATURE_HEADER_MINOR, EVMS_FEATURE_HEADER_PATCHLEVEL);
							rc = EINVAL;
						}
					}

				} else {
					LOG_WARNING("Feature header is version %d.%d.%d.  The Engine handles version %d.%d.%d or compatible.\n", fh->version.major, fh->version.minor, fh->version.patchlevel, EVMS_FEATURE_HEADER_MAJOR, EVMS_FEATURE_HEADER_MINOR, EVMS_FEATURE_HEADER_PATCHLEVEL);
					rc = EINVAL;
				}

			} else {
				LOG_WARNING("Feature header is version %d.%d.%d.  The Engine handles version %d.%d.%d or compatible.\n", fh->version.major, fh->version.minor, fh->version.patchlevel, EVMS_FEATURE_HEADER_MAJOR, EVMS_FEATURE_HEADER_MINOR, EVMS_FEATURE_HEADER_PATCHLEVEL);
				rc = EINVAL;
			}

		} else {
			LOG_DEBUG("Bad CRC. old(%#x) new(%#x)\n", old_crc, new_crc);
			rc = EINVAL;
		}

	} else {
		LOG_DEBUG("Sector does not have a feature header signature.\n");
		rc = ENOENT;
	}

	LOG_PROC_EXIT_INT(rc);
	return rc;
}


/*
 * Allocate a buffers and read in the primary and secondary feature headers.
 * Check for a valid feature header -- has a signature and the CRC is
 * correct.
 *
 * Returns:
 * non-0    something really bad happened, e.g, memory allocation failure.
 *
 * For the primary and secondary return codes and feature header pointers:
 *
 * *rc      *feature_header
 * ---      -----------------
 * 0        pointer to buffer   if good feature header
 * ENOENT   NULL                if no signature
 * EINVAL   NULL                if has signature but CRC is bad
 *                              or version is bad
 * error    NULL                if any other error occurs
 */
static int read_feature_header(storage_object_t        * object,
			       evms_feature_header_t * * primary_feature_header,
			       int                     * primary_rc,
			       evms_feature_header_t * * secondary_feature_header,
			       int                     * secondary_rc) {

	int rc = 0;
	evms_feature_header_t * primary_fh = NULL;
	evms_feature_header_t * secondary_fh = NULL;

	LOG_PROC_ENTRY();

	LOG_DEBUG("Look for feature headers on object %s.\n", object->name);

	secondary_fh = engine_alloc(sizeof(evms_feature_header_t) * 2);
	primary_fh = engine_alloc(sizeof(evms_feature_header_t));

	if ((primary_fh != NULL) &&
	    (secondary_fh != NULL)) {
		rc = object->plugin->functions.plugin->read(object,
							    object->size - (EVMS_FEATURE_HEADER_SECTORS * 2),
							    EVMS_FEATURE_HEADER_SECTORS * 2,
							    secondary_fh);
		if (rc == 0) {
			/*
			 * The sectors for both potential feature headers were read in
			 * successfully.  Copy the memory from the primary feature
			 * header location to the primary_fh buffer.
			 */

			memcpy(primary_fh, secondary_fh + 1, sizeof(evms_feature_header_t));

			/* Validate the feature headers. */
			*primary_rc = validate_feature_header(primary_fh);
			*secondary_rc = validate_feature_header(secondary_fh);

		} else {
			/*
			 * An error occurred trying to read both feature headers.
			 * Try to read them separately.
			 */

			*primary_rc = object->plugin->functions.plugin->read(object,
									     object->size - EVMS_FEATURE_HEADER_SECTORS,
									     EVMS_FEATURE_HEADER_SECTORS,
									     primary_fh);
			if (*primary_rc == 0) {
				/* Validate the primary feature header. */
				*primary_rc = validate_feature_header(primary_fh);
			}

			*secondary_rc = object->plugin->functions.plugin->read(object,
									       object->size - (EVMS_FEATURE_HEADER_SECTORS * 2),
									       EVMS_FEATURE_HEADER_SECTORS,
									       secondary_fh);
			if (*secondary_rc == 0) {
				/* Validate the secondary feature header. */
				*secondary_rc = validate_feature_header(secondary_fh);
			}
		}

		/*
		 * If either of the feature headers had and error on reading or on
		 * validation, free the feature header memory; it's useless.
		 */
		if (*primary_rc != 0) {
			engine_free(primary_fh);
			primary_fh = NULL;
		}
		if (*secondary_rc != 0) {
			engine_free(secondary_fh);
			secondary_fh = NULL;
		}

	} else {
		LOG_CRITICAL("Error allocating memory to read in a feature header.\n");
		if (primary_fh != NULL) {
			engine_free(primary_fh);
			primary_fh = NULL;
		}
		if (secondary_fh != NULL) {
			engine_free(secondary_fh);
			secondary_fh = NULL;
		}
		rc = ENOMEM;
	}

	*primary_feature_header   = primary_fh;
	*secondary_feature_header = secondary_fh;

	/* Any over all error code is returned as individual error codes as well. */
	if (rc != 0) {
		*primary_rc   = rc;
		*secondary_rc = rc;
	}

	LOG_PROC_EXIT_INT(rc);
	return rc;
}


/*
 * Read in the feature header at the top of the object, if there is one, and
 * put a pointer to it into the storage_object_t.
 */
static void get_feature_header(storage_object_t * obj) {

	int secondary_rc = 0;
	int primary_rc = 0;
	evms_feature_header_t * primary_fh = NULL;
	evms_feature_header_t * secondary_fh = NULL;

	LOG_PROC_ENTRY();

	LOG_DEBUG("Examining object %s for feature headers.\n", obj->name);

	/* If the object already has a feature header, then we're finished. */
	if (obj->feature_header == NULL) {

		/* Read in the feature headers at the back of the object. */
		read_feature_header(obj, &primary_fh, &primary_rc, &secondary_fh, &secondary_rc);

		switch (primary_rc) {
			case 0:
				LOG_DEBUG("Primary feature header read successfully.\n");

				switch (secondary_rc) {
					case 0:
						LOG_DEBUG("Secondary feature header read successfully.\n");

						/*
						 * Both feature headers were read in successfully.
						 * Check the sequence numbers.
						 */
						if (secondary_fh->sequence_number == primary_fh->sequence_number) {

							/*
							 * The sequence numbers match.  Life is good.  Put
							 * the primary copy (arbitrary choice) of the
							 * feature header into the storage object and free
							 * the secondary copy.
							 */
							LOG_DEBUG("Primary and secondary feature headers match.\n");
							obj->feature_header = primary_fh;
							engine_free(secondary_fh);

						} else {
							/*
							 * The sequence numbers don't match.  Use the
							 * feature header with the highest sequence number,
							 * throw out the other one, and mark the object's
							 * feature header dirty so that the feature headers
							 * will be rewritten to disk.
							 */
							if (primary_fh->sequence_number > secondary_fh->sequence_number) {
								LOG_WARNING("Primary feature header for object %s has a higher sequence number (%"PRIu64") than the secondary feature header (%"PRIu64").  Using the primary copy.\n",
									    primary_fh->object_name, primary_fh->sequence_number, secondary_fh->sequence_number);
								obj->feature_header = primary_fh;
								engine_free(secondary_fh);

							} else {
								LOG_WARNING("Secondary feature header for object %s has a higher sequence number (%"PRIu64") than the primary feature header (%"PRIu64").  Using the secondary copy.\n",
									    secondary_fh->object_name, secondary_fh->sequence_number, primary_fh->sequence_number);
								obj->feature_header = secondary_fh;
								engine_free(primary_fh);
							}

							obj->flags |= SOFLAG_FEATURE_HEADER_DIRTY;
						}
						break;

					case ENOENT:
					case EINVAL:
						/*
						 * The primary feature header was found, but the
						 * secondary feature header either had a bad signature
						 * or a bad CRC.  Keep the primary feature header.
						 * Assume that the secondary feature header is corrupt
						 * and needs to be rewritten.
						 */
						LOG_WARNING("Secondary feature header not found on object %s.  Using the primary copy.\n",
							    obj->name);
						obj->feature_header = primary_fh;
						obj->flags |= SOFLAG_FEATURE_HEADER_DIRTY;
						break;

					default:
						/*
						 * We got the primary feature header OK but some
						 * unexpected error occurred when reading the secondary
						 * feature header.  If we get an unexpected error we
						 * don't use any of the feature headers.
						 */
						engine_user_message(NULL, NULL,
								    _("Error code %d (%s) when reading the secondary copy of feature header on object %s.  Ignoring good primary feature header.\n"),
								    secondary_rc, strerror(secondary_rc), obj->name);
						engine_free(primary_fh);
						break;
				}
				break;

			case ENOENT:
			case EINVAL:
				switch (secondary_rc) {
					case 0:
						/*
						 * The primary feature header either had a bad
						 * signature or a bad CRC.  The secondary feature header
						 * was read successfully.  Keep the secondary feature
						 * header.  Assume the primary feature header is
						 * corrupt and needs to be rewritten.
						 */
						LOG_WARNING("Primary feature header not found on object %s.\n", obj->name);
						LOG_WARNING("Secondary feature header read successfully.  Using the secondary copy.\n");
						obj->feature_header = secondary_fh;
						obj->flags |= SOFLAG_FEATURE_HEADER_DIRTY;
						break;

					case ENOENT:
					case EINVAL:
						/*
						 * Both the primary and secondary feature headers either
						 * had a bad signature or a bad CRC.  Assume that there
						 * are no feature headers on this object.
						 */
						LOG_DEBUG("Primary feature header not found on object %s.\n", obj->name);
						LOG_DEBUG("Secondary feature header not found on object %s.\n", obj->name);
						break;

					default:
						/*
						 * The primary feature header either has a bad
						 * signature or the CRC is bad.  Regardless, we got an
						 * unexpected error code when reading the secondary
						 * feature header.  If we get an unexpected error we
						 * don't use any of the feature headers.
						 */
						LOG_WARNING("Primary feature header not found on object %s.\n", obj->name);
						engine_user_message(NULL, NULL,
								    _("Error code %d (%s) when reading the secondary copy of feature header on object %s.\n"),
								    secondary_rc, strerror(secondary_rc), obj->name);
						break;
				}
				break;

			default:
				/*
				 * We got an unexpected error when reading the primary feature
				 * header.  If we get an unexpected error we don't use any of
				 * the feature headers.
				 */
				engine_user_message(NULL, NULL,
						    _("Error code %d (%s) when reading the primary copy of feature header on object %s.\n"),
						    primary_rc, strerror(primary_rc), obj->name);

				switch (secondary_rc) {
					case 0:
						engine_user_message(NULL, NULL,
								    _("Ignoring good secondary feature header.\n"));
						engine_free(secondary_fh);
						break;

					case ENOENT:
					case EINVAL:
						LOG_WARNING("Secondary feature header not found.\n");
						break;

					default:
						/*
						 * We got an unexpected error when reading the
						 * secondary feature header.
						 */
						engine_user_message(NULL, NULL,
								    _("Error code %d (%s) when reading the secondary copy of feature header on object %s.\n"),
								    secondary_rc, strerror(secondary_rc), obj->name);
						break;
				}
				break;
		}
	}

	LOG_PROC_EXIT_VOID();
	return;
}


static void get_greatest_object_depth(storage_object_t * obj,
				      u_int32_t        * object_depth) {

	LOG_PROC_ENTRY();

	if (obj->feature_header != NULL) {
		if (obj->feature_header->object_depth > *object_depth) {
			*object_depth = obj->feature_header->object_depth;
		}
	}

	LOG_PROC_EXIT_VOID();
	return;
}


static int discover_objects_by_plugin(list_anchor_t object_list,
				      list_anchor_t resulting_object_list,
				      boolean report_status) {

	int rc = 0;
	int status_count = 0;
	STATIC_LIST_DECL(plugin_object_list);
	STATIC_LIST_DECL(new_object_list);
	list_element_t iter;
	storage_object_t * obj;

	LOG_PROC_ENTRY();

	/*
	 * Loop through the object list finding groups of objects that have the
	 * same top level feature.  Extract the objects with the same top level
	 * feature, put them into a separate list, and pass the list to the
	 * feature for discovery.
	 *
	 * Can't use LIST_FOR_EACH_SAFE() to do the loop because multiple
	 * objects may be removed from the list by extract_plugin_objects().
	 */
	for (obj = first_thing(object_list, &iter);
	    iter != NULL;
	    obj = first_thing(object_list, &iter)) {

		plugin_record_t * feature;
		list_element_t el = NULL;

		remove_element(iter);

		/*
		 * If the object has a feature header, start the discovery
		 * process on the object.
		 */
		if (obj->feature_header != NULL) {
			rc = engine_get_plugin_by_ID(obj->feature_header->feature_id, &feature);

			if (rc == 0) {
				/*
				 * Skip associative class features.  They are handled
				 * later after the volumes have been discovered.
				 */

				if (GetPluginType(feature->id) != EVMS_ASSOCIATIVE_FEATURE) {
					LOG_DEBUG("Extract objects for %s.\n", feature->short_name);

					/* Clear out the plugin_object_list */
					delete_all_elements(&plugin_object_list);

					/*
					 * Put the extracted object on the list of objects
					 * for the plug-in.
					 */
					LOG_DEBUG("Extract object %s.\n", obj->name);
					el = insert_element(&plugin_object_list,
							    iter,
							    INSERT_AFTER,
							    NULL);

					if (el != NULL) {

						/*
						 * Clear the new_object_list to receive the
						 * new objects from the next call to a
						 * plug-in's discover();
						 */
						delete_all_elements(&new_object_list);

						/*
						 * Pull the other objects that have the same
						 * feature ID off the list and add them to the
						 * plugin_object_list.
						 */
						rc = extract_plugin_objects(object_list,
									    feature->id,
									    &plugin_object_list);

						if (rc == 0) {
							list_element_t tmp_iter;
							storage_object_t * tmp_obj;

							if (report_status) {
								status_message(_("Discovering EVMS objects%s\n"),
									       status_dots(&status_count));
							}
							rc = feature->functions.plugin->discover(&plugin_object_list, &new_object_list, TRUE);

							/*
							 * Remove any corrupt objects from the
							 * list.
							 */
							remove_corrupt_objects(&new_object_list);

							/* Make sure we have a dev node for each object. */
							LIST_FOR_EACH(&new_object_list, tmp_iter, tmp_obj) {
								if (tmp_obj->flags & SOFLAG_ACTIVE) {
									make_object_dev_node(tmp_obj);
								}
							}

							propigate_cluster_info(&new_object_list);

							/*
							 * If discover() succeeded, put the new
							 * objects on the resulting_object_list.
							 */
							if (rc == 0) {
								merge_lists(resulting_object_list,
									    &new_object_list,
									    NULL,
									    NULL);

								discover_replace_objects(resulting_object_list);
							}
						}
					}

				} else {
					/*
					 * We hit an associative class feature.  Build a
					 * temporary volume structure to hold the volume
					 * information from the feature header structure and
					 * hang it off of the object so that we will know
					 * how to create the real volume when the
					 * associative feature discovers its object.
					 */
					logical_volume_t * volume = (logical_volume_t *) engine_alloc(sizeof(logical_volume_t));

					if (volume != NULL) {

						volume->serial_number = obj->feature_header->volume_serial_number;
						strcpy(volume->name, obj->feature_header->volume_name);

						/* (Ab)Use the consuming_private_data field in
						 * the object to hold the temporary volume.
						 * If we put it in the volume field, later code
						 * won't be able to find this object as a top
						 * object since it will think it is part of a
						 * volume.
						 */
						obj->consuming_private_data = volume;

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

			} else {
				LOG_WARNING("Error code %d when trying to get plug-in for feature ID %d (%#x) from feature header in object %s.\n", rc, obj->feature_header->feature_id, obj->feature_header->feature_id, obj->name);
			}

		} else {
			LOG_WARNING("Object %s does not contain a feature header.\n", obj->name);
		}

		if (el == NULL) {
			/*
			 * iter1 was  not insterted into the
			 * plugin_object_list.  Free the element.
			 */
			memset(iter, 0, sizeof(*iter));
			engine_free(iter);
		}
	}

	LOG_PROC_EXIT_INT(rc);
	return rc;
}


/*
 * Check if the feature header hanging off an object has the specified depth.
 * If so, put the object on the result_list and return TRUE so that the
 * object will be removed from the current list.  Else, return FALSE so that
 * the object stays on the list.
 */
static void extract_objects_by_depth(list_anchor_t list,
				     list_anchor_t result_list,
				     u_int32_t     depth) {

	list_element_t iter1;
	list_element_t iter2;
	storage_object_t * obj;

	LOG_PROC_ENTRY();

	LOG_DEBUG("Extracting objects of depth %u.\n", depth);

	LIST_FOR_EACH_SAFE(list, iter1, iter2, obj) {
		if (obj->feature_header != NULL) {
			if (obj->feature_header->object_depth == depth) {
				LOG_DEBUG("Extract object %s.\n", obj->name);
				remove_element(iter1);

				insert_element(result_list,
					       iter1,
					       INSERT_AFTER,
					       NULL);
			}

		} else {
			LOG_WARNING("Object %s does not have a feature header.  It is being removed from the list.\n", obj->name);
			delete_element(iter1);
		}
	}

	LOG_PROC_EXIT_VOID();
	return;
}


/*
 * Given a VolObj list, do the discovery of the objects that make up the volume.
 */
static void discover_objects_for_volume(volume_objects_t * pVolObj, boolean report_status) {

	int rc = 0;
	STATIC_LIST_DECL(object_list);
	u_int32_t   depth = 0;
	list_element_t iter;
	storage_object_t * obj;

	LOG_PROC_ENTRY();

	/* Find the greatest object_depth of the objects in the list. */
	LIST_FOR_EACH(pVolObj->objects, iter, obj) {
		get_greatest_object_depth(obj, &depth);
	}

	while ((rc == 0) && (depth > 0)) {

		/*
		 * Get the feature header for the top level feature contained
		 * within each object on the list.  The feature header is hung off a
		 * pointer in the storage object.  We ignore errors that may occur
		 * during the reading of feature headers.  If the error was bad
		 * enough, then there won't be a feature header and thus won't be
		 * anything to discover on that object.
		 */
		LIST_FOR_EACH(pVolObj->objects, iter, obj) {
			get_feature_header(obj);
		}

		/*
		 * Clear out the working object_list for this iteration through
		 * the loop.
		 */
		delete_all_elements(&object_list);

		/*
		 * Build a list of objects that have the same object_depth
		 * in the feature header.  The objects are extracted from the
		 * pVolObj->objects list.
		 */
		extract_objects_by_depth(pVolObj->objects, &object_list, depth);

		/*
		 * Given the list of objects with the same object_depth,
		 * group the objects by plug-in and pass each group to its
		 * plug-in for discovery.  Objects resulting from the
		 * discovery are put back on the pVolObj->objects list for
		 * the next pass through this loop.
		 */
		rc = discover_objects_by_plugin(&object_list, pVolObj->objects, report_status);
		if (rc == 0) {
			/* Move to the next depth. */
			depth--;
		}
	}

	if (rc == 0) {
		uint count;

		/*
		 * If there is only one item in the list, then it is a candidate for
		 * becoming a volume.
		 */
		count = list_count(pVolObj->objects);
		if (count == 1) {
			obj = first_thing(pVolObj->objects, NULL);

			if (obj != NULL) {

				/*
				 * A flag in the feature header (flags were copied to
				 * the volume_objects_t structure) says whether this
				 * object should be made into a volume or left as an
				 * object.
				 */
				if (!(pVolObj->flags & EVMS_VOLUME_DATA_OBJECT)) {
					rc = make_evms_volume_for_object(obj,
									 pVolObj->volume_name,
									 pVolObj->volume_serial);
					get_volume_dev_node_and_fsim(obj->volume);
				}
			}
		}
	}

	LOG_PROC_EXIT_VOID();
}


/*
 * Put an object onto a VolObj list (a list of objects which volume comprises).
 * If a VolObj list has not yet been made for the volume, make one.
 */
static int add_object_to_VolObj_list(storage_object_t * obj, volume_objects_t * * pVolObjList) {
	int rc = 0;
	volume_objects_t * pVolObj;

	LOG_PROC_ENTRY();

	pVolObj = *pVolObjList;
	while ((pVolObj != NULL) && (pVolObj->volume_serial != obj->feature_header->volume_serial_number)) {
		pVolObj = pVolObj->next;
	}

	if (pVolObj == NULL) {
		LOG_DEBUG("Create new VolObj list for volume %s.\n", obj->feature_header->volume_name);
		pVolObj = (volume_objects_t *) engine_alloc(sizeof(volume_objects_t));

		if (pVolObj != NULL) {
			pVolObj->flags = obj->feature_header->flags;
			pVolObj->volume_serial = obj->feature_header->volume_serial_number;
			memcpy(pVolObj->volume_name, obj->feature_header->volume_name, sizeof(pVolObj->volume_name));
			pVolObj->objects = allocate_list();

			pVolObj->next = *pVolObjList;
			*pVolObjList = pVolObj;

		} else {
			rc = ENOMEM;
			LOG_CRITICAL("Error when allocating memory for a volume_object structure.\n");
		}
	}

	if (rc == 0) {
		list_element_t el;

		LOG_DEBUG("Add object %s to volume object list for volume %s.\n", obj->name, pVolObj->volume_name);
		el = insert_thing(pVolObj->objects,
				  obj,
				  INSERT_AFTER,
				  NULL);

		if (el == NULL) {
			LOG_CRITICAL("Error inserting object %s into the object list for volume %s.\n",
				     obj->name, pVolObj->volume_name);
			rc = ENOMEM;
		}
	}

	LOG_PROC_EXIT_INT(rc);
	return rc;
}


/*
 * Initial processing of objects.
 * Objects with stop data have been removed from the list.  Non-data type
 * objects have been removed from the list.
 * If the object does not have a feature header, make it into a compatibility
 * volume.
 * If the object has a feature header on it, check if it is the special
 * feature header that is put on objects that don't have features but are made
 * into EVMS volumes.  If so, make the object into an EVMS volume.
 * Otherwise, the object has EVMS features on it.  Put the object into a list
 * of object that are grouped by volume.
 */
static int process_object(storage_object_t   * obj,
			  volume_objects_t * * pVolObjList) {

	int rc = 0;

	LOG_PROC_ENTRY();

	LOG_DEBUG("Processing object %s.\n", obj->name);

	/*
	 * If the object does not have a feature header, then make it into a
	 * compatibility volume.
	 */
	if (obj->feature_header == NULL) {
		make_compatibility_volume_for_object(obj);
		get_volume_dev_node_and_fsim(obj->volume);

	} else {
		/*
		 * The object has a feature header.  If the feature header has
		 * an object_depth of 1 and a feature_id of 0, then it is a
		 * feature header for making an EVMS volume.
		 */

		if ((obj->feature_header->object_depth == 1) &&
		    (obj->feature_header->feature_id == EVMS_VOLUME_FEATURE_ID)) {

			if (!(obj->feature_header->flags & EVMS_VOLUME_DATA_OBJECT)) {
				sector_count_t usable_size = obj->size - EVMS_FEATURE_HEADER_SECTORS * 2;
				usable_size = round_down_to_hard_sector(usable_size, obj);
				
				/* Don't let errors stop the discovery process. */
				make_evms_volume_for_object(obj,
							    obj->feature_header->volume_name,
							    obj->feature_header->volume_serial_number);
				if (obj->volume != NULL) {
					get_volume_dev_node_and_fsim(obj->volume);

					/*
					 * If the device for the volume is
					 * bigger than the device for its object
					 * and the file system on the device
					 * fits within the object size resize
					 * the volume to the object size.
					 */
					if (obj->volume->vol_size < usable_size) {
						if (obj->volume->fs_size <= usable_size) {
							obj->volume->vol_size = usable_size;
							obj->volume->flags |= VOLFLAG_NEEDS_ACTIVATE;

						} else {
							LOG_SERIOUS("The DM device for volume %s is of size %"PRIu64" which is "
								    "bigger than the DM device for the volume's object %s of size %"PRIu64".  "
								    "I'm deactivating the volume.\n",
								    obj->volume->name, obj->volume->vol_size,
								    obj->name, obj->size);

							obj->volume->flags |= VOLFLAG_NEEDS_DEACTIVATE;
						}
					}
				}
			}

		} else {
			/*
			 * This object has feature(s) on it.  Put it into a
			 * VolObjList.
			 */
			rc = add_object_to_VolObj_list(obj, pVolObjList);

			if (rc != 0) {
				LOG_CRITICAL("Error when putting object %s into the VolObj list.  Return code was %d.\n", obj->name, rc);
			}
		}
	}

	LOG_PROC_EXIT_INT(rc);
	return rc;
}


/*
 * Scan a list, removing any object that does not have have its data type
 * set to DATA_TYPE.
 */
static void remove_non_data_objects(list_anchor_t list) {

	list_element_t iter1;
	list_element_t iter2;
	storage_object_t * obj;

	LOG_PROC_ENTRY();

	LIST_FOR_EACH_SAFE(list, iter1, iter2, obj) {
		LOG_DEBUG("Object %s has a data_type of %d.\n", obj->name, obj->data_type);

		/*
		 * If the object is not a data object (e.g., it is a metadata object)
		 * delete it from the list.
		 */
		if (obj->data_type != DATA_TYPE) {
			delete_element(iter1);
		}
	}

	LOG_PROC_EXIT_VOID();
	return;
}


/*
 * Scan a list, removing any object that has stop data on it.
 */
static evms_feature_header_t stop_data_fh;

static void remove_stop_data_objects(list_anchor_t list) {

	int rc;
	list_element_t iter1;
	list_element_t iter2;
	storage_object_t * obj;

	LOG_PROC_ENTRY();

	LIST_FOR_EACH_SAFE(list, iter1, iter2, obj) {
		LOG_DEBUG("Request to remove object %s if it has stop data.\n", obj->name);

		/* Default, if anything fails, is that there are no stop data. */
		obj->flags &= ~SOFLAG_HAS_STOP_DATA;

		/* Only data objects can have stop data. */
		if (obj->data_type == DATA_TYPE) {

			/* Check for stop data at the end of the object. */
			rc = obj->plugin->functions.plugin->read(obj, obj->size - 1, 1, &stop_data_fh);

			if (rc == 0) {
				if (validate_feature_header(&stop_data_fh) == 0) {
					/*
					 * validate_feature_header() converts the in-memory
					 * feature header to CPU endian format.
					 */
					if (stop_data_fh.flags & EVMS_VOLUME_DATA_STOP) {
						LOG_DEBUG("Object %s has stop data on it.\n", obj->name);
						obj->flags |= SOFLAG_HAS_STOP_DATA;
						delete_element(iter1);
					}
				}
			}
		}
	}

	LOG_PROC_EXIT_VOID();
	return;
}


static int discover_evms_objects(list_anchor_t object_list, boolean report_status) {

	int rc = 0;
	volume_objects_t * VolObjList = NULL;
	list_element_t iter;
	storage_object_t * obj;

	LOG_PROC_ENTRY();

	/*
	 * Pull any metadata objects off the list.
	 * Pull any free space objects off the object_list.
	 */
	remove_non_data_objects(object_list);

	/* Remove from the list any objects with stop data on them. */
	remove_stop_data_objects(object_list);

	/*
	 * Get the feature headers for the top level feature contained
	 * within each object on the list.  The feature header if found,
	 * is hung off a pointer in the storage object.
	 */
	LIST_FOR_EACH(object_list, iter, obj) {
		get_feature_header(obj);
	}

	/* Process the objects on the object_list. */
	LIST_FOR_EACH(object_list, iter, obj) {
		rc = process_object(obj, &VolObjList);
		if (rc != 0) {
			break;
		}
	}

	if (rc == 0) {

		/*
		 * Do the feature discovery for each group of objects
		 * that belong to a volume.
		 */
		volume_objects_t * pVolObj;

		for (pVolObj = VolObjList; ((rc == 0) && (pVolObj != NULL)); pVolObj = pVolObj->next) {
			discover_objects_for_volume(pVolObj, report_status);
		}

		/* Clean up our lists. */
		while (VolObjList != NULL) {
			pVolObj = VolObjList;
			VolObjList = pVolObj->next;

			destroy_list(pVolObj->objects);
			engine_free(pVolObj);
		}
	}

	LOG_PROC_EXIT_INT(rc);
	return rc;
}


static int discover_associative_features(list_anchor_t object_list, boolean report_status) {

	int rc = 0;
	int status_count = 0;
	logical_volume_t * volume;
	plugin_record_t * feature;
	list_element_t iter1;
	list_element_t iter2;
	storage_object_t * obj;

	LOG_PROC_ENTRY();

	LIST_FOR_EACH_SAFE(object_list, iter1, iter2, obj) {
		LOG_DEBUG("Examining object %s.\n", obj->name);

		/*
		 * Check if a feature header even exists since this might be a topmost
		 * object that is not a volume.
		 */
		if (obj->feature_header == NULL) {
			LOG_DEBUG("Object does not have a feature header.\n");
			continue;
		}

		/*
		 * Filter out objects that have an associative class feature for a
		 * parent.
		 */
		if (GetPluginType(obj->feature_header->feature_id) != EVMS_ASSOCIATIVE_FEATURE) {
			LOG_DEBUG("Object's feature header is not for an associative class feature.\n");
			continue;
		}

		/*
		 * Extract the temporary volume structure that was attached to the
		 * object when it was discovered.
		 */

		volume = (logical_volume_t *) obj->consuming_private_data;

		if (volume == NULL) {
			LOG_DEBUG("Object has an associative class feature header, but does not have a volume structure attached.\n");
			continue;
		}

		obj->consuming_private_data = NULL;

		rc = engine_get_plugin_by_ID(obj->feature_header->feature_id, &feature);

		if (rc == 0) {

			/*
			 * discover() takes a list_anchor_t for input objects, so we have
			 * to build them, even though we know there is only one
			 * object for the discover().
			 */

			STATIC_LIST_DECL(object_list);
			list_element_t el;

			el = insert_thing(&object_list,
					  obj,
					  INSERT_AFTER,
					  NULL);

			if (el != NULL) {
				STATIC_LIST_DECL(new_object_list);
				list_element_t iter;
				storage_object_t * tmp_obj;

				if (report_status) {
					status_message(_("Discovering EVMS associative objects%s\n"),
						       status_dots(&status_count));
				}
				rc = feature->functions.plugin->discover(&object_list, &new_object_list, TRUE);

				LOG_DEBUG("Return code from %s discovery() was %d.\n", feature->short_name, rc);

				/* Remove any corrupt objects from the list. */
				remove_corrupt_objects(&new_object_list);

				LIST_FOR_EACH(&new_object_list, iter, tmp_obj) {
					if (tmp_obj->flags & SOFLAG_ACTIVE) {
						make_object_dev_node(tmp_obj);
					}
				}

				propigate_cluster_info(&new_object_list);

				if (rc == 0) {
					storage_object_t * new_object;

					new_object = first_thing(&new_object_list, NULL);

					/*
					 * Make a volume for the new object if the
					 * feature header doesn't say to make it
					 * into an object.
					 */
					if (!(obj->feature_header->flags & EVMS_VOLUME_DATA_OBJECT)) {

						/* Make a volume for the object. */
						rc = make_evms_volume_for_object(new_object,
										 volume->name,
										 volume->serial_number);
						get_volume_dev_node_and_fsim(obj->volume);
					}

					/*
					 * Delete this object
					 * from the list.
					 */
					delete_element(iter1);

				} else {
					/*
					 * If the associative feature failed to do
					 * its discovery, that's not a catastrophic
					 * error worth killing the Engine.  Clear
					 * the error code.
					 */
					if (rc == EVMS_FEATURE_FATAL_ERROR) {
						rc = 0;
					}
				}
			}

		} else {
			LOG_DEBUG("Unable to find the plug-in for feature ID %d.\n", obj->feature_header->feature_id);
		}

		/* Release the temporary volume structure. */
		engine_free(volume);
	}

	LOG_PROC_EXIT_INT(rc);
	return rc;
}


int compare_objects(void * thing1,
		    void * thing2,
		    void * user_data) {

	int result = 0;

	storage_object_t * obj1 = (storage_object_t *) thing1;
	storage_object_t * obj2 = (storage_object_t *) thing2;

	LOG_PROC_ENTRY();

	/* First sort by plug-in */
	result = strcmp(obj1->plugin->short_name, obj2->plugin->short_name);

	if (result == 0) {
		if (obj1->object_type == obj2->object_type) {
			switch (obj1->object_type) {
				case DISK:
				case REGION:
				case EVMS_OBJECT:
					/*
					 * If the objects are produced from containers,
					 * sort by the container name.
					 */
					if ((obj1->producing_container != NULL) &&
					    (obj2->producing_container != NULL)) {
						result = strcmp(obj1->producing_container->name,
								obj2->producing_container->name);
					}

					if (result == 0) {
						/* Sort by data type. */
						if (obj1->data_type != obj2->data_type) {
							switch (obj1->data_type) {
								case DATA_TYPE:
									/* Data objects come first. */
									result = -1;
									break;
								case FREE_SPACE_TYPE:
									/* Free space objects are second. */
									if (obj2->data_type == DATA_TYPE) {
										result = 1;
									} else {
										result = -1;
									}
									break;
								case META_DATA_TYPE:
									/* Metadata objects are last. */
									result = 1;
									break;
								default:
									break;
							}

						} else {
							/* Objects have the same data type. */
							/* Sort by name. */
							result = strcmp(obj1->name, obj2->name);
						}
					}
					break;

				case SEGMENT:
					{
						/* First, sort by disk. */
						storage_object_t * child1;
						storage_object_t * child2;

						child1 = first_thing(obj1->child_objects, NULL);
						child2 = first_thing(obj2->child_objects, NULL);

						if (child1 != NULL && child2 != NULL) {
							result = strcmp(child1->name, child2->name);

							if (result == 0) {
								/* Segments are sorted by starting LBA. */
								if (obj1->start < obj2->start) {
									result = -1;
								} else if (obj1->start > obj2->start) {
									result = 1;
								} else {
									result = 0;
								}
							}
						} else {
							result = strcmp(obj1->name, obj2->name);
						}
					}
					break;

				default:
					break;
			}

		} else {
			/*
			 * Objects are not of the same type.  This should not happen.
			 * Just in case, compare the object names.
			 */
			result = strcmp(obj1->name, obj2->name);
		}
	}

	LOG_PROC_EXIT_INT(result);
	return(result);
}


int compare_containers(void * thing1,
		       void * thing2,
		       void * user_data) {

	int result = 0;

	storage_container_t * con1 = (storage_container_t *) thing1;
	storage_container_t * con2 = (storage_container_t *) thing2;

	LOG_PROC_ENTRY();

	/* First sort by plug-in */
	result = strcmp(con1->plugin->short_name, con2->plugin->short_name);

	if (result == 0) {
		/* Next, sort by container name. */
		result = strcmp(con1->name, con2->name);
	}

	LOG_PROC_EXIT_INT(result);
	return(result);
}


int compare_volumes(void * thing1,
		    void * thing2,
		    void * user_data) {

	int result = 0;

	logical_volume_t * vol1 = (logical_volume_t *) thing1;
	logical_volume_t * vol2 = (logical_volume_t *) thing2;

	LOG_PROC_ENTRY();

	/* Sort by volume name. */
	result = strcmp(vol1->name, vol2->name);

	LOG_PROC_EXIT_INT(result);
	return(result);
}


/* Incremental size to allocate for the name array */
#define ALLOC_SIZE  4096

/* Mask to check if the number of array entries is on an allocation boundary */
#define ALLOC_MASK  ((ALLOC_SIZE / sizeof(char *)) - 1)

/*
 * Insert a name into the array, keeping the array sorted.
 */
static int insert_name(char * name, char * * * p_array, int * p_count) {

	int i;
	char * * array;
	int count = *p_count;
	int low;
	int curr;
	int high;
	int comp;

	LOG_PROC_ENTRY();

	LOG_DEBUG("Array has %u entries.\n", *p_count);
	LOG_DEBUG("Insert name \"%s\" into the array.\n", name);

	if (!(count & ALLOC_MASK)) {
		u_int32_t size = (count * (ALLOC_SIZE / sizeof(char *))) + ALLOC_SIZE;
		/*
		 * The array has a number of entries that fills up an entire
		 * allocated space.  (Includes the initial array size of 0.)
		 * We need more space for the array.
		 */
		LOG_DEBUG("Allocate more space for the array: room enough for %d entries.\n",
			  size);
		*p_array = engine_realloc(*p_array, size);
	}

	if (*p_array == NULL) {
		LOG_CRITICAL("Error allocating memory for the name list.\n");
		LOG_PROC_EXIT_INT(ENOMEM);
		return ENOMEM;
	}

	array = *p_array;

	/* Binary search for the insertion slot */
	low = 0;
	high = count - 1;
	curr = (high + low) / 2;
	comp = 1;

	while ((low <= high) && (comp != 0)) {

		comp = strcmp(name, array[curr]);
		if (comp != 0) {
			if (comp < 0) {
				high = curr - 1;

			} else {
				low = curr + 1;
			}
			curr = (high + low) / 2;
		}
	}

	/*
	 * "low" is the index of the entry before which the new entry belongs.
	 * Move all the entries from "low" to "count -1" up one slot in the
	 * array to make room for the new entry.
	 */
	for (i = count; i > low; i--) {
		array[i] = array[i - 1];
	}

	array[low] = name;
	(*p_count)++;

	LOG_DEBUG("Array has %u entries.\n", *p_count);

	LOG_PROC_EXIT_INT(0);
	return 0;
}


/*
 * Remove a name from the array.
 */
static int remove_name(char * name, char * * array, int * p_count) {

	int rc = 0;
	int i;
	int count = *p_count;
	int low;
	int curr;
	int high;
	int comp;

	LOG_PROC_ENTRY();

	LOG_DEBUG("Array has %u entries.\n", *p_count);
	LOG_DEBUG("Remove name \"%s\" from the array.\n", name);

	/* Binary search for the node name */
	low = 0;
	high = count - 1;
	curr = (high + low) / 2;
	comp = 1;

	while ((low <= high) && (comp != 0)) {

		comp = strcmp(name, array[curr]);
		if (comp != 0) {
			if (comp < 0) {
				high = curr - 1;
			} else {
				low = curr + 1;
			}
			curr = (high + low) / 2;
		}
	}

	if (comp == 0) {
		/* Found a match. */
		engine_free(array[curr]);

		/* Remove the match by collapsing the array. */
		for (i = curr + 1; i < count; i++) {
			array[i-1] = array[i];
		}

		array[count - 1] = NULL;
		(*p_count)--;

	} else {
		/* Did not find a match. */
		LOG_DEBUG("Name not found in the array.\n");
		rc = ENOENT;
	}

	LOG_DEBUG("Array has %u entries.\n", *p_count);

	LOG_PROC_EXIT_INT(rc);
	return rc;
}


/*
 * Get the names of the block device dev nodes in a given directory.
 * Return a pointer to an array of dev node name strings and a count of names
 * in the array.
 */
static int get_dev_node_names(char * dir_name, char * * * name_list, int * count) {

	int rc = 0;
	struct stat stat_buf;
	DIR * dir;
	struct dirent * dir_ent;
	char * new_name;
	char name_buf[PATH_MAX];
	int dir_path_len;

	LOG_PROC_ENTRY();

	LOG_DEBUG("Get device nodes in directory %s.\n", dir_name);

	dir = opendir(dir_name);

	if (dir == NULL) {
		LOG_PROC_EXIT_INT(errno);
		return errno;
	}

	strcpy(name_buf, dir_name);
	dir_path_len = strlen(name_buf);
	name_buf[dir_path_len] = '/';
	dir_path_len++;

	for (dir_ent = readdir(dir);
	     (rc == 0) && (dir_ent != NULL);
	     dir_ent = readdir(dir)) {
		
		/* Skip the "." and ".." entries. */
		if ((strcmp(dir_ent->d_name,".") == 0) ||
		    (strcmp(dir_ent->d_name,"..") == 0)) {
			continue;
		}

		/*
		 * Build the full dev node name by appending the
		 * dir_ent->d_name to the directory name.
		 */
		strcpy(&name_buf[dir_path_len], dir_ent->d_name);

		if (stat(name_buf, &stat_buf) == 0) {
			/*
			 * If this entry is a directory, call this function
			 * recursively to process the subdirectory.
			 */
			if (S_ISDIR(stat_buf.st_mode)) {
				int prev_count = *count;
				
				get_dev_node_names(name_buf, name_list, count);

				if (prev_count == *count) {
					/*
					 * Nothing found in the directory.
					 * Remove it.
					 */
					LOG_DEBUG("Remove empty directory: %s\n", name_buf);
					rmdir(name_buf);
				}

			} else {
				if (S_ISBLK(stat_buf.st_mode)) {
					new_name = engine_strdup(name_buf);
					if (new_name != NULL) {
						rc = insert_name(new_name, name_list, count);

					} else {
						LOG_CRITICAL("Error getting memory for a name string.\n");
						rc = ENOMEM;
					}
				}
			}
		}
	}

	closedir(dir);

	LOG_PROC_EXIT_INT(rc);
	return rc;
}


static void remove_dm_mapping(char * name) {

	if (strstr(name, EVMS_OBJECT_NODE_PATH) != NULL) {
		/*
		 * It's a dev node for an object.
		 * Build a dummy object to check if device-mapper has
		 * a map for the object name.  If so, deactivate the
		 * mapping.
		 */
		storage_object_t obj;

		memset(&obj, '\0', sizeof(storage_object_t));
		strcpy(obj.name, name + EVMS_OBJECT_NODE_PATH_LEN);

		dm_update_status(&obj);

		/*
		 * Object will be marked active if it has a
		 * device_mapper mapping.
		 */
		if (obj.flags & SOFLAG_ACTIVE) {
			dm_deactivate(&obj);
		}

	} else {
		/*
		 * It's a dev node for a volume.
		 * Build a dummy volume to check if device-mapper has
		 * a map for the volume name.  If so, deactivate the
		 * mapping.
		 */
		logical_volume_t vol;

		memset(&vol, '\0', sizeof(logical_volume_t));
		strcpy(vol.name, name);

		dm_update_volume_status(&vol);

		/*
		 * Volume will be marked active if it has a
		 * device_mapper mapping.
		 */
		if (vol.flags & VOLFLAG_ACTIVE) {
			dm_deactivate_volume(&vol);
		}
	}
}


static void cleanup_empty_dirs(char * dir_name) {

	int rc = 0;
	struct stat stat_buf;
	DIR * dir;
	struct dirent * dir_ent;
	char name_buf[PATH_MAX];
	int dir_path_len;
	boolean dir_empty = TRUE;

	LOG_PROC_ENTRY();

	LOG_DEBUG("Dir: %s\n", dir_name);

	dir = opendir(dir_name);

	if (dir == NULL) {
		LOG_PROC_EXIT_VOID();
		return;
	}

	strcpy(name_buf, dir_name);
	dir_path_len = strlen(name_buf);
	name_buf[dir_path_len] = '/';
	dir_path_len++;

	for (dir_ent = readdir(dir);
	     (rc == 0) && (dir_ent != NULL);
	     dir_ent = readdir(dir)) {
		
		/* Skip the "." and ".." entries. */
		if ((strcmp(dir_ent->d_name,".") == 0) ||
		    (strcmp(dir_ent->d_name,"..") == 0)) {
			continue;
		}

		/*
		 * Build the full dev node name by appending the
		 * dir_ent->d_name to the directory name.
		 */
		strcpy(&name_buf[dir_path_len], dir_ent->d_name);

		if (stat(name_buf, &stat_buf) == 0) {
			/*
			 * If this entry is a directory, call this function
			 * recursively to process the subdirectory.
			 */
			if (S_ISDIR(stat_buf.st_mode)) {
				
				cleanup_empty_dirs(name_buf);

				/*
				 * If the subdirectory was empty it will have
				 * been deleted. Check if it is still there.
				 */
				if (stat(name_buf, &stat_buf) != 0) {

					/*
					 * It's gone.  Don't know what the
					 * removal of the subdirecotry did to
					 * our walk through this directory.
					 * Rewind and start over.
					 */
					rewinddir(dir);

				} else {
					/*
					 * The subdirectory is still there.
					 * This directory is not empty.
					 */
					dir_empty = FALSE;
				}

			} else {
				/*
				 * It's not a subdirectory.  Something else is
				 * in this directory.  It's not empty.
				 */
				dir_empty = FALSE;
			}
		}
	}

	closedir(dir);

	if (dir_empty) {
		LOG_DEBUG("Removing empty directory %s.\n", dir_name);
		rmdir(dir_name);
	}

	LOG_PROC_EXIT_VOID();
	return;
}


static void cleanup_dev_evms_tree(void) {

	int num_nodes = 0;
	int i;
	char * * dev_nodes = NULL;
	logical_volume_t * vol;
	storage_object_t * obj;
	list_anchor_t lists[] = {&disks_list,
			         &segments_list,
			         &regions_list,
			         &EVMS_objects_list,
			         NULL};
	char obj_name_buf[EVMS_OBJECT_NODE_PATH_LEN + EVMS_NAME_SIZE + 1] = EVMS_OBJECT_NODE_PATH;
	int status_count = 0;
	list_element_t iter;

	LOG_PROC_ENTRY();
	
	sync_volumes();

	status_message(_("Cleaning up the /dev/evms tree%s\n"), status_dots(&status_count));

	/* Get an array of all the block dev nodes in the entire /dev/evms tree. */
	get_dev_node_names(EVMS_DEV_NODE_DIR, &dev_nodes, &num_nodes);

	status_message(_("Cleaning up the /dev/evms tree%s\n"), status_dots(&status_count));

	/*
	 * dev_nodes contains a sorted array of nodes in the /dev/evms tree.
	 * Remove the entries for the volumes and object dev nodes that we know
	 * about.
	 */
	LIST_FOR_EACH(&volumes_list, iter, vol) {
		if (vol->flags & VOLFLAG_ACTIVE) {
			remove_name(vol->name, dev_nodes, &num_nodes);
		}
	}

	for (i = 0; lists[i] != NULL; i++) {
		status_message(_("Cleaning up the /dev/evms tree%s\n"), status_dots(&status_count));
		LIST_FOR_EACH(lists[i], iter, obj) {
			if (obj->flags & SOFLAG_ACTIVE) {
				strcpy(&obj_name_buf[EVMS_OBJECT_NODE_PATH_LEN], obj->name);
				ensure_dev_node(obj_name_buf, obj->dev_major, obj->dev_minor);
				remove_name(obj_name_buf, dev_nodes, &num_nodes);
			}
		}
	}
	status_message(_("Cleaning up the /dev/evms tree%s\n"), status_dots(&status_count));

	/*
	 * All of the known nodes have been removed from the list.  What is left in
	 * the list are nodes for which we do not have a corresponding volume or
	 * object node.  unlink all the remaining nodes.
	 */
	for (i = 0; i < num_nodes; i++) {
		LOG_DETAILS("Removing dev node %s.\n", dev_nodes[i]);
		remove_dm_mapping(dev_nodes[i]);
		unlink(dev_nodes[i]);
		engine_free(dev_nodes[i]);
	}

	engine_free(dev_nodes);

	cleanup_empty_dirs(EVMS_DEV_NODE_DIR);

	status_message(_("Finished cleaning up the /dev/evms tree.\n"));

	LOG_PROC_EXIT_VOID();
	return;
}


int discover(list_anchor_t objects, boolean report_status) {

	int rc = 0;
	STATIC_LIST_DECL(discover_list);
	list_element_t iter1;
	list_element_t iter2;
	storage_object_t * obj;
	list_anchor_t discover_objects = copy_list(objects);

	boolean old_discover_in_progress;

	LOG_PROC_ENTRY();

	old_discover_in_progress = discover_in_progress;
	discover_in_progress = TRUE;

	/* Filter out disks and segments for segment discovery. */
	LIST_FOR_EACH_SAFE(objects, iter1, iter2, obj) {
		if (obj->object_type & (DISK | SEGMENT)) {
			remove_element(iter1);
			insert_element(&discover_list, iter1, INSERT_AFTER, NULL);
		}
	}

	rc = discover_segments(&discover_list, report_status);

	if (rc == 0) {
		sort_list(&segments_list, compare_objects, NULL);

		/* Add in regions for region discovery. */
		LIST_FOR_EACH_SAFE(objects, iter1, iter2, obj) {
			if (obj->object_type == REGION) {
				remove_element(iter1);
				insert_element(&discover_list, iter1, INSERT_AFTER, NULL);
			}
		}

		rc = discover_regions(&discover_list, report_status);

		if (rc == 0) {
			sort_list(&regions_list, compare_objects, NULL);

			/* Any objects left on the object list will be EVMS objects.
			 * Put any remaining objects on the discover_list.
			 */
			merge_lists(&discover_list, objects, NULL, NULL);

			rc = discover_evms_objects(&discover_list, report_status);
			if (rc == 0) {

				list_anchor_t top_object_list;

				rc = engine_get_object_list(0,
							    DATA_TYPE,
							    NULL,
							    NULL,
							    TOPMOST,
							    &top_object_list);

				if (rc == 0) {
					rc = discover_associative_features(top_object_list, report_status);
					if (rc == 0) {

						sort_list(&EVMS_objects_list, compare_objects, NULL);
						sort_list(&containers_list, compare_containers, NULL);
						sort_list(&volumes_list, compare_volumes, NULL);
					}

					destroy_list(top_object_list);
				}
			}
		}
	}

	discover_in_progress = old_discover_in_progress;

	mark_for_activation(discover_objects);
	destroy_list(discover_objects);

	LOG_PROC_EXIT_INT(rc);
	return rc;
}


int initial_discovery(void) {

	int rc = 0;
	STATIC_LIST_DECL(object_list);
	struct timezone tz;
	struct timeval discover_start_time;
	struct timeval discover_stop_time;
	struct timeval discover_time;
	int discover_hours;
	int discover_minutes;

	gettimeofday(&discover_start_time, &tz);

	LOG_PROC_ENTRY();

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

	if (rc == 0) {

		discover_in_progress = TRUE;

		rc = discover_logical_disks(&object_list, TRUE);
		sort_list(&disks_list, compare_objects, NULL);

		if (rc == 0) {
			rc = discover(&object_list, TRUE);
			if (replace_plugin != NULL) {
				replace_plugin->functions.plugin->discover(NULL, NULL, TRUE);
			}
			cleanup_dev_evms_tree();
		}

		discover_in_progress = FALSE;

	} else {
		engine_user_message(NULL, NULL,
				    _("Error creating the %s directory: %s\n"),
				    EVMS_OBJECT_NODE_DIR, strerror(rc));
	}

	/*
	 * Tell the Local Disk Manager to turn of its cache now that discovery
	 * is complete.
	 */
	if (local_disk_manager != NULL) {
		local_disk_manager->functions.plugin->plugin_function(NULL, LDM_Stop_Caching, NULL, NULL);
	}

	gettimeofday(&discover_stop_time, &tz);

	discover_time.tv_sec = discover_stop_time.tv_sec - discover_start_time.tv_sec;
	discover_time.tv_usec = discover_stop_time.tv_usec - discover_start_time.tv_usec;
	if (discover_time.tv_usec < 0) {
		discover_time.tv_sec --;
		discover_time.tv_usec += 1000000;
	}

	discover_hours = discover_time.tv_sec / 3600;
	discover_time.tv_sec %= 3600;
	discover_minutes = discover_time.tv_sec / 60;
	discover_time.tv_sec %= 60;

	LOG_DEFAULT("Discovery time: %02d:%02d:%02ld.%06ld\n", discover_hours, discover_minutes, discover_time.tv_sec, discover_time.tv_usec);

	status_message(_("Discovery finished.\n"));

	LOG_PROC_EXIT_INT(rc);
	return rc;
}


static void mark_siblings_for_rediscover(storage_object_t * obj) {

	list_element_t iter1;
	list_element_t iter2;
	storage_object_t * parent;
	storage_object_t * sibling;

	LOG_PROC_ENTRY();

	LOG_DEBUG("Request to mark the siblings of %s for rediscover.\n", obj->name);

	if (obj->consuming_container != NULL) {
		LIST_FOR_EACH(obj->consuming_container->objects_consumed, iter2, sibling) {
			LOG_DEBUG("Mark object %s for rediscover.\n", sibling->name);
			sibling->flags |= SOFLAG_REDISCOVER;
		}

	} else if (!list_empty(obj->parent_objects)) {
		LIST_FOR_EACH(obj->parent_objects, iter1, parent) {
			LIST_FOR_EACH(parent->child_objects, iter2, sibling) {
				LOG_DEBUG("Mark object %s for rediscover.\n", sibling->name);
				sibling->flags |= SOFLAG_REDISCOVER;
			}
		}
	}

	LOG_PROC_EXIT_VOID();
	return;
}


int evms_mark_for_rediscover(char * name) {

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

	LOG_PROC_ENTRY();

	if (name == NULL) {
		LOG_DEBUG("Request to mark everything for rediscover.\n");

		/* Rediscover everything. Throw out everything but the disks. */
		LIST_FOR_EACH(&disks_list, iter, obj) {
			engine_discard(obj->parent_objects);
			if (obj->volume != NULL) {
				discard_volume(obj->volume);
				obj->volume = NULL;
			}
			obj->flags |= SOFLAG_REDISCOVER;
		}

		LOG_PROC_EXIT_INT(0);
		return 0;
	}

	LOG_DEBUG("Request to mark %s for rediscover.\n", name);

	/* Must find the name by brute force search. */
	for (i = 0; object_lists[i] != NULL; i++) {
		LIST_FOR_EACH(object_lists[i], iter, obj) {
			if (strcmp(obj->name, name) == 0) {
				mark_siblings_for_rediscover(obj);

				engine_discard(obj->parent_objects);
				if (obj->volume != NULL) {
					discard_volume(obj->volume);
					obj->volume = NULL;
				}
				LOG_DEBUG("Mark object %s for rediscover.\n", obj->name);
				obj->flags |= SOFLAG_REDISCOVER;

				LOG_PROC_EXIT_INT(0);
				return 0;
			}
		}
	}

	LIST_FOR_EACH(&containers_list, iter, con) {
		if (strcmp(con->name, name) == 0) {
			engine_discard(con->objects_produced);
			LOG_DEBUG("Mark container %s for rediscover.\n", con->name);
			con->flags |= SCFLAG_REDISCOVER;

			LOG_PROC_EXIT_INT(0);
			return 0;
		}
	}

	LIST_FOR_EACH(&volumes_list, iter, vol) {
		if (strcmp(vol->name, name) == 0) {
			LOG_DEBUG("Mark volume %s for rediscover.\n", vol->name);
			vol->flags |= VOLFLAG_REDISCOVER;

			LOG_PROC_EXIT_INT(0);
			return 0;
		}
	}

	LOG_DEBUG("%s not found.\n", name);

	LOG_PROC_EXIT_INT(ENOENT);
	return ENOENT;
}


int rediscover(void) {

	int rc = 0;
	STATIC_LIST_DECL(plugin_discard_list);
	STATIC_LIST_DECL(discover_list);
	STATIC_LIST_DECL(object_list);
	list_element_t iter1;
	list_element_t iter2;
	storage_object_t * obj;
	storage_container_t * con;

	LOG_PROC_ENTRY();

	LIST_FOR_EACH(&disks_list, iter1, obj) {
		if (obj->flags & SOFLAG_REDISCOVER) {
			insert_thing(&discover_list, obj, INSERT_AFTER | EXCLUSIVE_INSERT, NULL);
			obj->flags &= ~SOFLAG_REDISCOVER;
		}
	}

	LIST_FOR_EACH(&segments_list, iter1, obj) {
		if (obj->flags & SOFLAG_REDISCOVER) {
			insert_thing(&discover_list, obj, INSERT_AFTER | EXCLUSIVE_INSERT, NULL);
			obj->flags &= ~SOFLAG_REDISCOVER;
		}
	}
	LIST_FOR_EACH(&regions_list, iter1, obj) {
		if (obj->flags & SOFLAG_REDISCOVER) {
			insert_thing(&discover_list, obj, INSERT_AFTER | EXCLUSIVE_INSERT, NULL);
			obj->flags &= ~SOFLAG_REDISCOVER;
		}
	}
	LIST_FOR_EACH(&EVMS_objects_list, iter1, obj) {
		if (obj->flags & SOFLAG_REDISCOVER) {
			insert_thing(&discover_list, obj, INSERT_AFTER | EXCLUSIVE_INSERT, NULL);
			obj->flags &= ~SOFLAG_REDISCOVER;
		}
	}
	LIST_FOR_EACH_SAFE(&containers_list, iter1, iter2, con) {
		if (con->flags & SCFLAG_REDISCOVER) {
			list_element_t iter;
			storage_object_t * child;

			/* Put the container's consumed objects on the discover list. */
			LIST_FOR_EACH(con->objects_consumed, iter, child) {
				insert_thing(&discover_list, child, INSERT_AFTER | EXCLUSIVE_INSERT, NULL);
			}

			rc = con->plugin->container_functions->discard_container(con);
		}
	}

	if (!list_empty(&discover_list)) {
		rc = discover(&discover_list, FALSE);
	}

	LOG_PROC_EXIT_INT(rc);
	return rc;
}


int evms_rediscover(void) {

	int rc = 0;
	int tmp_rc = 0;

	LOG_PROC_ENTRY();

	/* Deactivate any volumes that were discarded. */
	tmp_rc = deactivate_volumes_on_list(&volume_delete_list);
	if (tmp_rc != 0) {
		LOG_WARNING("deactivate_volumes_on_list() returned error code %d: %s\n", tmp_rc, evms_strerror(tmp_rc));
		if (rc == 0) {
			rc = tmp_rc;
		}
	} else {
		delete_volumes();
	}

	/* Deactivate any objects that were discarded. */
	tmp_rc = deactivate_objects_on_list(&deactivate_list);
	if (tmp_rc != 0) {
		LOG_WARNING("deactivate_objects_on_list() returned error code %d: %s\n", tmp_rc, evms_strerror(tmp_rc));
		if (rc == 0) {
			rc = tmp_rc;
		}
	} else {
		delete_objects();
	}

	rc = rediscover();
	if (rc != 0) {
		LOG_WARNING("rediscover() returned error code %d: %s\n", rc, evms_strerror(rc));
	}

	tmp_rc = activate();
	if (tmp_rc != 0) {
		LOG_WARNING("activate() returned error code %d: %s\n", tmp_rc, evms_strerror(tmp_rc));
		if (rc == 0) {
			rc = tmp_rc;
		}
	}

	cleanup_dev_evms_tree();

	LOG_PROC_EXIT_INT(rc);
	return rc;
}