6b78d530bb66ce6045bdcb1968339760f7b09ef1
[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 <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         ViceLog(5, ("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             ViceLog(5, ("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             ViceLog(5, ("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         return code;
467     }
468
469     /* next, we do a sanity check to see if the guy sending us the database is
470      * the guy we think is the sync site.  It turns out that we might not have
471      * decided yet that someone's the sync site, but they could have enough
472      * votes from others to be sync site anyway, and could send us the database
473      * in advance of getting our votes.  This is fine, what we're really trying
474      * to check is that some authenticated bogon isn't sending a random database
475      * into another configuration.  This could happen on a bad configuration
476      * screwup.  Thus, we only object if we're sure we know who the sync site
477      * is, and it ain't the guy talking to us.
478      */
479     offset = uvote_GetSyncSite();
480     tconn = rx_ConnectionOf(rxcall);
481     tpeer = rx_PeerOf(tconn);
482     otherHost = ubikGetPrimaryInterfaceAddr(rx_HostOf(tpeer));
483     if (offset && offset != otherHost) {
484         /* we *know* this is the wrong guy */
485         char sync_hoststr[16];
486         ViceLog(0,
487             ("Ubik: Refusing synchronization with server %s since it is not the sync-site (%s).\n",
488              afs_inet_ntoa_r(otherHost, hoststr),
489              afs_inet_ntoa_r(offset, sync_hoststr)));
490         return USYNC;
491     }
492
493     DBHOLD(dbase);
494
495     /* abort any active trans that may scribble over the database */
496     urecovery_AbortAll(dbase);
497
498     ViceLog(0, ("Ubik: Synchronize database with server %s\n",
499                afs_inet_ntoa_r(otherHost, hoststr)));
500
501     offset = 0;
502     UBIK_VERSION_LOCK;
503     epoch = tversion.epoch = 0;         /* start off by labelling in-transit db as invalid */
504     (*dbase->setlabel) (dbase, file, &tversion);        /* setlabel does sync */
505     snprintf(pbuffer, sizeof(pbuffer), "%s.DB%s%d.TMP",
506              ubik_dbase->pathName, (file<0)?"SYS":"",
507              (file<0)?-file:file);
508     fd = open(pbuffer, O_CREAT | O_RDWR | O_TRUNC, 0600);
509     if (fd < 0) {
510         code = errno;
511         goto failed_locked;
512     }
513     code = lseek(fd, HDRSIZE, 0);
514     if (code != HDRSIZE) {
515         close(fd);
516         goto failed_locked;
517     }
518     pass = 0;
519     memcpy(&ubik_dbase->version, &tversion, sizeof(struct ubik_version));
520     UBIK_VERSION_UNLOCK;
521     while (length > 0) {
522         tlen = (length > sizeof(tbuffer) ? sizeof(tbuffer) : length);
523 #if !defined(AFS_PTHREAD_ENV)
524         if (pass % 4 == 0)
525             IOMGR_Poll();
526 #endif
527         code = rx_Read(rxcall, tbuffer, tlen);
528         if (code != tlen) {
529             ViceLog(5, ("Rx-read length error=%d\n", code));
530             code = BULK_ERROR;
531             close(fd);
532             goto failed;
533         }
534         code = write(fd, tbuffer, tlen);
535         pass++;
536         if (code != tlen) {
537             ViceLog(5, ("write failed error=%d\n", code));
538             code = UIOERROR;
539             close(fd);
540             goto failed;
541         }
542         offset += tlen;
543         length -= tlen;
544     }
545     code = close(fd);
546     if (code)
547         goto failed;
548
549     /* sync data first, then write label and resync (resync done by setlabel call).
550      * This way, good label is only on good database. */
551     snprintf(tbuffer, sizeof(tbuffer), "%s.DB%s%d",
552              ubik_dbase->pathName, (file<0)?"SYS":"", (file<0)?-file:file);
553 #ifdef AFS_NT40_ENV
554     snprintf(pbuffer, sizeof(pbuffer), "%s.DB%s%d.OLD",
555              ubik_dbase->pathName, (file<0)?"SYS":"", (file<0)?-file:file);
556     code = unlink(pbuffer);
557     if (!code)
558         code = rename(tbuffer, pbuffer);
559     snprintf(pbuffer, sizeof(pbuffer), "%s.DB%s%d.TMP",
560              ubik_dbase->pathName, (file<0)?"SYS":"", (file<0)?-file:file);
561 #endif
562     if (!code)
563         code = rename(pbuffer, tbuffer);
564     UBIK_VERSION_LOCK;
565     if (!code) {
566         (*ubik_dbase->open) (ubik_dbase, file);
567         code = (*ubik_dbase->setlabel) (dbase, file, avers);
568     }
569 #ifdef AFS_NT40_ENV
570     snprintf(pbuffer, sizeof(pbuffer), "%s.DB%s%d.OLD",
571              ubik_dbase->pathName, (file<0)?"SYS":"", (file<0)?-file:file);
572     unlink(pbuffer);
573 #endif
574     memcpy(&ubik_dbase->version, avers, sizeof(struct ubik_version));
575     udisk_Invalidate(dbase, file);      /* new dbase, flush disk buffers */
576 #ifdef AFS_PTHREAD_ENV
577     opr_Assert(pthread_cond_broadcast(&dbase->version_cond) == 0);
578 #else
579     LWP_NoYieldSignal(&dbase->version);
580 #endif
581
582 failed_locked:
583     UBIK_VERSION_UNLOCK;
584
585 failed:
586     if (code) {
587         if (pbuffer[0] != '\0')
588             unlink(pbuffer);
589
590         /* Failed to sync. Allow reads again for now. */
591         if (dbase != NULL) {
592             UBIK_VERSION_LOCK;
593             tversion.epoch = epoch;
594             (*dbase->setlabel) (dbase, file, &tversion);
595             UBIK_VERSION_UNLOCK;
596         }
597         ViceLog(0, ("Ubik: Synchronize database with server %s failed (error = %d)\n",
598              afs_inet_ntoa_r(otherHost, hoststr), code));
599     } else {
600         uvote_set_dbVersion(*avers);
601         ViceLog(0, ("Ubik: Synchronize database completed\n"));
602     }
603     DBRELE(dbase);
604     return code;
605 }
606
607
608 afs_int32
609 SDISK_Probe(struct rx_call *rxcall)
610 {
611     return 0;
612 }
613
614 /*!
615  * \brief Update remote machines addresses in my server list
616  *
617  * Send back my addresses to caller of this RPC
618  * \return zero on success, else 1.
619  */
620 afs_int32
621 SDISK_UpdateInterfaceAddr(struct rx_call *rxcall,
622                           UbikInterfaceAddr *inAddr,
623                           UbikInterfaceAddr *outAddr)
624 {
625     struct ubik_server *ts, *tmp;
626     afs_uint32 remoteAddr;      /* in net byte order */
627     int i, j, found = 0, probableMatch = 0;
628     char hoststr[16];
629
630     UBIK_ADDR_LOCK;
631     /* copy the output parameters */
632     for (i = 0; i < UBIK_MAX_INTERFACE_ADDR; i++)
633         outAddr->hostAddr[i] = ntohl(ubik_host[i]);
634
635     remoteAddr = htonl(inAddr->hostAddr[0]);
636     for (ts = ubik_servers; ts; ts = ts->next)
637         if (ts->addr[0] == remoteAddr) {        /* both in net byte order */
638             probableMatch = 1;
639             break;
640         }
641
642     if (probableMatch) {
643         /* verify that all addresses in the incoming RPC are
644          ** not part of other server entries in my CellServDB
645          */
646         for (i = 0; !found && (i < UBIK_MAX_INTERFACE_ADDR)
647              && inAddr->hostAddr[i]; i++) {
648             remoteAddr = htonl(inAddr->hostAddr[i]);
649             for (tmp = ubik_servers; (!found && tmp); tmp = tmp->next) {
650                 if (ts == tmp)  /* this is my server */
651                     continue;
652                 for (j = 0; (j < UBIK_MAX_INTERFACE_ADDR) && tmp->addr[j];
653                      j++)
654                     if (remoteAddr == tmp->addr[j]) {
655                         found = 1;
656                         break;
657                     }
658             }
659         }
660     }
661
662     /* if (probableMatch) */
663     /* inconsistent addresses in CellServDB */
664     if (!probableMatch || found) {
665         ViceLog(0, ("Inconsistent Cell Info from server:\n"));
666         for (i = 0; i < UBIK_MAX_INTERFACE_ADDR && inAddr->hostAddr[i]; i++)
667             ViceLog(0, ("... %s\n", afs_inet_ntoa_r(htonl(inAddr->hostAddr[i]), hoststr)));
668         fflush(stdout);
669         fflush(stderr);
670         printServerInfo();
671         UBIK_ADDR_UNLOCK;
672         return UBADHOST;
673     }
674
675     /* update our data structures */
676     for (i = 1; i < UBIK_MAX_INTERFACE_ADDR; i++)
677         ts->addr[i] = htonl(inAddr->hostAddr[i]);
678
679     ViceLog(0, ("ubik: A Remote Server has addresses:\n"));
680     for (i = 0; i < UBIK_MAX_INTERFACE_ADDR && ts->addr[i]; i++)
681         ViceLog(0, ("... %s\n", afs_inet_ntoa_r(ts->addr[i], hoststr)));
682
683     UBIK_ADDR_UNLOCK;
684
685     /*
686      * The most likely cause of a DISK_UpdateInterfaceAddr RPC
687      * is because the server was restarted.  Reset its state
688      * so that no DISK_Begin RPCs will be issued until the
689      * known database version is current.
690      */
691     UBIK_BEACON_LOCK;
692     ts->beaconSinceDown = 0;
693     ts->currentDB = 0;
694     urecovery_LostServer(ts);
695     UBIK_BEACON_UNLOCK;
696     return 0;
697 }
698
699 static void
700 printServerInfo(void)
701 {
702     struct ubik_server *ts;
703     int i, j = 1;
704     char hoststr[16];
705
706     ViceLog(0, ("Local CellServDB:\n"));
707     for (ts = ubik_servers; ts; ts = ts->next, j++) {
708         ViceLog(0, ("  Server %d:\n", j));
709         for (i = 0; (i < UBIK_MAX_INTERFACE_ADDR) && ts->addr[i]; i++)
710             ViceLog(0, ("  ... %s\n", afs_inet_ntoa_r(ts->addr[i], hoststr)));
711     }
712 }
713
714 afs_int32
715 SDISK_SetVersion(struct rx_call *rxcall, struct ubik_tid *atid,
716                  struct ubik_version *oldversionp,
717                  struct ubik_version *newversionp)
718 {
719     afs_int32 code = 0;
720
721     if ((code = ubik_CheckAuth(rxcall))) {
722         return (code);
723     }
724     DBHOLD(ubik_dbase);
725     if (!ubik_currentTrans) {
726         code = USYNC;
727         goto done;
728     }
729     /* sanity check to make sure only write trans appear here */
730     if (ubik_currentTrans->type != UBIK_WRITETRANS) {
731         code = UBADTYPE;
732         goto done;
733     }
734
735     /* Should not get this for the sync site */
736     if (ubeacon_AmSyncSite()) {
737         code = UDEADLOCK;
738         goto done;
739     }
740
741     urecovery_CheckTid(atid, 0);
742     if (!ubik_currentTrans) {
743         code = USYNC;
744         goto done;
745     }
746
747     /* Set the label if our version matches the sync-site's. Also set the label
748      * if our on-disk version matches the old version, and our view of the
749      * sync-site's version matches the new version. This suggests that
750      * ubik_dbVersion was updated while the sync-site was setting the new
751      * version, and it already told us via VOTE_Beacon. */
752     if (uvote_eq_dbVersion(*oldversionp)
753         || (uvote_eq_dbVersion(*newversionp)
754             && vcmp(ubik_dbase->version, *oldversionp) == 0)) {
755         UBIK_VERSION_LOCK;
756         code = (*ubik_dbase->setlabel) (ubik_dbase, 0, newversionp);
757         if (!code) {
758             ubik_dbase->version = *newversionp;
759             uvote_set_dbVersion(*newversionp);
760         }
761         UBIK_VERSION_UNLOCK;
762     } else {
763         code = USYNC;
764     }
765 done:
766     DBRELE(ubik_dbase);
767     return code;
768 }