src/ubik/remote.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 <assert.h>
  16
  17 #include <afs/opr.h>
  18 #ifdef AFS_PTHREAD_ENV
  19 # include <opr/lock.h>
  20 #else
  21 # include <opr/lockstub.h>
  22 #endif
  23
  24 #include <lock.h>
  25 #include <rx/xdr.h>
  26 #include <rx/rx.h>
  27 #include <afs/afsutil.h>
  28
  29 #define UBIK_INTERNALS
  30 #include "ubik.h"
  31 #include "ubik_int.h"
  32
  33 static void printServerInfo(void);
  34
  35 /*! \file
  36  * routines for handling requests remotely-submitted by the sync site.  These are
  37  * only write transactions (we don't propagate read trans), and there is at most one
  38  * write transaction extant at any one time.
  39  */
  40
  41 struct ubik_trans *ubik_currentTrans = 0;
  42
  43
  44
  45 /* the rest of these guys handle remote execution of write
  46  * transactions: this is the code executed on the other servers when a
  47  * sync site is executing a write transaction.
  48  */
  49 afs_int32
  50 SDISK_Begin(struct rx_call *rxcall, struct ubik_tid *atid)
  51 {
  52     afs_int32 code;
  53
  54     if ((code = ubik_CheckAuth(rxcall))) {
  55         return code;
  56     }
  57     DBHOLD(ubik_dbase);
  58     if (urecovery_AllBetter(ubik_dbase, 0) == 0) {
  59         code = UNOQUORUM;
  60         goto out;
  61     }
  62     urecovery_CheckTid(atid, 1);
  63     code = udisk_begin(ubik_dbase, UBIK_WRITETRANS, &ubik_currentTrans);
  64     if (!code && ubik_currentTrans) {
  65         /* label this trans with the right trans id */
  66         ubik_currentTrans->tid.epoch = atid->epoch;
  67         ubik_currentTrans->tid.counter = atid->counter;
  68     }
  69   out:
  70     DBRELE(ubik_dbase);
  71     return code;
  72 }
  73
  74
  75 afs_int32
  76 SDISK_Commit(struct rx_call *rxcall, struct ubik_tid *atid)
  77 {
  78     afs_int32 code;
  79
  80     if ((code = ubik_CheckAuth(rxcall))) {
  81         return code;
  82     }
  83     ObtainWriteLock(&ubik_dbase->cache_lock);
  84     DBHOLD(ubik_dbase);
  85     if (!ubik_currentTrans) {
  86         code = USYNC;
  87         goto done;
  88     }
  89     /*
  90      * sanity check to make sure only write trans appear here
  91      */
  92     if (ubik_currentTrans->type != UBIK_WRITETRANS) {
  93         code = UBADTYPE;
  94         goto done;
  95     }
  96
  97     urecovery_CheckTid(atid, 0);
  98     if (!ubik_currentTrans) {
  99         code = USYNC;
 100         goto done;
 101     }
 102
 103     code = udisk_commit(ubik_currentTrans);
 104     if (code == 0) {
 105         /* sync site should now match */
 106         uvote_set_dbVersion(ubik_dbase->version);
 107     }
 108 done:
 109     DBRELE(ubik_dbase);
 110     ReleaseWriteLock(&ubik_dbase->cache_lock);
 111     return code;
 112 }
 113
 114 afs_int32
 115 SDISK_ReleaseLocks(struct rx_call *rxcall, struct ubik_tid *atid)
 116 {
 117     afs_int32 code;
 118
 119     if ((code = ubik_CheckAuth(rxcall))) {
 120         return code;
 121     }
 122
 123     DBHOLD(ubik_dbase);
 124
 125     if (!ubik_currentTrans) {
 126         code = USYNC;
 127         goto done;
 128     }
 129     /* sanity check to make sure only write trans appear here */
 130     if (ubik_currentTrans->type != UBIK_WRITETRANS) {
 131         code = UBADTYPE;
 132         goto done;
 133     }
 134
 135     urecovery_CheckTid(atid, 0);
 136     if (!ubik_currentTrans) {
 137         code = USYNC;
 138         goto done;
 139     }
 140
 141     /* If the thread is not waiting for lock - ok to end it */
 142     if (ubik_currentTrans->locktype != LOCKWAIT) {
 143         udisk_end(ubik_currentTrans);
 144     }
 145     ubik_currentTrans = (struct ubik_trans *)0;
 146 done:
 147     DBRELE(ubik_dbase);
 148     return code;
 149 }
 150
 151 afs_int32
 152 SDISK_Abort(struct rx_call *rxcall, struct ubik_tid *atid)
 153 {
 154     afs_int32 code;
 155
 156     if ((code = ubik_CheckAuth(rxcall))) {
 157         return code;
 158     }
 159     DBHOLD(ubik_dbase);
 160     if (!ubik_currentTrans) {
 161         code = USYNC;
 162         goto done;
 163     }
 164     /* sanity check to make sure only write trans appear here  */
 165     if (ubik_currentTrans->type != UBIK_WRITETRANS) {
 166         code = UBADTYPE;
 167         goto done;
 168     }
 169
 170     urecovery_CheckTid(atid, 0);
 171     if (!ubik_currentTrans) {
 172         code = USYNC;
 173         goto done;
 174     }
 175
 176     code = udisk_abort(ubik_currentTrans);
 177     /* If the thread is not waiting for lock - ok to end it */
 178     if (ubik_currentTrans->locktype != LOCKWAIT) {
 179         udisk_end(ubik_currentTrans);
 180     }
 181     ubik_currentTrans = (struct ubik_trans *)0;
 182 done:
 183     DBRELE(ubik_dbase);
 184     return code;
 185 }
 186
 187 /* apos and alen are not used */
 188 afs_int32
 189 SDISK_Lock(struct rx_call *rxcall, struct ubik_tid *atid,
 190            afs_int32 afile, afs_int32 apos, afs_int32 alen, afs_int32 atype)
 191 {
 192     afs_int32 code;
 193     struct ubik_trans *ubik_thisTrans;
 194
 195     if ((code = ubik_CheckAuth(rxcall))) {
 196         return code;
 197     }
 198     DBHOLD(ubik_dbase);
 199     if (!ubik_currentTrans) {
 200         code = USYNC;
 201         goto done;
 202     }
 203     /* sanity check to make sure only write trans appear here */
 204     if (ubik_currentTrans->type != UBIK_WRITETRANS) {
 205         code = UBADTYPE;
 206         goto done;
 207     }
 208     if (alen != 1) {
 209         code = UBADLOCK;
 210         goto done;
 211     }
 212     urecovery_CheckTid(atid, 0);
 213     if (!ubik_currentTrans) {
 214         code = USYNC;
 215         goto done;
 216     }
 217
 218     ubik_thisTrans = ubik_currentTrans;
 219     code = ulock_getLock(ubik_currentTrans, atype, 1);
 220
 221     /* While waiting, the transaction may have been ended/
 222      * aborted from under us (urecovery_CheckTid). In that
 223      * case, end the transaction here.
 224      */
 225     if (!code && (ubik_currentTrans != ubik_thisTrans)) {
 226         udisk_end(ubik_thisTrans);
 227         code = USYNC;
 228     }
 229 done:
 230     DBRELE(ubik_dbase);
 231     return code;
 232 }
 233
 234 /*!
 235  * \brief Write a vector of data
 236  */
 237 afs_int32
 238 SDISK_WriteV(struct rx_call *rxcall, struct ubik_tid *atid,
 239              iovec_wrt *io_vector, iovec_buf *io_buffer)
 240 {
 241     afs_int32 code, i, offset;
 242     struct ubik_iovec *iovec;
 243     char *iobuf;
 244
 245     if ((code = ubik_CheckAuth(rxcall))) {
 246         return code;
 247     }
 248     DBHOLD(ubik_dbase);
 249     if (!ubik_currentTrans) {
 250         code = USYNC;
 251         goto done;
 252     }
 253     /* sanity check to make sure only write trans appear here */
 254     if (ubik_currentTrans->type != UBIK_WRITETRANS) {
 255         code = UBADTYPE;
 256         goto done;
 257     }
 258
 259     urecovery_CheckTid(atid, 0);
 260     if (!ubik_currentTrans) {
 261         code = USYNC;
 262         goto done;
 263     }
 264
 265     iovec = (struct ubik_iovec *)io_vector->iovec_wrt_val;
 266     iobuf = (char *)io_buffer->iovec_buf_val;
 267     for (i = 0, offset = 0; i < io_vector->iovec_wrt_len; i++) {
 268         /* Sanity check for going off end of buffer */
 269         if ((offset + iovec[i].length) > io_buffer->iovec_buf_len) {
 270             code = UINTERNAL;
 271         } else {
 272             code =
 273                 udisk_write(ubik_currentTrans, iovec[i].file, &iobuf[offset],
 274                             iovec[i].position, iovec[i].length);
 275         }
 276         if (code)
 277             break;
 278
 279         offset += iovec[i].length;
 280     }
 281 done:
 282     DBRELE(ubik_dbase);
 283     return code;
 284 }
 285
 286 afs_int32
 287 SDISK_Write(struct rx_call *rxcall, struct ubik_tid *atid,
 288             afs_int32 afile, afs_int32 apos, bulkdata *adata)
 289 {
 290     afs_int32 code;
 291
 292     if ((code = ubik_CheckAuth(rxcall))) {
 293         return code;
 294     }
 295     DBHOLD(ubik_dbase);
 296     if (!ubik_currentTrans) {
 297         code = USYNC;
 298         goto done;
 299     }
 300     /* sanity check to make sure only write trans appear here */
 301     if (ubik_currentTrans->type != UBIK_WRITETRANS) {
 302         code = UBADTYPE;
 303         goto done;
 304     }
 305
 306     urecovery_CheckTid(atid, 0);
 307     if (!ubik_currentTrans) {
 308         code = USYNC;
 309         goto done;
 310     }
 311     code =
 312         udisk_write(ubik_currentTrans, afile, adata->bulkdata_val, apos,
 313                     adata->bulkdata_len);
 314 done:
 315     DBRELE(ubik_dbase);
 316     return code;
 317 }
 318
 319 afs_int32
 320 SDISK_Truncate(struct rx_call *rxcall, struct ubik_tid *atid,
 321                afs_int32 afile, afs_int32 alen)
 322 {
 323     afs_int32 code;
 324
 325     if ((code = ubik_CheckAuth(rxcall))) {
 326         return code;
 327     }
 328     DBHOLD(ubik_dbase);
 329     if (!ubik_currentTrans) {
 330         code = USYNC;
 331         goto done;
 332     }
 333     /* sanity check to make sure only write trans appear here */
 334     if (ubik_currentTrans->type != UBIK_WRITETRANS) {
 335         code = UBADTYPE;
 336         goto done;
 337     }
 338
 339     urecovery_CheckTid(atid, 0);
 340     if (!ubik_currentTrans) {
 341         code = USYNC;
 342         goto done;
 343     }
 344     code = udisk_truncate(ubik_currentTrans, afile, alen);
 345 done:
 346     DBRELE(ubik_dbase);
 347     return code;
 348 }
 349
 350 afs_int32
 351 SDISK_GetVersion(struct rx_call *rxcall,
 352                  struct ubik_version *aversion)
 353 {
 354     afs_int32 code;
 355
 356     if ((code = ubik_CheckAuth(rxcall))) {
 357         return code;
 358     }
 359
 360     /*
 361      * If we are the sync site, recovery shouldn't be running on any
 362      * other site. We shouldn't be getting this RPC as long as we are
 363      * the sync site.  To prevent any unforseen activity, we should
 364      * reject this RPC until we have recognized that we are not the
 365      * sync site anymore, and/or if we have any pending WRITE
 366      * transactions that have to complete. This way we can be assured
 367      * that this RPC would not block any pending transactions that
 368      * should either fail or pass. If we have recognized the fact that
 369      * we are not the sync site any more, all write transactions would
 370      * fail with UNOQUORUM anyway.
 371      */
 372     DBHOLD(ubik_dbase);
 373     if (ubeacon_AmSyncSite()) {
 374         DBRELE(ubik_dbase);
 375         return UDEADLOCK;
 376     }
 377
 378     code = (*ubik_dbase->getlabel) (ubik_dbase, 0, aversion);
 379     DBRELE(ubik_dbase);
 380     if (code) {
 381         /* tell other side there's no dbase */
 382         aversion->epoch = 0;
 383         aversion->counter = 0;
 384     }
 385     return 0;
 386 }
 387
 388 afs_int32
 389 SDISK_GetFile(struct rx_call *rxcall, afs_int32 file,
 390               struct ubik_version *version)
 391 {
 392     afs_int32 code;
 393     struct ubik_dbase *dbase;
 394     afs_int32 offset;
 395     struct ubik_stat ubikstat;
 396     char tbuffer[256];
 397     afs_int32 tlen;
 398     afs_int32 length;
 399
 400     if ((code = ubik_CheckAuth(rxcall))) {
 401         return code;
 402     }
 403     dbase = ubik_dbase;
 404     DBHOLD(dbase);
 405     code = (*dbase->stat) (dbase, file, &ubikstat);
 406     if (code < 0) {
 407         DBRELE(dbase);
 408         return code;
 409     }
 410     length = ubikstat.size;
 411     tlen = htonl(length);
 412     code = rx_Write(rxcall, (char *)&tlen, sizeof(afs_int32));
 413     if (code != sizeof(afs_int32)) {
 414         DBRELE(dbase);
 415         ubik_dprint("Rx-write length error=%d\n", code);
 416         return BULK_ERROR;
 417     }
 418     offset = 0;
 419     while (length > 0) {
 420         tlen = (length > sizeof(tbuffer) ? sizeof(tbuffer) : length);
 421         code = (*dbase->read) (dbase, file, tbuffer, offset, tlen);
 422         if (code != tlen) {
 423             DBRELE(dbase);
 424             ubik_dprint("read failed error=%d\n", code);
 425             return UIOERROR;
 426         }
 427         code = rx_Write(rxcall, tbuffer, tlen);
 428         if (code != tlen) {
 429             DBRELE(dbase);
 430             ubik_dprint("Rx-write length error=%d\n", code);
 431             return BULK_ERROR;
 432         }
 433         length -= tlen;
 434         offset += tlen;
 435     }
 436     code = (*dbase->getlabel) (dbase, file, version);   /* return the dbase, too */
 437     DBRELE(dbase);
 438     return code;
 439 }
 440
 441 afs_int32
 442 SDISK_SendFile(struct rx_call *rxcall, afs_int32 file,
 443                afs_int32 length, struct ubik_version *avers)
 444 {
 445     afs_int32 code;
 446     struct ubik_dbase *dbase = NULL;
 447     char tbuffer[1024];
 448     afs_int32 offset;
 449     struct ubik_version tversion;
 450     int tlen;
 451     struct rx_peer *tpeer;
 452     struct rx_connection *tconn;
 453     afs_uint32 otherHost = 0;
 454     char hoststr[16];
 455     char pbuffer[1028];
 456     int fd = -1;
 457     afs_int32 epoch = 0;
 458     afs_int32 pass;
 459
 460     /* send the file back to the requester */
 461
 462     dbase = ubik_dbase;
 463     pbuffer[0] = '\0';
 464
 465     if ((code = ubik_CheckAuth(rxcall))) {
 466         DBHOLD(dbase);
 467         goto failed;
 468     }
 469
 470     /* next, we do a sanity check to see if the guy sending us the database is
 471      * the guy we think is the sync site.  It turns out that we might not have
 472      * decided yet that someone's the sync site, but they could have enough
 473      * votes from others to be sync site anyway, and could send us the database
 474      * in advance of getting our votes.  This is fine, what we're really trying
 475      * to check is that some authenticated bogon isn't sending a random database
 476      * into another configuration.  This could happen on a bad configuration
 477      * screwup.  Thus, we only object if we're sure we know who the sync site
 478      * is, and it ain't the guy talking to us.
 479      */
 480     offset = uvote_GetSyncSite();
 481     tconn = rx_ConnectionOf(rxcall);
 482     tpeer = rx_PeerOf(tconn);
 483     otherHost = ubikGetPrimaryInterfaceAddr(rx_HostOf(tpeer));
 484     if (offset && offset != otherHost) {
 485         /* we *know* this is the wrong guy */
 486         code = USYNC;
 487         DBHOLD(dbase);
 488         goto failed;
 489     }
 490
 491     DBHOLD(dbase);
 492
 493     /* abort any active trans that may scribble over the database */
 494     urecovery_AbortAll(dbase);
 495
 496     ubik_print("Ubik: Synchronize database with server %s\n",
 497                afs_inet_ntoa_r(otherHost, hoststr));
 498
 499     offset = 0;
 500     UBIK_VERSION_LOCK;
 501     epoch = tversion.epoch = 0;         /* start off by labelling in-transit db as invalid */
 502     (*dbase->setlabel) (dbase, file, &tversion);        /* setlabel does sync */
 503     snprintf(pbuffer, sizeof(pbuffer), "%s.DB%s%d.TMP",
 504              ubik_dbase->pathName, (file<0)?"SYS":"",
 505              (file<0)?-file:file);
 506     fd = open(pbuffer, O_CREAT | O_RDWR | O_TRUNC, 0600);
 507     if (fd < 0) {
 508         code = errno;
 509         goto failed_locked;
 510     }
 511     code = lseek(fd, HDRSIZE, 0);
 512     if (code != HDRSIZE) {
 513         close(fd);
 514         goto failed_locked;
 515     }
 516     pass = 0;
 517     memcpy(&ubik_dbase->version, &tversion, sizeof(struct ubik_version));
 518     UBIK_VERSION_UNLOCK;
 519     while (length > 0) {
 520         tlen = (length > sizeof(tbuffer) ? sizeof(tbuffer) : length);
 521 #if !defined(AFS_PTHREAD_ENV)
 522         if (pass % 4 == 0)
 523             IOMGR_Poll();
 524 #endif
 525         code = rx_Read(rxcall, tbuffer, tlen);
 526         if (code != tlen) {
 527             ubik_dprint("Rx-read length error=%d\n", code);
 528             code = BULK_ERROR;
 529             close(fd);
 530             goto failed;
 531         }
 532         code = write(fd, tbuffer, tlen);
 533         pass++;
 534         if (code != tlen) {
 535             ubik_dprint("write failed error=%d\n", code);
 536             code = UIOERROR;
 537             close(fd);
 538             goto failed;
 539         }
 540         offset += tlen;
 541         length -= tlen;
 542     }
 543     code = close(fd);
 544     if (code)
 545         goto failed;
 546
 547     /* sync data first, then write label and resync (resync done by setlabel call).
 548      * This way, good label is only on good database. */
 549     snprintf(tbuffer, sizeof(tbuffer), "%s.DB%s%d",
 550              ubik_dbase->pathName, (file<0)?"SYS":"", (file<0)?-file:file);
 551 #ifdef AFS_NT40_ENV
 552     snprintf(pbuffer, sizeof(pbuffer), "%s.DB%s%d.OLD",
 553              ubik_dbase->pathName, (file<0)?"SYS":"", (file<0)?-file:file);
 554     code = unlink(pbuffer);
 555     if (!code)
 556         code = rename(tbuffer, pbuffer);
 557     snprintf(pbuffer, sizeof(pbuffer), "%s.DB%s%d.TMP",
 558              ubik_dbase->pathName, (file<0)?"SYS":"", (file<0)?-file:file);
 559 #endif
 560     if (!code)
 561         code = rename(pbuffer, tbuffer);
 562     UBIK_VERSION_LOCK;
 563     if (!code) {
 564         (*ubik_dbase->open) (ubik_dbase, file);
 565         code = (*ubik_dbase->setlabel) (dbase, file, avers);
 566     }
 567 #ifdef AFS_NT40_ENV
 568     snprintf(pbuffer, sizeof(pbuffer), "%s.DB%s%d.OLD",
 569              ubik_dbase->pathName, (file<0)?"SYS":"", (file<0)?-file:file);
 570     unlink(pbuffer);
 571 #endif
 572     memcpy(&ubik_dbase->version, avers, sizeof(struct ubik_version));
 573     udisk_Invalidate(dbase, file);      /* new dbase, flush disk buffers */
 574 #ifdef AFS_PTHREAD_ENV
 575     opr_Assert(pthread_cond_broadcast(&dbase->version_cond) == 0);
 576 #else
 577     LWP_NoYieldSignal(&dbase->version);
 578 #endif
 579
 580 failed_locked:
 581     UBIK_VERSION_UNLOCK;
 582
 583 failed:
 584     if (code) {
 585         if (pbuffer[0] != '\0')
 586             unlink(pbuffer);
 587
 588         /* Failed to sync. Allow reads again for now. */
 589         if (dbase != NULL) {
 590             UBIK_VERSION_LOCK;
 591             tversion.epoch = epoch;
 592             (*dbase->setlabel) (dbase, file, &tversion);
 593             UBIK_VERSION_UNLOCK;
 594         }
 595         ubik_print
 596             ("Ubik: Synchronize database with server %s failed (error = %d)\n",
 597              afs_inet_ntoa_r(otherHost, hoststr), code);
 598     } else {
 599         uvote_set_dbVersion(*avers);
 600         ubik_print("Ubik: Synchronize database completed\n");
 601     }
 602     DBRELE(dbase);
 603     return code;
 604 }
 605
 606
 607 afs_int32
 608 SDISK_Probe(struct rx_call *rxcall)
 609 {
 610     return 0;
 611 }
 612
 613 /*!
 614  * \brief Update remote machines addresses in my server list
 615  *
 616  * Send back my addresses to caller of this RPC
 617  * \return zero on success, else 1.
 618  */
 619 afs_int32
 620 SDISK_UpdateInterfaceAddr(struct rx_call *rxcall,
 621                           UbikInterfaceAddr *inAddr,
 622                           UbikInterfaceAddr *outAddr)
 623 {
 624     struct ubik_server *ts, *tmp;
 625     afs_uint32 remoteAddr;      /* in net byte order */
 626     int i, j, found = 0, probableMatch = 0;
 627     char hoststr[16];
 628
 629     UBIK_ADDR_LOCK;
 630     /* copy the output parameters */
 631     for (i = 0; i < UBIK_MAX_INTERFACE_ADDR; i++)
 632         outAddr->hostAddr[i] = ntohl(ubik_host[i]);
 633
 634     remoteAddr = htonl(inAddr->hostAddr[0]);
 635     for (ts = ubik_servers; ts; ts = ts->next)
 636         if (ts->addr[0] == remoteAddr) {        /* both in net byte order */
 637             probableMatch = 1;
 638             break;
 639         }
 640
 641     if (probableMatch) {
 642         /* verify that all addresses in the incoming RPC are
 643          ** not part of other server entries in my CellServDB
 644          */
 645         for (i = 0; !found && (i < UBIK_MAX_INTERFACE_ADDR)
 646              && inAddr->hostAddr[i]; i++) {
 647             remoteAddr = htonl(inAddr->hostAddr[i]);
 648             for (tmp = ubik_servers; (!found && tmp); tmp = tmp->next) {
 649                 if (ts == tmp)  /* this is my server */
 650                     continue;
 651                 for (j = 0; (j < UBIK_MAX_INTERFACE_ADDR) && tmp->addr[j];
 652                      j++)
 653                     if (remoteAddr == tmp->addr[j]) {
 654                         found = 1;
 655                         break;
 656                     }
 657             }
 658         }
 659     }
 660
 661     /* if (probableMatch) */
 662     /* inconsistent addresses in CellServDB */
 663     if (!probableMatch || found) {
 664         ubik_print("Inconsistent Cell Info from server:\n");
 665         for (i = 0; i < UBIK_MAX_INTERFACE_ADDR && inAddr->hostAddr[i]; i++)
 666             ubik_print("... %s\n", afs_inet_ntoa_r(htonl(inAddr->hostAddr[i]), hoststr));
 667         fflush(stdout);
 668         fflush(stderr);
 669         printServerInfo();
 670         UBIK_ADDR_UNLOCK;
 671         return UBADHOST;
 672     }
 673
 674     /* update our data structures */
 675     for (i = 1; i < UBIK_MAX_INTERFACE_ADDR; i++)
 676         ts->addr[i] = htonl(inAddr->hostAddr[i]);
 677
 678     ubik_print("ubik: A Remote Server has addresses:\n");
 679     for (i = 0; i < UBIK_MAX_INTERFACE_ADDR && ts->addr[i]; i++)
 680         ubik_print("... %s\n", afs_inet_ntoa_r(ts->addr[i], hoststr));
 681
 682     UBIK_ADDR_UNLOCK;
 683
 684     /*
 685      * The most likely cause of a DISK_UpdateInterfaceAddr RPC
 686      * is because the server was restarted.  Reset its state
 687      * so that no DISK_Begin RPCs will be issued until the
 688      * known database version is current.
 689      */
 690     UBIK_BEACON_LOCK;
 691     ts->beaconSinceDown = 0;
 692     ts->currentDB = 0;
 693     urecovery_LostServer(ts);
 694     UBIK_BEACON_UNLOCK;
 695     return 0;
 696 }
 697
 698 static void
 699 printServerInfo(void)
 700 {
 701     struct ubik_server *ts;
 702     int i, j = 1;
 703     char hoststr[16];
 704
 705     ubik_print("Local CellServDB:\n");
 706     for (ts = ubik_servers; ts; ts = ts->next, j++) {
 707         ubik_print("  Server %d:\n", j);
 708         for (i = 0; (i < UBIK_MAX_INTERFACE_ADDR) && ts->addr[i]; i++)
 709             ubik_print("  ... %s\n", afs_inet_ntoa_r(ts->addr[i], hoststr));
 710     }
 711 }
 712
 713 afs_int32
 714 SDISK_SetVersion(struct rx_call *rxcall, struct ubik_tid *atid,
 715                  struct ubik_version *oldversionp,
 716                  struct ubik_version *newversionp)
 717 {
 718     afs_int32 code = 0;
 719
 720     if ((code = ubik_CheckAuth(rxcall))) {
 721         return (code);
 722     }
 723     DBHOLD(ubik_dbase);
 724     if (!ubik_currentTrans) {
 725         code = USYNC;
 726         goto done;
 727     }
 728     /* sanity check to make sure only write trans appear here */
 729     if (ubik_currentTrans->type != UBIK_WRITETRANS) {
 730         code = UBADTYPE;
 731         goto done;
 732     }
 733
 734     /* Should not get this for the sync site */
 735     if (ubeacon_AmSyncSite()) {
 736         code = UDEADLOCK;
 737         goto done;
 738     }
 739
 740     urecovery_CheckTid(atid, 0);
 741     if (!ubik_currentTrans) {
 742         code = USYNC;
 743         goto done;
 744     }
 745
 746     /* Set the label if our version matches the sync-site's. Also set the label
 747      * if our on-disk version matches the old version, and our view of the
 748      * sync-site's version matches the new version. This suggests that
 749      * ubik_dbVersion was updated while the sync-site was setting the new
 750      * version, and it already told us via VOTE_Beacon. */
 751     if (uvote_eq_dbVersion(*oldversionp)
 752         || (uvote_eq_dbVersion(*newversionp)
 753             && vcmp(ubik_dbase->version, *oldversionp) == 0)) {
 754         UBIK_VERSION_LOCK;
 755         code = (*ubik_dbase->setlabel) (ubik_dbase, 0, newversionp);
 756         if (!code) {
 757             ubik_dbase->version = *newversionp;
 758             uvote_set_dbVersion(*newversionp);
 759         }
 760         UBIK_VERSION_UNLOCK;
 761     } else {
 762         code = USYNC;
 763     }
 764 done:
 765     DBRELE(ubik_dbase);
 766     return code;
 767 }