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