389-ds-base/ldap/servers/slapd/back-ldbm/import-threads.c

/** BEGIN COPYRIGHT BLOCK
 * 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; version 2 of the License.
 * 
 * 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.
 * 
 * In addition, as a special exception, Red Hat, Inc. gives You the additional
 * right to link the code of this Program with code not covered under the GNU
 * General Public License ("Non-GPL Code") and to distribute linked combinations
 * including the two, subject to the limitations in this paragraph. Non-GPL Code
 * permitted under this exception must only link to the code of this Program
 * through those well defined interfaces identified in the file named EXCEPTION
 * found in the source code files (the "Approved Interfaces"). The files of
 * Non-GPL Code may instantiate templates or use macros or inline functions from
 * the Approved Interfaces without causing the resulting work to be covered by
 * the GNU General Public License. Only Red Hat, Inc. may make changes or
 * additions to the list of Approved Interfaces. You must obey the GNU General
 * Public License in all respects for all of the Program code and other code used
 * in conjunction with the Program except the Non-GPL Code covered by this
 * exception. If you modify this file, you may extend this exception to your
 * version of the file, but you are not obligated to do so. If you do not wish to
 * provide this exception without modification, you must delete this exception
 * statement from your version and license this file solely under the GPL without
 * exception. 
 * 
 * 
 * Copyright (C) 2001 Sun Microsystems, Inc. Used by permission.
 * Copyright (C) 2005 Red Hat, Inc.
 * All rights reserved.
 * END COPYRIGHT BLOCK **/
/*
 * the threads that make up an import:
 * producer (1)
 * foreman (1)
 * worker (N: 1 for each index)
 *
 * a wire import (aka "fast replica" import) won't have a producer thread.
 */

#include "back-ldbm.h"
#include "vlv_srch.h"
#include "import.h"
#ifdef XP_WIN32
#define STDIN_FILENO 0
#endif


static struct backentry *import_make_backentry(Slapi_Entry *e, ID id)
{
    struct backentry *ep = backentry_alloc();

    if (NULL != ep) {
	ep->ep_entry = e;
	ep->ep_id = id;
    }
    return ep;
}

static void import_decref_entry(struct backentry *ep)
{
    PR_AtomicDecrement(&(ep->ep_refcnt));
    PR_ASSERT(ep->ep_refcnt >= 0);
}

/* generate uniqueid if requested */
static void import_generate_uniqueid(ImportJob *job, Slapi_Entry *e)
{
    const char *uniqueid = slapi_entry_get_uniqueid(e);
    int rc;

    if (!uniqueid && (job->uuid_gen_type != SLAPI_UNIQUEID_GENERATE_NONE)) {
        char *newuniqueid;

        /* generate id based on dn */
        if (job->uuid_gen_type == SLAPI_UNIQUEID_GENERATE_NAME_BASED) {
            char *dn = slapi_entry_get_dn(e);

            rc = slapi_uniqueIDGenerateFromNameString(&newuniqueid,
                job->uuid_namespace, dn, strlen(dn));
        } else {
            /* time based */
            rc = slapi_uniqueIDGenerateString(&newuniqueid);        
        }

        if (rc == UID_SUCCESS) {
            slapi_entry_set_uniqueid (e, newuniqueid);
        } else {
            char ebuf[BUFSIZ];
            LDAPDebug( LDAP_DEBUG_ANY,
                       "import_generate_uniqueid: failed to generate "
                       "uniqueid for %s; error=%d.\n", 
                       escape_string(slapi_entry_get_dn_const(e), ebuf), rc, 0 );
        }                
    }
}


/**********  BETTER LDIF PARSER  **********/


/* like the function in libldif, except this one doesn't need to use
 * FILE (which breaks on various platforms for >4G files or large numbers
 * of open files)
 */
#define LDIF_BUFFER_SIZE 8192

typedef struct {
    char *b;		/* buffer */
    size_t size;	/* how full the buffer is */
    size_t offset;	/* where the current entry starts */
} ldif_context;

static void import_init_ldif(ldif_context *c)
{
    c->size = c->offset = 0;
    c->b = NULL;
}

static void import_free_ldif(ldif_context *c)
{
    if (c->b)
	FREE(c->b);
    import_init_ldif(c);
}

static char *import_get_entry(ldif_context *c, int fd, int *lineno)
{
    int ret;
    int done = 0, got_lf = 0;
    size_t bufSize = 0, bufOffset = 0, i;
    char *buf = NULL;

    while (!done) {

	/* If there's no data in the buffer, get some */
	if ((c->size == 0) || (c->offset == c->size)) {
	    /* Do we even have a buffer ? */
	    if (! c->b) {
		c->b = slapi_ch_malloc(LDIF_BUFFER_SIZE);
		if (! c->b)
		    return NULL;
	    }
	    ret = read(fd, c->b, LDIF_BUFFER_SIZE);
	    if (ret < 0) {
		/* Must be error */
		goto error;
	    } else if (ret == 0) {
		/* eof */
		if (buf) {
		    /* last entry */
		    buf[bufOffset] = 0;
		    return buf;
		}
		return NULL;
	    } else {
		/* read completed OK */
		c->size = ret;
		c->offset = 0;
	    }
	}

        /* skip blank lines at start of entry */
        if (bufOffset == 0) {
            size_t n;
            char *p;

            for (n = c->offset, p = c->b + n; n < c->size; n++, p++) {
                if (!(*p == '\r' || *p == '\n' || *p == ' '|| *p == '\t')) 
                    break;
            }
            c->offset = n;
            if (c->offset == c->size) continue;
        }

	i = c->offset;
	while (!done && (i < c->size)) {
	    /* scan forward in the buffer, looking for the end of the entry */
	    while ((i < c->size) && (c->b[i] != '\n'))
		i++;

	    if ((i < c->size) && (c->b[i] == '\n')) {
		if (got_lf && ((i == 0) || ((i == 1) && (c->b[0] == '\r')))) {
		    /* saw an lf at the end of the last buffer */
		    i++, (*lineno)++;
		    done = 1;
		    got_lf = 0;
		    break;
		}
		got_lf = 0;
		(*lineno)++;
		/* is this the end?  (need another linefeed) */
		if (++i < c->size) {
		    if (c->b[i] == '\n') {
			/* gotcha! */
			i++, (*lineno)++;
			done = 1;
		    } else if (c->b[i] == '\r') {
			if (++i < c->size) {
			    if (c->b[i] == '\n') {
				/* gotcha! (nt) */
				i++, (*lineno)++;
				done = 1;
			    }
			} else {
			    got_lf = 1;
			}
		    }
		} else {
		    /* lf at the very end of the buffer */
		    got_lf = 1;
		}
	    }
	}

	/* copy what we did so far into the output buffer */
	/* (first, make sure the output buffer is large enough) */
	if (bufSize - bufOffset < i - c->offset + 1) {
	    char *newbuf = NULL;
	    size_t newsize = (buf ? bufSize*2 : LDIF_BUFFER_SIZE);

	    newbuf = slapi_ch_malloc(newsize);
	    if (! newbuf)
		goto error;
	    /* copy over the old data (if there was any) */
	    if (buf) {
		memmove(newbuf, buf, bufOffset);
		slapi_ch_free((void **)&buf);
	    }
	    buf = newbuf;
	    bufSize = newsize;
	}
	memmove(buf + bufOffset, c->b + c->offset, i - c->offset);
	bufOffset += (i - c->offset);
	c->offset = i;
    }

    /* add terminating NUL char */
    buf[bufOffset] = 0;
    return buf;

error:
    if (buf)
	slapi_ch_free((void **)&buf);
    return NULL;
}


/**********  THREADS  **********/

/*
 * Description:
 * 1) return the ldif version #
 * 2) replace "version: 1" with "#ersion: 1"
 *    to pretend like a comment for the str2entry
 */
static int
import_get_version(char *str)
{
    char *s;
    char *type;
    char *valuecharptr;
    char *mystr, *ms;
    int offset;
    int	valuelen;
    int my_version = 0;
    int retmalloc = 0;

    if ((s = strstr(str, "version:")) == NULL)
	return 0;

    offset = s - str;
    mystr = ms = slapi_ch_strdup(str);
    while ( (s = ldif_getline( &ms )) != NULL ) {
        char *errmsg = NULL;
	if ( (retmalloc = ldif_parse_line( s, &type, &valuecharptr, &valuelen, &errmsg )) >= 0 ) {
	    if (!strcasecmp(type, "version")) {
		my_version = atoi(valuecharptr);
		*(str + offset) = '#';
		/* the memory below was not allocated by the slapi_ch_ functions */
		if (errmsg) PR_smprintf_free(errmsg);
		if (retmalloc) slapi_ch_free((void **) &valuecharptr);
		break;
	    } 
	} else if ( errmsg != NULL ) {
	    LDAPDebug( LDAP_DEBUG_PARSE, "%s", errmsg, 0, 0 );
	}
	/* the memory below was not allocated by the slapi_ch_ functions */
	if (errmsg) slapi_ch_free((void **) &errmsg);
	if (retmalloc) slapi_ch_free((void **) &valuecharptr);
    }

    slapi_ch_free((void **)&mystr);
    return my_version;
}

/* producer thread:
 * read through the given file list, parsing entries (str2entry), assigning
 * them IDs and queueing them on the entry FIFO.  other threads will do
 * the indexing.
 */
void import_producer(void *param)
{
    ImportWorkerInfo *info = (ImportWorkerInfo *)param;
    ImportJob *job = info->job;
    ID id = job->first_ID, id_filestart = id;
    Slapi_Entry *e = NULL;
    struct backentry *ep = NULL, *old_ep = NULL;
    ldbm_instance *inst = job->inst;
    PRIntervalTime sleeptime;
    char *estr = NULL;
    int str2entry_flags =
    SLAPI_STR2ENTRY_TOMBSTONE_CHECK |
    SLAPI_STR2ENTRY_REMOVEDUPVALS |
    SLAPI_STR2ENTRY_EXPAND_OBJECTCLASSES |
    SLAPI_STR2ENTRY_ADDRDNVALS |
    SLAPI_STR2ENTRY_NOT_WELL_FORMED_LDIF;
    int finished = 0;
    int detected_eof = 0;
    int fd, curr_file, curr_lineno;
    char *curr_filename = NULL;
    int idx;
    ldif_context c;
    int my_version = 0;
    size_t newesize = 0;

    PR_ASSERT(info != NULL);
    PR_ASSERT(inst != NULL);
    
    if ( job->flags & FLAG_ABORT ) {
        goto error;
    }

    sleeptime = PR_MillisecondsToInterval(import_sleep_time);

    /* pause until we're told to run */
    while ((info->command == PAUSE) && !(job->flags & FLAG_ABORT)) {
        info->state = WAITING;
        DS_Sleep(sleeptime);
    }
    info->state = RUNNING;
    import_init_ldif(&c);

    /* jumpstart by opening the first file */
    curr_file = 0;
    fd = -1;
    detected_eof = finished = 0;

    /* we loop around reading the input files and processing each entry
     * as we read it.
     */
    while (! finished) {
        Slapi_Attr *attr = NULL;
        int flags = 0;
        int prev_lineno = 0;
        int lines_in_entry = 0;

        if (job->flags & FLAG_ABORT) { 
            goto error;
        }

        /* move on to next file? */
        if (detected_eof) {
            /* check if the file can still be read, whine if so... */
            if (read(fd, (void *)&idx, 1) > 0) {
                import_log_notice(job, "WARNING: Unexpected end of file found "
                                  "at line %d of file \"%s\"", curr_lineno,
                                  curr_filename);
            }

            if (fd == STDIN_FILENO) {
                import_log_notice(job, "Finished scanning file stdin (%lu "
                                  "entries)", (u_long)(id-id_filestart));
            } else {
                import_log_notice(job, "Finished scanning file \"%s\" (%lu "
                                  "entries)", curr_filename, (u_long)(id-id_filestart));
            }
            close(fd);
            fd = -1;
            detected_eof = 0;
            id_filestart = id;
            curr_file++;
            if (job->task) {
                job->task->task_progress++;
                slapi_task_status_changed(job->task);
            }
            if (job->input_filenames[curr_file] == NULL) {
                /* done! */
                finished = 1;
                break;
            }
        }

        /* separate from above, because this is also triggered when we
         * start (to open the first file)
         */
        if (fd < 0) {
            curr_lineno = 0;
            curr_filename = job->input_filenames[curr_file];
            if (strcmp(curr_filename, "-") == 0) {
                fd = STDIN_FILENO;
            } else {
                int o_flag = O_RDONLY;
#ifdef XP_WIN32
                /* 613041 Somehow the windows low level io lose "\n"
                   at a very particular situation using O_TEXT mode read.
                   I think it is a windows bug for O_TEXT mode read.
                   Use O_BINARY instead, which honestly returns chars
                   without any translation.
                */
                o_flag |= O_BINARY;
#endif
                fd = dblayer_open_huge_file(curr_filename, o_flag, 0);
            }
            if (fd < 0) {
                import_log_notice(job, "Could not open LDIF file \"%s\"",
                                  curr_filename);
                goto error;
            }
            if (fd == STDIN_FILENO) {
                import_log_notice(job, "Processing file stdin");
            } else {
                import_log_notice(job, "Processing file \"%s\"", curr_filename);
            }
        }
         if (job->flags & FLAG_ABORT) {   
             goto error;
         }


        while ((info->command == PAUSE)  && !(job->flags & FLAG_ABORT)){
            info->state = WAITING;
            DS_Sleep(sleeptime);
        }
        info->state = RUNNING;

        prev_lineno = curr_lineno;
        estr = import_get_entry(&c, fd, &curr_lineno);

        lines_in_entry = curr_lineno - prev_lineno;
        if (!estr) {
            /* error reading entry, or end of file */
            detected_eof = 1;
            continue;
        }

        if (0 == my_version && strstr(estr, "version:")) {
            my_version = import_get_version(estr);
                str2entry_flags |= SLAPI_STR2ENTRY_INCLUDE_VERSION_STR;
        }

        /* If there are more than so many lines in the entry, we tell
         * str2entry to optimize for a large entry.
         */
        if (lines_in_entry > STR2ENTRY_ATTRIBUTE_PRESENCE_CHECK_THRESHOLD) {
            flags = str2entry_flags | SLAPI_STR2ENTRY_BIGENTRY;
        } else {
            flags = str2entry_flags;
        }
        e = slapi_str2entry(estr, flags);
        FREE(estr);
        if (! e) {
                if (!(str2entry_flags & SLAPI_STR2ENTRY_INCLUDE_VERSION_STR))
                        import_log_notice(job, "WARNING: skipping bad LDIF entry "
                              "ending line %d of file \"%s\"", curr_lineno,
                              curr_filename);
            continue;
        }
        if (0 == my_version) {
            /* after the first entry version string won't be given */
            my_version = -1;
        }

        if (! import_entry_belongs_here(e, inst->inst_be)) {
            /* silently skip */
            if (e) {
                job->not_here_skipped++;
                slapi_entry_free(e);
            }
            continue;
        }

        if (slapi_entry_schema_check(NULL, e) != 0) {
            char ebuf[BUFSIZ];
            import_log_notice(job, "WARNING: skipping entry \"%s\" which "
                              "violates schema, ending line %d of file "
                              "\"%s\"", escape_string(slapi_entry_get_dn(e), ebuf),
                              curr_lineno, curr_filename);
            if (e)
                slapi_entry_free(e);
                job->skipped++;
            continue;
        }

        /* generate uniqueid if necessary */
        import_generate_uniqueid(job, e);

        ep = import_make_backentry(e, id);
        if (!ep)
            goto error;

        /* check for include/exclude subtree lists */
        if (! ldbm_back_ok_to_dump(backentry_get_ndn(ep),
                                   job->include_subtrees,
                                   job->exclude_subtrees)) {
            backentry_free(&ep);
            continue;
        }

        /* not sure what this does, but it looked like it could be
         * simplified.  if it's broken, it's my fault.  -robey 
         */
        if (slapi_entry_attr_find(ep->ep_entry, "userpassword", &attr) == 0) {
            Slapi_Value **va = attr_get_present_values(attr);

            pw_encodevals( (Slapi_Value **)va ); /* jcm - cast away const */
        }

         if (job->flags & FLAG_ABORT) { 
             goto error;
         }


        /* Now we have this new entry, all decoded
         * Next thing we need to do is:
         * (1) see if the appropriate fifo location contains an
         *     entry which had been processed by the indexers.
         *     If so, proceed.
         *     If not, spin waiting for it to become free.
         * (2) free the old entry and store the new one there.
         * (3) Update the job progress indicators so the indexers
         *     can use the new entry.
         */
        idx = id % job->fifo.size;
        old_ep = job->fifo.item[idx].entry;
        if (old_ep) {
            /* for the slot to be recycled, it needs to be already absorbed
             * by the foreman (id >= ready_ID), and all the workers need to
             * be finished with it (refcount = 0).
             */
            while (((old_ep->ep_refcnt > 0) ||
                    (old_ep->ep_id >= job->ready_ID))
                   && (info->command != ABORT) && !(job->flags & FLAG_ABORT)) {
                info->state = WAITING;
                DS_Sleep(sleeptime);
            }
            if (job->flags & FLAG_ABORT){
                goto error;
            }
            info->state = RUNNING;
            PR_ASSERT(old_ep == job->fifo.item[idx].entry);
            job->fifo.item[idx].entry = NULL;
            if (job->fifo.c_bsize > job->fifo.item[idx].esize)
                job->fifo.c_bsize -= job->fifo.item[idx].esize;
            else
                job->fifo.c_bsize = 0;
            backentry_free(&old_ep);
        }

        newesize = (slapi_entry_size(ep->ep_entry) + sizeof(struct backentry));
        if (newesize > job->fifo.bsize) {    /* entry too big */
            char ebuf[BUFSIZ];
            import_log_notice(job, "WARNING: skipping entry \"%s\" "
                    "ending line %d of file \"%s\"",
                    escape_string(slapi_entry_get_dn(e), ebuf),
                    curr_lineno, curr_filename);
            import_log_notice(job, "REASON: entry too large (%d bytes) for "
                    "the buffer size (%d bytes)", newesize, job->fifo.bsize);
            backentry_free(&ep);
            job->skipped++;
            continue;
        }
        /* Now check if fifo has enough space for the new entry */
        if ((job->fifo.c_bsize + newesize) > job->fifo.bsize) {
            import_wait_for_space_in_fifo( job, newesize );
        }

        /* We have enough space */
        job->fifo.item[idx].filename = curr_filename;
        job->fifo.item[idx].line = curr_lineno;
        job->fifo.item[idx].entry = ep;
        job->fifo.item[idx].bad = 0;
        job->fifo.item[idx].esize = newesize;

        /* Add the entry size to total fifo size */
        job->fifo.c_bsize += ep->ep_entry? job->fifo.item[idx].esize : 0;

        /* Update the job to show our progress */
        job->lead_ID = id;
        if ((id - info->first_ID) <= job->fifo.size) {
            job->trailing_ID = info->first_ID;
        } else {
            job->trailing_ID = id - job->fifo.size;
        }

        /* Update our progress meter too */
        info->last_ID_processed = id;
        id++;
        if (job->flags & FLAG_ABORT){
            goto error;
        }
        if (info->command == STOP) {
            if (fd >= 0)
                close(fd);
            finished = 1;
        }
    }

    import_free_ldif(&c);
    info->state = FINISHED;
    return;

error:
    info->state = ABORTED;
}

/* producer thread for re-indexing:
 * read id2entry, parsing entries (str2entry) (needed???), assigning
 * them IDs (again, needed???) and queueing them on the entry FIFO. 
 * other threads will do the indexing -- same as in import.
 */
void index_producer(void *param)
{
    ImportWorkerInfo *info = (ImportWorkerInfo *)param;
    ImportJob *job = info->job;
    ID id = job->first_ID;
    Slapi_Entry *e = NULL;
    struct backentry *ep = NULL, *old_ep = NULL;
    ldbm_instance *inst = job->inst;
    PRIntervalTime sleeptime;
    int finished = 0;
    int idx;

    /* vars for Berkeley DB */
    DB_ENV *env = NULL;
    DB *db = NULL;
    DBC *dbc = NULL;
    DBT key = {0};
    DBT data = {0};
    int db_rval = -1;
    backend *be = inst->inst_be;
    int isfirst = 1;
    int curr_entry = 0;
    size_t newesize = 0;

    PR_ASSERT(info != NULL);
    PR_ASSERT(inst != NULL);
    PR_ASSERT(be != NULL);
    
    if ( job->flags & FLAG_ABORT )
        goto error;

    sleeptime = PR_MillisecondsToInterval(import_sleep_time);

    /* pause until we're told to run */
    while ((info->command == PAUSE) && !(job->flags & FLAG_ABORT)) {
        info->state = WAITING;
        DS_Sleep(sleeptime);
    }
    info->state = RUNNING;

    /* open id2entry with dedicated db env and db handler */
    if ( dblayer_get_aux_id2entry( be, &db, &env ) != 0  || db == NULL ||
         env == NULL) {
        LDAPDebug( LDAP_DEBUG_ANY, "Could not open id2entry\n", 0, 0, 0 );
        goto error;
    }

    /* get a cursor to we can walk over the table */
    db_rval = db->cursor(db, NULL, &dbc, 0);
    if ( 0 != db_rval ) {
        LDAPDebug( LDAP_DEBUG_ANY,
                   "Failed to get cursor for reindexing\n", 0, 0, 0 );
        dblayer_release_id2entry(be, db);
        goto error;
    }

    /* we loop around reading the input files and processing each entry
     * as we read it.
     */
    finished = 0;
    while (!finished) {
        Slapi_Attr *attr = NULL;
        ID temp_id;

        if (job->flags & FLAG_ABORT) {   
            goto error;
        }
        while ((info->command == PAUSE)  && !(job->flags & FLAG_ABORT)){
            info->state = WAITING;
            DS_Sleep(sleeptime);
        }
        info->state = RUNNING;

        key.flags = DB_DBT_MALLOC;
        data.flags = DB_DBT_MALLOC;
        if (isfirst)
        {
            db_rval = dbc->c_get(dbc, &key, &data, DB_FIRST);
            isfirst = 0;
        }
        else
        {
            db_rval = dbc->c_get(dbc, &key, &data, DB_NEXT);
        }
        
        if (0 != db_rval) {
            if (DB_NOTFOUND != db_rval) {
                LDAPDebug(LDAP_DEBUG_ANY, "%s: Failed to read database, "
                    "errno=%d (%s)\n", inst->inst_name, db_rval,
                    dblayer_strerror(db_rval));
                if (job->task) {
                    slapi_task_log_notice(job->task,
                        "%s: Failed to read database, err %d (%s)",
                        inst->inst_name, db_rval,
                        dblayer_strerror(db_rval));
                }
            }
            break;
        }
        curr_entry++;
        temp_id = id_stored_to_internal((char *)key.data);
        free(key.data);

        /* call post-entry plugin */
        plugin_call_entryfetch_plugins((char **) &data.dptr, &data.dsize);
        e = slapi_str2entry(data.data, 0);
        if ( NULL == e ) {
            if (job->task) {
                slapi_task_log_notice(job->task,
                    "%s: WARNING: skipping badly formatted entry (id %lu)",
                    inst->inst_name, (u_long)temp_id);
            }
            LDAPDebug(LDAP_DEBUG_ANY,
                "%s: WARNING: skipping badly formatted entry (id %lu)\n",
                inst->inst_name, (u_long)temp_id, 0);
            continue;
        } 
        free(data.data);

        /* generate uniqueid if necessary */
        import_generate_uniqueid(job, e);

        ep = import_make_backentry(e, temp_id);
        if (!ep)
            goto error;

        /* not sure what this does, but it looked like it could be
         * simplified.  if it's broken, it's my fault.  -robey 
         */
        if (slapi_entry_attr_find(ep->ep_entry, "userpassword", &attr) == 0) {
            Slapi_Value **va = attr_get_present_values(attr);

            pw_encodevals( (Slapi_Value **)va ); /* jcm - cast away const */
        }

        if (job->flags & FLAG_ABORT)
             goto error;

        /* Now we have this new entry, all decoded
         * Next thing we need to do is:
         * (1) see if the appropriate fifo location contains an
         *     entry which had been processed by the indexers.
         *     If so, proceed.
         *     If not, spin waiting for it to become free.
         * (2) free the old entry and store the new one there.
         * (3) Update the job progress indicators so the indexers
         *     can use the new entry.
         */
        idx = id % job->fifo.size;
        old_ep = job->fifo.item[idx].entry;
        if (old_ep) {
            /* for the slot to be recycled, it needs to be already absorbed
             * by the foreman (id >= ready_ID), and all the workers need to
             * be finished with it (refcount = 0).
             */
            while (((old_ep->ep_refcnt > 0) ||
                    (old_ep->ep_id >= job->ready_ID))
                   && (info->command != ABORT) && !(job->flags & FLAG_ABORT)) {
                info->state = WAITING;
                DS_Sleep(sleeptime);
            }
            if (job->flags & FLAG_ABORT)
                goto error;

            info->state = RUNNING;
            PR_ASSERT(old_ep == job->fifo.item[idx].entry);
            job->fifo.item[idx].entry = NULL;
            if (job->fifo.c_bsize > job->fifo.item[idx].esize)
                job->fifo.c_bsize -= job->fifo.item[idx].esize;
            else
                job->fifo.c_bsize = 0;
            backentry_free(&old_ep);
        }

        newesize = (slapi_entry_size(ep->ep_entry) + sizeof(struct backentry));
        if (newesize > job->fifo.bsize) {    /* entry too big */
            char ebuf[BUFSIZ];
            import_log_notice(job, "WARNING: skipping entry \"%s\"",
                    escape_string(slapi_entry_get_dn(e), ebuf));
            import_log_notice(job, "REASON: entry too large (%d bytes) for "
                    "the buffer size (%d bytes)", newesize, job->fifo.bsize);
            backentry_free(&ep);
            job->skipped++;
            continue;
        }
        /* Now check if fifo has enough space for the new entry */
        if ((job->fifo.c_bsize + newesize) > job->fifo.bsize) {
            import_wait_for_space_in_fifo( job, newesize );
        }

        /* We have enough space */
        job->fifo.item[idx].filename = ID2ENTRY LDBM_FILENAME_SUFFIX;
        job->fifo.item[idx].line = curr_entry;
        job->fifo.item[idx].entry = ep;
        job->fifo.item[idx].bad = 0;
        job->fifo.item[idx].esize = newesize;

        /* Add the entry size to total fifo size */
        job->fifo.c_bsize += ep->ep_entry? job->fifo.item[idx].esize : 0;

        /* Update the job to show our progress */
        job->lead_ID = id;
        if ((id - info->first_ID) <= job->fifo.size) {
            job->trailing_ID = info->first_ID;
        } else {
            job->trailing_ID = id - job->fifo.size;
        }

        /* Update our progress meter too */
        info->last_ID_processed = id;
        id++;
        if (job->flags & FLAG_ABORT)
            goto error;
        if (info->command == STOP)
        {
            finished = 1;
        }
    }

    dbc->c_close(dbc);
    dblayer_release_aux_id2entry( be, db, env );
    info->state = FINISHED;
    return;

error:
    dbc->c_close(dbc);
    dblayer_release_aux_id2entry( be, db, env );
    info->state = ABORTED;
}

static void
import_wait_for_space_in_fifo(ImportJob *job, size_t new_esize)
{
    struct backentry *temp_ep = NULL;
    size_t i;
    int slot_found;
    PRIntervalTime sleeptime;

    sleeptime = PR_MillisecondsToInterval(import_sleep_time);

    /* Now check if fifo has enough space for the new entry */
    while ((job->fifo.c_bsize + new_esize) > job->fifo.bsize) {
        for ( i = 0, slot_found = 0 ; i < job->fifo.size ; i++ ) {
            temp_ep = job->fifo.item[i].entry;
            if (temp_ep) {
                if (temp_ep->ep_refcnt == 0 && temp_ep->ep_id < job->ready_ID) {
                    job->fifo.item[i].entry = NULL;
                    if (job->fifo.c_bsize > job->fifo.item[i].esize)
                        job->fifo.c_bsize -= job->fifo.item[i].esize;
                    else
                        job->fifo.c_bsize = 0;
                    backentry_free(&temp_ep);
                    slot_found = 1;
                }
            }
        }
        if ( slot_found == 0 )
            DS_Sleep(sleeptime);
    }
}

/* helper function for the foreman: */
static int foreman_do_parentid(ImportJob *job, struct backentry *entry,
                               struct attrinfo *parentid_ai)
{
    backend *be = job->inst->inst_be;
    Slapi_Value **svals = NULL;
    Slapi_Attr *attr = NULL;
    int idl_disposition = 0;
    int ret = 0;

    if (slapi_entry_attr_find(entry->ep_entry, "parentid", &attr) == 0) {
        svals = attr_get_present_values(attr);
        ret = index_addordel_values_ext_sv(be, "parentid", svals, NULL, entry->ep_id,
            BE_INDEX_ADD, NULL, &idl_disposition, NULL);
        if (idl_disposition != IDL_INSERT_NORMAL) {
            char *attr_value = slapi_value_get_berval(svals[0])->bv_val;
            ID parent_id = atol(attr_value);

            if (idl_disposition == IDL_INSERT_NOW_ALLIDS) {
                import_subcount_mother_init(job->mothers, parent_id,
                    idl_get_allidslimit(parentid_ai)+1);
            } else if (idl_disposition == IDL_INSERT_ALLIDS) {
                import_subcount_mother_count(job->mothers, parent_id);
            }
        }
        if (ret != 0) {
            import_log_notice(job, "ERROR: Can't update parentid index "
                              "(error %d)", ret);
            return ret;
        }
    }

    return 0;
}

/* helper function for the foreman: */
static int foreman_do_entrydn(ImportJob *job, FifoItem *fi)
{
    backend *be = job->inst->inst_be;
    struct berval bv;
    int err = 0, ret = 0;
    IDList *IDL;

    /* insert into the entrydn index */
    bv.bv_val = (void*)backentry_get_ndn(fi->entry);   /* jcm - Had to cast away const */
    bv.bv_len = strlen(bv.bv_val);

    /* We need to check here whether the DN is already present in
     * the entrydn index. If it is then the input ldif 
     * contained a duplicate entry, which it isn't allowed to */
    /* Due to popular demand, we only warn on this, given the 
     * tendency for customers to want to import dirty data */
    /* So, we do an index read first */
    err = 0;
    IDL = index_read(be, "entrydn", indextype_EQUALITY, &bv, NULL, &err);

    /* Did this work ? */
    if (NULL != IDL) {
        /* IMPOSTER ! Get thee hence... */
        import_log_notice(job, "WARNING: Skipping duplicate entry "
                          "\"%s\" found at line %d of file \"%s\"",
                          slapi_entry_get_dn(fi->entry->ep_entry),
                          fi->line, fi->filename);
        idl_free(IDL);
        /* skip this one */
        fi->bad = 1;
        job->skipped++;
        return -1;      /* skip to next entry */
    }
    if ((ret = index_addordel_string(be, "entrydn", 
                                     bv.bv_val,
                                     fi->entry->ep_id,
                                     BE_INDEX_ADD|BE_INDEX_NORMALIZED, NULL)) != 0) {
        import_log_notice(job, "Error writing entrydn index "
                          "(error %d: %s)",
                          ret, dblayer_strerror(ret));
        return ret;
    }
    return 0;
}

/* foreman thread:
 * i go through the FIFO just like the other worker threads, but i'm 
 * responsible for the interrelated indexes: entrydn, id2entry, and the
 * operational attributes (plus the parentid index).
 */
void import_foreman(void *param)
{
    ImportWorkerInfo *info = (ImportWorkerInfo *)param;
    ImportJob *job = info->job;
    ldbm_instance *inst = job->inst;
    backend *be = inst->inst_be;
    PRIntervalTime sleeptime;
    int finished = 0;
    ID id = info->first_ID;
    int ret = 0;
    struct attrinfo *parentid_ai;
    Slapi_PBlock *pb = slapi_pblock_new();
	int shift = 0;

    PR_ASSERT(info != NULL);
    PR_ASSERT(inst != NULL);

    if (job->flags & FLAG_ABORT) { 
        goto error;
    }

    /* the pblock is used only by add_op_attrs */
    slapi_pblock_set(pb, SLAPI_BACKEND, be);
    sleeptime = PR_MillisecondsToInterval(import_sleep_time);
    info->state = RUNNING;

    ainfo_get(be, "parentid", &parentid_ai);

    while (! finished) {
        FifoItem *fi = NULL;
        int parent_status = 0;
         

        if (job->flags & FLAG_ABORT) { 
            goto error;
        }

        while ( ((info->command == PAUSE) || (id > job->lead_ID)) &&
                (info->command != STOP) && (info->command != ABORT)  && !(job->flags & FLAG_ABORT)) {
            /* Check to see if we've been told to stop */
            info->state = WAITING;
            DS_Sleep(sleeptime);
        }
        if (info->command == STOP) {
            finished = 1;
            continue;
        }
        if (job->flags & FLAG_ABORT) {
            goto error;
        }

        info->state = RUNNING;

        /* Read that entry from the cache */
        fi = import_fifo_fetch(job, id, 0, shift);
        if (! fi) {
            import_log_notice(job, "WARNING: entry id %d is missing", id);
			shift++;
            continue;
        }

        /* first, fill in any operational attributes */
        /* add_op_attrs wants a pblock for some reason. */
        if (add_op_attrs(pb, inst->inst_li, fi->entry, &parent_status) != 0) {
            import_log_notice(job, "ERROR: Could not add op attrs to "
                              "entry ending at line %d of file \"%s\"",
                              fi->line, fi->filename);
            goto error;
        }

        if (! slapi_entry_flag_is_set(fi->entry->ep_entry,
                                      SLAPI_ENTRY_FLAG_TOMBSTONE)) {
            /*
             * Only check for a parent and add to the entry2dn index if
             * the entry is not a tombstone.
             */
             if (job->flags & FLAG_ABORT) {       
                 goto error;
             }

            if (parent_status == IMPORT_ADD_OP_ATTRS_NO_PARENT) {
                /* If this entry is a suffix entry, this is not a problem */
                /* However, if it is not, this is an error---it means that
                 * someone tried to import an entry before importing its parent
                 * we reject the entry but carry on since we've not stored
                 * anything related to this entry.
                 */
                if (! slapi_be_issuffix(inst->inst_be, backentry_get_sdn(fi->entry))) {
                    import_log_notice(job, "WARNING: Skipping entry \"%s\" "
                                      "which has no parent, ending at line %d "
                                      "of file \"%s\"",
                                      slapi_entry_get_dn(fi->entry->ep_entry),
                                      fi->line, fi->filename);
                    /* skip this one */
                    fi->bad = 1;
                    job->skipped++;
                    goto cont;      /* below */
                }
            }
             if (job->flags & FLAG_ABORT) {
                 goto error;
             }


            /* insert into the entrydn index */
            ret = foreman_do_entrydn(job, fi);
            if (ret == -1)
                goto cont;      /* skip entry */
            if (ret != 0)
                goto error; 
        }

        if (job->flags & FLAG_ABORT) {
            goto error;
        }

        if (!(job->flags & FLAG_REINDEXING))/* reindex reads data from id2entry */
        {
            /* insert into the id2entry index
             * (that isn't really an index -- it's the storehouse of the entries
             * themselves.)
             */
            if ((ret = id2entry_add_ext(be, fi->entry, NULL, job->encrypt)) != 0) {
                /* DB_RUNRECOVERY usually occurs if disk fills */
                if (LDBM_OS_ERR_IS_DISKFULL(ret)) {
                import_log_notice(job, "ERROR: OUT OF SPACE ON DISK or FILE TOO LARGE -- "
                                  "Could not store the entry ending at line "
                                  "%d of file \"%s\"",
                                  fi->line, fi->filename);
                } else if (ret == DB_RUNRECOVERY) {
                    import_log_notice(job, "FATAL ERROR: (LARGEFILE SUPPORT NOT ENABLED? OUT OF SPACE ON DISK?) -- "
                                  "Could not store the entry ending at line "
                                  "%d of file \"%s\"",
                                  fi->line, fi->filename);
                } else {
                    import_log_notice(job, "ERROR: Could not store the entry "
                                  "ending at line %d of file \"%s\" -- "
                                  "error %d", fi->line, fi->filename, ret);
                }
                goto error;
            }
        }

        if (job->flags & FLAG_ABORT) { 
            goto error;
        }

        if (! slapi_entry_flag_is_set(fi->entry->ep_entry,
                                      SLAPI_ENTRY_FLAG_TOMBSTONE)) {
            /* parentid index
             * (we have to do this here, because the parentID is dependent on
             * looking up by entrydn.)
             * Only add to the parent index if the entry is not a tombstone.
             */
            ret = foreman_do_parentid(job, fi->entry, parentid_ai);
            if (ret != 0)
                goto error;
            
            /* Lastly, before we're finished with the entry, pass it to the 
               vlv code to see whether it's within the scope a VLV index. */
            vlv_grok_new_import_entry(fi->entry, be);
        }
         if (job->flags & FLAG_ABORT) { 
             goto error;
         }


        /* Remove the entry from the cache (caused by id2entry_add) */
    if (!(job->flags & FLAG_REINDEXING))/* reindex reads data from id2entry */
        cache_remove(&inst->inst_cache, fi->entry);
        fi->entry->ep_refcnt = job->number_indexers;

    cont:
         if (job->flags & FLAG_ABORT) { 
             goto error;
         }

        job->ready_ID = id;
        info->last_ID_processed = id;
        id++;

        if (job->flags & FLAG_ABORT){
            goto error;
        }
    }

    slapi_pblock_destroy(pb);
    info->state = FINISHED;
    return;

error:
    slapi_pblock_destroy(pb);
    info->state = ABORTED;
}


/* worker thread:
 * given an attribute, this worker plows through the entry FIFO, building
 * up the attribute index.
 */
void import_worker(void *param)
{
    ImportWorkerInfo *info = (ImportWorkerInfo *)param;
    ImportJob *job = info->job;
    ldbm_instance *inst = job->inst;
    backend *be = inst->inst_be;
    PRIntervalTime sleeptime;
    int finished = 0;
    ID id = info->first_ID;
    int ret = 0;
    int idl_disposition = 0;
    struct vlvIndex* vlv_index = NULL;
    void *substring_key_buffer = NULL;
    FifoItem *fi;
    int is_objectclass_attribute;
    int is_nsuniqueid_attribute;
    void *attrlist_cursor;
    
    PR_ASSERT(NULL != info);
    PR_ASSERT(NULL != inst);

    if (job->flags & FLAG_ABORT) {
        goto error;
    }

    if (INDEX_VLV == info->index_info->ai->ai_indexmask) {
        vlv_index = vlv_find_indexname(info->index_info->name, be);
        if (NULL == vlv_index) {
            goto error;
        }
    }

    /*
     * If the entry is a Tombstone, then we only add it to the nsuniqeid index
     * and the idlist for (objectclass=tombstone). These two flags are just
     * handy for working out what to do in this case.
     */
    is_objectclass_attribute =
        (strcasecmp(info->index_info->name, "objectclass") == 0);
    is_nsuniqueid_attribute = 
        (strcasecmp(info->index_info->name, SLAPI_ATTR_UNIQUEID) == 0);

    if (1 != idl_get_idl_new()) {
        /* Is there substring indexing going on here ? */
        if ( (INDEX_SUB & info->index_info->ai->ai_indexmask) &&
             (info->index_buffer_size > 0) ) {
            /* Then make a key buffer thing */
            ret = index_buffer_init(info->index_buffer_size, 0, 
                                    &substring_key_buffer);
            if (0 != ret) {
                import_log_notice(job, "IMPORT FAIL 1 (error %d)", ret);
            }
        }
    }

    sleeptime = PR_MillisecondsToInterval(import_sleep_time);
    info->state = RUNNING;
    info->last_ID_processed = id-1;

    while (! finished) {
        struct backentry *ep = NULL;
        Slapi_Value **svals = NULL;
        Slapi_Attr *attr = NULL;
        
        if (job->flags & FLAG_ABORT) {
            goto error;
        }

        /* entry can be NULL if it turned out to be bogus */
        while (!finished && !ep) {
                /* This worker thread must wait if the command flag is "PAUSE" or
                 * the entry corresponds to the current entry treated by the foreman
                 * thread, and the state is neither STOP nor ABORT
                 */
            while (((info->command == PAUSE) || (id > job->ready_ID)) &&
                   (info->command != STOP) && (info->command != ABORT) && !(job->flags & FLAG_ABORT)) {
                /* Check to see if we've been told to stop */
                info->state = WAITING;
                DS_Sleep(sleeptime);                
            }

            if (info->command == STOP) {
                finished = 1;
                continue;
            }
            if (job->flags & FLAG_ABORT) {
                goto error;
            }

            info->state = RUNNING;

            /* Read that entry from the cache */
            fi = import_fifo_fetch(job, id, 1, 0);
            ep = fi ? fi->entry : NULL;
            if (!ep) {
                /* skipping an entry that turned out to be bad */
                info->last_ID_processed = id;
                id++;
            }
        }
        if (finished)
            continue;

        if (! slapi_entry_flag_is_set(fi->entry->ep_entry,
                                      SLAPI_ENTRY_FLAG_TOMBSTONE)) {
            /* This is not a tombstone entry. */
            /* Is this a VLV index ? */

            if (job->flags & FLAG_ABORT) {
                goto error;
            }
            
            if (INDEX_VLV == info->index_info->ai->ai_indexmask) {
                /* Yes, call VLV code -- needs pblock to find backend */
                Slapi_PBlock *pb = slapi_pblock_new();

                PR_ASSERT(NULL != vlv_index);
                slapi_pblock_set(pb, SLAPI_BACKEND, be);
                vlv_update_index(vlv_index, NULL, inst->inst_li, pb, NULL, ep);
                slapi_pblock_destroy(pb);
            } else {
                /* No, process regular index */
                /* Look for the attribute we're indexing and its subtypes */
                /* For each attr write to the index */
                attrlist_cursor = NULL;
                while ((attr = attrlist_find_ex(ep->ep_entry->e_attrs,
                                                info->index_info->name,
                                                NULL,
                                                NULL,
                                                &attrlist_cursor)) != NULL) {

                    if (job->flags & FLAG_ABORT) {
                        goto error;
                    }
                    if(valueset_isempty(&(attr->a_present_values))) continue;
                    svals = attr_get_present_values(attr);
                    ret = index_addordel_values_ext_sv(be, info->index_info->name,
                        svals, NULL, ep->ep_id, BE_INDEX_ADD | (job->encrypt ? 0 : BE_INDEX_DONT_ENCRYPT), NULL, &idl_disposition,
                        substring_key_buffer);

                    if (0 != ret) {
                        /* Something went wrong, eg disk filled up */
                        goto error;
                    }
                }
            }
        } else {
            /* This is a Tombstone entry... we only add it to the nsuniqeid
             * index and the idlist for (objectclass=nstombstone).
             */
            if (job->flags & FLAG_ABORT) {
                        goto error;
            }
            if (is_nsuniqueid_attribute) {
                ret = index_addordel_string(be, SLAPI_ATTR_UNIQUEID,
                    slapi_entry_get_uniqueid(ep->ep_entry), ep->ep_id,
                    BE_INDEX_ADD, NULL);
                if (0 != ret) {
                    /* Something went wrong, eg disk filled up */
                    goto error;
                }
            }
            if (is_objectclass_attribute) {
                ret = index_addordel_string(be, SLAPI_ATTR_OBJECTCLASS,
                    SLAPI_ATTR_VALUE_TOMBSTONE, ep->ep_id, BE_INDEX_ADD, NULL);
                if (0 != ret) {
                    /* Something went wrong, eg disk filled up */
                    goto error;
                }
            }
        }
        import_decref_entry(ep);
        info->last_ID_processed = id;
        id++;

        if (job->flags & FLAG_ABORT) {
            goto error;
        }
    }
    
     if (job->flags & FLAG_ABORT) {
         goto error;
     }


    /* If we were buffering index keys, now flush them */
    if (substring_key_buffer) {
        ret = index_buffer_flush(substring_key_buffer,
                                 inst->inst_be, NULL, 
                                 info->index_info->ai);
        if (0 != ret) {
            goto error;
        }
        index_buffer_terminate(substring_key_buffer);
    }
    info->state = FINISHED;
    return;

error:
    if (ret == DB_RUNRECOVERY) {
        LDAPDebug(LDAP_DEBUG_ANY,"cannot import; database recovery needed\n",
                  0,0,0);
    } else if (ret == DB_LOCK_DEADLOCK) {
        /* can this occur? */
    }

    info->state = ABORTED;
}



/*
 * import entries to a backend, over the wire -- entries will arrive
 * asynchronously, so this method has no "producer" thread.  instead, the
 * front-end drops new entries in as they arrive.
 *
 * this is sometimes called "fast replica initialization".
 *
 * some of this code is duplicated from ldif2ldbm, but i don't think we
 * can avoid it.
 */
static int bulk_import_start(Slapi_PBlock *pb)
{
    struct ldbminfo *li = NULL;
    ImportJob *job = NULL;
    backend *be = NULL;
    PRThread *thread = NULL;
    int ret = 0;

    job = CALLOC(ImportJob);
    if (job == NULL) {
        LDAPDebug(LDAP_DEBUG_ANY, "not enough memory to do import job\n",
                  0, 0, 0);
        return -1;
    }

    slapi_pblock_get(pb, SLAPI_BACKEND, &be);
    PR_ASSERT(be != NULL);
    li = (struct ldbminfo *)(be->be_database->plg_private);
    job->inst = (ldbm_instance *)be->be_instance_info;

    /* check if an import/restore is already ongoing... */
    PR_Lock(job->inst->inst_config_mutex);
    if (job->inst->inst_flags & INST_FLAG_BUSY) {
        PR_Unlock(job->inst->inst_config_mutex);
        LDAPDebug(LDAP_DEBUG_ANY, "ldbm: '%s' is already in the middle of "
                  "another task and cannot be disturbed.\n",
                  job->inst->inst_name, 0, 0);
        FREE(job);
        return SLAPI_BI_ERR_BUSY;
    }
    job->inst->inst_flags |= INST_FLAG_BUSY;
    PR_Unlock(job->inst->inst_config_mutex);

    /* take backend offline */
    slapi_mtn_be_disable(be);

    /* get uniqueid info */
    slapi_pblock_get(pb, SLAPI_LDIF2DB_GENERATE_UNIQUEID, &job->uuid_gen_type);
    if (job->uuid_gen_type == SLAPI_UNIQUEID_GENERATE_NAME_BASED) {
        char *namespaceid;

        slapi_pblock_get(pb, SLAPI_LDIF2DB_NAMESPACEID, &namespaceid);
        job->uuid_namespace = slapi_ch_strdup(namespaceid);
    }

    job->flags = 0;     /* don't use files */
    job->flags |= FLAG_INDEX_ATTRS;
    job->flags |= FLAG_ONLINE;
    job->starting_ID = 1;
    job->first_ID = 1;

    job->mothers = CALLOC(import_subcount_stuff);
    /* how much space should we allocate to index buffering? */
    job->job_index_buffer_size = import_get_index_buffer_size();
    if (job->job_index_buffer_size == 0) {
	/* 10% of the allocated cache size + one meg */
	job->job_index_buffer_size = (job->inst->inst_li->li_dbcachesize/10) +
            (1024*1024);
    }
    import_subcount_stuff_init(job->mothers);
    job->wire_lock = PR_NewLock();
    job->wire_cv = PR_NewCondVar(job->wire_lock);

    /* COPIED from ldif2ldbm.c : */

    /* shutdown this instance of the db */
    cache_clear(&job->inst->inst_cache);
    dblayer_instance_close(be);

    /* Delete old database files */
    dblayer_delete_instance_dir(be);
    /* it's okay to fail -- it might already be gone */

    /* dblayer_instance_start will init the id2entry index. */
    /* it also (finally) fills in inst_dir_name */
    ret = dblayer_instance_start(be, DBLAYER_IMPORT_MODE);
    if (ret != 0)
        goto fail;

    /* END OF COPIED SECTION */

    PR_Lock(job->wire_lock);
    vlv_init(job->inst);

    /* create thread for import_main, so we can return */
    thread = PR_CreateThread(PR_USER_THREAD, import_main, (void *)job,
                             PR_PRIORITY_NORMAL, PR_GLOBAL_THREAD,
                             PR_JOINABLE_THREAD,
                             SLAPD_DEFAULT_THREAD_STACKSIZE);
    if (thread == NULL) {
	PRErrorCode prerr = PR_GetError();
        LDAPDebug(LDAP_DEBUG_ANY, "unable to spawn import thread, "
				SLAPI_COMPONENT_NAME_NSPR " error %d (%s)\n",
				prerr, slapd_pr_strerror(prerr), 0);
        PR_Unlock(job->wire_lock);
        ret = -2;
        goto fail;
    }

    job->main_thread = thread;
    slapi_set_object_extension(li->li_bulk_import_object, pb->pb_conn,
        li->li_bulk_import_handle, job);

    /* wait for the import_main to signal that it's ready for entries */
    /* (don't want to send the success code back to the LDAP client until
     * we're ready for the adds to start rolling in)
     */
    PR_WaitCondVar(job->wire_cv, PR_INTERVAL_NO_TIMEOUT);
    PR_Unlock(job->wire_lock);

    return 0;

fail:
    PR_Lock(job->inst->inst_config_mutex);
    job->inst->inst_flags &= ~INST_FLAG_BUSY;
    PR_Unlock(job->inst->inst_config_mutex);
    import_free_job(job);
    FREE(job);
    return ret;
}

/* returns 0 on success, or < 0 on error 
 *
 * on error, the import process is aborted -- so if this returns an error,
 * don't try to queue any more entries or you'll be sorry.
 *
 * flag_block in used to know if this thread should block when
 * the fifo is full or return an error LDAP_BUSY
 * Typically, import done on from the GUI or the command line will
 * block while online import as used by the replication total update
 * will not block
 */
static int bulk_import_queue(ImportJob *job, Slapi_Entry *entry, int flag_block)
{
    struct backentry *ep = NULL, *old_ep = NULL;
    int idx;
    ID id = 0;
    Slapi_Attr *attr = NULL;
    size_t newesize = 0;

    PR_Lock(job->wire_lock);
	/* Let's do this inside the lock !*/
    id = job->lead_ID + 1;
    /* generate uniqueid if necessary */
    import_generate_uniqueid(job, entry);

    /* make into backentry */
    ep = import_make_backentry(entry, id);
    if (!ep) {
        import_abort_all(job, 1);
        PR_Unlock(job->wire_lock);
        return -1;
    }

    /* encode the password */
    if (slapi_entry_attr_find(ep->ep_entry, "userpassword", &attr) == 0) {
        Slapi_Value **va = attr_get_present_values(attr);

        pw_encodevals( (Slapi_Value **)va ); /* jcm - had to cast away const */
    }

    /* Now we have this new entry, all decoded
     * Next thing we need to do is:
     * (1) see if the appropriate fifo location contains an
     *     entry which had been processed by the indexers.
     *     If so, proceed.
     *     If not, spin waiting for it to become free.
     * (2) free the old entry and store the new one there.
     * (3) Update the job progress indicators so the indexers
     *     can use the new entry.
     */
    idx = id % job->fifo.size;
    old_ep = job->fifo.item[idx].entry;
    if (old_ep) {
        while ((old_ep->ep_refcnt > 0) && !(job->flags & FLAG_ABORT))
        {
            if (flag_block)
                DS_Sleep(PR_MillisecondsToInterval(import_sleep_time));
            else
            {
				/* DBBD: Argh -- why not just block, what's the benefit to this ?? */
				/* I think that to support pipelining in the transport, we need to block here, */
				/* Otherwise evil things could happen where we say we're busy for operation N, but */
				/* Not for operation N+1, but the sender doesn't find out about this until after sending */
				/* Operation N+2 etc. Seems possible to end up with children processed before parents which won't work. */
                PR_Unlock(job->wire_lock);
                return LDAP_BUSY;
            }
        }

        /* the producer could be running thru the fifo while
         * everyone else is cycling to a new pass...
         * double-check that this entry is < ready_ID
         */
        while ((old_ep->ep_id >= job->ready_ID) && !(job->flags & FLAG_ABORT))
        {
            if (flag_block)
                DS_Sleep(PR_MillisecondsToInterval(import_sleep_time));
            else
            {
                PR_Unlock(job->wire_lock);
                return LDAP_BUSY;
            }
        }

        if (job->flags & FLAG_ABORT) {
            PR_Unlock(job->wire_lock);
            return -2;
        }

        PR_ASSERT(old_ep == job->fifo.item[idx].entry);
        job->fifo.item[idx].entry = NULL;
        if (job->fifo.c_bsize > job->fifo.item[idx].esize)
            job->fifo.c_bsize -= job->fifo.item[idx].esize;
        else
            job->fifo.c_bsize = 0;
        backentry_free(&old_ep);
    }

    newesize = (slapi_entry_size(ep->ep_entry) + sizeof(struct backentry));
    if (newesize > job->fifo.bsize) {    /* entry too big */
        char ebuf[BUFSIZ];
        import_log_notice(job, "WARNING: skipping entry \"%s\"",
                    escape_string(slapi_entry_get_dn(ep->ep_entry), ebuf));
        import_log_notice(job, "REASON: entry too large (%d bytes) for "
                    "the buffer size (%d bytes)", newesize, job->fifo.bsize);
        backentry_free(&ep);
        PR_Unlock(job->wire_lock);
        return -1;
    }
    /* Now check if fifo has enough space for the new entry */
    if ((job->fifo.c_bsize + newesize) > job->fifo.bsize) {
        import_wait_for_space_in_fifo( job, newesize );
    }

    /* We have enough space */
    job->fifo.item[idx].filename = "(bulk import)";
    job->fifo.item[idx].line = 0;
    job->fifo.item[idx].entry = ep;
    job->fifo.item[idx].bad = 0;
    job->fifo.item[idx].esize = newesize;

    /* Add the entry size to total fifo size */
    job->fifo.c_bsize += ep->ep_entry? job->fifo.item[idx].esize : 0;

    /* Update the job to show our progress */
    job->lead_ID = id;
    if ((id - job->starting_ID) <= job->fifo.size) {
        job->trailing_ID = job->starting_ID;
    } else {
        job->trailing_ID = id - job->fifo.size;
    }

    PR_Unlock(job->wire_lock);
    return 0;
}

void *factory_constructor(void *object, void *parent)
{
    return NULL;
}

void factory_destructor(void *extension, void *object, void *parent)
{
    ImportJob *job = (ImportJob *)extension;
    PRThread *thread;

    if (extension == NULL)
        return;

    /* connection was destroyed while we were still storing the extension --
     * this is bad news and means we have a bulk import that needs to be
     * aborted!
     */
    thread = job->main_thread;
    LDAPDebug(LDAP_DEBUG_ANY, "ERROR bulk import abandoned\n",
              0, 0, 0);
    import_abort_all(job, 1);
    /* wait for import_main to finish... */
    PR_JoinThread(thread);
    /* extension object is free'd by import_main */
    return;
}

/* plugin entry function for replica init */
int ldbm_back_wire_import(Slapi_PBlock *pb)
{
    struct ldbminfo *li;
    backend *be = NULL;
    ImportJob *job;
    PRThread *thread;
    int state;

    slapi_pblock_get(pb, SLAPI_BACKEND, &be);
    PR_ASSERT(be != NULL);
    li = (struct ldbminfo *)(be->be_database->plg_private);
    slapi_pblock_get(pb, SLAPI_BULK_IMPORT_STATE, &state);
    if (state == SLAPI_BI_STATE_START) {
        /* starting a new import */
        return bulk_import_start(pb);
    }

    PR_ASSERT(pb->pb_conn != NULL);
    if (pb->pb_conn != NULL) {
        job = (ImportJob *)slapi_get_object_extension(li->li_bulk_import_object, pb->pb_conn, li->li_bulk_import_handle);
    }

    if ((job == NULL) || (pb->pb_conn == NULL)) {
        /* import might be aborting */
        return -1;
    }

    if (state == SLAPI_BI_STATE_ADD) {
        /* continuing previous import */
        if (! import_entry_belongs_here(pb->pb_import_entry,
                                        job->inst->inst_be)) {
            /* silently skip */
            return 0;
        }
		/* These days, we don't want to return LDAP_BUSY (it makes pipelineing impossible 
		and actually doesn't achieve anything anyway). So we pass '1' for the block flag. */
        return bulk_import_queue(job, pb->pb_import_entry,
                                        1);
    }

    thread = job->main_thread;

    if (state == SLAPI_BI_STATE_DONE) {
        /* finished with an import */
        job->flags |= FLAG_PRODUCER_DONE;
        /* "job" struct may vanish at any moment after we set the DONE
         * flag, so keep a copy of the thread id in 'thread' for safekeeping.
         */
        /* wait for import_main to finish... */
        PR_JoinThread(thread);
        slapi_set_object_extension(li->li_bulk_import_object, pb->pb_conn,
            li->li_bulk_import_handle, NULL);
        return 0;
    }

    /* ??? unknown state */
    LDAPDebug(LDAP_DEBUG_ANY,
              "ERROR: ldbm_back_wire_import: unknown state %d\n",
              state, 0, 0);
    return -1;
}

/*
 * backup index configuration
 * this function is called from dblayer_backup (ldbm2archive)
 * [547427] index config must not change between backup and restore
 */
#define DSE_INDEX "dse_index.ldif"
#define DSE_INSTANCE "dse_instance.ldif"
#define DSE_INDEX_FILTER "(objectclass=nsIndex)"
#define DSE_INSTANCE_FILTER "(objectclass=nsBackendInstance)"
static int
dse_conf_backup_core(struct ldbminfo *li, char *dest_dir, char *file_name, char *filter)
{
    Slapi_PBlock *srch_pb = NULL;
    Slapi_Entry **entries = NULL;
    Slapi_Entry **ep = NULL;
    Slapi_Attr *attr = NULL;
    char *attr_name;
    char *filename = NULL;
    PRFileDesc *prfd = NULL;
    int rval = 0;
    int dlen = 0;
    PRInt32 prrval;
    char tmpbuf[BUFSIZ];
    char *tp = NULL;
    
    dlen = strlen(dest_dir);
    if (0 == dlen)
    {
        filename = file_name;
    }
    else
    {
        filename = slapi_ch_smprintf("%s/%s", dest_dir, file_name);
    }
    LDAPDebug(LDAP_DEBUG_TRACE, "dse_conf_backup(%s): backup file %s\n",
              filter, filename, 0);

    /* Open the file to write */
    if ((prfd = PR_Open(filename, PR_RDWR | PR_CREATE_FILE | PR_TRUNCATE,
                        SLAPD_DEFAULT_FILE_MODE)) == NULL)
    {
        LDAPDebug(LDAP_DEBUG_ANY,
                "dse_conf_backup(%s): open %s failed: (%s)\n",
                filter, filename, slapd_pr_strerror(PR_GetError()));
        rval = -1;
        goto out;
    }

    srch_pb = slapi_pblock_new();
    slapi_search_internal_set_pb(srch_pb, li->li_plugin->plg_dn,
        LDAP_SCOPE_SUBTREE, filter, NULL, 0, NULL, NULL, li->li_identity, 0);
    slapi_search_internal_pb(srch_pb);
    slapi_pblock_get(srch_pb, SLAPI_PLUGIN_INTOP_SEARCH_ENTRIES, &entries);
    for (ep = entries; ep != NULL && *ep != NULL; ep++)
    {
        size_t l = strlen(slapi_entry_get_dn_const(*ep)) + 5 /* "dn: \n" */;
        LDAPDebug(LDAP_DEBUG_TRACE, "\ndn: %s\n", 
                 slapi_entry_get_dn_const(*ep), 0, 0);

        if (l <= BUFSIZ)
            tp = tmpbuf;
        else
            tp = (char *)slapi_ch_malloc(l);    /* should be very rare ... */
        sprintf(tp, "dn: %s\n", slapi_entry_get_dn_const(*ep));
        prrval = PR_Write(prfd, tp, l);
        if ((size_t)prrval != l)
        {
            LDAPDebug(LDAP_DEBUG_ANY,
                "dse_conf_backup(%s): write %s failed: %d (%s)\n",
                filter, PR_GetError(), slapd_pr_strerror(PR_GetError()));
            rval = -1;
            if (l > BUFSIZ)
                slapi_ch_free_string(&tp);
            goto out;
        }
        if (l > BUFSIZ)
            slapi_ch_free_string(&tp);

        for (slapi_entry_first_attr(*ep, &attr); attr;
             slapi_entry_next_attr(*ep, attr, &attr))
        {
            int i;
            Slapi_Value *sval = NULL;
            const struct berval *attr_val;
            int attr_name_len;

            slapi_attr_get_type(attr, &attr_name);
            /* numsubordinates should not be backed up */
            if (!strcasecmp("numsubordinates", attr_name))
                continue;
            attr_name_len = strlen(attr_name);
            for (i = slapi_attr_first_value(attr, &sval); i != -1;
                 i = slapi_attr_next_value(attr, i, &sval))
            {
                attr_val = slapi_value_get_berval(sval);
                l = strlen(attr_val->bv_val) + attr_name_len + 3; /* : \n" */
                LDAPDebug(LDAP_DEBUG_TRACE, "%s: %s\n", attr_name,
                            attr_val->bv_val, 0);
                if (l <= BUFSIZ)
                    tp = tmpbuf;
                else
                    tp = (char *)slapi_ch_malloc(l);
                sprintf(tp, "%s: %s\n", attr_name, attr_val->bv_val);
                prrval = PR_Write(prfd, tp, l);
                if ((size_t)prrval != l)
                {
                    LDAPDebug(LDAP_DEBUG_ANY,
                        "dse_conf_backup(%s): write %s failed: %d (%s)\n",
                        filter, PR_GetError(), slapd_pr_strerror(PR_GetError()));
                    rval = -1;
                    if (l > BUFSIZ)
                        slapi_ch_free_string(&tp);
                    goto out;
                }
                if (l > BUFSIZ)
                    slapi_ch_free_string(&tp);
            }
        }
        if (ep+1 != NULL && *(ep+1) != NULL)
        {
            prrval = PR_Write(prfd, "\n", 1);
            if ((int)prrval != 1)
            {
                LDAPDebug(LDAP_DEBUG_ANY,
                    "dse_conf_backup(%s): write %s failed: %d (%s)\n",
                    filter, PR_GetError(), slapd_pr_strerror(PR_GetError()));
                rval = -1;
                goto out;
            }
        }
    }

out:
    slapi_free_search_results_internal(srch_pb);
    if (srch_pb)
    {
        slapi_pblock_destroy(srch_pb);
    }

    if (0 != dlen)
    {
        slapi_ch_free_string(&filename);
    }

    if (prfd)
    {
        prrval = PR_Close(prfd);
        if (PR_SUCCESS != prrval)
        {
            LDAPDebug( LDAP_DEBUG_ANY,
                "Fatal Error---Failed to back up dse indexes %d (%s)\n",
                PR_GetError(), slapd_pr_strerror(PR_GetError()), 0);
            rval = -1;
        }
    }

    return rval;
}

int
dse_conf_backup(struct ldbminfo *li, char *dest_dir)
{
    int rval = 0;
    rval  = dse_conf_backup_core(li, dest_dir, DSE_INSTANCE, DSE_INSTANCE_FILTER);
    rval += dse_conf_backup_core(li, dest_dir, DSE_INDEX, DSE_INDEX_FILTER);
    return rval;
}

/*
 * read the backed up index configuration
 * adjust them if the current configuration is different from it.
 * this function is called from dblayer_restore (archive2ldbm)
 * these functions are placed here to borrow import_get_entry
 * [547427] index config must not change between backup and restore
 */
int
dse_conf_verify_core(struct ldbminfo *li, char *src_dir, char *file_name, char *filter, char *log_str, char *entry_filter)
{
    char *filename = NULL;
    int rval = 0;
    ldif_context c;
    int fd = -1;
    int curr_lineno = 0;
    int finished = 0;
    int backup_entry_len = 256;
	char *search_scope = NULL;
    Slapi_Entry **backup_entries = NULL;
    Slapi_Entry **bep = NULL;
    Slapi_Entry **curr_entries = NULL;
    Slapi_PBlock srch_pb;
    
    filename = slapi_ch_smprintf("%s/%s", src_dir, file_name);

    if (PR_SUCCESS != PR_Access(filename, PR_ACCESS_READ_OK))
    {
        LDAPDebug(LDAP_DEBUG_ANY,
            "Warning: config backup file %s not found in backup\n",
            file_name, 0, 0);
        rval = 0;
        goto out;
    }

    fd = dblayer_open_huge_file(filename, O_RDONLY, 0);
    if (fd < 0)
    {
        LDAPDebug(LDAP_DEBUG_ANY,
            "Warning: can't open config backup file: %s\n", filename, 0, 0);
        rval = -1;
        goto out;
    }

    import_init_ldif(&c);
    bep = backup_entries = (Slapi_Entry **)slapi_ch_calloc(1,
                            backup_entry_len * sizeof(Slapi_Entry *));

    while (!finished)
    {
        char *estr = NULL;
        Slapi_Entry *e = NULL;
        estr = import_get_entry(&c, fd, &curr_lineno);

        if (!estr)
            break;
		
		if (entry_filter != NULL) /* Single instance restoration */
		{
			if (!(int)strstr(estr, entry_filter))
				continue;
		}

        e = slapi_str2entry(estr, 0);
        slapi_ch_free_string(&estr);
        if (!e) {
            LDAPDebug(LDAP_DEBUG_ANY, "WARNING: skipping bad LDIF entry "
                "ending line %d of file \"%s\"", curr_lineno, filename, 0);
            continue;
        }
        if (bep - backup_entries >= backup_entry_len)
        {
            backup_entries = (Slapi_Entry **)slapi_ch_realloc((char *)backup_entries, 
                            2 * backup_entry_len * sizeof(Slapi_Entry *));
            bep = backup_entries + backup_entry_len;
            backup_entry_len *= 2;
        }
        *bep = e;
        bep++;
    }
    /* 623986: terminate the list if we reallocated backup_entries */
    if (backup_entry_len > 256)
       *bep = NULL;

    pblock_init(&srch_pb);

	if (entry_filter != NULL)
	{ /* Single instance restoration */
        search_scope = slapi_ch_smprintf("%s,%s", entry_filter, li->li_plugin->plg_dn);
	} else { /* Normal restoration */
        search_scope = slapi_ch_strdup(li->li_plugin->plg_dn);
	}

    slapi_search_internal_set_pb(&srch_pb, search_scope,
        LDAP_SCOPE_SUBTREE, filter, NULL, 0, NULL, NULL, li->li_identity, 0);
    slapi_search_internal_pb(&srch_pb);
    slapi_pblock_get(&srch_pb, SLAPI_PLUGIN_INTOP_SEARCH_ENTRIES, &curr_entries);


    if (0 != slapi_entries_diff(backup_entries, curr_entries, 1 /* test_all */,
                                log_str, 1 /* force_update */, li->li_identity))
    {
        LDAPDebug(LDAP_DEBUG_ANY, "WARNING!!: current %s is "
                  "different from backed up configuration; "
                  "The backup is restored.\n", log_str, 0, 0);
    }

    slapi_free_search_results_internal(&srch_pb);
    pblock_done(&srch_pb);
    import_free_ldif(&c);
out:
    for (bep = backup_entries; bep && *bep; bep++)
        slapi_entry_free(*bep);
    slapi_ch_free((void **)&backup_entries);

    slapi_ch_free_string(&filename);

	slapi_ch_free_string(&search_scope);


    if (fd > 0)
        close(fd);

    return rval;
}

int
dse_conf_verify(struct ldbminfo *li, char *src_dir, char *bename)
{
    int rval;
	char *entry_filter = NULL;
	char *instance_entry_filter = NULL;
	
	if (bename != NULL) /* This was a restore of a single backend */
	{
		/* Entry filter string */
        entry_filter = slapi_ch_smprintf("cn=%s", bename);

		/* Instance search filter */
        instance_entry_filter = slapi_ch_smprintf("(&%s(cn=%s))", DSE_INSTANCE_FILTER, bename);
	} else {
	    instance_entry_filter = slapi_ch_strdup(DSE_INSTANCE_FILTER);
	}

	rval  = dse_conf_verify_core(li, src_dir, DSE_INSTANCE, instance_entry_filter,
                "Instance Config", entry_filter);
    rval += dse_conf_verify_core(li, src_dir, DSE_INDEX, DSE_INDEX_FILTER,
                "Index Config", entry_filter);

	slapi_ch_free_string(&entry_filter);
	slapi_ch_free_string(&instance_entry_filter);

    return rval;
}