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