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
  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(struct ubik_dbase *adbase, int areadAny)
 112 {
 113     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     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(struct ubik_tid *atid, int abortalways)
 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 || abortalways) {
 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(struct ubik_dbase *adbase)
 207 {
 208     afs_int32 opcode;
 209     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_print("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_print("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_print("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(struct ubik_dbase *adbase)
 372 {
 373     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         CV_BROADCAST(&adbase->version_cond);
 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(struct ubik_dbase *adbase)
 403 {
 404     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;
 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 fd = -1;
 465     afs_int32 pass;
 466 #endif
 467
 468     /* otherwise, begin interaction */
 469     urecovery_state = 0;
 470     lastProbeTime = 0;
 471     while (1) {
 472         /* Run through this loop every 4 seconds */
 473         tv.tv_sec = 4;
 474         tv.tv_usec = 0;
 475 #ifdef AFS_PTHREAD_ENV
 476         select(0, 0, 0, 0, &tv);
 477 #else
 478         IOMGR_Select(0, 0, 0, 0, &tv);
 479 #endif
 480
 481         ubik_dprint("recovery running in state %x\n", urecovery_state);
 482
 483         /* Every 30 seconds, check all the down servers and mark them
 484          * as up if they respond. When a server comes up or found to
 485          * not be current, then re-find the the best database and
 486          * propogate it.
 487          */
 488         if ((now = FT_ApproxTime()) > 30 + lastProbeTime) {
 489
 490 #ifdef AFS_PTHREAD_ENV
 491             DBHOLD(ubik_dbase);
 492 #endif
 493
 494             for (ts = ubik_servers, doingRPC = 0; ts; ts = ts->next) {
 495                 if (!ts->up) {
 496                     doingRPC = 1;
 497                     code = DoProbe(ts);
 498                     if (code == 0) {
 499                         ts->up = 1;
 500                         urecovery_state &= ~UBIK_RECFOUNDDB;
 501                     }
 502                 } else if (!ts->currentDB) {
 503                     urecovery_state &= ~UBIK_RECFOUNDDB;
 504                 }
 505             }
 506
 507 #ifdef AFS_PTHREAD_ENV
 508             DBRELE(ubik_dbase);
 509 #endif
 510
 511             if (doingRPC)
 512                 now = FT_ApproxTime();
 513             lastProbeTime = now;
 514         }
 515
 516         /* Mark whether we are the sync site */
 517         if (!ubeacon_AmSyncSite()) {
 518             urecovery_state &= ~UBIK_RECSYNCSITE;
 519             continue;           /* nothing to do */
 520         }
 521         urecovery_state |= UBIK_RECSYNCSITE;
 522
 523         /* If a server has just come up or if we have not found the
 524          * most current database, then go find the most current db.
 525          */
 526         if (!(urecovery_state & UBIK_RECFOUNDDB)) {
 527             bestServer = (struct ubik_server *)0;
 528             bestDBVersion.epoch = 0;
 529             bestDBVersion.counter = 0;
 530             for (ts = ubik_servers; ts; ts = ts->next) {
 531                 if (!ts->up)
 532                     continue;   /* don't bother with these guys */
 533                 if (ts->isClone)
 534                     continue;
 535                 code = DISK_GetVersion(ts->disk_rxcid, &ts->version);
 536                 if (code == 0) {
 537                     /* perhaps this is the best version */
 538                     if (vcmp(ts->version, bestDBVersion) > 0) {
 539                         /* new best version */
 540                         bestDBVersion = ts->version;
 541                         bestServer = ts;
 542                     }
 543                 }
 544             }
 545             /* take into consideration our version. Remember if we,
 546              * the sync site, have the best version. Also note that
 547              * we may need to send the best version out.
 548              */
 549             if (vcmp(ubik_dbase->version, bestDBVersion) >= 0) {
 550                 bestDBVersion = ubik_dbase->version;
 551                 bestServer = (struct ubik_server *)0;
 552                 urecovery_state |= UBIK_RECHAVEDB;
 553             } else {
 554                 /* Clear the flag only when we know we have to retrieve
 555                  * the db. Because urecovery_AllBetter() looks at it.
 556                  */
 557                 urecovery_state &= ~UBIK_RECHAVEDB;
 558             }
 559             urecovery_state |= UBIK_RECFOUNDDB;
 560             urecovery_state &= ~UBIK_RECSENTDB;
 561         }
 562 #if defined(UBIK_PAUSE)
 563         /* it's not possible for UBIK_RECFOUNDDB not to be set here.
 564          * However, we might have lost UBIK_RECSYNCSITE, and that
 565          * IS important.
 566          */
 567         if (!(urecovery_state & UBIK_RECSYNCSITE))
 568             continue;           /* lost sync */
 569 #else
 570         if (!(urecovery_state & UBIK_RECFOUNDDB))
 571             continue;           /* not ready */
 572 #endif /* UBIK_PAUSE */
 573
 574         /* If we, the sync site, do not have the best db version, then
 575          * go and get it from the server that does.
 576          */
 577         if ((urecovery_state & UBIK_RECHAVEDB) || !bestServer) {
 578             urecovery_state |= UBIK_RECHAVEDB;
 579         } else {
 580             /* we don't have the best version; we should fetch it. */
 581             DBHOLD(ubik_dbase);
 582             urecovery_AbortAll(ubik_dbase);
 583
 584             /* Rx code to do the Bulk fetch */
 585             file = 0;
 586             offset = 0;
 587             rxcall = rx_NewCall(bestServer->disk_rxcid);
 588
 589             ubik_print("Ubik: Synchronize database with server %s\n",
 590                        afs_inet_ntoa_r(bestServer->addr[0], hoststr));
 591
 592             code = StartDISK_GetFile(rxcall, file);
 593             if (code) {
 594                 ubik_dprint("StartDiskGetFile failed=%d\n", code);
 595                 goto FetchEndCall;
 596             }
 597             nbytes = rx_Read(rxcall, (char *)&length, sizeof(afs_int32));
 598             length = ntohl(length);
 599             if (nbytes != sizeof(afs_int32)) {
 600                 ubik_dprint("Rx-read length error=%d\n", code = BULK_ERROR);
 601                 code = EIO;
 602                 goto FetchEndCall;
 603             }
 604
 605 #ifdef OLD_URECOVERY
 606             /* Truncate the file first */
 607             code = (*ubik_dbase->truncate) (ubik_dbase, file, 0);
 608             if (code) {
 609                 ubik_dprint("truncate io error=%d\n", code);
 610                 goto FetchEndCall;
 611             }
 612             tversion.counter = 0;
 613 #endif
 614             /* give invalid label during file transit */
 615             tversion.epoch = 0;
 616             code = (*ubik_dbase->setlabel) (ubik_dbase, file, &tversion);
 617             if (code) {
 618                 ubik_dprint("setlabel io error=%d\n", code);
 619                 goto FetchEndCall;
 620             }
 621 #ifndef OLD_URECOVERY
 622             afs_snprintf(pbuffer, sizeof(pbuffer), "%s.DB%s%d.TMP", ubik_dbase->pathName, (file<0)?"SYS":"", (file<0)?-file:file);
 623             fd = open(pbuffer, O_CREAT | O_RDWR | O_TRUNC, 0600);
 624             if (fd < 0) {
 625                 code = errno;
 626                 goto FetchEndCall;
 627             }
 628             code = lseek(fd, HDRSIZE, 0);
 629             if (code != HDRSIZE) {
 630                 close(fd);
 631                 goto FetchEndCall;
 632             }
 633 #endif
 634
 635             pass = 0;
 636             while (length > 0) {
 637                 tlen = (length > sizeof(tbuffer) ? sizeof(tbuffer) : length);
 638 #ifndef AFS_PTHREAD_ENV
 639                 if (pass % 4 == 0)
 640                     IOMGR_Poll();
 641 #endif
 642                 nbytes = rx_Read(rxcall, tbuffer, tlen);
 643                 if (nbytes != tlen) {
 644                     ubik_dprint("Rx-read bulk error=%d\n", code = BULK_ERROR);
 645                     code = EIO;
 646                     close(fd);
 647                     goto FetchEndCall;
 648                 }
 649 #ifdef OLD_URECOVERY
 650                 nbytes =
 651                     (*ubik_dbase->write) (ubik_dbase, file, tbuffer, offset,
 652                                           tlen);
 653 #else
 654                 nbytes = write(fd, tbuffer, tlen);
 655                 pass++;
 656 #endif
 657                 if (nbytes != tlen) {
 658                     code = UIOERROR;
 659                     close(fd);
 660                     goto FetchEndCall;
 661                 }
 662                 offset += tlen;
 663                 length -= tlen;
 664             }
 665 #ifndef OLD_URECOVERY
 666             code = close(fd);
 667             if (code)
 668                 goto FetchEndCall;
 669 #endif
 670             code = EndDISK_GetFile(rxcall, &tversion);
 671           FetchEndCall:
 672             tcode = rx_EndCall(rxcall, code);
 673             if (!code)
 674                 code = tcode;
 675             if (!code) {
 676                 /* we got a new file, set up its header */
 677                 urecovery_state |= UBIK_RECHAVEDB;
 678                 memcpy(&ubik_dbase->version, &tversion,
 679                        sizeof(struct ubik_version));
 680 #ifdef OLD_URECOVERY
 681                 (*ubik_dbase->sync) (ubik_dbase, 0);    /* get data out first */
 682 #else
 683                 afs_snprintf(tbuffer, sizeof(tbuffer), "%s.DB%s%d", ubik_dbase->pathName, (file<0)?"SYS":"", (file<0)?-file:file);
 684 #ifdef AFS_NT40_ENV
 685                 afs_snprintf(pbuffer, sizeof(pbuffer), "%s.DB%s%d.OLD", ubik_dbase->pathName, (file<0)?"SYS":"", (file<0)?-file:file);
 686                 code = unlink(pbuffer);
 687                 if (!code)
 688                     code = rename(tbuffer, pbuffer);
 689                 afs_snprintf(pbuffer, sizeof(pbuffer), "%s.DB%s%d.TMP", ubik_dbase->pathName, (file<0)?"SYS":"", (file<0)?-file:file);
 690 #endif
 691                 if (!code)
 692                     code = rename(pbuffer, tbuffer);
 693                 if (!code) {
 694                     (*ubik_dbase->open) (ubik_dbase, file);
 695 #endif
 696                     /* after data is good, sync disk with correct label */
 697                     code =
 698                         (*ubik_dbase->setlabel) (ubik_dbase, 0,
 699                                                  &ubik_dbase->version);
 700 #ifndef OLD_URECOVERY
 701                 }
 702 #ifdef AFS_NT40_ENV
 703                 afs_snprintf(pbuffer, sizeof(pbuffer), "%s.DB%s%d.OLD", ubik_dbase->pathName, (file<0)?"SYS":"", (file<0)?-file:file);
 704                 unlink(pbuffer);
 705 #endif
 706 #endif
 707             }
 708             if (code) {
 709 #ifndef OLD_URECOVERY
 710                 unlink(pbuffer);
 711                 /*
 712                  * We will effectively invalidate the old data forever now.
 713                  * Unclear if we *should* but we do.
 714                  */
 715 #endif
 716                 ubik_dbase->version.epoch = 0;
 717                 ubik_dbase->version.counter = 0;
 718                 ubik_print("Ubik: Synchronize database failed (error = %d)\n",
 719                            code);
 720             } else {
 721                 ubik_print("Ubik: Synchronize database completed\n");
 722                 urecovery_state |= UBIK_RECHAVEDB;
 723             }
 724             udisk_Invalidate(ubik_dbase, 0);    /* data has changed */
 725 #ifdef AFS_PTHREAD_ENV
 726             CV_BROADCAST(&ubik_dbase->version_cond);
 727 #else
 728             LWP_NoYieldSignal(&ubik_dbase->version);
 729 #endif
 730             DBRELE(ubik_dbase);
 731         }
 732 #if defined(UBIK_PAUSE)
 733         if (!(urecovery_state & UBIK_RECSYNCSITE))
 734             continue;           /* lost sync */
 735 #endif /* UBIK_PAUSE */
 736         if (!(urecovery_state & UBIK_RECHAVEDB))
 737             continue;           /* not ready */
 738
 739         /* If the database was newly initialized, then when we establish quorum, write
 740          * a new label. This allows urecovery_AllBetter() to allow access for reads.
 741          * Setting it to 2 also allows another site to come along with a newer
 742          * database and overwrite this one.
 743          */
 744         if (ubik_dbase->version.epoch == 1) {
 745             DBHOLD(ubik_dbase);
 746             urecovery_AbortAll(ubik_dbase);
 747             ubik_epochTime = 2;
 748             ubik_dbase->version.epoch = ubik_epochTime;
 749             ubik_dbase->version.counter = 1;
 750             code =
 751                 (*ubik_dbase->setlabel) (ubik_dbase, 0, &ubik_dbase->version);
 752             udisk_Invalidate(ubik_dbase, 0);    /* data may have changed */
 753 #ifdef AFS_PTHREAD_ENV
 754             CV_BROADCAST(&ubik_dbase->version_cond);
 755 #else
 756             LWP_NoYieldSignal(&ubik_dbase->version);
 757 #endif
 758             DBRELE(ubik_dbase);
 759         }
 760
 761         /* Check the other sites and send the database to them if they
 762          * do not have the current db.
 763          */
 764         if (!(urecovery_state & UBIK_RECSENTDB)) {
 765             /* now propagate out new version to everyone else */
 766             dbok = 1;           /* start off assuming they all worked */
 767
 768             DBHOLD(ubik_dbase);
 769             /*
 770              * Check if a write transaction is in progress. We can't send the
 771              * db when a write is in progress here because the db would be
 772              * obsolete as soon as it goes there. Also, ops after the begin
 773              * trans would reach the recepient and wouldn't find a transaction
 774              * pending there.  Frankly, I don't think it's possible to get past
 775              * the write-lock above if there is a write transaction in progress,
 776              * but then, it won't hurt to check, will it?
 777              */
 778             if (ubik_dbase->flags & DBWRITING) {
 779                 struct timeval tv;
 780                 int safety = 0;
 781                 tv.tv_sec = 0;
 782                 tv.tv_usec = 50000;
 783                 while ((ubik_dbase->flags & DBWRITING) && (safety < 500)) {
 784                     DBRELE(ubik_dbase);
 785                     /* sleep for a little while */
 786 #ifdef AFS_PTHREAD_ENV
 787                     select(0, 0, 0, 0, &tv);
 788 #else
 789                     IOMGR_Select(0, 0, 0, 0, &tv);
 790 #endif
 791                     tv.tv_usec += 10000;
 792                     safety++;
 793                     DBHOLD(ubik_dbase);
 794                 }
 795             }
 796
 797             for (ts = ubik_servers; ts; ts = ts->next) {
 798                 inAddr.s_addr = ts->addr[0];
 799                 if (!ts->up) {
 800                     ubik_dprint("recovery cannot send version to %s\n",
 801                                 afs_inet_ntoa_r(inAddr.s_addr, hoststr));
 802                     dbok = 0;
 803                     continue;
 804                 }
 805                 ubik_dprint("recovery sending version to %s\n",
 806                             afs_inet_ntoa_r(inAddr.s_addr, hoststr));
 807                 if (vcmp(ts->version, ubik_dbase->version) != 0) {
 808                     ubik_dprint("recovery stating local database\n");
 809
 810                     /* Rx code to do the Bulk Store */
 811                     code = (*ubik_dbase->stat) (ubik_dbase, 0, &ubikstat);
 812                     if (!code) {
 813                         length = ubikstat.size;
 814                         file = offset = 0;
 815                         rxcall = rx_NewCall(ts->disk_rxcid);
 816                         code =
 817                             StartDISK_SendFile(rxcall, file, length,
 818                                                &ubik_dbase->version);
 819                         if (code) {
 820                             ubik_dprint("StartDiskSendFile failed=%d\n",
 821                                         code);
 822                             goto StoreEndCall;
 823                         }
 824                         while (length > 0) {
 825                             tlen =
 826                                 (length >
 827                                  sizeof(tbuffer) ? sizeof(tbuffer) : length);
 828                             nbytes =
 829                                 (*ubik_dbase->read) (ubik_dbase, file,
 830                                                      tbuffer, offset, tlen);
 831                             if (nbytes != tlen) {
 832                                 ubik_dprint("Local disk read error=%d\n",
 833                                             code = UIOERROR);
 834                                 goto StoreEndCall;
 835                             }
 836                             nbytes = rx_Write(rxcall, tbuffer, tlen);
 837                             if (nbytes != tlen) {
 838                                 ubik_dprint("Rx-write bulk error=%d\n", code =
 839                                             BULK_ERROR);
 840                                 goto StoreEndCall;
 841                             }
 842                             offset += tlen;
 843                             length -= tlen;
 844                         }
 845                         code = EndDISK_SendFile(rxcall);
 846                       StoreEndCall:
 847                         code = rx_EndCall(rxcall, code);
 848                     }
 849                     if (code == 0) {
 850                         /* we set a new file, process its header */
 851                         ts->version = ubik_dbase->version;
 852                         ts->currentDB = 1;
 853                     } else
 854                         dbok = 0;
 855                 } else {
 856                     /* mark file up to date */
 857                     ts->currentDB = 1;
 858                 }
 859             }
 860             DBRELE(ubik_dbase);
 861             if (dbok)
 862                 urecovery_state |= UBIK_RECSENTDB;
 863         }
 864     }
 865     return NULL;
 866 }
 867
 868 /*!
 869  * \brief send a Probe to all the network address of this server
 870  *
 871  * \return 0 if success, else return 1
 872  */
 873 int
 874 DoProbe(struct ubik_server *server)
 875 {
 876     struct rx_connection *conns[UBIK_MAX_INTERFACE_ADDR];
 877     struct rx_connection *connSuccess = 0;
 878     int i, j, success_i = -1;
 879     afs_uint32 addr;
 880     char buffer[32];
 881     char hoststr[16];
 882     extern afs_int32 ubikSecIndex;
 883     extern struct rx_securityClass *ubikSecClass;
 884
 885     for (i = 0; (addr = server->addr[i]) && (i < UBIK_MAX_INTERFACE_ADDR);
 886          i++) {
 887         conns[i] =
 888             rx_NewConnection(addr, ubik_callPortal, DISK_SERVICE_ID,
 889                              ubikSecClass, ubikSecIndex);
 890
 891         /* user requirement to use only the primary interface */
 892         if (ubikPrimaryAddrOnly) {
 893             i = 1;
 894             break;
 895         }
 896     }
 897     osi_Assert(i);                      /* at least one interface address for this server */
 898
 899 #ifdef AFS_PTHREAD_ENV
 900     DBRELE(ubik_dbase);
 901 #endif
 902
 903     multi_Rx(conns, i) {
 904         multi_DISK_Probe();
 905         if (!multi_error) {     /* first success */
 906             success_i = multi_i;
 907
 908             multi_Abort;
 909         }
 910     } multi_End_Ignore;
 911
 912 #ifdef AFS_PTHREAD_ENV
 913     DBHOLD(ubik_dbase);
 914 #endif
 915
 916     if (success_i >= 0) {
 917         addr = server->addr[success_i]; /* successful interface addr */
 918
 919         if (server->disk_rxcid) /* destroy existing conn */
 920             rx_DestroyConnection(server->disk_rxcid);
 921         if (server->vote_rxcid)
 922             rx_DestroyConnection(server->vote_rxcid);
 923
 924         /* make new connections */
 925         server->disk_rxcid = conns[success_i];
 926         server->vote_rxcid = rx_NewConnection(addr, ubik_callPortal,
 927                                               VOTE_SERVICE_ID, ubikSecClass,
 928                                               ubikSecIndex);
 929
 930         connSuccess = conns[success_i];
 931         strcpy(buffer, afs_inet_ntoa_r(server->addr[0], hoststr));
 932
 933         ubik_print("ubik:server %s is back up: will be contacted through %s\n",
 934              buffer, afs_inet_ntoa_r(addr, hoststr));
 935     }
 936
 937     /* Destroy all connections except the one on which we succeeded */
 938     for (j = 0; j < i; j++)
 939         if (conns[j] != connSuccess)
 940             rx_DestroyConnection(conns[j]);
 941
 942     if (!connSuccess)
 943         ubik_dprint("ubik:server %s still down\n",
 944                     afs_inet_ntoa_r(server->addr[0], hoststr));
 945
 946     if (connSuccess)
 947         return 0;               /* success */
 948     else
 949         return 1;               /* failure */
 950 }