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