ubik sync client error recovery
[openafs.git] / src / ubik / recovery.c
index cbcaae9..cc5d951 100644 (file)
@@ -1,7 +1,7 @@
 /*
  * Copyright 2000, International Business Machines Corporation and others.
  * All Rights Reserved.
- * 
+ *
  * This software has been released under the terms of the IBM Public
  * License.  For details, see the LICENSE file in the top-level source
  * directory or online at http://www.openafs.org/dl/license10.html
 #include <afsconfig.h>
 #include <afs/param.h>
 
-RCSID
-    ("$Header$");
+#include <roken.h>
 
 #include <sys/types.h>
+#include <string.h>
+#include <stdarg.h>
+#include <errno.h>
+
 #ifdef AFS_NT40_ENV
 #include <winsock2.h>
 #include <time.h>
+#include <fcntl.h>
 #else
 #include <sys/file.h>
 #include <netinet/in.h>
 #include <sys/time.h>
 #endif
-#include <assert.h>
+
 #include <lock.h>
-#ifdef HAVE_STRING_H
-#include <string.h>
-#else
-#ifdef HAVE_STRINGS_H
-#include <strings.h>
-#endif
-#endif
 #include <rx/xdr.h>
 #include <rx/rx.h>
-#include <errno.h>
 #include <afs/afsutil.h>
+#include <afs/cellconfig.h>
 
 #define UBIK_INTERNALS
 #include "ubik.h"
 #include "ubik_int.h"
 
-/* This module is responsible for determining when the system has
+/*! \file
+ * This module is responsible for determining when the system has
  * recovered to the point that it can handle new transactions.  It
  * replays logs, polls to determine the current dbase after a crash,
  * and distributes the new database to the others.
- */
-
-/* The sync site associates a version number with each database.  It
+ *
+ * The sync site associates a version number with each database.  It
  * broadcasts the version associated with its current dbase in every
  * one of its beacon messages.  When the sync site send a dbase to a
  * server, it also sends the db's version.  A non-sync site server can
  * tell if it has the right dbase version by simply comparing the
- * version from the beacon message (uvote_dbVersion) with the version
- * associated with the database (ubik_dbase->version).  The sync site
+ * version from the beacon message \p uvote_dbVersion with the version
+ * associated with the database \p ubik_dbase->version.  The sync site
  * itself simply has one counter to keep track of all of this (again
- * ubik_dbase->version).
- */
-
-/* sync site: routine called when the sync site loses its quorum; this
+ * \p ubik_dbase->version).
+ *
+ * sync site: routine called when the sync site loses its quorum; this
  * procedure is called "up" from the beacon package.  It resyncs the
  * dbase and nudges the recovery daemon to try to propagate out the
  * changes.  It also resets the recovery daemon's state, since
@@ -66,33 +62,47 @@ RCSID
  * servers.
  */
 
-/* if this flag is set, then ubik will use only the primary address 
-** ( the address specified in the CellServDB) to contact other 
-** ubik servers. Ubik recovery will not try opening connections 
-** to the alternate interface addresses. 
-*/
+/*!
+ * if this flag is set, then ubik will use only the primary address
+ * (the address specified in the CellServDB) to contact other
+ * ubik servers. Ubik recovery will not try opening connections
+ * to the alternate interface addresses.
+ */
 int ubikPrimaryAddrOnly;
 
 int
 urecovery_ResetState(void)
 {
     urecovery_state = 0;
+#if !defined(AFS_PTHREAD_ENV)
+    /*  No corresponding LWP_WaitProcess found anywhere for this -- klm */
     LWP_NoYieldSignal(&urecovery_state);
+#endif
     return 0;
 }
 
-/* sync site: routine called when a non-sync site server goes down; restarts recovery
- * process to send missing server the new db when it comes back up.
- * This routine should not do anything with variables used by non-sync site servers. 
+/*!
+ * \brief sync site
+ *
+ * routine called when a non-sync site server goes down; restarts recovery
+ * process to send missing server the new db when it comes back up for
+ * non-sync site servers.
+ *
+ * \note This routine should not do anything with variables used by non-sync site servers.
  */
 int
-urecovery_LostServer(void)
+urecovery_LostServer(struct ubik_server *ts)
 {
+    ubeacon_ReinitServer(ts);
+#if !defined(AFS_PTHREAD_ENV)
+    /*  No corresponding LWP_WaitProcess found anywhere for this -- klm */
     LWP_NoYieldSignal(&urecovery_state);
+#endif
     return 0;
 }
 
-/* return true iff we have a current database (called by both sync
+/*!
+ * return true iff we have a current database (called by both sync
  * sites and non-sync sites) How do we determine this?  If we're the
  * sync site, we wait until recovery has finished fetching and
  * re-labelling its dbase (it may still be trying to propagate it out
@@ -101,11 +111,11 @@ urecovery_LostServer(void)
  * and we must have a currently-good sync site.
  */
 int
-urecovery_AllBetter(register struct ubik_dbase *adbase, int areadAny)
+urecovery_AllBetter(struct ubik_dbase *adbase, int areadAny)
 {
-    register afs_int32 rcode;
+    afs_int32 rcode;
 
-    ubik_dprint("allbetter checking\n");
+    ubik_dprint_25("allbetter checking\n");
     rcode = 0;
 
 
@@ -128,30 +138,34 @@ urecovery_AllBetter(register struct ubik_dbase *adbase, int areadAny)
        rcode = 1;
     }
 
-    ubik_dprint("allbetter: returning %d\n", rcode);
+    ubik_dprint_25("allbetter: returning %d\n", rcode);
     return rcode;
 }
 
-/* abort all transactions on this database */
+/*!
+ * \brief abort all transactions on this database
+ */
 int
 urecovery_AbortAll(struct ubik_dbase *adbase)
 {
-    register struct ubik_trans *tt;
+    struct ubik_trans *tt;
     for (tt = adbase->activeTrans; tt; tt = tt->next) {
        udisk_abort(tt);
     }
     return 0;
 }
 
-/* this routine aborts the current remote transaction, if any, if the tid is wrong */
+/*!
+ * \brief this routine aborts the current remote transaction, if any, if the tid is wrong
+ */
 int
-urecovery_CheckTid(register struct ubik_tid *atid)
+urecovery_CheckTid(struct ubik_tid *atid, int abortalways)
 {
     if (ubik_currentTrans) {
        /* there is remote write trans, see if we match, see if this
         * is a new transaction */
        if (atid->epoch != ubik_currentTrans->tid.epoch
-           || atid->counter > ubik_currentTrans->tid.counter) {
+           || atid->counter > ubik_currentTrans->tid.counter || abortalways) {
            /* don't match, abort it */
            /* If the thread is not waiting for lock - ok to end it */
 #if !defined(UBIK_PAUSE)
@@ -164,20 +178,23 @@ urecovery_CheckTid(register struct ubik_tid *atid)
            ubik_currentTrans = (struct ubik_trans *)0;
        }
     }
-    return;
+    return 0;
 }
 
-/* log format is defined here, and implicitly in disk.c
+/*!
+ * \brief replay logs
+ *
+ * log format is defined here, and implicitly in disk.c
  *
  * 4 byte opcode, followed by parameters, each 4 bytes long.  All integers
  * are in logged in network standard byte order, in case we want to move logs
  * from machine-to-machine someday.
  *
- * Begin transaction: opcode
- * Commit transaction: opcode, version (8 bytes)
- * Truncate file: opcode, file number, length
- * Abort transaction: opcode
- * Write data: opcode, file, position, length, <length> data bytes
+ * Begin transaction: opcode \n
+ * Commit transaction: opcode, version (8 bytes) \n
+ * Truncate file: opcode, file number, length \n
+ * Abort transaction: opcode \n
+ * Write data: opcode, file, position, length, <length> data bytes \n
  *
  * A very simple routine, it just replays the log.  Note that this is a new-value only log, which
  * implies that no uncommitted data is written to the dbase: one writes data to the log, including
@@ -188,13 +205,11 @@ urecovery_CheckTid(register struct ubik_tid *atid)
  * abort and the remaining dbase on the disk is exactly the right dbase, without having to read
  * the log.
  */
-
-/* replay logs */
 static int
-ReplayLog(register struct ubik_dbase *adbase)
+ReplayLog(struct ubik_dbase *adbase)
 {
     afs_int32 opcode;
-    register afs_int32 code, tpos;
+    afs_int32 code, tpos;
     int logIsGood;
     afs_int32 len, thisSize, tfile, filePos;
     afs_int32 buffer[4];
@@ -210,10 +225,11 @@ ReplayLog(register struct ubik_dbase *adbase)
     /* for now, assume that all ops in log pertain to one transaction; see if there's a commit */
     while (1) {
        code =
-           (*adbase->read) (adbase, LOGFILE, &opcode, tpos,
+           (*adbase->read) (adbase, LOGFILE, (char *)&opcode, tpos,
                             sizeof(afs_int32));
        if (code != sizeof(afs_int32))
            break;
+       opcode = ntohl(opcode);
        if (opcode == LOGNEW) {
            /* handle begin trans */
            tpos += sizeof(afs_int32);
@@ -225,7 +241,7 @@ ReplayLog(register struct ubik_dbase *adbase)
        } else if (opcode == LOGTRUNCATE) {
            tpos += 4;
            code =
-               (*adbase->read) (adbase, LOGFILE, buffer, tpos,
+               (*adbase->read) (adbase, LOGFILE, (char *)buffer, tpos,
                                 2 * sizeof(afs_int32));
            if (code != 2 * sizeof(afs_int32))
                break;          /* premature eof or io error */
@@ -233,15 +249,15 @@ ReplayLog(register struct ubik_dbase *adbase)
        } else if (opcode == LOGDATA) {
            tpos += 4;
            code =
-               (*adbase->read) (adbase, LOGFILE, buffer, tpos,
+               (*adbase->read) (adbase, LOGFILE, (char *)buffer, tpos,
                                 3 * sizeof(afs_int32));
            if (code != 3 * sizeof(afs_int32))
                break;
            /* otherwise, skip over the data bytes, too */
-           tpos += buffer[2] + 3 * sizeof(afs_int32);
+           tpos += ntohl(buffer[2]) + 3 * sizeof(afs_int32);
        } else {
-           ubik_dprint("corrupt log opcode (%d) at position %d\n", opcode,
-                       tpos);
+           ubik_print("corrupt log opcode (%d) at position %d\n", opcode,
+                      tpos);
            break;              /* corrupt log! */
        }
     }
@@ -253,31 +269,38 @@ ReplayLog(register struct ubik_dbase *adbase)
        syncFile = -1;
        while (1) {
            code =
-               (*adbase->read) (adbase, LOGFILE, &opcode, tpos,
+               (*adbase->read) (adbase, LOGFILE, (char *)&opcode, tpos,
                                 sizeof(afs_int32));
            if (code != sizeof(afs_int32))
                break;
+           opcode = ntohl(opcode);
            if (opcode == LOGNEW) {
                /* handle begin trans */
                tpos += sizeof(afs_int32);
            } else if (opcode == LOGABORT)
                panic("log abort\n");
            else if (opcode == LOGEND) {
+               struct ubik_version version;
                tpos += 4;
                code =
-                   (*adbase->read) (adbase, LOGFILE, buffer, tpos,
+                   (*adbase->read) (adbase, LOGFILE, (char *)buffer, tpos,
                                     2 * sizeof(afs_int32));
                if (code != 2 * sizeof(afs_int32))
                    return UBADLOG;
-               code = (*adbase->setlabel) (adbase, 0, buffer);
+               version.epoch = ntohl(buffer[0]);
+               version.counter = ntohl(buffer[1]);
+               code = (*adbase->setlabel) (adbase, 0, &version);
                if (code)
                    return code;
+               ubik_print("Successfully replayed log for interrupted "
+                          "transaction; db version is now %ld.%ld\n",
+                          (long) version.epoch, (long) version.counter);
                logIsGood = 1;
                break;          /* all done now */
            } else if (opcode == LOGTRUNCATE) {
                tpos += 4;
                code =
-                   (*adbase->read) (adbase, LOGFILE, buffer, tpos,
+                   (*adbase->read) (adbase, LOGFILE, (char *)buffer, tpos,
                                     2 * sizeof(afs_int32));
                if (code != 2 * sizeof(afs_int32))
                    break;      /* premature eof or io error */
@@ -290,7 +313,7 @@ ReplayLog(register struct ubik_dbase *adbase)
            } else if (opcode == LOGDATA) {
                tpos += 4;
                code =
-                   (*adbase->read) (adbase, LOGFILE, buffer, tpos,
+                   (*adbase->read) (adbase, LOGFILE, (char *)buffer, tpos,
                                     3 * sizeof(afs_int32));
                if (code != 3 * sizeof(afs_int32))
                    break;
@@ -313,12 +336,12 @@ ReplayLog(register struct ubik_dbase *adbase)
                    thisSize = (len > sizeof(data) ? sizeof(data) : len);
                    /* copy sizeof(data) buffer bytes at a time */
                    code =
-                       (*adbase->read) (adbase, LOGFILE, data, tpos,
+                       (*adbase->read) (adbase, LOGFILE, (char *)data, tpos,
                                         thisSize);
                    if (code != thisSize)
                        return UBADLOG;
                    code =
-                       (*adbase->write) (adbase, tfile, data, filePos,
+                       (*adbase->write) (adbase, tfile, (char *)data, filePos,
                                          thisSize);
                    if (code != thisSize)
                        return UBADLOG;
@@ -327,8 +350,8 @@ ReplayLog(register struct ubik_dbase *adbase)
                    len -= thisSize;
                }
            } else {
-               ubik_dprint("corrupt log opcode (%d) at position %d\n",
-                           opcode, tpos);
+               ubik_print("corrupt log opcode (%d) at position %d\n",
+                          opcode, tpos);
                break;          /* corrupt log! */
            }
        }
@@ -338,7 +361,7 @@ ReplayLog(register struct ubik_dbase *adbase)
            if (code)
                return code;
        } else {
-           ubik_dprint("Log read error on pass 2\n");
+           ubik_print("Log read error on pass 2\n");
            return UBADLOG;
        }
     }
@@ -348,13 +371,15 @@ ReplayLog(register struct ubik_dbase *adbase)
     return code;
 }
 
-/* Called at initialization to figure out version of the dbase we really have.
+/*! \brief
+ * Called at initialization to figure out version of the dbase we really have.
+ *
  * This routine is called after replaying the log; it reads the restored labels.
  */
 static int
-InitializeDB(register struct ubik_dbase *adbase)
+InitializeDB(struct ubik_dbase *adbase)
 {
-    register afs_int32 code;
+    afs_int32 code;
 
     code = (*adbase->getlabel) (adbase, 0, &adbase->version);
     if (code) {
@@ -368,18 +393,24 @@ InitializeDB(register struct ubik_dbase *adbase)
            adbase->version.counter = 0;
            (*adbase->setlabel) (adbase, 0, &adbase->version);
        }
+#ifdef AFS_PTHREAD_ENV
+       CV_BROADCAST(&adbase->version_cond);
+#else
        LWP_NoYieldSignal(&adbase->version);
+#endif
     }
     return 0;
 }
 
-/* initialize the local dbase
+/*!
+ * \brief initialize the local ubik_dbase
+ *
  * We replay the logs and then read the resulting file to figure out what version we've really got.
  */
 int
-urecovery_Initialize(register struct ubik_dbase *adbase)
+urecovery_Initialize(struct ubik_dbase *adbase)
 {
-    register afs_int32 code;
+    afs_int32 code;
 
     code = ReplayLog(adbase);
     if (code)
@@ -388,12 +419,14 @@ urecovery_Initialize(register struct ubik_dbase *adbase)
     return code;
 }
 
-/* Main interaction loop for the recovery manager
+/*!
+ * \brief Main interaction loop for the recovery manager
+ *
  * The recovery light-weight process only runs when you're the
  * synchronization site.  It performs the following tasks, if and only
  * if the prerequisite tasks have been performed successfully (it
  * keeps track of which ones have been performed in its bit map,
- * urecovery_state).
+ * \p urecovery_state).
  *
  * First, it is responsible for probing that all servers are up.  This
  * is the only operation that must be performed even if this is not
@@ -417,42 +450,56 @@ urecovery_Initialize(register struct ubik_dbase *adbase)
  * requests.  However, the recovery module still has one more task:
  * propagating the dbase out to everyone who is up in the network.
  */
-int
-urecovery_Interact(void)
+void *
+urecovery_Interact(void *dummy)
 {
     afs_int32 code, tcode;
     struct ubik_server *bestServer = NULL;
     struct ubik_server *ts;
     int dbok, doingRPC, now;
-    afs_int32 lastProbeTime, lastDBVCheck;
+    afs_int32 lastProbeTime;
     /* if we're the sync site, the best db version we've found yet */
     static struct ubik_version bestDBVersion;
     struct ubik_version tversion;
     struct timeval tv;
     int length, tlen, offset, file, nbytes;
     struct rx_call *rxcall;
-    char tbuffer[256];
+    char tbuffer[1024];
     struct ubik_stat ubikstat;
     struct in_addr inAddr;
+    char hoststr[16];
+#ifndef OLD_URECOVERY
+    char pbuffer[1028];
+    int fd = -1;
+    afs_int32 pass;
+#endif
 
     /* otherwise, begin interaction */
     urecovery_state = 0;
     lastProbeTime = 0;
-    lastDBVCheck = 0;
     while (1) {
        /* Run through this loop every 4 seconds */
        tv.tv_sec = 4;
        tv.tv_usec = 0;
+#ifdef AFS_PTHREAD_ENV
+       select(0, 0, 0, 0, &tv);
+#else
        IOMGR_Select(0, 0, 0, 0, &tv);
+#endif
 
        ubik_dprint("recovery running in state %x\n", urecovery_state);
 
        /* Every 30 seconds, check all the down servers and mark them
         * as up if they respond. When a server comes up or found to
-        * not be current, then re-find the the best database and 
+        * not be current, then re-find the the best database and
         * propogate it.
         */
        if ((now = FT_ApproxTime()) > 30 + lastProbeTime) {
+
+#ifdef AFS_PTHREAD_ENV
+           DBHOLD(ubik_dbase);
+#endif
+
            for (ts = ubik_servers, doingRPC = 0; ts; ts = ts->next) {
                if (!ts->up) {
                    doingRPC = 1;
@@ -465,6 +512,11 @@ urecovery_Interact(void)
                    urecovery_state &= ~UBIK_RECFOUNDDB;
                }
            }
+
+#ifdef AFS_PTHREAD_ENV
+           DBRELE(ubik_dbase);
+#endif
+
            if (doingRPC)
                now = FT_ApproxTime();
            lastProbeTime = now;
@@ -477,7 +529,7 @@ urecovery_Interact(void)
        }
        urecovery_state |= UBIK_RECSYNCSITE;
 
-       /* If a server has just come up or if we have not found the 
+       /* If a server has just come up or if we have not found the
         * most current database, then go find the most current db.
         */
        if (!(urecovery_state & UBIK_RECFOUNDDB)) {
@@ -513,7 +565,6 @@ urecovery_Interact(void)
                 */
                urecovery_state &= ~UBIK_RECHAVEDB;
            }
-           lastDBVCheck = FT_ApproxTime();
            urecovery_state |= UBIK_RECFOUNDDB;
            urecovery_state &= ~UBIK_RECSENTDB;
        }
@@ -536,11 +587,7 @@ urecovery_Interact(void)
            urecovery_state |= UBIK_RECHAVEDB;
        } else {
            /* we don't have the best version; we should fetch it. */
-#if defined(UBIK_PAUSE)
            DBHOLD(ubik_dbase);
-#else
-           ObtainWriteLock(&ubik_dbase->versionLock);
-#endif /* UBIK_PAUSE */
            urecovery_AbortAll(ubik_dbase);
 
            /* Rx code to do the Bulk fetch */
@@ -549,14 +596,14 @@ urecovery_Interact(void)
            rxcall = rx_NewCall(bestServer->disk_rxcid);
 
            ubik_print("Ubik: Synchronize database with server %s\n",
-                      afs_inet_ntoa(bestServer->addr[0]));
+                      afs_inet_ntoa_r(bestServer->addr[0], hoststr));
 
            code = StartDISK_GetFile(rxcall, file);
            if (code) {
                ubik_dprint("StartDiskGetFile failed=%d\n", code);
                goto FetchEndCall;
            }
-           nbytes = rx_Read(rxcall, &length, sizeof(afs_int32));
+           nbytes = rx_Read(rxcall, (char *)&length, sizeof(afs_int32));
            length = ntohl(length);
            if (nbytes != sizeof(afs_int32)) {
                ubik_dprint("Rx-read length error=%d\n", code = BULK_ERROR);
@@ -564,40 +611,71 @@ urecovery_Interact(void)
                goto FetchEndCall;
            }
 
-           /* Truncate the file firest */
+#ifdef OLD_URECOVERY
+           /* Truncate the file first */
            code = (*ubik_dbase->truncate) (ubik_dbase, file, 0);
            if (code) {
                ubik_dprint("truncate io error=%d\n", code);
                goto FetchEndCall;
            }
-
+           tversion.counter = 0;
+#endif
            /* give invalid label during file transit */
            tversion.epoch = 0;
-           tversion.counter = 0;
            code = (*ubik_dbase->setlabel) (ubik_dbase, file, &tversion);
            if (code) {
                ubik_dprint("setlabel io error=%d\n", code);
                goto FetchEndCall;
            }
+#ifndef OLD_URECOVERY
+           afs_snprintf(pbuffer, sizeof(pbuffer), "%s.DB%s%d.TMP", ubik_dbase->pathName, (file<0)?"SYS":"", (file<0)?-file:file);
+           fd = open(pbuffer, O_CREAT | O_RDWR | O_TRUNC, 0600);
+           if (fd < 0) {
+               code = errno;
+               goto FetchEndCall;
+           }
+           code = lseek(fd, HDRSIZE, 0);
+           if (code != HDRSIZE) {
+               close(fd);
+               goto FetchEndCall;
+           }
+#endif
 
+           pass = 0;
            while (length > 0) {
                tlen = (length > sizeof(tbuffer) ? sizeof(tbuffer) : length);
+#ifndef AFS_PTHREAD_ENV
+               if (pass % 4 == 0)
+                   IOMGR_Poll();
+#endif
                nbytes = rx_Read(rxcall, tbuffer, tlen);
                if (nbytes != tlen) {
                    ubik_dprint("Rx-read bulk error=%d\n", code = BULK_ERROR);
                    code = EIO;
+                   close(fd);
                    goto FetchEndCall;
                }
+#ifdef OLD_URECOVERY
                nbytes =
                    (*ubik_dbase->write) (ubik_dbase, file, tbuffer, offset,
                                          tlen);
+#else
+               nbytes = write(fd, tbuffer, tlen);
+               pass++;
+#endif
                if (nbytes != tlen) {
                    code = UIOERROR;
+                   close(fd);
                    goto FetchEndCall;
                }
                offset += tlen;
                length -= tlen;
            }
+#ifndef OLD_URECOVERY
+           code = close(fd);
+           if (code)
+               goto FetchEndCall;
+#endif
            code = EndDISK_GetFile(rxcall, &tversion);
          FetchEndCall:
            tcode = rx_EndCall(rxcall, code);
@@ -608,27 +686,57 @@ urecovery_Interact(void)
                urecovery_state |= UBIK_RECHAVEDB;
                memcpy(&ubik_dbase->version, &tversion,
                       sizeof(struct ubik_version));
+#ifdef OLD_URECOVERY
                (*ubik_dbase->sync) (ubik_dbase, 0);    /* get data out first */
-               /* after data is good, sync disk with correct label */
-               code =
-                   (*ubik_dbase->setlabel) (ubik_dbase, 0,
-                                            &ubik_dbase->version);
+#else
+               afs_snprintf(tbuffer, sizeof(tbuffer), "%s.DB%s%d", ubik_dbase->pathName, (file<0)?"SYS":"", (file<0)?-file:file);
+#ifdef AFS_NT40_ENV
+               afs_snprintf(pbuffer, sizeof(pbuffer), "%s.DB%s%d.OLD", ubik_dbase->pathName, (file<0)?"SYS":"", (file<0)?-file:file);
+               code = unlink(pbuffer);
+               if (!code)
+                   code = rename(tbuffer, pbuffer);
+               afs_snprintf(pbuffer, sizeof(pbuffer), "%s.DB%s%d.TMP", ubik_dbase->pathName, (file<0)?"SYS":"", (file<0)?-file:file);
+#endif
+               if (!code)
+                   code = rename(pbuffer, tbuffer);
+               if (!code) {
+                   (*ubik_dbase->open) (ubik_dbase, file);
+#endif
+                   /* after data is good, sync disk with correct label */
+                   code =
+                       (*ubik_dbase->setlabel) (ubik_dbase, 0,
+                                                &ubik_dbase->version);
+#ifndef OLD_URECOVERY
+               }
+#ifdef AFS_NT40_ENV
+               afs_snprintf(pbuffer, sizeof(pbuffer), "%s.DB%s%d.OLD", ubik_dbase->pathName, (file<0)?"SYS":"", (file<0)?-file:file);
+               unlink(pbuffer);
+#endif
+#endif
            }
            if (code) {
+#ifndef OLD_URECOVERY
+               unlink(pbuffer);
+               /*
+                * We will effectively invalidate the old data forever now.
+                * Unclear if we *should* but we do.
+                */
+#endif
                ubik_dbase->version.epoch = 0;
                ubik_dbase->version.counter = 0;
                ubik_print("Ubik: Synchronize database failed (error = %d)\n",
                           code);
            } else {
                ubik_print("Ubik: Synchronize database completed\n");
+               urecovery_state |= UBIK_RECHAVEDB;
            }
            udisk_Invalidate(ubik_dbase, 0);    /* data has changed */
+#ifdef AFS_PTHREAD_ENV
+           CV_BROADCAST(&ubik_dbase->version_cond);
+#else
            LWP_NoYieldSignal(&ubik_dbase->version);
-#if defined(UBIK_PAUSE)
+#endif
            DBRELE(ubik_dbase);
-#else
-           ReleaseWriteLock(&ubik_dbase->versionLock);
-#endif /* UBIK_PAUSE */
        }
 #if defined(UBIK_PAUSE)
        if (!(urecovery_state & UBIK_RECSYNCSITE))
@@ -643,11 +751,7 @@ urecovery_Interact(void)
         * database and overwrite this one.
         */
        if (ubik_dbase->version.epoch == 1) {
-#if defined(UBIK_PAUSE)
            DBHOLD(ubik_dbase);
-#else
-           ObtainWriteLock(&ubik_dbase->versionLock);
-#endif /* UBIK_PAUSE */
            urecovery_AbortAll(ubik_dbase);
            ubik_epochTime = 2;
            ubik_dbase->version.epoch = ubik_epochTime;
@@ -655,12 +759,12 @@ urecovery_Interact(void)
            code =
                (*ubik_dbase->setlabel) (ubik_dbase, 0, &ubik_dbase->version);
            udisk_Invalidate(ubik_dbase, 0);    /* data may have changed */
+#ifdef AFS_PTHREAD_ENV
+           CV_BROADCAST(&ubik_dbase->version_cond);
+#else
            LWP_NoYieldSignal(&ubik_dbase->version);
-#if defined(UBIK_PAUSE)
+#endif
            DBRELE(ubik_dbase);
-#else
-           ReleaseWriteLock(&ubik_dbase->versionLock);
-#endif /* UBIK_PAUSE */
        }
 
        /* Check the other sites and send the database to them if they
@@ -670,11 +774,7 @@ urecovery_Interact(void)
            /* now propagate out new version to everyone else */
            dbok = 1;           /* start off assuming they all worked */
 
-#if defined(UBIK_PAUSE)
            DBHOLD(ubik_dbase);
-#else
-           ObtainWriteLock(&ubik_dbase->versionLock);
-#endif /* UBIK_PAUSE */
            /*
             * Check if a write transaction is in progress. We can't send the
             * db when a write is in progress here because the db would be
@@ -690,20 +790,16 @@ urecovery_Interact(void)
                tv.tv_sec = 0;
                tv.tv_usec = 50000;
                while ((ubik_dbase->flags & DBWRITING) && (safety < 500)) {
-#if defined(UBIK_PAUSE)
                    DBRELE(ubik_dbase);
-#else
-                   ReleaseWriteLock(&ubik_dbase->versionLock);
-#endif /* UBIK_PAUSE */
                    /* sleep for a little while */
+#ifdef AFS_PTHREAD_ENV
+                   select(0, 0, 0, 0, &tv);
+#else
                    IOMGR_Select(0, 0, 0, 0, &tv);
+#endif
                    tv.tv_usec += 10000;
                    safety++;
-#if defined(UBIK_PAUSE)
                    DBHOLD(ubik_dbase);
-#else
-                   ObtainWriteLock(&ubik_dbase->versionLock);
-#endif /* UBIK_PAUSE */
                }
            }
 
@@ -711,12 +807,12 @@ urecovery_Interact(void)
                inAddr.s_addr = ts->addr[0];
                if (!ts->up) {
                    ubik_dprint("recovery cannot send version to %s\n",
-                               afs_inet_ntoa(inAddr.s_addr));
+                               afs_inet_ntoa_r(inAddr.s_addr, hoststr));
                    dbok = 0;
                    continue;
                }
                ubik_dprint("recovery sending version to %s\n",
-                           afs_inet_ntoa(inAddr.s_addr));
+                           afs_inet_ntoa_r(inAddr.s_addr, hoststr));
                if (vcmp(ts->version, ubik_dbase->version) != 0) {
                    ubik_dprint("recovery stating local database\n");
 
@@ -770,29 +866,28 @@ urecovery_Interact(void)
                    ts->currentDB = 1;
                }
            }
-#if defined(UBIK_PAUSE)
            DBRELE(ubik_dbase);
-#else
-           ReleaseWriteLock(&ubik_dbase->versionLock);
-#endif /* UBIK_PAUSE */
            if (dbok)
                urecovery_state |= UBIK_RECSENTDB;
        }
     }
+    return NULL;
 }
 
-/*
-** send a Probe to all the network address of this server 
-** Return 0  if success, else return 1
-*/
+/*!
+ * \brief send a Probe to all the network address of this server
+ *
+ * \return 0 if success, else return 1
+ */
 int
 DoProbe(struct ubik_server *server)
 {
     struct rx_connection *conns[UBIK_MAX_INTERFACE_ADDR];
     struct rx_connection *connSuccess = 0;
-    int i, j;
+    int i, j, success_i = -1;
     afs_uint32 addr;
     char buffer[32];
+    char hoststr[16];
     extern afs_int32 ubikSecIndex;
     extern struct rx_securityClass *ubikSecClass;
 
@@ -808,32 +903,46 @@ DoProbe(struct ubik_server *server)
            break;
        }
     }
-    assert(i);                 /* at least one interface address for this server */
+    osi_Assert(i);                     /* at least one interface address for this server */
+
+#ifdef AFS_PTHREAD_ENV
+    DBRELE(ubik_dbase);
+#endif
 
     multi_Rx(conns, i) {
        multi_DISK_Probe();
        if (!multi_error) {     /* first success */
-           addr = server->addr[multi_i];       /* successful interface addr */
-
-           if (server->disk_rxcid)     /* destroy existing conn */
-               rx_DestroyConnection(server->disk_rxcid);
-           if (server->vote_rxcid)
-               rx_DestroyConnection(server->vote_rxcid);
-
-           /* make new connections */
-           server->disk_rxcid = conns[multi_i];
-           server->vote_rxcid = rx_NewConnection(addr, ubik_callPortal, VOTE_SERVICE_ID, ubikSecClass, ubikSecIndex);  /* for vote reqs */
-
-           connSuccess = conns[multi_i];
-           strcpy(buffer, (char *)afs_inet_ntoa(server->addr[0]));
-           ubik_print
-               ("ubik:server %s is back up: will be contacted through %s\n",
-                buffer, afs_inet_ntoa(addr));
+           success_i = multi_i;
 
            multi_Abort;
        }
     } multi_End_Ignore;
 
+#ifdef AFS_PTHREAD_ENV
+    DBHOLD(ubik_dbase);
+#endif
+
+    if (success_i >= 0) {
+       addr = server->addr[success_i]; /* successful interface addr */
+
+       if (server->disk_rxcid) /* destroy existing conn */
+           rx_DestroyConnection(server->disk_rxcid);
+       if (server->vote_rxcid)
+           rx_DestroyConnection(server->vote_rxcid);
+
+       /* make new connections */
+       server->disk_rxcid = conns[success_i];
+       server->vote_rxcid = rx_NewConnection(addr, ubik_callPortal,
+                                             VOTE_SERVICE_ID, ubikSecClass,
+                                             ubikSecIndex);
+
+       connSuccess = conns[success_i];
+       strcpy(buffer, afs_inet_ntoa_r(server->addr[0], hoststr));
+
+       ubik_print("ubik:server %s is back up: will be contacted through %s\n",
+            buffer, afs_inet_ntoa_r(addr, hoststr));
+    }
+
     /* Destroy all connections except the one on which we succeeded */
     for (j = 0; j < i; j++)
        if (conns[j] != connSuccess)
@@ -841,7 +950,7 @@ DoProbe(struct ubik_server *server)
 
     if (!connSuccess)
        ubik_dprint("ubik:server %s still down\n",
-                   afs_inet_ntoa(server->addr[0]));
+                   afs_inet_ntoa_r(server->addr[0], hoststr));
 
     if (connSuccess)
        return 0;               /* success */