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