/*
 *
 *   Copyright (c) International Business Machines  Corp., 2001
 *
 *   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 <unistd.h>
#include <sys/stat.h>
#include <sys/time.h>

#include <fullengine.h>
#include "discover.h"
#include "engine.h"
#include "internalAPI.h"
#include "crc.h"
#include "volume.h"
#include "commit.h"
#include "message.h"


BOOLEAN discover_in_progress = FALSE;


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


/*
 * remove_corrupt_object() is used as the function of PruneList().  The caller
 * can issue a PruneList() on a list of storage_object_t structures and can
 * specify the pruning function as remove_corrupt_object.  This function will
 * check if a storage_object_t has its SOFLAG_CORRUPT set.  If so, it will
 * return TRUE so that the object is removed from the current list.
 */
BOOLEAN remove_corrupt_object(ADDRESS   object,
                              TAG       object_tag,
                              uint      object_size,
                              ADDRESS   object_handle,
                              ADDRESS   parameters,
                              BOOLEAN * free_memory,
                              uint    * error) {
    BOOLEAN result = FALSE;

    LOG_PROC_ENTRY();

    /* Never free item memory. */
    *free_memory = FALSE;

    /* Only handle storage objects. */
    if ((object_tag == DISK_TAG) ||
        (object_tag == SEGMENT_TAG) |
        (object_tag == REGION_TAG) ||
        (object_tag == EVMS_OBJECT)) {

        storage_object_t * obj = (storage_object_t *) object;

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

        if (obj->flags & SOFLAG_CORRUPT) {
            result = TRUE;
        }
    }

    /*
     * Don't stop PruneList, even if we encounter an error.  The error was
     * logged.
     */
    *error = DLIST_SUCCESS;

    LOG_PROC_EXIT_BOOLEAN(result);
    return result;
}


static int set_not_claimed(ADDRESS object,
                           TAG     object_tag,
                           uint    object_size,
                           ADDRESS object_handle,
                           ADDRESS parameters) {

    storage_object_t * obj = (storage_object_t *) object;

    LOG_PROC_ENTRY();

    /* Safety check */
    if ((object_tag == DISK_TAG) ||
        (object_tag == SEGMENT_TAG) ||
        (object_tag == REGION_TAG)) {

        LOG_DEBUG("Setting SOFLAG_NOT_CLAIMED flag on object %s.\n", obj->name);
        obj->flags |= SOFLAG_NOT_CLAIMED;
    }

    LOG_PROC_EXIT_INT(0);
    return 0;
}


static BOOLEAN remove_unclaimed_object(ADDRESS   object,
                                       TAG       object_tag,
                                       uint      object_size,
                                       ADDRESS   object_handle,
                                       ADDRESS   parameters,
                                       BOOLEAN * free_memory,
                                       uint    * error) {
    BOOLEAN result = FALSE;
    storage_object_t * obj = (storage_object_t *) object;
    dlist_t unclaimed_object_list = (dlist_t) parameters;
    int rc;

    LOG_PROC_ENTRY();

    /* Never free item memory */
    *free_memory = FALSE;

    /* Safety check */
    if ((object_tag == DISK_TAG) ||
        (object_tag == SEGMENT_TAG) ||
        (object_tag == REGION_TAG)) {

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

        if (obj->flags & SOFLAG_NOT_CLAIMED) {
            ADDRESS trash;

            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);
            rc = InsertObject(unclaimed_object_list,
                              sizeof(storage_object_t),
                              object,
                              object_tag,
                              NULL,
                              AppendToList,
                              FALSE,
                              &trash);

            if (rc != DLIST_SUCCESS) {
                LOG_WARNING("Error code %d when putting object %s on the unclaimed object list.\n", rc, obj->name);
            }

            /*
             * Return TRUE so that the object is removed from the list that is
             * being processed.
             */
            result = TRUE;

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

    LOG_PROC_EXIT_BOOLEAN(result);
    *error = 0;
    return result;
}


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

static int discover_logical_disks(dlist_t object_list) {

    int rc = 0;
    plugin_record_t * pPlugRec = NULL;
    dlist_t result_object_list = CreateList();

    LOG_PROC_ENTRY();

    if (result_object_list != NULL) {

        /*
         * Call all of the device manager plug-ins to have them discover their
         * disks.
         */
        rc = GoToStartOfList(PluginList);
        if (rc == DLIST_SUCCESS) {
            rc = GetObject(PluginList,
                           sizeof(plugin_record_t),
                           PLUGIN_TAG,
                           NULL,
                           TRUE,
                           (ADDRESS *) &pPlugRec);

            while (pPlugRec != NULL) {
                if (GetPluginType(pPlugRec->id) == EVMS_DEVICE_MANAGER) {
                    LOG_DEBUG("Calling discover() in %s.\n", pPlugRec->short_name);
                    pPlugRec->functions.plugin->discover(object_list, result_object_list, TRUE);
                    LOG_DEBUG("Return code from discover() is %d.\n", rc);

                    /* Remove any corrupt objects from the list. */
                    PruneList(result_object_list,
                              remove_corrupt_object,
                              NULL);

                    /* Remove any remaining objects from the source list. */
                    DeleteAllItems(object_list, FALSE);

                    /*
                     * Put the objects from the result list onto the source
                     * list.
                     */
                    AppendList(object_list, result_object_list);
                }

                rc = GetNextObject(PluginList,
                                   sizeof(plugin_record_t),
                                   PLUGIN_TAG,
                                   (ADDRESS *) &pPlugRec);
            }

            if ((rc == DLIST_END_OF_LIST) ||
                (rc == DLIST_EMPTY)) {
                rc = DLIST_SUCCESS;
            }
        }

        DestroyList(&result_object_list, FALSE);

    } else {
        /* There was an error in creating the result_object_list. */
        LOG_CRITICAL("Error allocating memory for the result object list.\n");
        rc = ENOMEM;
    }

    LOG_PROC_EXIT_INT(rc);
    return rc;
}


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

static int discover_segments(dlist_t object_list) {

    int rc = 0;
    plugin_record_t * pPlugRec = NULL;
    dlist_t result_object_list = CreateList();
    dlist_t unclaimed_object_list = CreateList();
    BOOLEAN finished = FALSE;
    BOOLEAN last_pass = FALSE;

    LOG_PROC_ENTRY();

    if ((result_object_list != NULL) &&
        (unclaimed_object_list != NULL)) {

        while ((rc == 0) && (!finished || last_pass)) {

            /*
             * 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
             */

            ForEachItem(object_list,
                        set_not_claimed,
                        NULL,
                        TRUE);

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

            rc = GoToStartOfList(PluginList);
            if (rc == DLIST_SUCCESS) {
                rc = GetObject(PluginList,
                               sizeof(plugin_record_t),
                               PLUGIN_TAG,
                               NULL,
                               TRUE,
                               (ADDRESS *) &pPlugRec);

                while (pPlugRec != NULL) {

                    if (GetPluginType(pPlugRec->id) == EVMS_SEGMENT_MANAGER) {
                        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("Return code from discover() is %d.\n", rc);

                        /* Remove any corrupt objects from the list. */
                        PruneList(result_object_list,
                                  remove_corrupt_object,
                                  NULL);

                        /*
                         * 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.
                         */
                        DeleteAllItems(object_list, FALSE);

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

                    rc = GetNextObject(PluginList,
                                       sizeof(plugin_record_t),
                                       PLUGIN_TAG,
                                       (ADDRESS *) &pPlugRec);
                }

                if ((rc == DLIST_END_OF_LIST) ||
                    (rc == DLIST_EMPTY)) {
                    rc = DLIST_SUCCESS;
                }
            }

            /*
             * 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.
             */
            PruneList(object_list,
                      remove_unclaimed_object,
                      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.
         */
        CopyList(object_list, unclaimed_object_list, InsertAtStart);

        DestroyList(&result_object_list, FALSE);
        DestroyList(&unclaimed_object_list, FALSE);

    } else {
        /*
         * There was an error in creating the result_object_list or the
         * unclaimed_object_list.
         */
        if (result_object_list == NULL) {
            LOG_CRITICAL("Error allocating memory for the result object list.\n");
        } else {
            DestroyList(&result_object_list, FALSE);
        }
        if (unclaimed_object_list == NULL) {
            LOG_CRITICAL("Error allocating memory for the unclaimed object list.\n");
        } else {
            DestroyList(&unclaimed_object_list, FALSE);
        }
        rc = ENOMEM;
    }

    LOG_PROC_EXIT_INT(rc);
    return rc;
}


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

static int discover_regions(dlist_t object_list) {

    int rc = 0;
    plugin_record_t * pPlugRec = NULL;
    dlist_t result_object_list = CreateList();
    dlist_t unclaimed_object_list = CreateList();
    BOOLEAN finished = FALSE;
    BOOLEAN last_pass = FALSE;

    LOG_PROC_ENTRY();

    if ((result_object_list != NULL) &&
        (unclaimed_object_list != NULL)) {

        while ((rc == 0) && (!finished || last_pass)) {

            /*
             * 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
             */

            ForEachItem(object_list,
                        set_not_claimed,
                        NULL,
                        TRUE);

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

            rc = GoToStartOfList(PluginList);
            if (rc == DLIST_SUCCESS) {
                rc = GetObject(PluginList,
                               sizeof(plugin_record_t),
                               PLUGIN_TAG,
                               NULL,
                               TRUE,
                               (ADDRESS *) &pPlugRec);

                while (pPlugRec != NULL) {

                    if (GetPluginType(pPlugRec->id) == EVMS_REGION_MANAGER) {
                        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("Return code from discover() is %d.\n", rc);

                        /* Remove any corrupt objects from the list. */
                        PruneList(result_object_list,
                                  remove_corrupt_object,
                                  NULL);

                        /*
                         * 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.
                         */
                        DeleteAllItems(object_list, FALSE);

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

                    rc = GetNextObject(PluginList,
                                       sizeof(plugin_record_t),
                                       PLUGIN_TAG,
                                       (ADDRESS *) &pPlugRec);
                }

                if ((rc == DLIST_END_OF_LIST) ||
                    (rc == DLIST_EMPTY)) {
                    rc = DLIST_SUCCESS;
                }
            }

            /*
             * 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.
             */
            PruneList(object_list,
                      remove_unclaimed_object,
                      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.
         */
        CopyList(object_list, unclaimed_object_list, InsertAtStart);

        DestroyList(&result_object_list, FALSE);
        DestroyList(&unclaimed_object_list, FALSE);

    } else {
        /*
         * There was an error in creating the result_object_list or the
         * unclaimed_object_list.
         */
        if (result_object_list == NULL) {
            LOG_CRITICAL("Error allocating memory for the result object list.\n");
        } else {
            DestroyList(&result_object_list, FALSE);
        }
        if (unclaimed_object_list == NULL) {
            LOG_CRITICAL("Error allocating memory for the unclaimed object list.\n");
        } else {
            DestroyList(&unclaimed_object_list, FALSE);
        }
        rc = ENOMEM;
    }

    LOG_PROC_EXIT_INT(rc);
    return rc;
}


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

/*
 * Set the volume pointer in an object.  If the object is an EVMS object, set
 * the volume pointer in its child objects.
 * This function has its parameters structured so that it can be called by
 * ForEachItem() to process a dlist_t of objects.
 */
int set_volume_in_object(ADDRESS object,
                         TAG     object_tag,
                         uint    object_size,
                         ADDRESS object_handle,
                         ADDRESS parameters) {

    logical_volume_t * volume = (logical_volume_t *) parameters;
    storage_object_t * obj = (storage_object_t *) object;

    LOG_PROC_ENTRY();

    /*
     * This function can be called for a child of any object.
     */
    switch (object_tag) {
        case DISK_TAG:
        case SEGMENT_TAG:
        case REGION_TAG:
        case EVMS_OBJECT_TAG:
            {
                /* 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 != volume) {
                    if (obj->feature_header != NULL) {
                        obj->flags |= SOFLAG_FEATURE_HEADER_DIRTY;
                    }
                }

                obj->volume = volume;

                /*
                 * 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 (object_tag == DISK_TAG) {
                    uint parent_count = 0;

                    GetListSize(obj->parent_objects, &parent_count);
                    if (parent_count >1) {
                        obj->volume = NULL;

                    } else {
                        if (parent_count == 1) {
                            storage_object_t * parent_object;
                            uint size;
                            TAG tag;
                            int rc;

                            rc = BlindGetObject(obj->parent_objects,
                                                &size,
                                                &tag,
                                                NULL,
                                                FALSE,
                                                (ADDRESS *) &parent_object);

                            if (rc == DLIST_SUCCESS) {

                                /* If the parent is a segment manager,
                                 * clear the volume pointer in the object.
                                 */
                                if (GetPluginType(parent_object->plugin->id) == EVMS_SEGMENT_MANAGER) {
                                    obj->volume = NULL;
                                }

                            } else {
                                /*
                                 * Couldn't get the parent object.
                                 * Assume the worst; clear the volume pointer.
                                 */
                                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, (volume != 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) {
                    ForEachItem(obj->child_objects,
                                set_volume_in_object,
                                volume,
                                TRUE);
                }
            }
            break;

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

    LOG_PROC_EXIT_INT(DLIST_SUCCESS);
    return DLIST_SUCCESS;
}


/*
 * 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) {

    int rc = 0;
    BOOLEAN found = FALSE;
    plugin_record_t * pPlugRec = NULL;

    LOG_PROC_ENTRY();

    volume->file_system_manager = NULL;

    rc = GoToStartOfList(PluginList);
    if (rc == DLIST_SUCCESS) {
        rc = GetObject(PluginList,
                       sizeof(plugin_record_t),
                       PLUGIN_TAG,
                       NULL,
                       TRUE,
                       (ADDRESS *) &pPlugRec);

        while ((pPlugRec != NULL) && (!found)) {

            if (GetPluginType(pPlugRec->id) == EVMS_FILESYSTEM_INTERFACE_MODULE) {
                found = (pPlugRec->functions.fsim->is_this_yours(volume) == 0);

                if (found) {
                    volume->file_system_manager = pPlugRec;
                    pPlugRec->functions.fsim->get_fs_size(volume, &volume->fs_size);
                    pPlugRec->functions.fsim->get_fs_limits(volume, &volume->min_fs_size,
                                                                    &volume->max_fs_size,
                                                                    &volume->max_vol_size);

                    LOG_DEBUG("Volume %s belongs to %s.\n", volume->name, pPlugRec->short_name);
                }
            }

            if (!found) {
                /* Set default sizes for volumes that don't have an FSIM. */
                volume->fs_size = volume->vol_size;
                volume->min_fs_size = 0;
                volume->max_fs_size = -1;
                volume->max_vol_size = -1;
            }

            rc = GetNextObject(PluginList,
                               sizeof(plugin_record_t),
                               PLUGIN_TAG,
                               (ADDRESS *) &pPlugRec);
        }

        if (volume->file_system_manager == NULL) {
            LOG_DEBUG("No FSIM claimed volume %s.\n", volume->name);
        }
    }

    LOG_PROC_EXIT_VOID();
}


/*
 * The VolumeList is sorted by minor number.  Insert the new volume structure
 * into the appropriate place.
 */
int insert_new_volume_into_volume_list(logical_volume_t * new_volume) {
    int rc = 0;

    LOG_PROC_ENTRY();

    rc = GoToStartOfList(VolumeList);

    if (rc == DLIST_SUCCESS) {
        logical_volume_t * curr_volume;

        rc = GetObject(VolumeList,
                       sizeof(logical_volume_t),
                       VOLUME_TAG,
                       NULL,
                       FALSE,
                       (ADDRESS *) &curr_volume);

        while ((rc == DLIST_SUCCESS) && (curr_volume != NULL) &&
               (curr_volume->minor_number < new_volume->minor_number)) {
            rc = GetNextObject(VolumeList,
                               sizeof(logical_volume_t),
                               VOLUME_TAG,
                               (ADDRESS *) &curr_volume);

        }

        if (rc == DLIST_SUCCESS) {
            ADDRESS trash;
            /*
             * We found the first volume that has a minor number greater than
             * the minor number of the new volume.  Insert the new volume before
             * the current volume.
             */
            rc = InsertObject(VolumeList,
                              sizeof(logical_volume_t),
                              new_volume,
                              VOLUME_TAG,
                              NULL,
                              InsertBefore,
                              FALSE,
                              &trash);

            if (rc != DLIST_SUCCESS) {
                LOG_CRITICAL("Received error code %d when inserting volume %s into the VolumeList.\n", rc, new_volume->name);
            }

        } else {
            if ((rc == DLIST_EMPTY) ||
                (rc == DLIST_END_OF_LIST)) {
                ADDRESS trash;

                /*
                 * The search went OK, but we did not find a volume structure
                 * with a minor number greater than the minor number of the new
                 * volume.  Insert the new volume at the end of the list.
                 */
                rc = InsertObject(VolumeList,
                                  sizeof(logical_volume_t),
                                  new_volume,
                                  VOLUME_TAG,
                                  NULL,
                                  AppendToList,
                                  FALSE,
                                  &trash);
            } else {
                LOG_CRITICAL("Received error code %d when searching the VolumeList.\n", rc);
            }
        }
    }

    LOG_PROC_EXIT_INT(rc);
    return rc;
}


/*
 * Make a new volume structure for an object.
 * The volume name, minor number, and initial flags for the logical_volume_t
 * are provided as parameters.
 * Register the volume name.  Find an FSIM for the volume, if one exists.
 * Put the volume into the VolumeList.  Set the volume pointer in all the
 * child objects of the volume.
 */
int make_volume(storage_object_t * object, char * name, u_int32_t minor, uint flags, u_int64_t serial) {
    int rc = 0;

    LOG_PROC_ENTRY();

    LOG_DEBUG("Request to make volume %s: minor %d; flags %x; serial number %lld.\n", name, minor,flags, serial);

    /* Try to register the volume name. */
    rc = engine_register_name(name);

    if (rc == 0) {
        logical_volume_t * volume = (logical_volume_t *) calloc(1, sizeof(logical_volume_t));

        if (volume != NULL) {

            /*
             * Initial volume size is the object size, minus any feature headers
             * that may be on the object.  Some FSIM's need to know how big the
             * volume is so they know where to look to see if their file system
             * is installed.
             */
            if (object->feature_header == NULL) {
                volume->vol_size = object->size;
            } else {
                volume->vol_size = object->size - (FEATURE_HEADER_SECTORS * 2);
            }

            /* Default - assume the volume can be shrunk to oblivion. */
            volume->min_fs_size = 0;

            /* Default - assume the volume can grow to be as big as possible. */
            volume->max_fs_size = -1;
            volume->max_vol_size = -1;

            volume->object = object;
            volume->minor_number = minor;
            volume->serial_number = serial;
            volume->flags = flags;
            if (object->flags & SOFLAG_READ_ONLY) {
                volume->flags |= VOLFLAG_READ_ONLY;
            }
            strncpy(volume->name, name, sizeof(volume->name) - 1);

            if (discover_in_progress) {
                if (hasa_dev_node(volume->name,volume->minor_number) == 0) {
                    find_fsim_for_volume(volume);

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

                volume->original_fsim = volume->file_system_manager;
                volume->original_vol_size = volume->vol_size;
            }

            rc = insert_new_volume_into_volume_list(volume);

            if (rc == DLIST_SUCCESS) {
                rc = set_volume_in_object(object,
                                          get_tag_for_object(object),
                                          sizeof(storage_object_t),
                                          NULL,
                                          volume);
            }

        } else {
            LOG_CRITICAL("Failed to get memory for a new logical volume structure.\n");
            rc = ENOMEM;
        }
    }

    LOG_PROC_EXIT_INT(rc);
    return rc;
}

/*
 * Get the minor number for an existing volume.  This function searches the list
 * of volume names and their corresponding minor numbers which was obtained from
 * the kernel.
 */
static int get_volume_minor_number(char * name, int * minor_number) {
    int rc = 0;

    LOG_PROC_ENTRY();

    *minor_number = 0;    /* For safety, in case we fail. */

    rc = GoToStartOfList(VolumeDataList);

    if (rc == DLIST_SUCCESS) {
        evms_volume_data_t * vol_data;
        BOOLEAN found = FALSE;

        rc = GetObject(VolumeDataList,
                       sizeof(evms_volume_data_t),
                       VOLUME_DATA_TAG,
                       NULL,
                       FALSE,
                       (ADDRESS *) &vol_data);

        while ((rc == DLIST_SUCCESS) && (vol_data != NULL) && !found) {
            if (strcmp(vol_data->volume_name, name) == 0) {
                LOG_DEBUG("Volume %s is minor number %d.\n", vol_data->volume_name, vol_data->minor);

                *minor_number = vol_data->minor;
                found = TRUE;

            } else {
                rc = GetNextObject(VolumeDataList,
                                   sizeof(evms_volume_data_t),
                                   VOLUME_DATA_TAG,
                                   (ADDRESS *) &vol_data);
            }
        }

        if ((rc == DLIST_EMPTY) || (rc == DLIST_END_OF_LIST) || !found) {
            LOG_WARNING("No match found.\n");
            rc = ENOENT;
        }
    }

    LOG_PROC_EXIT_INT(rc);
    return rc;
}


/*
 * Make a volume for a compatibility object.
 */
static int make_volume_for_compatibility_object(storage_object_t * object) {

    int rc = 0;
    int minor;
    char * pVolName = object->name;


    LOG_PROC_ENTRY();

    /* Prepend the EVMS dev node prefix if it isn't already there. */
    if (strncmp(pVolName, EVMS_DEV_NODE_PATH, strlen(EVMS_DEV_NODE_PATH)) != 0) {
        pVolName = malloc(strlen(pVolName) + strlen(EVMS_DEV_NODE_PATH) + 1);
        if (pVolName != NULL) {
            strcpy(pVolName, EVMS_DEV_NODE_PATH);
            strcat(pVolName, object->name);
        } else {
            LOG_CRITICAL("Could not get memory for building a volume name for object %s.\n", object->name);
            rc = ENOMEM;
        }
    }

    if (rc == 0) {
        u_int32_t flags = VOLFLAG_COMPATIBILITY;

        /* Find the minor number for the volume. */
        rc = get_volume_minor_number(pVolName, &minor);

        if (rc != 0) {
            if (rc == ENOENT) {
                LOG_WARNING("The Engine discovered volume %s but the volume is not exported by the EVMS kernel.\n", pVolName);

                /*
                 * Give the volume an out of bounds minor number so that it
                 * won't conflict with any other compatibility volumes that
                 * might be discovered.  The minor number will be reset by
                 * sync_volume_minors_with_kernel() that is called at the
                 * end of discovery.
                 */
                minor = MAX_EVMS_VOLUMES + 1;

                /* Mark this volume as new and dirty. */
                flags |= (VOLFLAG_NEW | VOLFLAG_DIRTY);

                /*
                 * Clear the error code so that the code below will make a
                 * a volume for this object.
                 */
                rc = 0;
            }
        }

        if (rc == 0) {
            rc = make_volume(object, pVolName, minor, flags, 0);
            if (rc != 0) {
                LOG_WARNING("Could not make volume %s for object %s.  Return code was %d.\n", pVolName, object->name, rc);
            }

        } else {
            LOG_WARNING("Could not find a minor number for volume %s.  Return code was %d.\n", pVolName, rc);
        }
    }

    LOG_PROC_EXIT_INT(rc);
    return rc;
}


/*
 * Compare the minor numbers in two logical_volume_t structures.
 * This function is called by SortList to sort the VolumeList by minor number.
 */
static int compare_volume_minors(ADDRESS Object1,
                                 TAG     Object1Tag,
                                 ADDRESS Object2,
                                 TAG     Object2Tag,
                                 uint  * Error) {
    int result = 0;

    LOG_PROC_ENTRY();

    /* Assume things will go well. */
    *Error = DLIST_SUCCESS;

    if ((Object1Tag == VOLUME_TAG) &&
        (Object2Tag == VOLUME_TAG)) {
        logical_volume_t * volume1 = (logical_volume_t *) Object1;
        logical_volume_t * volume2 = (logical_volume_t *) Object2;

        if (volume1->minor_number == volume2->minor_number) {
            /*
             * This should not happen.  We should not have two volumes with
             * the same minor number.  Even so, we will return a zero
             * indicating that the objects are the same.
             */
            LOG_WARNING("Volumes %s and %s have the same minor number %d.\n", volume1->name, volume2->name, volume1->minor_number);
            result = 0;

        } else {
            if (volume1->minor_number < volume2->minor_number) {
                result = -1;
            } else {
                result = 1;
            }
        }

    } else {
        *Error = DLIST_ITEM_TAG_WRONG;
    }

    LOG_PROC_EXIT_BOOLEAN_INT(result, *Error);
    return result;
}


/*
 * Make sure the Engine's volume structure has its minor number agree with what
 * the kernel says the volume's minor number is.
 * This function has its parameters structured so that it can be called
 * by ForEachItem to process a dlist_t of volumes.
 */
static BOOLEAN sync_volume_minor_with_kernel(ADDRESS   object,
                                             TAG       object_tag,
                                             uint      object_size,
                                             ADDRESS   object_handle,
                                             ADDRESS   parameters,
                                             BOOLEAN * free_memory,
                                             uint    * error) {
    BOOLEAN result = FALSE;
    int rc = 0;

    LOG_PROC_ENTRY();

    /* Never free item memory */
    *free_memory = FALSE;

    if (object_tag == VOLUME_TAG) {
        logical_volume_t * volume = (logical_volume_t *) object;

        BOOLEAN found = FALSE;
        evms_volume_data_t * vol_data;

        rc = GoToStartOfList(VolumeDataList);

        if (rc == DLIST_SUCCESS) {
            rc = GetObject(VolumeDataList,
                           sizeof(evms_volume_data_t),
                           VOLUME_DATA_TAG,
                           NULL,
                           FALSE,
                           (ADDRESS *) &vol_data);

            while ((rc == DLIST_SUCCESS) && (vol_data != NULL) && !found) {

                /*
                 * Do the sync if this volume data structure contains the
                 * information for the volume given in the parameters.  We know
                 * they are for the same volume if the names match.
                 */
                if (strcmp(vol_data->volume_name, volume->name) == 0) {

                    if (vol_data->minor != volume->minor_number) {

                        /*
                         * Update our volume structure to use the kernel's minor
                         * number for the volume.
                         */
                        volume->minor_number = vol_data->minor;

                        /* If the volume is an EVMS volume then mark the volume
                         * dirty so that the new minor number will be saved in
                         * the feature headers on disk.
                         */
                        if (!(volume->flags & VOLFLAG_COMPATIBILITY)) {
                            volume->flags |= VOLFLAG_DIRTY;

                            /* We have stuff to write to disk. */
                            changes_pending = TRUE;
                        }
                    }

                    /*
                     * We found the volume data for the volume.  End the search.
                     */
                    found = TRUE;

                } else {
                    rc = GetNextObject(VolumeDataList,
                                       sizeof(evms_volume_data_t),
                                       VOLUME_DATA_TAG,
                                       (ADDRESS *) &vol_data);
                }
            }

            if (!found) {
                /*
                 * If the volume is an EVMS volume, let it keep its requested
                 * minor number and mark the volume new, dirty, etc.  If the
                 * volume ever gets recognized by the kernel, the Engine will
                 * sync the volume's minor number with what the kernel says it
                 * is.
                 */
                if (!(volume->flags & VOLFLAG_COMPATIBILITY)) {

                    volume->flags |= (VOLFLAG_NEW | VOLFLAG_DIRTY);
                    changes_pending = TRUE;

                } else {
                    /* It's a compatibility volume.
                     * First, remove the volume from the VolumeList so that it
                     * won't be found by the functions that search for unused
                     * minor numbers.
                     */
                    DeleteObject(VolumeList, volume);

                    rc = get_compatibility_minor_number(&volume->minor_number);

                    if (rc == 0) {
                        volume->flags |= (VOLFLAG_NEW | VOLFLAG_DIRTY);
                        changes_pending = TRUE;
                        rc = insert_new_volume_into_volume_list(volume);

                        if (rc != 0) {
                            LOG_WARNING("Error code %d when inserting volume \"&s\" into the VolumeList.\n", rc, volume->name);
                        }

                    } else {
                        LOG_WARNING("Error code %d when getting a minor number for volume \"%s\".  The volume will be removed and its storage object, %s,  made available.\n", rc, volume->name, volume->object->name);
                    }

                    if (rc != 0) {
                        LOG_DEBUG("Removing volume \"%s\" due to an error when attempting to set its minor number.\n", volume->name);

                        /* Set the volume pointer to NULL in all the objects. */
                        set_volume_in_object(volume->object,
                                             get_tag_for_object(volume->object),
                                             sizeof(storage_object_t),
                                             NULL,
                                             NULL);

                        /* Free the logical_volume_t structure. */
                        free(volume);

                        /*
                         * Return TRUE so that the volume is removed from the
                         * VolumeList.
                         */
                        result = TRUE;
                    }
                }
            }
        }

        /* If we are keeping the volume... */
        if (!result) {
            /* Check if there is a dev node for the volume. */
            rc = hasa_dev_node(volume->name, volume->minor_number);
            if (rc != 0) {
                if ((rc == ENOENT) || (rc == EEXIST)) {

                    /*
                     * dev node does not exist or it does exist and it has the
                     * wrong minor number.
                     */
                    volume->flags |= VOLFLAG_NEEDS_DEV_NODE;
                    changes_pending = TRUE;
                }

            } else {
                /* Volume has a dev node.  Make sure the flag is off. */
                volume->flags &= ~VOLFLAG_NEEDS_DEV_NODE;
            }
        }
    }

    /* Always return no error so that the PruneList processor doesn't stop. */
    *error = DLIST_SUCCESS;
    LOG_PROC_EXIT_BOOLEAN(result);
    return result;
}


/*
 * Make sure that all of the minor numbers that the Engine has for volumes
 * agree with the minor numbers that the kernel has associated with the
 * volumes.
 */
int sync_volume_minors_with_kernel(void) {
    int rc = 0;

    LOG_PROC_ENTRY();

    /*
     * Loop through the volume list and make sure each volume's minor number is
     * in sync with the kernel.  Remove volumes for which we cannot assign a
     * minor number.
     */
    PruneList(VolumeList,
                sync_volume_minor_with_kernel,
                NULL);

    /* Sort the VolumeList by minor number. */
    rc = SortList(VolumeList, compare_volume_minors);

    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_int32_t               volume_system_id;
    u_int64_t               volume_serial;
    char                    volume_name[EVMS_VOLUME_NAME_SIZE];
    dlist_t                 objects;
    struct volume_objects * next;
} volume_objects_t;


/*
 * Get the tag used to put this object in a dlist_t.
 */
TAG get_tag_for_object(storage_object_t * object) {

    TAG tag = 0;

    LOG_PROC_ENTRY();

    switch (object->object_type) {
        case DISK:
            tag = DISK_TAG;
            break;

        case SEGMENT:
            tag = SEGMENT_TAG;
            break;

        case REGION:
            tag = REGION_TAG;
            break;

        case EVMS_OBJECT:
            tag = EVMS_OBJECT_TAG;
            break;

        default:
            break;

    }

    LOG_PROC_EXIT_INT(tag);
    return tag;
}


/*
 * The structure of the parameters passed to extract_plugin_object() in its
 * Parameters pointer.
 */
typedef struct extract_object_parms_s {
    plugin_id_t id;
    dlist_t     plugin_object_list;
} extract_object_parms_t;


/*
 * If this object is managed by the plug-in specified in the parameters, then
 * put it into the plugin_object_list and return TRUE to remove it from the
 * current list.
 * This function has its parameters structured so that it can be called by
 * PruneList.
 */
static BOOLEAN extract_plugin_object(ADDRESS   Object,
                                     TAG       ObjectTag,
                                     uint      ObjectSize,
                                     ADDRESS   ObjectHandle,
                                     ADDRESS   Parameters,
                                     BOOLEAN * FreeMemory,
                                     uint    * Error) {
    BOOLEAN result = FALSE;
    storage_object_t * object = (storage_object_t *) Object;
    extract_object_parms_t * parms = (extract_object_parms_t *) Parameters;

    LOG_PROC_ENTRY();

    /* Never free the object memory. */

    *FreeMemory = FALSE;

    /*
     * If this object has the same plug-in ID as the one we're looking for
     * (specified in the parms) then add it to the list specified in the parms
     * and return TRUE so that it is removed from the main list that is being
     * traversed.  Else, don't put the object in the list and return FALSE so
     * that the object remains in the main list.
     */

    if (object->feature_header->feature_id == parms->id) {
        ADDRESS handle;

        *Error = InsertObject(parms->plugin_object_list,
                              sizeof(storage_object_t),
                              Object,
                              EVMS_OBJECT_TAG,
                              NULL,
                              AppendToList,
                              FALSE,
                              &handle);

        result = TRUE;

    } else {
        *Error = DLIST_SUCCESS;
    }

    LOG_PROC_EXIT_BOOLEAN_INT(result, *Error);
    return result;
}


/*
 * 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 = calculate_CRC(0xFFFFFFFFL, 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) &&
                        (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 = malloc(sizeof(evms_feature_header_t) * 2);
    primary_fh = malloc(sizeof(evms_feature_header_t));

    if ((primary_fh != NULL) &&
        (secondary_fh != NULL)) {
        rc = object->plugin->functions.plugin->read(object,
                                                    object->size - (FEATURE_HEADER_SECTORS * 2),
                                                    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 - FEATURE_HEADER_SECTORS,
                                                                 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 - (FEATURE_HEADER_SECTORS * 2),
                                                                   FEATURE_HEADER_SECTORS * 2,
                                                                   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) {
            free(primary_fh);
            primary_fh = NULL;
        }
        if (*secondary_rc != 0) {
            free(secondary_fh);
            secondary_fh = NULL;
        }

    } else {
        LOG_CRITICAL("Error allocating memory to read in a feature header.\n");
        if (primary_fh != NULL) {
            free(primary_fh);
            primary_fh = NULL;
        }
        if (secondary_fh != NULL) {
            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.
 * This function has its parameters structured so that it can be called by
 * ForEachItem to process a list of objects.
 */
static int get_feature_header(ADDRESS Object,
                              TAG     ObjectTag,
                              uint    ObjectSize,
                              ADDRESS ObjectHandle,
                              ADDRESS Parameters) {

    int rc = 0;
    int secondary_rc = 0;
    int primary_rc = 0;
    storage_object_t * object = (storage_object_t *) Object;
    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", object->name);

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

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

        switch (secondary_rc) {
            case 0:
                switch (primary_rc) {
                    case 0:
                        /*
                         * 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("Secondary and primary feature headers match.\n");
                            object->feature_header = primary_fh;
                            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 (secondary_fh->sequence_number > primary_fh->sequence_number) {
                                LOG_WARNING("Secondary feature header for object %s has a higher sequence number (%lld) than the primary feature header (%lld).  Using the secondary copy.\n", secondary_fh->object_name, secondary_fh->sequence_number, primary_fh->sequence_number);
                                object->feature_header = secondary_fh;
                                free(primary_fh);

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

                            object->flags |= SOFLAG_FEATURE_HEADER_DIRTY;
                        }
                        break;

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

                    default:
                        /*
                         * We got the secondary feature header OK but some
                         * unexpected error occurred when reading the primary
                         * feature header.  Free the secondary feature header
                         * and return the error code from the primary feature
                         * header to the caller.
                         */
                        LOG_WARNING("Error code %d when reading the primary copy of feature header on object %s.\n", rc, object->name);
                        if (secondary_fh != NULL) {
                            free(secondary_fh);
                        }

                        rc = primary_rc;
                        break;
                }
                break;

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

                    case ENOENT:
                    case EINVAL:
                        /*
                         * Both the secondary and primary feature headers either
                         * had a bad signature or a bad CRC.  Assume that there
                         * are no feature headers on this object.  Return
                         * success.
                         */
                        LOG_DEBUG("No feature headers found on object.\n");
                        rc = 0;
                        break;

                    default:
                        /*
                         * The secondary feature header either has a bad
                         * signature or the CRC is bad.  Regardless, we got an
                         * unexpected error code when reading the primary
                         * feature header.  Return the primary error code to the
                         * caller.
                         */
                        rc = primary_rc;
                        break;
                }
                break;

            default:
                /*
                 * We got an unexpected error when reading the secondary feature
                 * header.  Return the error code to the caller.  Free the
                 * primary feature header if we got one.
                 */
                LOG_WARNING("Error code %d when reading the secondary copy of feature header on object %s.\n", rc, object->name);
                if (primary_fh != NULL) {
                    free(primary_fh);
                }

                rc = secondary_rc;
                break;
        }
    }

    LOG_PROC_EXIT_INT(rc);
    return rc;
}


static int get_greatest_object_depth(ADDRESS object,
                                     TAG     object_tag,
                                     uint    object_size,
                                     ADDRESS object_handle,
                                     ADDRESS parameters) {

    storage_object_t * obj = (storage_object_t *) object;
    u_int32_t * object_depth = (u_int32_t *) parameters;

    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_INT(DLIST_SUCCESS);
    return DLIST_SUCCESS;
}


static int discover_objects_by_plugin(dlist_t object_list, dlist_t resulting_object_list) {

    int rc = 0;

    dlist_t plugin_object_list = CreateList();
    dlist_t new_object_list = CreateList();

    LOG_PROC_ENTRY();

    if ((plugin_object_list != NULL) &&
        (new_object_list != NULL)) {
        /*
         * 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.
         */
        while (!ListEmpty(object_list) && (rc == 0)) {
            uint               size;
            TAG                tag;
            storage_object_t * obj;

            rc = BlindExtractObject(object_list,
                                    &size,
                                    &tag,
                                    NULL,
                                    (ADDRESS *) &obj);

            if ((rc == DLIST_SUCCESS) && (obj != NULL)) {
                plugin_record_t * feature;

                /*
                 * 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) {
                            ADDRESS trash;

                            /* Clear out the plugin_object_list */
                            DeleteAllItems(plugin_object_list, FALSE);

                            /*
                             * Put the extracted object on the list of objects
                             * for the plug-in.
                             */
                            rc = InsertObject(plugin_object_list,
                                              size,
                                              obj,
                                              tag,
                                              NULL,
                                              AppendToList,
                                              FALSE,
                                              &trash);

                            if (rc == DLIST_SUCCESS) {
                                extract_object_parms_t parms;

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

                                /*
                                 * Pull the other objects that have the same
                                 * feature ID off the list and add them to the
                                 * plugin_object_list.
                                 */
                                parms.id = feature->id;
                                parms.plugin_object_list = plugin_object_list;

                                rc = PruneList(object_list,
                                               extract_plugin_object,
                                               (ADDRESS) &parms);

                                if (rc == DLIST_SUCCESS) {

                                    rc = feature->functions.plugin->discover(plugin_object_list, new_object_list, TRUE);

                                    /*
                                     * Remove any corrupt objects from the
                                     * list.
                                     */
                                    PruneList(new_object_list,
                                              remove_corrupt_object,
                                              NULL);

                                    /*
                                     * If discover() succeeded, put the new
                                     * objects on the resulting_object_list.
                                     */
                                    if (rc == 0) {
                                        CopyList(resulting_object_list,
                                                 new_object_list,
                                                 AppendToList);
                                    }
                                }

                                /*
                                 * We have handled any errors.  Clear the
                                 * error code so that the loop will not
                                 * terminate.
                                 */
                                rc = 0;
                            }

                        } 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 *) calloc(1, sizeof(logical_volume_t));

                            if (volume != NULL) {

                                volume->minor_number = obj->feature_header->volume_system_id;
                                volume->serial_number = obj->feature_header->volume_serial_number;

                                /*
                                 * The feature header contains a relative volume
                                 * name.  Prepend the dev node path to the
                                 * volume name.
                                 */
                                strcpy(volume->name, EVMS_DEV_NODE_PATH);
                                strncat(volume->name, obj->feature_header->volume_name, EVMS_VOLUME_NAME_SIZE);

                                /* (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);
                }
            }
        }

    } else {
        /* Failed to create one or more lists. */
        if (plugin_object_list == NULL) {
            LOG_CRITICAL("Error allocating memory for the plugin_object_list.\n");
        }
        if (new_object_list == NULL) {
            LOG_CRITICAL("Error allocating memory for the new_object_list.\n");
        }
        rc = ENOMEM;
    }

    if (plugin_object_list != NULL) DestroyList(&plugin_object_list, FALSE);
    if (new_object_list != NULL) DestroyList(&new_object_list, FALSE);

    LOG_PROC_EXIT_INT(rc);
    return rc;
}


typedef struct extract_by_depth_parms_s {
    dlist_t     result_list;
    u_int32_t   depth;
} extract_by_depth_parms_t;


/*
 * 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.
 * This function has its parameters structured so that it can be called by the
 * PruneList() dlist processor on a list of objects.
 */
static BOOLEAN extract_object_by_depth(ADDRESS   object,
                                       TAG       object_tag,
                                       uint      object_size,
                                       ADDRESS   object_handle,
                                       ADDRESS   parameters,
                                       BOOLEAN * free_memory,
                                       uint    * error) {
    BOOLEAN result = FALSE;
    storage_object_t * obj = (storage_object_t *) object;
    extract_by_depth_parms_t * parms = (extract_by_depth_parms_t *) parameters;

    /* Never free item memory. */
    *free_memory = FALSE;

    if (obj->feature_header != NULL) {
        if (obj->feature_header->object_depth == parms->depth) {
            ADDRESS trash;

            *error = InsertObject(parms->result_list,
                                  sizeof(storage_object_t),
                                  object,
                                  get_tag_for_object(obj),
                                  NULL,
                                  AppendToList,
                                  FALSE,
                                  &trash);
            result = TRUE;
        }
    }

    LOG_PROC_EXIT_BOOLEAN_INT(result, *error);
    return result;

}


/*
 * 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) {

    int rc = 0;
    extract_by_depth_parms_t extract_parms;
    dlist_t                  object_list = CreateList();

    LOG_PROC_ENTRY();

    extract_parms.depth = 0;
    extract_parms.result_list = object_list;

    if (object_list != NULL) {

        /* Find the greatest object_depth of the objects in the list. */
        ForEachItem(pVolObj->objects,
                    get_greatest_object_depth,
                    &extract_parms.depth,
                    TRUE);

        while ((rc == DLIST_SUCCESS) && (extract_parms.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.
             */
            ForEachItem(pVolObj->objects,
                        get_feature_header,
                        NULL,
                        TRUE);

            /*
             * Clear out the working object_list for this iteration through
             * the loop.
             */
            DeleteAllItems(object_list, FALSE);

            /*
             * Build a list of objects that have the same object_depth
             * in the feature header.  The objects are extracted from the
             * pVolObj->objects list.
             */
            rc = PruneList(pVolObj->objects,
                           extract_object_by_depth,
                           &extract_parms);

            if (rc == DLIST_SUCCESS) {

                /*
                 * 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);

                if (rc == 0) {
                    /* Move to the next depth. */
                    extract_parms.depth--;
                }
            }
        }

        if (rc == DLIST_SUCCESS) {

            /*
             * If there is only one item in the list, then it is a candidate for
             * becoming a volume.
             */
            uint list_size = 0;

            GetListSize(pVolObj->objects, &list_size);

            if (list_size == 1) {
                storage_object_t * obj;

                rc = GetObject(pVolObj->objects,
                               sizeof(storage_object_t),
                               EVMS_OBJECT_TAG,
                               NULL,
                               FALSE,
                               (ADDRESS *) &obj);

                if (rc == DLIST_SUCCESS) {

                    /*
                     * The feature header for this object is hanging off of
                     * its child object(s).  Get a child object so that we
                     * can examine the feature header.
                     */
                    uint size;
                    TAG  tag;
                    storage_object_t * child_object;

                    rc = BlindGetObject(obj->child_objects,
                                        &size,
                                        &tag,
                                        NULL,
                                        FALSE,
                                        (ADDRESS *) &child_object);

                    if (rc == DLIST_SUCCESS) {

                        /*
                         * A flag in the feature header says whether this
                         * object should be made into a volume or left as
                         * an object.
                         */
                        if (!(child_object->feature_header->flags & EVMS_VOLUME_DATA_OBJECT)) {

                            /*
                             * The feature header contains a relative volume name.
                             * Prepend the dev node path to the volume name.
                             */
                            char volume_name[EVMS_VOLUME_NAME_SIZE];
                            int minor;
                            u_int32_t flags = 0;
                            strcpy(volume_name, EVMS_DEV_NODE_PATH);
                            strncat(volume_name, pVolObj->volume_name, EVMS_VOLUME_NAME_SIZE);

                            /*
                             * Check if the kernel is exporting this volume.
                             * If not, mark the volume new and dirty.
                             */
                            if (get_volume_minor_number(volume_name, &minor) != 0) {
                                flags |= (VOLFLAG_NEW | VOLFLAG_DIRTY);
                            }

                            /* Make a volume for the object. */
                            rc = make_volume(obj,
                                             volume_name,
                                             pVolObj->volume_system_id,
                                             flags,
                                             pVolObj->volume_serial);
                        }
                    }
                }
            }
        }

        DestroyList(&object_list, FALSE);

    } else {
        /* We couldn't create an object_list. */
        rc = ENOMEM;
    }

    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 *) malloc(sizeof(volume_objects_t));

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

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

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

    if (rc == 0) {
        ADDRESS handle;

        LOG_DEBUG("Add object %s to volume object list for volume %s.\n", obj->name, pVolObj->volume_name);
        rc = InsertObject(pVolObj->objects,
                          sizeof(storage_object_t),
                          obj,
                          get_tag_for_object(obj),
                          NULL,
                          AppendToList,
                          TRUE,
                          &handle);
    }

    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(ADDRESS object,
                          TAG     object_tag,
                          uint    object_size,
                          ADDRESS object_handle,
                          ADDRESS parameters) {
    int rc = 0;
    storage_object_t * obj = (storage_object_t *) object;
    volume_objects_t * * pVolObjList = (volume_objects_t * *) parameters;

    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_volume_for_compatibility_object(obj);

    } 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)) {
                /*
                 * The feature header contains a relative volume name.
                 * Prepend the dev node path to the volume name.
                 */
                char volume_name[EVMS_VOLUME_NAME_SIZE];
                int minor;
                u_int32_t flags = 0;
                strcpy(volume_name, EVMS_DEV_NODE_PATH);
                strncat(volume_name, obj->feature_header->volume_name, EVMS_VOLUME_NAME_SIZE);

                /*
                 * Check if the kernel is exporting this volume.
                 * If not, mark the volume new and dirty.
                 */
                if (get_volume_minor_number(volume_name, &minor) != 0) {
                    flags |= (VOLFLAG_NEW | VOLFLAG_DIRTY);
                }

                /* Make a volume for the object. */
                rc = make_volume(obj,
                                 volume_name,
                                 obj->feature_header->volume_system_id,
                                 flags,
                                 obj->feature_header->volume_serial_number);

                if (rc != 0) {
                    LOG_WARNING("Error when making a volume for object %s.  Return code was %d.\n", obj->name, rc);

                    /* Don't let errors stop the discovery process. */
                    rc = 0;
                }
            }

        } 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;
}


/*
 * Check if an object has the data_type of DATA_TYPE.  If so, return FALSE to
 * leave it on the list.  If not, return TRUE so that it will be removed from
 * the list.
 * This function has its parameters structured so that it can be called by the
 * PruneList() dlist processor on a list of objects.
 */
static BOOLEAN remove_non_data_object(ADDRESS   object,
                                      TAG       object_tag,
                                      uint      object_size,
                                      ADDRESS   object_handle,
                                      ADDRESS   parameters,
                                      BOOLEAN * free_memory,
                                      uint    * error) {
    BOOLEAN result = FALSE;
    int rc = 0;
    storage_object_t * obj = (storage_object_t *) object;

    LOG_PROC_ENTRY();

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

    /* Never free the item memory. */
    *free_memory = FALSE;

    /*
     * If the object is a free space object, return TRUE to remove it from this
     * list.
     */
    if (obj->data_type == FREE_SPACE_TYPE) {
        result = TRUE;

    } else {
        /*
         * If the object is not a data object (e.g., it is a meta data object)
         * return TRUE to remove it from this list.
         */
        if (obj->data_type != DATA_TYPE) {
            result = TRUE;
        }
    }

    LOG_PROC_EXIT_BOOLEAN_INT(result, rc);
    *error = rc;
    return result;
}


/* A place to read in data to check if it is stop data. */
static evms_feature_header_t stop_data;


/*
 * Check if an object has stop data on it.  If so, return TRUE so that it will
 * be removed from its current list.  Else, return FALSE so that the object
 * stays in the list.
 * This function has its parameters structured so that it can be called by the
 * PruneList() dlist processor on a list of objects.
 */
static BOOLEAN remove_stop_data_object(ADDRESS   object,
                                       TAG       object_tag,
                                       uint      object_size,
                                       ADDRESS   object_handle,
                                       ADDRESS   parameters,
                                       BOOLEAN * free_memory,
                                       uint    * error) {
    BOOLEAN result = FALSE;
    int rc = 0;
    storage_object_t * obj = (storage_object_t *) object;

    LOG_PROC_ENTRY();

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

    /* Never free the item memory. */
    *free_memory = FALSE;

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

    if (rc == 0) {
        /* Do some validation. */

        if (DISK_TO_CPU32(stop_data.signature) == EVMS_FEATURE_HEADER_SIGNATURE) {
            u_int32_t old_crc;
            u_int32_t new_crc;

            old_crc = DISK_TO_CPU32(stop_data.crc);
            stop_data.crc = 0;
            new_crc = calculate_CRC(0xFFFFFFFFL, &stop_data, sizeof(evms_feature_header_t));

            if ((old_crc == new_crc) || (old_crc == EVMS_MAGIC_CRC)) {
                if (stop_data.flags & EVMS_VOLUME_DATA_STOP) {
                    result = TRUE;
                }
            }
        }
    }

    /* Don't allow an error to kill the processing of the list. */
    rc = 0;

    LOG_PROC_EXIT_BOOLEAN_INT(result, rc);
    *error = rc;
    return result;
}


static int discover_evms_objects(dlist_t object_list) {
    int rc = 0;

    dlist_t result_object_list;

    LOG_PROC_ENTRY();

    result_object_list = CreateList();

    if (result_object_list != NULL) {

        /*
         * Pull any meta data objects off the list.
         * Pull any free space objects off the object_list.
         */
        rc = PruneList(object_list,
                       remove_non_data_object,
                       NULL);

        if (rc == DLIST_SUCCESS) {

            /* Remove from the list any objects with stop data on them. */
            rc = PruneList(object_list,
                           remove_stop_data_object,
                           NULL);

            if (rc == DLIST_SUCCESS) {
                volume_objects_t * VolObjList = NULL;


                /*
                 * 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.
                 */
                ForEachItem(object_list,
                            get_feature_header,
                            NULL,
                            TRUE);

                /* Process the objects on the object_list. */
                rc = ForEachItem(object_list,
                                 process_object,
                                 &VolObjList,
                                 TRUE);

                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);
                    }

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

                        DestroyList(&(pVolObj->objects), FALSE);
                        free(pVolObj);
                    }
                }
            }
        }

    } else {
        rc = ENOMEM;
    }

    LOG_PROC_EXIT_INT(rc);
    return rc;
}


static BOOLEAN extract_associative_object(ADDRESS   Object,
                                          TAG       ObjectTag,
                                          uint      ObjectSize,
                                          ADDRESS   ObjectHandle,
                                          ADDRESS   Parameters,
                                          BOOLEAN * FreeMemory,
                                          uint    * Error) {

    storage_object_t * obj = (storage_object_t *) Object;
    BOOLEAN result = FALSE;
    int rc = 0;

    LOG_PROC_ENTRY();

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

    /* Never free the item memory. */

    *FreeMemory = FALSE;

    /*
     * Check if a feature header even exists since this might be a topmost
     * object that is not a volume.
     */
    if (obj->feature_header != NULL) {

        /*
         * Filter out objects that have an associative class feature for a
         * parent.
         */
        if (GetPluginType(obj->feature_header->feature_id) == EVMS_ASSOCIATIVE_FEATURE) {

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

            logical_volume_t * volume = (logical_volume_t *) obj->consuming_private_data;

            if (volume != NULL) {
                plugin_record_t * feature;

                obj->consuming_private_data = NULL;

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

                if (rc == 0) {

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

                    dlist_t object_list = CreateList();

                    if (object_list != NULL) {
                        ADDRESS handle;

                        rc = InsertObject(object_list,
                                          sizeof(storage_object_t),
                                          obj,
                                          EVMS_OBJECT_TAG,
                                          NULL,
                                          AppendToList,
                                          FALSE,
                                          &handle);

                        if (rc == DLIST_SUCCESS) {
                            dlist_t new_object_list = CreateList();

                            if (new_object_list != NULL) {
                                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. */
                                PruneList(new_object_list,
                                          remove_corrupt_object,
                                          NULL);

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

                                    rc = GetObject(new_object_list,
                                                   sizeof(storage_object_t),
                                                   EVMS_OBJECT_TAG,
                                                   NULL,
                                                   FALSE,
                                                   (ADDRESS *) &new_object);

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

                                        int minor;
                                        u_int32_t flags = 0;

                                        /*
                                         * Check if the kernel is exporting this
                                         * volume.  If not, mark the volume new
                                         * and dirty.
                                         */
                                        if (get_volume_minor_number(volume->name, &minor) != 0) {
                                            flags |= (VOLFLAG_NEW | VOLFLAG_DIRTY);
                                        }

                                        /* Make a volume for the object. */
                                        rc = make_volume(new_object,
                                                         volume->name,
                                                         volume->minor_number,
                                                         flags,
                                                         volume->serial_number);
                                    }

                                    /*
                                     * Return TRUE so that this object will be
                                     * removed from the list.
                                     */
                                    result = TRUE;

                                    if ((rc == DLIST_END_OF_LIST) ||
                                        (rc == DLIST_EMPTY)) {
                                        rc = DLIST_SUCCESS;
                                    }

                                } 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;
                                    }
                                }

                                DestroyList(&new_object_list, FALSE);

                            } else {
                                LOG_CRITICAL("Error allocating memory for a resulting objects list.\n");
                                rc = ENOMEM;
                            }
                        }

                        DestroyList(&object_list, FALSE);

                    } else {
                        LOG_CRITICAL("Error allocating memory for an input objects list.\n");
                        rc = ENOMEM;
                    }

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

                /* Release the temporary volume structure. */
                free(volume);

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

        } else {
            LOG_DEBUG("Object's feature header is not for an associative class feature.\n");
        }

    } else {
        LOG_DEBUG("Object does not have a feature header.\n");
    }


    LOG_PROC_EXIT_BOOLEAN_INT(result, rc);
    *Error = rc;
    return result;
}


static int discover_associative_features(dlist_t object_list) {
    int rc = 0;

    LOG_PROC_ENTRY();

    /*
     * Use PruneList to run the ObjectList checking for associative class
     * features, calling their Discover() functions, and making them into
     * volumes if necessary.  If the object is made into a volume it will be
     * removed from the ObjectList.
     */

    rc = PruneList(object_list,
                   extract_associative_object,
                   NULL);

    LOG_PROC_EXIT_INT(rc);
    return rc;
}


/*
 * Check if a given volume name matches the volume name in a logical_volume_t
 * in the Engine.
 * This function has its parameters structured so that it can be called by
 * the ForEachItem() dlist processor.
 * This function returns 0 if the name does not match, EEXIST if the name does
 * match.  A zero return code will allow ForEachItem() to continue processing
 * the list.  A non-zero return code stops ForEachItem(), which is what we want
 * to do if the volume name matches.
 */
int is_engine_volume(ADDRESS object,
                     TAG     object_tag,
                     uint    object_size,
                     ADDRESS object_handle,
                     ADDRESS parameters) {
    int rc = 0;

    LOG_PROC_ENTRY();

    /* Safety check */
    if (object_tag == VOLUME) {
        logical_volume_t * vol = (logical_volume_t *) object;
        char * volume_name = (char *) parameters;

        LOG_DEBUG("Comparing %s and %s.\n", vol->name, volume_name);

        if (strcmp(vol->name, volume_name) == 0) {
            rc = EEXIST;
        }
    }

    LOG_PROC_EXIT_INT(rc);
    return rc;
}


int check_kernel_volume(ADDRESS object,
                        TAG     object_tag,
                        uint    object_size,
                        ADDRESS object_handle,
                        ADDRESS parameters) {
    int rc = 0;

    LOG_PROC_ENTRY();

    /* Safety check */
    if (object_tag == VOLUME_DATA_TAG) {
        evms_volume_data_t * volume_data = (evms_volume_data_t *) object;

        LOG_DEBUG("Checking volume %s\n", volume_data->volume_name);

        rc = ForEachItem(VolumeList,
                         is_engine_volume,
                         volume_data->volume_name,
                         TRUE);

        /*
         * is_engine_volume() returns 0 when it doesn't find a match.  Returning
         * 0 enables ForEachItem() to continue processing the list.
         * is_engine_volume() returns non-zero when it finds a match, which
         * aborts the ForEachItem() which halts the search.  Getting 0 from
         * ForEachItem() means an Engine volume with the same name was not
         * found.
         */
        if (rc == 0) {
            logical_volume_t * vol = calloc(1, sizeof(logical_volume_t));

            engine_user_message(NULL, NULL,
                                "WARNING: Volume \"%s\" was exported by the EVMS kernel but was not discovered by the EVMS Engine.  "
                                "The kernel's in memory copy of the volume is scheduled to be deleted when changes are committed.  "
                                "Deleting the kernel's in memory copy of the volume will not change any data on the disks.  "
                                "If the volume truly exists, the kernel will discover it after the changes have been committed.\n",
                                volume_data->volume_name);

            /*
             * Fill in the logical_volume_t structure with the necessary
             * information so that it can be added to the HardVolumeDeleteList.
             */
            if (vol != NULL) {
                ADDRESS trash;

                strcpy(vol->name, volume_data->volume_name);
                vol->minor_number = volume_data->minor;

                rc = InsertObject(HardVolumeDeleteList,
                                  sizeof(logical_volume_t),
                                  vol,
                                  VOLUME_TAG,
                                  NULL,
                                  AppendToList,
                                  FALSE,
                                  &trash);

                if (rc == DLIST_SUCCESS) {
                    /*
                     * The volume was successfully added to the
                     * HardVolumeDeletList.  Mark that changes are pending
                     * so that UIs can know there is something to be
                     * committed.
                     */
                    changes_pending = TRUE;

                } else {

                    LOG_WARNING("Error code %d when inserting volume %s into the HardVolumeDelete list.\n", rc, vol->name);
                    free(vol);

                    /*
                     * Clear the error code so that it doesn't stop the
                     * ForEachItem loop.
                     */
                    rc = 0;
                }

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

        } else {
            /*
             * If the ForEachItem on is_engine_volume returned EEXIST that
             * means a match was found, which is a good thing.  Set rc to zero
             * so that the ForEachItem on the VolumeDataList does not abort.
             */
            if (rc == EEXIST) {
                rc = 0;
            }
        }
    }

    LOG_PROC_EXIT_INT(rc);
    return rc;
}


/*
 * Run the kernel volume list and make sure we discovered the volume in the
 * Engine.  If the Engine did not discover the volume, display a warning
 * message that asks the user if the volume should be deleted or kept.
 */
void warn_if_kernel_volume_but_not_engine_volume(void) {

    LOG_PROC_ENTRY();

    ForEachItem(VolumeDataList,
                check_kernel_volume,
                NULL,
                TRUE);

    LOG_PROC_EXIT_VOID();
}


int do_discovery(void) {

    int rc = 0;
    dlist_t object_list;
    struct timezone tz;
    struct timeval discover_start_time;
    struct timeval discover_stop_time;
    struct timeval discover_time;

    gettimeofday(&discover_start_time, &tz);

    LOG_PROC_ENTRY();

    object_list = CreateList();
    if (object_list != NULL) {

        discover_in_progress = TRUE;

        rc = discover_logical_disks(object_list);
        if (rc == 0) {

            rc = discover_segments(object_list);
            if (rc == 0) {

                rc = discover_regions(object_list);
                if (rc == 0) {

                    rc = discover_evms_objects(object_list);
                    if (rc == 0) {

                        dlist_t top_object_list;

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

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

                                /* Display a warning message for any volume
                                 * exported by the EVMS kernel but was not
                                 * discovered by the Engine.
                                 */
                                warn_if_kernel_volume_but_not_engine_volume();

                                /*
                                 * Make sure our volume minors are in
                                 * sync with the kernel's view of volume
                                 * minors.
                                 */
                                rc = sync_volume_minors_with_kernel();
                            }

                            DestroyList(&top_object_list, FALSE);
                        }
                    }
                }
            }
        }

        discover_in_progress = FALSE;

        DestroyList(&object_list, FALSE);

    } else {
        rc = ENOMEM;
    }

    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;
    }
    LOG_DEBUG("Discovery took %d.%06d seconds.\n", discover_time.tv_sec, discover_time.tv_usec);

    LOG_PROC_EXIT_INT(rc);
    return rc;
}