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