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