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                 struct ubik_version version;
 281                 tpos += 4;
 282                 code =
 283                     (*adbase->read) (adbase, LOGFILE, (char *)buffer, tpos,
 284                                      2 * sizeof(afs_int32));
 285                 if (code != 2 * sizeof(afs_int32))
 286                     return UBADLOG;
 287                 version.epoch = ntohl(buffer[0]);
 288                 version.counter = ntohl(buffer[1]);
 289                 code = (*adbase->setlabel) (adbase, 0, &version);
 290                 if (code)
 291                     return code;
 292                 logIsGood = 1;
 293                 break;          /* all done now */
 294             } else if (opcode == LOGTRUNCATE) {
 295                 tpos += 4;
 296                 code =
 297                     (*adbase->read) (adbase, LOGFILE, (char *)buffer, tpos,
 298                                      2 * sizeof(afs_int32));
 299                 if (code != 2 * sizeof(afs_int32))
 300                     break;      /* premature eof or io error */
 301                 tpos += 2 * sizeof(afs_int32);
 302                 code =
 303                     (*adbase->truncate) (adbase, ntohl(buffer[0]),
 304                                          ntohl(buffer[1]));
 305                 if (code)
 306                     return code;
 307             } else if (opcode == LOGDATA) {
 308                 tpos += 4;
 309                 code =
 310                     (*adbase->read) (adbase, LOGFILE, (char *)buffer, tpos,
 311                                      3 * sizeof(afs_int32));
 312                 if (code != 3 * sizeof(afs_int32))
 313                     break;
 314                 tpos += 3 * sizeof(afs_int32);
 315                 /* otherwise, skip over the data bytes, too */
 316                 len = ntohl(buffer[2]); /* total number of bytes to copy */
 317                 filePos = ntohl(buffer[1]);
 318                 tfile = ntohl(buffer[0]);
 319                 /* try to minimize file syncs */
 320                 if (syncFile != tfile) {
 321                     if (syncFile >= 0)
 322                         code = (*adbase->sync) (adbase, syncFile);
 323                     else
 324                         code = 0;
 325                     syncFile = tfile;
 326                     if (code)
 327                         return code;
 328                 }
 329                 while (len > 0) {
 330                     thisSize = (len > sizeof(data) ? sizeof(data) : len);
 331                     /* copy sizeof(data) buffer bytes at a time */
 332                     code =
 333                         (*adbase->read) (adbase, LOGFILE, (char *)data, tpos,
 334                                          thisSize);
 335                     if (code != thisSize)
 336                         return UBADLOG;
 337                     code =
 338                         (*adbase->write) (adbase, tfile, (char *)data, filePos,
 339                                           thisSize);
 340                     if (code != thisSize)
 341                         return UBADLOG;
 342                     filePos += thisSize;
 343                     tpos += thisSize;
 344                     len -= thisSize;
 345                 }
 346             } else {
 347                 ubik_print("corrupt log opcode (%d) at position %d\n",
 348                            opcode, tpos);
 349                 break;          /* corrupt log! */
 350             }
 351         }
 352         if (logIsGood) {
 353             if (syncFile >= 0)
 354                 code = (*adbase->sync) (adbase, syncFile);
 355             if (code)
 356                 return code;
 357         } else {
 358             ubik_print("Log read error on pass 2\n");
 359             return UBADLOG;
 360         }
 361     }
 362
 363     /* now truncate the log, we're done with it */
 364     code = (*adbase->truncate) (adbase, LOGFILE, 0);
 365     return code;
 366 }
 367
 368 /*! \brief
 369  * Called at initialization to figure out version of the dbase we really have.
 370  *
 371  * This routine is called after replaying the log; it reads the restored labels.
 372  */
 373 static int
 374 InitializeDB(struct ubik_dbase *adbase)
 375 {
 376     afs_int32 code;
 377
 378     code = (*adbase->getlabel) (adbase, 0, &adbase->version);
 379     if (code) {
 380         /* try setting the label to a new value */
 381         adbase->version.epoch = 1;      /* value for newly-initialized db */
 382         adbase->version.counter = 1;
 383         code = (*adbase->setlabel) (adbase, 0, &adbase->version);
 384         if (code) {
 385             /* failed, try to set it back */
 386             adbase->version.epoch = 0;
 387             adbase->version.counter = 0;
 388             (*adbase->setlabel) (adbase, 0, &adbase->version);
 389         }
 390 #ifdef AFS_PTHREAD_ENV
 391         CV_BROADCAST(&adbase->version_cond);
 392 #else
 393         LWP_NoYieldSignal(&adbase->version);
 394 #endif
 395     }
 396     return 0;
 397 }
 398
 399 /*!
 400  * \brief initialize the local ubik_dbase
 401  *
 402  * We replay the logs and then read the resulting file to figure out what version we've really got.
 403  */
 404 int
 405 urecovery_Initialize(struct ubik_dbase *adbase)
 406 {
 407     afs_int32 code;
 408
 409     code = ReplayLog(adbase);
 410     if (code)
 411         return code;
 412     code = InitializeDB(adbase);
 413     return code;
 414 }
 415
 416 /*!
 417  * \brief Main interaction loop for the recovery manager
 418  *
 419  * The recovery light-weight process only runs when you're the
 420  * synchronization site.  It performs the following tasks, if and only
 421  * if the prerequisite tasks have been performed successfully (it
 422  * keeps track of which ones have been performed in its bit map,
 423  * \p urecovery_state).
 424  *
 425  * First, it is responsible for probing that all servers are up.  This
 426  * is the only operation that must be performed even if this is not
 427  * yet the sync site, since otherwise this site may not notice that
 428  * enough other machines are running to even elect this guy to be the
 429  * sync site.
 430  *
 431  * After that, the recovery process does nothing until the beacon and
 432  * voting modules manage to get this site elected sync site.
 433  *
 434  * After becoming sync site, recovery first attempts to find the best
 435  * database available in the network (it must do this in order to
 436  * ensure finding the latest committed data).  After finding the right
 437  * database, it must fetch this dbase to the sync site.
 438  *
 439  * After fetching the dbase, it relabels it with a new version number,
 440  * to ensure that everyone recognizes this dbase as the most recent
 441  * dbase.
 442  *
 443  * One the dbase has been relabelled, this machine can start handling
 444  * requests.  However, the recovery module still has one more task:
 445  * propagating the dbase out to everyone who is up in the network.
 446  */
 447 void *
 448 urecovery_Interact(void *dummy)
 449 {
 450     afs_int32 code, tcode;
 451     struct ubik_server *bestServer = NULL;
 452     struct ubik_server *ts;
 453     int dbok, doingRPC, now;
 454     afs_int32 lastProbeTime;
 455     /* if we're the sync site, the best db version we've found yet */
 456     static struct ubik_version bestDBVersion;
 457     struct ubik_version tversion;
 458     struct timeval tv;
 459     int length, tlen, offset, file, nbytes;
 460     struct rx_call *rxcall;
 461     char tbuffer[1024];
 462     struct ubik_stat ubikstat;
 463     struct in_addr inAddr;
 464     char hoststr[16];
 465 #ifndef OLD_URECOVERY
 466     char pbuffer[1028];
 467     int fd = -1;
 468     afs_int32 pass;
 469 #endif
 470
 471     /* otherwise, begin interaction */
 472     urecovery_state = 0;
 473     lastProbeTime = 0;
 474     while (1) {
 475         /* Run through this loop every 4 seconds */
 476         tv.tv_sec = 4;
 477         tv.tv_usec = 0;
 478 #ifdef AFS_PTHREAD_ENV
 479         select(0, 0, 0, 0, &tv);
 480 #else
 481         IOMGR_Select(0, 0, 0, 0, &tv);
 482 #endif
 483
 484         ubik_dprint("recovery running in state %x\n", urecovery_state);
 485
 486         /* Every 30 seconds, check all the down servers and mark them
 487          * as up if they respond. When a server comes up or found to
 488          * not be current, then re-find the the best database and
 489          * propogate it.
 490          */
 491         if ((now = FT_ApproxTime()) > 30 + lastProbeTime) {
 492
 493 #ifdef AFS_PTHREAD_ENV
 494             DBHOLD(ubik_dbase);
 495 #endif
 496
 497             for (ts = ubik_servers, doingRPC = 0; ts; ts = ts->next) {
 498                 if (!ts->up) {
 499                     doingRPC = 1;
 500                     code = DoProbe(ts);
 501                     if (code == 0) {
 502                         ts->up = 1;
 503                         urecovery_state &= ~UBIK_RECFOUNDDB;
 504                     }
 505                 } else if (!ts->currentDB) {
 506                     urecovery_state &= ~UBIK_RECFOUNDDB;
 507                 }
 508             }
 509
 510 #ifdef AFS_PTHREAD_ENV
 511             DBRELE(ubik_dbase);
 512 #endif
 513
 514             if (doingRPC)
 515                 now = FT_ApproxTime();
 516             lastProbeTime = now;
 517         }
 518
 519         /* Mark whether we are the sync site */
 520         if (!ubeacon_AmSyncSite()) {
 521             urecovery_state &= ~UBIK_RECSYNCSITE;
 522             continue;           /* nothing to do */
 523         }
 524         urecovery_state |= UBIK_RECSYNCSITE;
 525
 526         /* If a server has just come up or if we have not found the
 527          * most current database, then go find the most current db.
 528          */
 529         if (!(urecovery_state & UBIK_RECFOUNDDB)) {
 530             bestServer = (struct ubik_server *)0;
 531             bestDBVersion.epoch = 0;
 532             bestDBVersion.counter = 0;
 533             for (ts = ubik_servers; ts; ts = ts->next) {
 534                 if (!ts->up)
 535                     continue;   /* don't bother with these guys */
 536                 if (ts->isClone)
 537                     continue;
 538                 code = DISK_GetVersion(ts->disk_rxcid, &ts->version);
 539                 if (code == 0) {
 540                     /* perhaps this is the best version */
 541                     if (vcmp(ts->version, bestDBVersion) > 0) {
 542                         /* new best version */
 543                         bestDBVersion = ts->version;
 544                         bestServer = ts;
 545                     }
 546                 }
 547             }
 548             /* take into consideration our version. Remember if we,
 549              * the sync site, have the best version. Also note that
 550              * we may need to send the best version out.
 551              */
 552             if (vcmp(ubik_dbase->version, bestDBVersion) >= 0) {
 553                 bestDBVersion = ubik_dbase->version;
 554                 bestServer = (struct ubik_server *)0;
 555                 urecovery_state |= UBIK_RECHAVEDB;
 556             } else {
 557                 /* Clear the flag only when we know we have to retrieve
 558                  * the db. Because urecovery_AllBetter() looks at it.
 559                  */
 560                 urecovery_state &= ~UBIK_RECHAVEDB;
 561             }
 562             urecovery_state |= UBIK_RECFOUNDDB;
 563             urecovery_state &= ~UBIK_RECSENTDB;
 564         }
 565 #if defined(UBIK_PAUSE)
 566         /* it's not possible for UBIK_RECFOUNDDB not to be set here.
 567          * However, we might have lost UBIK_RECSYNCSITE, and that
 568          * IS important.
 569          */
 570         if (!(urecovery_state & UBIK_RECSYNCSITE))
 571             continue;           /* lost sync */
 572 #else
 573         if (!(urecovery_state & UBIK_RECFOUNDDB))
 574             continue;           /* not ready */
 575 #endif /* UBIK_PAUSE */
 576
 577         /* If we, the sync site, do not have the best db version, then
 578          * go and get it from the server that does.
 579          */
 580         if ((urecovery_state & UBIK_RECHAVEDB) || !bestServer) {
 581             urecovery_state |= UBIK_RECHAVEDB;
 582         } else {
 583             /* we don't have the best version; we should fetch it. */
 584             DBHOLD(ubik_dbase);
 585             urecovery_AbortAll(ubik_dbase);
 586
 587             /* Rx code to do the Bulk fetch */
 588             file = 0;
 589             offset = 0;
 590             rxcall = rx_NewCall(bestServer->disk_rxcid);
 591
 592             ubik_print("Ubik: Synchronize database with server %s\n",
 593                        afs_inet_ntoa_r(bestServer->addr[0], hoststr));
 594
 595             code = StartDISK_GetFile(rxcall, file);
 596             if (code) {
 597                 ubik_dprint("StartDiskGetFile failed=%d\n", code);
 598                 goto FetchEndCall;
 599             }
 600             nbytes = rx_Read(rxcall, (char *)&length, sizeof(afs_int32));
 601             length = ntohl(length);
 602             if (nbytes != sizeof(afs_int32)) {
 603                 ubik_dprint("Rx-read length error=%d\n", code = BULK_ERROR);
 604                 code = EIO;
 605                 goto FetchEndCall;
 606             }
 607
 608 #ifdef OLD_URECOVERY
 609             /* Truncate the file first */
 610             code = (*ubik_dbase->truncate) (ubik_dbase, file, 0);
 611             if (code) {
 612                 ubik_dprint("truncate io error=%d\n", code);
 613                 goto FetchEndCall;
 614             }
 615             tversion.counter = 0;
 616 #endif
 617             /* give invalid label during file transit */
 618             tversion.epoch = 0;
 619             code = (*ubik_dbase->setlabel) (ubik_dbase, file, &tversion);
 620             if (code) {
 621                 ubik_dprint("setlabel io error=%d\n", code);
 622                 goto FetchEndCall;
 623             }
 624 #ifndef OLD_URECOVERY
 625             afs_snprintf(pbuffer, sizeof(pbuffer), "%s.DB%s%d.TMP", ubik_dbase->pathName, (file<0)?"SYS":"", (file<0)?-file:file);
 626             fd = open(pbuffer, O_CREAT | O_RDWR | O_TRUNC, 0600);
 627             if (fd < 0) {
 628                 code = errno;
 629                 goto FetchEndCall;
 630             }
 631             code = lseek(fd, HDRSIZE, 0);
 632             if (code != HDRSIZE) {
 633                 close(fd);
 634                 goto FetchEndCall;
 635             }
 636 #endif
 637
 638             pass = 0;
 639             while (length > 0) {
 640                 tlen = (length > sizeof(tbuffer) ? sizeof(tbuffer) : length);
 641 #ifndef AFS_PTHREAD_ENV
 642                 if (pass % 4 == 0)
 643                     IOMGR_Poll();
 644 #endif
 645                 nbytes = rx_Read(rxcall, tbuffer, tlen);
 646                 if (nbytes != tlen) {
 647                     ubik_dprint("Rx-read bulk error=%d\n", code = BULK_ERROR);
 648                     code = EIO;
 649                     close(fd);
 650                     goto FetchEndCall;
 651                 }
 652 #ifdef OLD_URECOVERY
 653                 nbytes =
 654                     (*ubik_dbase->write) (ubik_dbase, file, tbuffer, offset,
 655                                           tlen);
 656 #else
 657                 nbytes = write(fd, tbuffer, tlen);
 658                 pass++;
 659 #endif
 660                 if (nbytes != tlen) {
 661                     code = UIOERROR;
 662                     close(fd);
 663                     goto FetchEndCall;
 664                 }
 665                 offset += tlen;
 666                 length -= tlen;
 667             }
 668 #ifndef OLD_URECOVERY
 669             code = close(fd);
 670             if (code)
 671                 goto FetchEndCall;
 672 #endif
 673             code = EndDISK_GetFile(rxcall, &tversion);
 674           FetchEndCall:
 675             tcode = rx_EndCall(rxcall, code);
 676             if (!code)
 677                 code = tcode;
 678             if (!code) {
 679                 /* we got a new file, set up its header */
 680                 urecovery_state |= UBIK_RECHAVEDB;
 681                 memcpy(&ubik_dbase->version, &tversion,
 682                        sizeof(struct ubik_version));
 683 #ifdef OLD_URECOVERY
 684                 (*ubik_dbase->sync) (ubik_dbase, 0);    /* get data out first */
 685 #else
 686                 afs_snprintf(tbuffer, sizeof(tbuffer), "%s.DB%s%d", ubik_dbase->pathName, (file<0)?"SYS":"", (file<0)?-file:file);
 687 #ifdef AFS_NT40_ENV
 688                 afs_snprintf(pbuffer, sizeof(pbuffer), "%s.DB%s%d.OLD", ubik_dbase->pathName, (file<0)?"SYS":"", (file<0)?-file:file);
 689                 code = unlink(pbuffer);
 690                 if (!code)
 691                     code = rename(tbuffer, pbuffer);
 692                 afs_snprintf(pbuffer, sizeof(pbuffer), "%s.DB%s%d.TMP", ubik_dbase->pathName, (file<0)?"SYS":"", (file<0)?-file:file);
 693 #endif
 694                 if (!code)
 695                     code = rename(pbuffer, tbuffer);
 696                 if (!code) {
 697                     (*ubik_dbase->open) (ubik_dbase, file);
 698 #endif
 699                     /* after data is good, sync disk with correct label */
 700                     code =
 701                         (*ubik_dbase->setlabel) (ubik_dbase, 0,
 702                                                  &ubik_dbase->version);
 703 #ifndef OLD_URECOVERY
 704                 }
 705 #ifdef AFS_NT40_ENV
 706                 afs_snprintf(pbuffer, sizeof(pbuffer), "%s.DB%s%d.OLD", ubik_dbase->pathName, (file<0)?"SYS":"", (file<0)?-file:file);
 707                 unlink(pbuffer);
 708 #endif
 709 #endif
 710             }
 711             if (code) {
 712 #ifndef OLD_URECOVERY
 713                 unlink(pbuffer);
 714                 /*
 715                  * We will effectively invalidate the old data forever now.
 716                  * Unclear if we *should* but we do.
 717                  */
 718 #endif
 719                 ubik_dbase->version.epoch = 0;
 720                 ubik_dbase->version.counter = 0;
 721                 ubik_print("Ubik: Synchronize database failed (error = %d)\n",
 722                            code);
 723             } else {
 724                 ubik_print("Ubik: Synchronize database completed\n");
 725                 urecovery_state |= UBIK_RECHAVEDB;
 726             }
 727             udisk_Invalidate(ubik_dbase, 0);    /* data has changed */
 728 #ifdef AFS_PTHREAD_ENV
 729             CV_BROADCAST(&ubik_dbase->version_cond);
 730 #else
 731             LWP_NoYieldSignal(&ubik_dbase->version);
 732 #endif
 733             DBRELE(ubik_dbase);
 734         }
 735 #if defined(UBIK_PAUSE)
 736         if (!(urecovery_state & UBIK_RECSYNCSITE))
 737             continue;           /* lost sync */
 738 #endif /* UBIK_PAUSE */
 739         if (!(urecovery_state & UBIK_RECHAVEDB))
 740             continue;           /* not ready */
 741
 742         /* If the database was newly initialized, then when we establish quorum, write
 743          * a new label. This allows urecovery_AllBetter() to allow access for reads.
 744          * Setting it to 2 also allows another site to come along with a newer
 745          * database and overwrite this one.
 746          */
 747         if (ubik_dbase->version.epoch == 1) {
 748             DBHOLD(ubik_dbase);
 749             urecovery_AbortAll(ubik_dbase);
 750             ubik_epochTime = 2;
 751             ubik_dbase->version.epoch = ubik_epochTime;
 752             ubik_dbase->version.counter = 1;
 753             code =
 754                 (*ubik_dbase->setlabel) (ubik_dbase, 0, &ubik_dbase->version);
 755             udisk_Invalidate(ubik_dbase, 0);    /* data may have changed */
 756 #ifdef AFS_PTHREAD_ENV
 757             CV_BROADCAST(&ubik_dbase->version_cond);
 758 #else
 759             LWP_NoYieldSignal(&ubik_dbase->version);
 760 #endif
 761             DBRELE(ubik_dbase);
 762         }
 763
 764         /* Check the other sites and send the database to them if they
 765          * do not have the current db.
 766          */
 767         if (!(urecovery_state & UBIK_RECSENTDB)) {
 768             /* now propagate out new version to everyone else */
 769             dbok = 1;           /* start off assuming they all worked */
 770
 771             DBHOLD(ubik_dbase);
 772             /*
 773              * Check if a write transaction is in progress. We can't send the
 774              * db when a write is in progress here because the db would be
 775              * obsolete as soon as it goes there. Also, ops after the begin
 776              * trans would reach the recepient and wouldn't find a transaction
 777              * pending there.  Frankly, I don't think it's possible to get past
 778              * the write-lock above if there is a write transaction in progress,
 779              * but then, it won't hurt to check, will it?
 780              */
 781             if (ubik_dbase->flags & DBWRITING) {
 782                 struct timeval tv;
 783                 int safety = 0;
 784                 tv.tv_sec = 0;
 785                 tv.tv_usec = 50000;
 786                 while ((ubik_dbase->flags & DBWRITING) && (safety < 500)) {
 787                     DBRELE(ubik_dbase);
 788                     /* sleep for a little while */
 789 #ifdef AFS_PTHREAD_ENV
 790                     select(0, 0, 0, 0, &tv);
 791 #else
 792                     IOMGR_Select(0, 0, 0, 0, &tv);
 793 #endif
 794                     tv.tv_usec += 10000;
 795                     safety++;
 796                     DBHOLD(ubik_dbase);
 797                 }
 798             }
 799
 800             for (ts = ubik_servers; ts; ts = ts->next) {
 801                 inAddr.s_addr = ts->addr[0];
 802                 if (!ts->up) {
 803                     ubik_dprint("recovery cannot send version to %s\n",
 804                                 afs_inet_ntoa_r(inAddr.s_addr, hoststr));
 805                     dbok = 0;
 806                     continue;
 807                 }
 808                 ubik_dprint("recovery sending version to %s\n",
 809                             afs_inet_ntoa_r(inAddr.s_addr, hoststr));
 810                 if (vcmp(ts->version, ubik_dbase->version) != 0) {
 811                     ubik_dprint("recovery stating local database\n");
 812
 813                     /* Rx code to do the Bulk Store */
 814                     code = (*ubik_dbase->stat) (ubik_dbase, 0, &ubikstat);
 815                     if (!code) {
 816                         length = ubikstat.size;
 817                         file = offset = 0;
 818                         rxcall = rx_NewCall(ts->disk_rxcid);
 819                         code =
 820                             StartDISK_SendFile(rxcall, file, length,
 821                                                &ubik_dbase->version);
 822                         if (code) {
 823                             ubik_dprint("StartDiskSendFile failed=%d\n",
 824                                         code);
 825                             goto StoreEndCall;
 826                         }
 827                         while (length > 0) {
 828                             tlen =
 829                                 (length >
 830                                  sizeof(tbuffer) ? sizeof(tbuffer) : length);
 831                             nbytes =
 832                                 (*ubik_dbase->read) (ubik_dbase, file,
 833                                                      tbuffer, offset, tlen);
 834                             if (nbytes != tlen) {
 835                                 ubik_dprint("Local disk read error=%d\n",
 836                                             code = UIOERROR);
 837                                 goto StoreEndCall;
 838                             }
 839                             nbytes = rx_Write(rxcall, tbuffer, tlen);
 840                             if (nbytes != tlen) {
 841                                 ubik_dprint("Rx-write bulk error=%d\n", code =
 842                                             BULK_ERROR);
 843                                 goto StoreEndCall;
 844                             }
 845                             offset += tlen;
 846                             length -= tlen;
 847                         }
 848                         code = EndDISK_SendFile(rxcall);
 849                       StoreEndCall:
 850                         code = rx_EndCall(rxcall, code);
 851                     }
 852                     if (code == 0) {
 853                         /* we set a new file, process its header */
 854                         ts->version = ubik_dbase->version;
 855                         ts->currentDB = 1;
 856                     } else
 857                         dbok = 0;
 858                 } else {
 859                     /* mark file up to date */
 860                     ts->currentDB = 1;
 861                 }
 862             }
 863             DBRELE(ubik_dbase);
 864             if (dbok)
 865                 urecovery_state |= UBIK_RECSENTDB;
 866         }
 867     }
 868     return NULL;
 869 }
 870
 871 /*!
 872  * \brief send a Probe to all the network address of this server
 873  *
 874  * \return 0 if success, else return 1
 875  */
 876 int
 877 DoProbe(struct ubik_server *server)
 878 {
 879     struct rx_connection *conns[UBIK_MAX_INTERFACE_ADDR];
 880     struct rx_connection *connSuccess = 0;
 881     int i, j, success_i = -1;
 882     afs_uint32 addr;
 883     char buffer[32];
 884     char hoststr[16];
 885     extern afs_int32 ubikSecIndex;
 886     extern struct rx_securityClass *ubikSecClass;
 887
 888     for (i = 0; (addr = server->addr[i]) && (i < UBIK_MAX_INTERFACE_ADDR);
 889          i++) {
 890         conns[i] =
 891             rx_NewConnection(addr, ubik_callPortal, DISK_SERVICE_ID,
 892                              ubikSecClass, ubikSecIndex);
 893
 894         /* user requirement to use only the primary interface */
 895         if (ubikPrimaryAddrOnly) {
 896             i = 1;
 897             break;
 898         }
 899     }
 900     osi_Assert(i);                      /* at least one interface address for this server */
 901
 902 #ifdef AFS_PTHREAD_ENV
 903     DBRELE(ubik_dbase);
 904 #endif
 905
 906     multi_Rx(conns, i) {
 907         multi_DISK_Probe();
 908         if (!multi_error) {     /* first success */
 909             success_i = multi_i;
 910
 911             multi_Abort;
 912         }
 913     } multi_End_Ignore;
 914
 915 #ifdef AFS_PTHREAD_ENV
 916     DBHOLD(ubik_dbase);
 917 #endif
 918
 919     if (success_i >= 0) {
 920         addr = server->addr[success_i]; /* successful interface addr */
 921
 922         if (server->disk_rxcid) /* destroy existing conn */
 923             rx_DestroyConnection(server->disk_rxcid);
 924         if (server->vote_rxcid)
 925             rx_DestroyConnection(server->vote_rxcid);
 926
 927         /* make new connections */
 928         server->disk_rxcid = conns[success_i];
 929         server->vote_rxcid = rx_NewConnection(addr, ubik_callPortal,
 930                                               VOTE_SERVICE_ID, ubikSecClass,
 931                                               ubikSecIndex);
 932
 933         connSuccess = conns[success_i];
 934         strcpy(buffer, afs_inet_ntoa_r(server->addr[0], hoststr));
 935
 936         ubik_print("ubik:server %s is back up: will be contacted through %s\n",
 937              buffer, afs_inet_ntoa_r(addr, hoststr));
 938     }
 939
 940     /* Destroy all connections except the one on which we succeeded */
 941     for (j = 0; j < i; j++)
 942         if (conns[j] != connSuccess)
 943             rx_DestroyConnection(conns[j]);
 944
 945     if (!connSuccess)
 946         ubik_dprint("ubik:server %s still down\n",
 947                     afs_inet_ntoa_r(server->addr[0], hoststr));
 948
 949     if (connSuccess)
 950         return 0;               /* success */
 951     else
 952         return 1;               /* failure */
 953 }