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