src/ubik/recovery.c

   1 /*
   2  * Copyright 2000, International Business Machines Corporation and others.
   3  * All Rights Reserved.
   4  *
   5  * This software has been released under the terms of the IBM Public
   6  * License.  For details, see the LICENSE file in the top-level source
   7  * directory or online at http://www.openafs.org/dl/license10.html
   8  */
   9
  10 #include <afsconfig.h>
  11 #include <afs/param.h>
  12
  13
  14 #include <sys/types.h>
  15 #include <string.h>
  16 #include <stdarg.h>
  17 #include <errno.h>
  18 #include <assert.h>
  19
  20 #ifdef AFS_NT40_ENV
  21 #include <winsock2.h>
  22 #include <time.h>
  23 #include <fcntl.h>
  24 #else
  25 #include <sys/file.h>
  26 #include <netinet/in.h>
  27 #include <sys/time.h>
  28 #endif
  29
  30 #include <lock.h>
  31 #include <rx/xdr.h>
  32 #include <rx/rx.h>
  33 #include <afs/afsutil.h>
  34
  35 #define UBIK_INTERNALS
  36 #include "ubik.h"
  37 #include "ubik_int.h"
  38
  39 /*! \file
  40  * This module is responsible for determining when the system has
  41  * recovered to the point that it can handle new transactions.  It
  42  * replays logs, polls to determine the current dbase after a crash,
  43  * and distributes the new database to the others.
  44  *
  45  * The sync site associates a version number with each database.  It
  46  * broadcasts the version associated with its current dbase in every
  47  * one of its beacon messages.  When the sync site send a dbase to a
  48  * server, it also sends the db's version.  A non-sync site server can
  49  * tell if it has the right dbase version by simply comparing the
  50  * version from the beacon message \p uvote_dbVersion with the version
  51  * associated with the database \p ubik_dbase->version.  The sync site
  52  * itself simply has one counter to keep track of all of this (again
  53  * \p ubik_dbase->version).
  54  *
  55  * sync site: routine called when the sync site loses its quorum; this
  56  * procedure is called "up" from the beacon package.  It resyncs the
  57  * dbase and nudges the recovery daemon to try to propagate out the
  58  * changes.  It also resets the recovery daemon's state, since
  59  * recovery must potentially find a new dbase to propagate out.  This
  60  * routine should not do anything with variables used by non-sync site
  61  * servers.
  62  */
  63
  64 /*!
  65  * if this flag is set, then ubik will use only the primary address
  66  * (the address specified in the CellServDB) to contact other
  67  * ubik servers. Ubik recovery will not try opening connections
  68  * to the alternate interface addresses.
  69  */
  70 int ubikPrimaryAddrOnly;
  71
  72 int
  73 urecovery_ResetState(void)
  74 {
  75     urecovery_state = 0;
  76 #if !defined(AFS_PTHREAD_ENV)
  77     /*  No corresponding LWP_WaitProcess found anywhere for this -- klm */
  78     LWP_NoYieldSignal(&urecovery_state);
  79 #endif
  80     return 0;
  81 }
  82
  83 /*!
  84  * \brief sync site
  85  *
  86  * routine called when a non-sync site server goes down; restarts recovery
  87  * process to send missing server the new db when it comes back up.
  88  *
  89  * \note This routine should not do anything with variables used by non-sync site servers.
  90  */
  91 int
  92 urecovery_LostServer(void)
  93 {
  94 #if !defined(AFS_PTHREAD_ENV)
  95     /*  No corresponding LWP_WaitProcess found anywhere for this -- klm */
  96     LWP_NoYieldSignal(&urecovery_state);
  97 #endif
  98     return 0;
  99 }
 100
 101 /*!
 102  * return true iff we have a current database (called by both sync
 103  * sites and non-sync sites) How do we determine this?  If we're the
 104  * sync site, we wait until recovery has finished fetching and
 105  * re-labelling its dbase (it may still be trying to propagate it out
 106  * to everyone else; that's THEIR problem).  If we're not the sync
 107  * site, then we must have a dbase labelled with the right version,
 108  * and we must have a currently-good sync site.
 109  */
 110 int
 111 urecovery_AllBetter(register struct ubik_dbase *adbase, int areadAny)
 112 {
 113     register afs_int32 rcode;
 114
 115     ubik_dprint_25("allbetter checking\n");
 116     rcode = 0;
 117
 118
 119     if (areadAny) {
 120         if (ubik_dbase->version.epoch > 1)
 121             rcode = 1;          /* Happy with any good version of database */
 122     }
 123
 124     /* Check if we're sync site and we've got the right data */
 125     else if (ubeacon_AmSyncSite() && (urecovery_state & UBIK_RECHAVEDB)) {
 126         rcode = 1;
 127     }
 128
 129     /* next, check if we're aux site, and we've ever been sent the
 130      * right data (note that if a dbase update fails, we won't think
 131      * that the sync site is still the sync site, 'cause it won't talk
 132      * to us until a timeout period has gone by.  When we recover, we
 133      * leave this clear until we get a new dbase */
 134     else if ((uvote_GetSyncSite() && (vcmp(ubik_dbVersion, ubik_dbase->version) == 0))) {       /* && order is important */
 135         rcode = 1;
 136     }
 137
 138     ubik_dprint_25("allbetter: returning %d\n", rcode);
 139     return rcode;
 140 }
 141
 142 /*!
 143  * \brief abort all transactions on this database
 144  */
 145 int
 146 urecovery_AbortAll(struct ubik_dbase *adbase)
 147 {
 148     register struct ubik_trans *tt;
 149     for (tt = adbase->activeTrans; tt; tt = tt->next) {
 150         udisk_abort(tt);
 151     }
 152     return 0;
 153 }
 154
 155 /*!
 156  * \brief this routine aborts the current remote transaction, if any, if the tid is wrong
 157  */
 158 int
 159 urecovery_CheckTid(register struct ubik_tid *atid)
 160 {
 161     if (ubik_currentTrans) {
 162         /* there is remote write trans, see if we match, see if this
 163          * is a new transaction */
 164         if (atid->epoch != ubik_currentTrans->tid.epoch
 165             || atid->counter > ubik_currentTrans->tid.counter) {
 166             /* don't match, abort it */
 167             /* If the thread is not waiting for lock - ok to end it */
 168 #if !defined(UBIK_PAUSE)
 169             if (ubik_currentTrans->locktype != LOCKWAIT) {
 170 #endif /* UBIK_PAUSE */
 171                 udisk_end(ubik_currentTrans);
 172 #if !defined(UBIK_PAUSE)
 173             }
 174 #endif /* UBIK_PAUSE */
 175             ubik_currentTrans = (struct ubik_trans *)0;
 176         }
 177     }
 178     return 0;
 179 }
 180
 181 /*!
 182  * \brief replay logs
 183  *
 184  * log format is defined here, and implicitly in disk.c
 185  *
 186  * 4 byte opcode, followed by parameters, each 4 bytes long.  All integers
 187  * are in logged in network standard byte order, in case we want to move logs
 188  * from machine-to-machine someday.
 189  *
 190  * Begin transaction: opcode \n
 191  * Commit transaction: opcode, version (8 bytes) \n
 192  * Truncate file: opcode, file number, length \n
 193  * Abort transaction: opcode \n
 194  * Write data: opcode, file, position, length, <length> data bytes \n
 195  *
 196  * A very simple routine, it just replays the log.  Note that this is a new-value only log, which
 197  * implies that no uncommitted data is written to the dbase: one writes data to the log, including
 198  * the commit record, then we allow data to be written through to the dbase.  In our particular
 199  * implementation, once a transaction is done, we write out the pages to the database, so that
 200  * our buffer package doesn't have to know about stable and uncommitted data in the memory buffers:
 201  * any changed data while there is an uncommitted write transaction can be zapped during an
 202  * abort and the remaining dbase on the disk is exactly the right dbase, without having to read
 203  * the log.
 204  */
 205 static int
 206 ReplayLog(register struct ubik_dbase *adbase)
 207 {
 208     afs_int32 opcode;
 209     register afs_int32 code, tpos;
 210     int logIsGood;
 211     afs_int32 len, thisSize, tfile, filePos;
 212     afs_int32 buffer[4];
 213     afs_int32 syncFile = -1;
 214     afs_int32 data[1024];
 215
 216     /* read the lock twice, once to see whether we have a transaction to deal
 217      * with that committed, (theoretically, we should support more than one
 218      * trans in the log at once, but not yet), and once replaying the
 219      * transactions.  */
 220     tpos = 0;
 221     logIsGood = 0;
 222     /* for now, assume that all ops in log pertain to one transaction; see if there's a commit */
 223     while (1) {
 224         code =
 225             (*adbase->read) (adbase, LOGFILE, (char *)&opcode, tpos,
 226                              sizeof(afs_int32));
 227         if (code != sizeof(afs_int32))
 228             break;
 229         opcode = ntohl(opcode);
 230         if (opcode == LOGNEW) {
 231             /* handle begin trans */
 232             tpos += sizeof(afs_int32);
 233         } else if (opcode == LOGABORT)
 234             break;
 235         else if (opcode == LOGEND) {
 236             logIsGood = 1;
 237             break;
 238         } else if (opcode == LOGTRUNCATE) {
 239             tpos += 4;
 240             code =
 241                 (*adbase->read) (adbase, LOGFILE, (char *)buffer, tpos,
 242                                  2 * sizeof(afs_int32));
 243             if (code != 2 * sizeof(afs_int32))
 244                 break;          /* premature eof or io error */
 245             tpos += 2 * sizeof(afs_int32);
 246         } else if (opcode == LOGDATA) {
 247             tpos += 4;
 248             code =
 249                 (*adbase->read) (adbase, LOGFILE, (char *)buffer, tpos,
 250                                  3 * sizeof(afs_int32));
 251             if (code != 3 * sizeof(afs_int32))
 252                 break;
 253             /* otherwise, skip over the data bytes, too */
 254             tpos += ntohl(buffer[2]) + 3 * sizeof(afs_int32);
 255         } else {
 256             ubik_dprint("corrupt log opcode (%d) at position %d\n", opcode,
 257                         tpos);
 258             break;              /* corrupt log! */
 259         }
 260     }
 261     if (logIsGood) {
 262         /* actually do the replay; log should go all the way through the commit record, since
 263          * we just read it above. */
 264         tpos = 0;
 265         logIsGood = 0;
 266         syncFile = -1;
 267         while (1) {
 268             code =
 269                 (*adbase->read) (adbase, LOGFILE, (char *)&opcode, tpos,
 270                                  sizeof(afs_int32));
 271             if (code != sizeof(afs_int32))
 272                 break;
 273             opcode = ntohl(opcode);
 274             if (opcode == LOGNEW) {
 275                 /* handle begin trans */
 276                 tpos += sizeof(afs_int32);
 277             } else if (opcode == LOGABORT)
 278                 panic("log abort\n");
 279             else if (opcode == LOGEND) {
 280                 tpos += 4;
 281                 code =
 282                     (*adbase->read) (adbase, LOGFILE, (char *)buffer, tpos,
 283                                      2 * sizeof(afs_int32));
 284                 if (code != 2 * sizeof(afs_int32))
 285                     return UBADLOG;
 286                 code = (*adbase->setlabel) (adbase, 0, (ubik_version *)buffer);
 287                 if (code)
 288                     return code;
 289                 logIsGood = 1;
 290                 break;          /* all done now */
 291             } else if (opcode == LOGTRUNCATE) {
 292                 tpos += 4;
 293                 code =
 294                     (*adbase->read) (adbase, LOGFILE, (char *)buffer, tpos,
 295                                      2 * sizeof(afs_int32));
 296                 if (code != 2 * sizeof(afs_int32))
 297                     break;      /* premature eof or io error */
 298                 tpos += 2 * sizeof(afs_int32);
 299                 code =
 300                     (*adbase->truncate) (adbase, ntohl(buffer[0]),
 301                                          ntohl(buffer[1]));
 302                 if (code)
 303                     return code;
 304             } else if (opcode == LOGDATA) {
 305                 tpos += 4;
 306                 code =
 307                     (*adbase->read) (adbase, LOGFILE, (char *)buffer, tpos,
 308                                      3 * sizeof(afs_int32));
 309                 if (code != 3 * sizeof(afs_int32))
 310                     break;
 311                 tpos += 3 * sizeof(afs_int32);
 312                 /* otherwise, skip over the data bytes, too */
 313                 len = ntohl(buffer[2]); /* total number of bytes to copy */
 314                 filePos = ntohl(buffer[1]);
 315                 tfile = ntohl(buffer[0]);
 316                 /* try to minimize file syncs */
 317                 if (syncFile != tfile) {
 318                     if (syncFile >= 0)
 319                         code = (*adbase->sync) (adbase, syncFile);
 320                     else
 321                         code = 0;
 322                     syncFile = tfile;
 323                     if (code)
 324                         return code;
 325                 }
 326                 while (len > 0) {
 327                     thisSize = (len > sizeof(data) ? sizeof(data) : len);
 328                     /* copy sizeof(data) buffer bytes at a time */
 329                     code =
 330                         (*adbase->read) (adbase, LOGFILE, (char *)data, tpos,
 331                                          thisSize);
 332                     if (code != thisSize)
 333                         return UBADLOG;
 334                     code =
 335                         (*adbase->write) (adbase, tfile, (char *)data, filePos,
 336                                           thisSize);
 337                     if (code != thisSize)
 338                         return UBADLOG;
 339                     filePos += thisSize;
 340                     tpos += thisSize;
 341                     len -= thisSize;
 342                 }
 343             } else {
 344                 ubik_dprint("corrupt log opcode (%d) at position %d\n",
 345                             opcode, tpos);
 346                 break;          /* corrupt log! */
 347             }
 348         }
 349         if (logIsGood) {
 350             if (syncFile >= 0)
 351                 code = (*adbase->sync) (adbase, syncFile);
 352             if (code)
 353                 return code;
 354         } else {
 355             ubik_dprint("Log read error on pass 2\n");
 356             return UBADLOG;
 357         }
 358     }
 359
 360     /* now truncate the log, we're done with it */
 361     code = (*adbase->truncate) (adbase, LOGFILE, 0);
 362     return code;
 363 }
 364
 365 /*! \brief
 366  * Called at initialization to figure out version of the dbase we really have.
 367  *
 368  * This routine is called after replaying the log; it reads the restored labels.
 369  */
 370 static int
 371 InitializeDB(register struct ubik_dbase *adbase)
 372 {
 373     register afs_int32 code;
 374
 375     code = (*adbase->getlabel) (adbase, 0, &adbase->version);
 376     if (code) {
 377         /* try setting the label to a new value */
 378         adbase->version.epoch = 1;      /* value for newly-initialized db */
 379         adbase->version.counter = 1;
 380         code = (*adbase->setlabel) (adbase, 0, &adbase->version);
 381         if (code) {
 382             /* failed, try to set it back */
 383             adbase->version.epoch = 0;
 384             adbase->version.counter = 0;
 385             (*adbase->setlabel) (adbase, 0, &adbase->version);
 386         }
 387 #ifdef AFS_PTHREAD_ENV
 388         assert(pthread_cond_broadcast(&adbase->version_cond) == 0);
 389 #else
 390         LWP_NoYieldSignal(&adbase->version);
 391 #endif
 392     }
 393     return 0;
 394 }
 395
 396 /*!
 397  * \brief initialize the local ubik_dbase
 398  *
 399  * We replay the logs and then read the resulting file to figure out what version we've really got.
 400  */
 401 int
 402 urecovery_Initialize(register struct ubik_dbase *adbase)
 403 {
 404     register afs_int32 code;
 405
 406     code = ReplayLog(adbase);
 407     if (code)
 408         return code;
 409     code = InitializeDB(adbase);
 410     return code;
 411 }
 412
 413 /*!
 414  * \brief Main interaction loop for the recovery manager
 415  *
 416  * The recovery light-weight process only runs when you're the
 417  * synchronization site.  It performs the following tasks, if and only
 418  * if the prerequisite tasks have been performed successfully (it
 419  * keeps track of which ones have been performed in its bit map,
 420  * \p urecovery_state).
 421  *
 422  * First, it is responsible for probing that all servers are up.  This
 423  * is the only operation that must be performed even if this is not
 424  * yet the sync site, since otherwise this site may not notice that
 425  * enough other machines are running to even elect this guy to be the
 426  * sync site.
 427  *
 428  * After that, the recovery process does nothing until the beacon and
 429  * voting modules manage to get this site elected sync site.
 430  *
 431  * After becoming sync site, recovery first attempts to find the best
 432  * database available in the network (it must do this in order to
 433  * ensure finding the latest committed data).  After finding the right
 434  * database, it must fetch this dbase to the sync site.
 435  *
 436  * After fetching the dbase, it relabels it with a new version number,
 437  * to ensure that everyone recognizes this dbase as the most recent
 438  * dbase.
 439  *
 440  * One the dbase has been relabelled, this machine can start handling
 441  * requests.  However, the recovery module still has one more task:
 442  * propagating the dbase out to everyone who is up in the network.
 443  */
 444 void *
 445 urecovery_Interact(void *dummy)
 446 {
 447     afs_int32 code, tcode;
 448     struct ubik_server *bestServer = NULL;
 449     struct ubik_server *ts;
 450     int dbok, doingRPC, now;
 451     afs_int32 lastProbeTime, lastDBVCheck;
 452     /* if we're the sync site, the best db version we've found yet */
 453     static struct ubik_version bestDBVersion;
 454     struct ubik_version tversion;
 455     struct timeval tv;
 456     int length, tlen, offset, file, nbytes;
 457     struct rx_call *rxcall;
 458     char tbuffer[1024];
 459     struct ubik_stat ubikstat;
 460     struct in_addr inAddr;
 461     char hoststr[16];
 462 #ifndef OLD_URECOVERY
 463     char pbuffer[1028];
 464     int flen, fd = -1;
 465     afs_int32 pass;
 466 #endif
 467
 468     /* otherwise, begin interaction */
 469     urecovery_state = 0;
 470     lastProbeTime = 0;
 471     lastDBVCheck = 0;
 472     while (1) {
 473         /* Run through this loop every 4 seconds */
 474         tv.tv_sec = 4;
 475         tv.tv_usec = 0;
 476 #ifdef AFS_PTHREAD_ENV
 477         select(0, 0, 0, 0, &tv);
 478 #else
 479         IOMGR_Select(0, 0, 0, 0, &tv);
 480 #endif
 481
 482         ubik_dprint("recovery running in state %x\n", urecovery_state);
 483
 484         /* Every 30 seconds, check all the down servers and mark them
 485          * as up if they respond. When a server comes up or found to
 486          * not be current, then re-find the the best database and
 487          * propogate it.
 488          */
 489         if ((now = FT_ApproxTime()) > 30 + lastProbeTime) {
 490             for (ts = ubik_servers, doingRPC = 0; ts; ts = ts->next) {
 491                 if (!ts->up) {
 492                     doingRPC = 1;
 493                     code = DoProbe(ts);
 494                     if (code == 0) {
 495                         ts->up = 1;
 496                         urecovery_state &= ~UBIK_RECFOUNDDB;
 497                     }
 498                 } else if (!ts->currentDB) {
 499                     urecovery_state &= ~UBIK_RECFOUNDDB;
 500                 }
 501             }
 502             if (doingRPC)
 503                 now = FT_ApproxTime();
 504             lastProbeTime = now;
 505         }
 506
 507         /* Mark whether we are the sync site */
 508         if (!ubeacon_AmSyncSite()) {
 509             urecovery_state &= ~UBIK_RECSYNCSITE;
 510             continue;           /* nothing to do */
 511         }
 512         urecovery_state |= UBIK_RECSYNCSITE;
 513
 514         /* If a server has just come up or if we have not found the
 515          * most current database, then go find the most current db.
 516          */
 517         if (!(urecovery_state & UBIK_RECFOUNDDB)) {
 518             bestServer = (struct ubik_server *)0;
 519             bestDBVersion.epoch = 0;
 520             bestDBVersion.counter = 0;
 521             for (ts = ubik_servers; ts; ts = ts->next) {
 522                 if (!ts->up)
 523                     continue;   /* don't bother with these guys */
 524                 if (ts->isClone)
 525                     continue;
 526                 code = DISK_GetVersion(ts->disk_rxcid, &ts->version);
 527                 if (code == 0) {
 528                     /* perhaps this is the best version */
 529                     if (vcmp(ts->version, bestDBVersion) > 0) {
 530                         /* new best version */
 531                         bestDBVersion = ts->version;
 532                         bestServer = ts;
 533                     }
 534                 }
 535             }
 536             /* take into consideration our version. Remember if we,
 537              * the sync site, have the best version. Also note that
 538              * we may need to send the best version out.
 539              */
 540             if (vcmp(ubik_dbase->version, bestDBVersion) >= 0) {
 541                 bestDBVersion = ubik_dbase->version;
 542                 bestServer = (struct ubik_server *)0;
 543                 urecovery_state |= UBIK_RECHAVEDB;
 544             } else {
 545                 /* Clear the flag only when we know we have to retrieve
 546                  * the db. Because urecovery_AllBetter() looks at it.
 547                  */
 548                 urecovery_state &= ~UBIK_RECHAVEDB;
 549             }
 550             lastDBVCheck = FT_ApproxTime();
 551             urecovery_state |= UBIK_RECFOUNDDB;
 552             urecovery_state &= ~UBIK_RECSENTDB;
 553         }
 554 #if defined(UBIK_PAUSE)
 555         /* it's not possible for UBIK_RECFOUNDDB not to be set here.
 556          * However, we might have lost UBIK_RECSYNCSITE, and that
 557          * IS important.
 558          */
 559         if (!(urecovery_state & UBIK_RECSYNCSITE))
 560             continue;           /* lost sync */
 561 #else
 562         if (!(urecovery_state & UBIK_RECFOUNDDB))
 563             continue;           /* not ready */
 564 #endif /* UBIK_PAUSE */
 565
 566         /* If we, the sync site, do not have the best db version, then
 567          * go and get it from the server that does.
 568          */
 569         if ((urecovery_state & UBIK_RECHAVEDB) || !bestServer) {
 570             urecovery_state |= UBIK_RECHAVEDB;
 571         } else {
 572             /* we don't have the best version; we should fetch it. */
 573             DBHOLD(ubik_dbase);
 574             urecovery_AbortAll(ubik_dbase);
 575
 576             /* Rx code to do the Bulk fetch */
 577             file = 0;
 578             offset = 0;
 579             rxcall = rx_NewCall(bestServer->disk_rxcid);
 580
 581             ubik_print("Ubik: Synchronize database with server %s\n",
 582                        afs_inet_ntoa_r(bestServer->addr[0], hoststr));
 583
 584             code = StartDISK_GetFile(rxcall, file);
 585             if (code) {
 586                 ubik_dprint("StartDiskGetFile failed=%d\n", code);
 587                 goto FetchEndCall;
 588             }
 589             nbytes = rx_Read(rxcall, (char *)&length, sizeof(afs_int32));
 590             length = ntohl(length);
 591             if (nbytes != sizeof(afs_int32)) {
 592                 ubik_dprint("Rx-read length error=%d\n", code = BULK_ERROR);
 593                 code = EIO;
 594                 goto FetchEndCall;
 595             }
 596
 597 #ifdef OLD_URECOVERY
 598             /* Truncate the file first */
 599             code = (*ubik_dbase->truncate) (ubik_dbase, file, 0);
 600             if (code) {
 601                 ubik_dprint("truncate io error=%d\n", code);
 602                 goto FetchEndCall;
 603             }
 604             tversion.counter = 0;
 605 #endif
 606             /* give invalid label during file transit */
 607             tversion.epoch = 0;
 608             code = (*ubik_dbase->setlabel) (ubik_dbase, file, &tversion);
 609             if (code) {
 610                 ubik_dprint("setlabel io error=%d\n", code);
 611                 goto FetchEndCall;
 612             }
 613 #ifndef OLD_URECOVERY
 614             flen = length;
 615             afs_snprintf(pbuffer, sizeof(pbuffer), "%s.DB%s%d.TMP", ubik_dbase->pathName, (file<0)?"SYS":"", (file<0)?-file:file);
 616             fd = open(pbuffer, O_CREAT | O_RDWR | O_TRUNC, 0600);
 617             if (fd < 0) {
 618                 code = errno;
 619                 goto FetchEndCall;
 620             }
 621             code = lseek(fd, HDRSIZE, 0);
 622             if (code != HDRSIZE) {
 623                 close(fd);
 624                 goto FetchEndCall;
 625             }
 626 #endif
 627
 628             pass = 0;
 629             while (length > 0) {
 630                 tlen = (length > sizeof(tbuffer) ? sizeof(tbuffer) : length);
 631 #ifndef AFS_PTHREAD_ENV
 632                 if (pass % 4 == 0)
 633                     IOMGR_Poll();
 634 #endif
 635                 nbytes = rx_Read(rxcall, tbuffer, tlen);
 636                 if (nbytes != tlen) {
 637                     ubik_dprint("Rx-read bulk error=%d\n", code = BULK_ERROR);
 638                     code = EIO;
 639                     close(fd);
 640                     goto FetchEndCall;
 641                 }
 642 #ifdef OLD_URECOVERY
 643                 nbytes =
 644                     (*ubik_dbase->write) (ubik_dbase, file, tbuffer, offset,
 645                                           tlen);
 646 #else
 647                 nbytes = write(fd, tbuffer, tlen);
 648                 pass++;
 649 #endif
 650                 if (nbytes != tlen) {
 651                     code = UIOERROR;
 652                     close(fd);
 653                     goto FetchEndCall;
 654                 }
 655                 offset += tlen;
 656                 length -= tlen;
 657             }
 658 #ifndef OLD_URECOVERY
 659             code = close(fd);
 660             if (code)
 661                 goto FetchEndCall;
 662 #endif
 663             code = EndDISK_GetFile(rxcall, &tversion);
 664           FetchEndCall:
 665             tcode = rx_EndCall(rxcall, code);
 666             if (!code)
 667                 code = tcode;
 668             if (!code) {
 669                 /* we got a new file, set up its header */
 670                 urecovery_state |= UBIK_RECHAVEDB;
 671                 memcpy(&ubik_dbase->version, &tversion,
 672                        sizeof(struct ubik_version));
 673 #ifdef OLD_URECOVERY
 674                 (*ubik_dbase->sync) (ubik_dbase, 0);    /* get data out first */
 675 #else
 676                 afs_snprintf(tbuffer, sizeof(tbuffer), "%s.DB%s%d", ubik_dbase->pathName, (file<0)?"SYS":"", (file<0)?-file:file);
 677 #ifdef AFS_NT40_ENV
 678                 afs_snprintf(pbuffer, sizeof(pbuffer), "%s.DB%s%d.OLD", ubik_dbase->pathName, (file<0)?"SYS":"", (file<0)?-file:file);
 679                 code = unlink(pbuffer);
 680                 if (!code)
 681                     code = rename(tbuffer, pbuffer);
 682                 afs_snprintf(pbuffer, sizeof(pbuffer), "%s.DB%s%d.TMP", ubik_dbase->pathName, (file<0)?"SYS":"", (file<0)?-file:file);
 683 #endif
 684                 if (!code)
 685                     code = rename(pbuffer, tbuffer);
 686                 if (!code) {
 687                     (*ubik_dbase->open) (ubik_dbase, 0);
 688 #endif
 689                     /* after data is good, sync disk with correct label */
 690                     code =
 691                         (*ubik_dbase->setlabel) (ubik_dbase, 0,
 692                                                  &ubik_dbase->version);
 693 #ifndef OLD_URECOVERY
 694                 }
 695 #ifdef AFS_NT40_ENV
 696                 afs_snprintf(pbuffer, sizeof(pbuffer), "%s.DB%s%d.OLD", ubik_dbase->pathName, (file<0)?"SYS":"", (file<0)?-file:file);
 697                 unlink(pbuffer);
 698 #endif
 699 #endif
 700             }
 701             if (code) {
 702 #ifndef OLD_URECOVERY
 703                 unlink(pbuffer);
 704                 /*
 705                  * We will effectively invalidate the old data forever now.
 706                  * Unclear if we *should* but we do.
 707                  */
 708 #endif
 709                 ubik_dbase->version.epoch = 0;
 710                 ubik_dbase->version.counter = 0;
 711                 ubik_print("Ubik: Synchronize database failed (error = %d)\n",
 712                            code);
 713             } else {
 714                 ubik_print("Ubik: Synchronize database completed\n");
 715                 urecovery_state |= UBIK_RECHAVEDB;
 716             }
 717             udisk_Invalidate(ubik_dbase, 0);    /* data has changed */
 718 #ifdef AFS_PTHREAD_ENV
 719             assert(pthread_cond_broadcast(&ubik_dbase->version_cond) == 0);
 720 #else
 721             LWP_NoYieldSignal(&ubik_dbase->version);
 722 #endif
 723             DBRELE(ubik_dbase);
 724         }
 725 #if defined(UBIK_PAUSE)
 726         if (!(urecovery_state & UBIK_RECSYNCSITE))
 727             continue;           /* lost sync */
 728 #endif /* UBIK_PAUSE */
 729         if (!(urecovery_state & UBIK_RECHAVEDB))
 730             continue;           /* not ready */
 731
 732         /* If the database was newly initialized, then when we establish quorum, write
 733          * a new label. This allows urecovery_AllBetter() to allow access for reads.
 734          * Setting it to 2 also allows another site to come along with a newer
 735          * database and overwrite this one.
 736          */
 737         if (ubik_dbase->version.epoch == 1) {
 738             DBHOLD(ubik_dbase);
 739             urecovery_AbortAll(ubik_dbase);
 740             ubik_epochTime = 2;
 741             ubik_dbase->version.epoch = ubik_epochTime;
 742             ubik_dbase->version.counter = 1;
 743             code =
 744                 (*ubik_dbase->setlabel) (ubik_dbase, 0, &ubik_dbase->version);
 745             udisk_Invalidate(ubik_dbase, 0);    /* data may have changed */
 746 #ifdef AFS_PTHREAD_ENV
 747             assert(pthread_cond_broadcast(&ubik_dbase->version_cond) == 0);
 748 #else
 749             LWP_NoYieldSignal(&ubik_dbase->version);
 750 #endif
 751             DBRELE(ubik_dbase);
 752         }
 753
 754         /* Check the other sites and send the database to them if they
 755          * do not have the current db.
 756          */
 757         if (!(urecovery_state & UBIK_RECSENTDB)) {
 758             /* now propagate out new version to everyone else */
 759             dbok = 1;           /* start off assuming they all worked */
 760
 761             DBHOLD(ubik_dbase);
 762             /*
 763              * Check if a write transaction is in progress. We can't send the
 764              * db when a write is in progress here because the db would be
 765              * obsolete as soon as it goes there. Also, ops after the begin
 766              * trans would reach the recepient and wouldn't find a transaction
 767              * pending there.  Frankly, I don't think it's possible to get past
 768              * the write-lock above if there is a write transaction in progress,
 769              * but then, it won't hurt to check, will it?
 770              */
 771             if (ubik_dbase->flags & DBWRITING) {
 772                 struct timeval tv;
 773                 int safety = 0;
 774                 tv.tv_sec = 0;
 775                 tv.tv_usec = 50000;
 776                 while ((ubik_dbase->flags & DBWRITING) && (safety < 500)) {
 777                     DBRELE(ubik_dbase);
 778                     /* sleep for a little while */
 779 #ifdef AFS_PTHREAD_ENV
 780                     select(0, 0, 0, 0, &tv);
 781 #else
 782                     IOMGR_Select(0, 0, 0, 0, &tv);
 783 #endif
 784                     tv.tv_usec += 10000;
 785                     safety++;
 786                     DBHOLD(ubik_dbase);
 787                 }
 788             }
 789
 790             for (ts = ubik_servers; ts; ts = ts->next) {
 791                 inAddr.s_addr = ts->addr[0];
 792                 if (!ts->up) {
 793                     ubik_dprint("recovery cannot send version to %s\n",
 794                                 afs_inet_ntoa_r(inAddr.s_addr, hoststr));
 795                     dbok = 0;
 796                     continue;
 797                 }
 798                 ubik_dprint("recovery sending version to %s\n",
 799                             afs_inet_ntoa_r(inAddr.s_addr, hoststr));
 800                 if (vcmp(ts->version, ubik_dbase->version) != 0) {
 801                     ubik_dprint("recovery stating local database\n");
 802
 803                     /* Rx code to do the Bulk Store */
 804                     code = (*ubik_dbase->stat) (ubik_dbase, 0, &ubikstat);
 805                     if (!code) {
 806                         length = ubikstat.size;
 807                         file = offset = 0;
 808                         rxcall = rx_NewCall(ts->disk_rxcid);
 809                         code =
 810                             StartDISK_SendFile(rxcall, file, length,
 811                                                &ubik_dbase->version);
 812                         if (code) {
 813                             ubik_dprint("StartDiskSendFile failed=%d\n",
 814                                         code);
 815                             goto StoreEndCall;
 816                         }
 817                         while (length > 0) {
 818                             tlen =
 819                                 (length >
 820                                  sizeof(tbuffer) ? sizeof(tbuffer) : length);
 821                             nbytes =
 822                                 (*ubik_dbase->read) (ubik_dbase, file,
 823                                                      tbuffer, offset, tlen);
 824                             if (nbytes != tlen) {
 825                                 ubik_dprint("Local disk read error=%d\n",
 826                                             code = UIOERROR);
 827                                 goto StoreEndCall;
 828                             }
 829                             nbytes = rx_Write(rxcall, tbuffer, tlen);
 830                             if (nbytes != tlen) {
 831                                 ubik_dprint("Rx-write bulk error=%d\n", code =
 832                                             BULK_ERROR);
 833                                 goto StoreEndCall;
 834                             }
 835                             offset += tlen;
 836                             length -= tlen;
 837                         }
 838                         code = EndDISK_SendFile(rxcall);
 839                       StoreEndCall:
 840                         code = rx_EndCall(rxcall, code);
 841                     }
 842                     if (code == 0) {
 843                         /* we set a new file, process its header */
 844                         ts->version = ubik_dbase->version;
 845                         ts->currentDB = 1;
 846                     } else
 847                         dbok = 0;
 848                 } else {
 849                     /* mark file up to date */
 850                     ts->currentDB = 1;
 851                 }
 852             }
 853             DBRELE(ubik_dbase);
 854             if (dbok)
 855                 urecovery_state |= UBIK_RECSENTDB;
 856         }
 857     }
 858     return NULL;
 859 }
 860
 861 /*!
 862  * \brief send a Probe to all the network address of this server
 863  *
 864  * \return 0 if success, else return 1
 865  */
 866 int
 867 DoProbe(struct ubik_server *server)
 868 {
 869     struct rx_connection *conns[UBIK_MAX_INTERFACE_ADDR];
 870     struct rx_connection *connSuccess = 0;
 871     int i, j;
 872     afs_uint32 addr;
 873     char buffer[32];
 874     char hoststr[16];
 875     extern afs_int32 ubikSecIndex;
 876     extern struct rx_securityClass *ubikSecClass;
 877
 878     for (i = 0; (addr = server->addr[i]) && (i < UBIK_MAX_INTERFACE_ADDR);
 879          i++) {
 880         conns[i] =
 881             rx_NewConnection(addr, ubik_callPortal, DISK_SERVICE_ID,
 882                              ubikSecClass, ubikSecIndex);
 883
 884         /* user requirement to use only the primary interface */
 885         if (ubikPrimaryAddrOnly) {
 886             i = 1;
 887             break;
 888         }
 889     }
 890     assert(i);                  /* at least one interface address for this server */
 891
 892     multi_Rx(conns, i) {
 893         multi_DISK_Probe();
 894         if (!multi_error) {     /* first success */
 895             addr = server->addr[multi_i];       /* successful interface addr */
 896
 897             if (server->disk_rxcid)     /* destroy existing conn */
 898                 rx_DestroyConnection(server->disk_rxcid);
 899             if (server->vote_rxcid)
 900                 rx_DestroyConnection(server->vote_rxcid);
 901
 902             /* make new connections */
 903             server->disk_rxcid = conns[multi_i];
 904             server->vote_rxcid = rx_NewConnection(addr, ubik_callPortal, VOTE_SERVICE_ID, ubikSecClass, ubikSecIndex);  /* for vote reqs */
 905
 906             connSuccess = conns[multi_i];
 907             strcpy(buffer, afs_inet_ntoa_r(server->addr[0], hoststr));
 908             ubik_print
 909                 ("ubik:server %s is back up: will be contacted through %s\n",
 910                  buffer, afs_inet_ntoa_r(addr, hoststr));
 911
 912             multi_Abort;
 913         }
 914     } multi_End_Ignore;
 915
 916     /* Destroy all connections except the one on which we succeeded */
 917     for (j = 0; j < i; j++)
 918         if (conns[j] != connSuccess)
 919             rx_DestroyConnection(conns[j]);
 920
 921     if (!connSuccess)
 922         ubik_dprint("ubik:server %s still down\n",
 923                     afs_inet_ntoa_r(server->addr[0], hoststr));
 924
 925     if (connSuccess)
 926         return 0;               /* success */
 927     else
 928         return 1;               /* failure */
 929 }