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