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