ubik: Set but not used variables
[openafs.git] / src / ubik / recovery.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
14 #include <sys/types.h>
15 #include <string.h>
16 #include <stdarg.h>
17 #include <errno.h>
18
19 #ifdef AFS_NT40_ENV
20 #include <winsock2.h>
21 #include <time.h>
22 #include <fcntl.h>
23 #else
24 #include <sys/file.h>
25 #include <netinet/in.h>
26 #include <sys/time.h>
27 #endif
28
29 #include <lock.h>
30 #include <rx/xdr.h>
31 #include <rx/rx.h>
32 #include <afs/afsutil.h>
33
34 #define UBIK_INTERNALS
35 #include "ubik.h"
36 #include "ubik_int.h"
37
38 /*! \file
39  * This module is responsible for determining when the system has
40  * recovered to the point that it can handle new transactions.  It
41  * replays logs, polls to determine the current dbase after a crash,
42  * and distributes the new database to the others.
43  *
44  * The sync site associates a version number with each database.  It
45  * broadcasts the version associated with its current dbase in every
46  * one of its beacon messages.  When the sync site send a dbase to a
47  * server, it also sends the db's version.  A non-sync site server can
48  * tell if it has the right dbase version by simply comparing the
49  * version from the beacon message \p uvote_dbVersion with the version
50  * associated with the database \p ubik_dbase->version.  The sync site
51  * itself simply has one counter to keep track of all of this (again
52  * \p ubik_dbase->version).
53  *
54  * sync site: routine called when the sync site loses its quorum; this
55  * procedure is called "up" from the beacon package.  It resyncs the
56  * dbase and nudges the recovery daemon to try to propagate out the
57  * changes.  It also resets the recovery daemon's state, since
58  * recovery must potentially find a new dbase to propagate out.  This
59  * routine should not do anything with variables used by non-sync site
60  * servers.
61  */
62
63 /*!
64  * if this flag is set, then ubik will use only the primary address
65  * (the address specified in the CellServDB) to contact other
66  * ubik servers. Ubik recovery will not try opening connections
67  * to the alternate interface addresses.
68  */
69 int ubikPrimaryAddrOnly;
70
71 int
72 urecovery_ResetState(void)
73 {
74     urecovery_state = 0;
75 #if !defined(AFS_PTHREAD_ENV)
76     /*  No corresponding LWP_WaitProcess found anywhere for this -- klm */
77     LWP_NoYieldSignal(&urecovery_state);
78 #endif
79     return 0;
80 }
81
82 /*!
83  * \brief sync site
84  *
85  * routine called when a non-sync site server goes down; restarts recovery
86  * process to send missing server the new db when it comes back up.
87  *
88  * \note This routine should not do anything with variables used by non-sync site servers.
89  */
90 int
91 urecovery_LostServer(void)
92 {
93 #if !defined(AFS_PTHREAD_ENV)
94     /*  No corresponding LWP_WaitProcess found anywhere for this -- klm */
95     LWP_NoYieldSignal(&urecovery_state);
96 #endif
97     return 0;
98 }
99
100 /*!
101  * return true iff we have a current database (called by both sync
102  * sites and non-sync sites) How do we determine this?  If we're the
103  * sync site, we wait until recovery has finished fetching and
104  * re-labelling its dbase (it may still be trying to propagate it out
105  * to everyone else; that's THEIR problem).  If we're not the sync
106  * site, then we must have a dbase labelled with the right version,
107  * and we must have a currently-good sync site.
108  */
109 int
110 urecovery_AllBetter(struct ubik_dbase *adbase, int areadAny)
111 {
112     afs_int32 rcode;
113
114     ubik_dprint_25("allbetter checking\n");
115     rcode = 0;
116
117
118     if (areadAny) {
119         if (ubik_dbase->version.epoch > 1)
120             rcode = 1;          /* Happy with any good version of database */
121     }
122
123     /* Check if we're sync site and we've got the right data */
124     else if (ubeacon_AmSyncSite() && (urecovery_state & UBIK_RECHAVEDB)) {
125         rcode = 1;
126     }
127
128     /* next, check if we're aux site, and we've ever been sent the
129      * right data (note that if a dbase update fails, we won't think
130      * that the sync site is still the sync site, 'cause it won't talk
131      * to us until a timeout period has gone by.  When we recover, we
132      * leave this clear until we get a new dbase */
133     else if ((uvote_GetSyncSite() && (vcmp(ubik_dbVersion, ubik_dbase->version) == 0))) {       /* && order is important */
134         rcode = 1;
135     }
136
137     ubik_dprint_25("allbetter: returning %d\n", rcode);
138     return rcode;
139 }
140
141 /*!
142  * \brief abort all transactions on this database
143  */
144 int
145 urecovery_AbortAll(struct ubik_dbase *adbase)
146 {
147     struct ubik_trans *tt;
148     for (tt = adbase->activeTrans; tt; tt = tt->next) {
149         udisk_abort(tt);
150     }
151     return 0;
152 }
153
154 /*!
155  * \brief this routine aborts the current remote transaction, if any, if the tid is wrong
156  */
157 int
158 urecovery_CheckTid(struct ubik_tid *atid, int abortalways)
159 {
160     if (ubik_currentTrans) {
161         /* there is remote write trans, see if we match, see if this
162          * is a new transaction */
163         if (atid->epoch != ubik_currentTrans->tid.epoch
164             || atid->counter > ubik_currentTrans->tid.counter || abortalways) {
165             /* don't match, abort it */
166             /* If the thread is not waiting for lock - ok to end it */
167 #if !defined(UBIK_PAUSE)
168             if (ubik_currentTrans->locktype != LOCKWAIT) {
169 #endif /* UBIK_PAUSE */
170                 udisk_end(ubik_currentTrans);
171 #if !defined(UBIK_PAUSE)
172             }
173 #endif /* UBIK_PAUSE */
174             ubik_currentTrans = (struct ubik_trans *)0;
175         }
176     }
177     return 0;
178 }
179
180 /*!
181  * \brief replay logs
182  *
183  * log format is defined here, and implicitly in disk.c
184  *
185  * 4 byte opcode, followed by parameters, each 4 bytes long.  All integers
186  * are in logged in network standard byte order, in case we want to move logs
187  * from machine-to-machine someday.
188  *
189  * Begin transaction: opcode \n
190  * Commit transaction: opcode, version (8 bytes) \n
191  * Truncate file: opcode, file number, length \n
192  * Abort transaction: opcode \n
193  * Write data: opcode, file, position, length, <length> data bytes \n
194  *
195  * A very simple routine, it just replays the log.  Note that this is a new-value only log, which
196  * implies that no uncommitted data is written to the dbase: one writes data to the log, including
197  * the commit record, then we allow data to be written through to the dbase.  In our particular
198  * implementation, once a transaction is done, we write out the pages to the database, so that
199  * our buffer package doesn't have to know about stable and uncommitted data in the memory buffers:
200  * any changed data while there is an uncommitted write transaction can be zapped during an
201  * abort and the remaining dbase on the disk is exactly the right dbase, without having to read
202  * the log.
203  */
204 static int
205 ReplayLog(struct ubik_dbase *adbase)
206 {
207     afs_int32 opcode;
208     afs_int32 code, tpos;
209     int logIsGood;
210     afs_int32 len, thisSize, tfile, filePos;
211     afs_int32 buffer[4];
212     afs_int32 syncFile = -1;
213     afs_int32 data[1024];
214
215     /* read the lock twice, once to see whether we have a transaction to deal
216      * with that committed, (theoretically, we should support more than one
217      * trans in the log at once, but not yet), and once replaying the
218      * transactions.  */
219     tpos = 0;
220     logIsGood = 0;
221     /* for now, assume that all ops in log pertain to one transaction; see if there's a commit */
222     while (1) {
223         code =
224             (*adbase->read) (adbase, LOGFILE, (char *)&opcode, tpos,
225                              sizeof(afs_int32));
226         if (code != sizeof(afs_int32))
227             break;
228         opcode = ntohl(opcode);
229         if (opcode == LOGNEW) {
230             /* handle begin trans */
231             tpos += sizeof(afs_int32);
232         } else if (opcode == LOGABORT)
233             break;
234         else if (opcode == LOGEND) {
235             logIsGood = 1;
236             break;
237         } else if (opcode == LOGTRUNCATE) {
238             tpos += 4;
239             code =
240                 (*adbase->read) (adbase, LOGFILE, (char *)buffer, tpos,
241                                  2 * sizeof(afs_int32));
242             if (code != 2 * sizeof(afs_int32))
243                 break;          /* premature eof or io error */
244             tpos += 2 * sizeof(afs_int32);
245         } else if (opcode == LOGDATA) {
246             tpos += 4;
247             code =
248                 (*adbase->read) (adbase, LOGFILE, (char *)buffer, tpos,
249                                  3 * sizeof(afs_int32));
250             if (code != 3 * sizeof(afs_int32))
251                 break;
252             /* otherwise, skip over the data bytes, too */
253             tpos += ntohl(buffer[2]) + 3 * sizeof(afs_int32);
254         } else {
255             ubik_print("corrupt log opcode (%d) at position %d\n", opcode,
256                        tpos);
257             break;              /* corrupt log! */
258         }
259     }
260     if (logIsGood) {
261         /* actually do the replay; log should go all the way through the commit record, since
262          * we just read it above. */
263         tpos = 0;
264         logIsGood = 0;
265         syncFile = -1;
266         while (1) {
267             code =
268                 (*adbase->read) (adbase, LOGFILE, (char *)&opcode, tpos,
269                                  sizeof(afs_int32));
270             if (code != sizeof(afs_int32))
271                 break;
272             opcode = ntohl(opcode);
273             if (opcode == LOGNEW) {
274                 /* handle begin trans */
275                 tpos += sizeof(afs_int32);
276             } else if (opcode == LOGABORT)
277                 panic("log abort\n");
278             else if (opcode == LOGEND) {
279                 tpos += 4;
280                 code =
281                     (*adbase->read) (adbase, LOGFILE, (char *)buffer, tpos,
282                                      2 * sizeof(afs_int32));
283                 if (code != 2 * sizeof(afs_int32))
284                     return UBADLOG;
285                 code = (*adbase->setlabel) (adbase, 0, (ubik_version *)buffer);
286                 if (code)
287                     return code;
288                 logIsGood = 1;
289                 break;          /* all done now */
290             } else if (opcode == LOGTRUNCATE) {
291                 tpos += 4;
292                 code =
293                     (*adbase->read) (adbase, LOGFILE, (char *)buffer, tpos,
294                                      2 * sizeof(afs_int32));
295                 if (code != 2 * sizeof(afs_int32))
296                     break;      /* premature eof or io error */
297                 tpos += 2 * sizeof(afs_int32);
298                 code =
299                     (*adbase->truncate) (adbase, ntohl(buffer[0]),
300                                          ntohl(buffer[1]));
301                 if (code)
302                     return code;
303             } else if (opcode == LOGDATA) {
304                 tpos += 4;
305                 code =
306                     (*adbase->read) (adbase, LOGFILE, (char *)buffer, tpos,
307                                      3 * sizeof(afs_int32));
308                 if (code != 3 * sizeof(afs_int32))
309                     break;
310                 tpos += 3 * sizeof(afs_int32);
311                 /* otherwise, skip over the data bytes, too */
312                 len = ntohl(buffer[2]); /* total number of bytes to copy */
313                 filePos = ntohl(buffer[1]);
314                 tfile = ntohl(buffer[0]);
315                 /* try to minimize file syncs */
316                 if (syncFile != tfile) {
317                     if (syncFile >= 0)
318                         code = (*adbase->sync) (adbase, syncFile);
319                     else
320                         code = 0;
321                     syncFile = tfile;
322                     if (code)
323                         return code;
324                 }
325                 while (len > 0) {
326                     thisSize = (len > sizeof(data) ? sizeof(data) : len);
327                     /* copy sizeof(data) buffer bytes at a time */
328                     code =
329                         (*adbase->read) (adbase, LOGFILE, (char *)data, tpos,
330                                          thisSize);
331                     if (code != thisSize)
332                         return UBADLOG;
333                     code =
334                         (*adbase->write) (adbase, tfile, (char *)data, filePos,
335                                           thisSize);
336                     if (code != thisSize)
337                         return UBADLOG;
338                     filePos += thisSize;
339                     tpos += thisSize;
340                     len -= thisSize;
341                 }
342             } else {
343                 ubik_print("corrupt log opcode (%d) at position %d\n",
344                            opcode, tpos);
345                 break;          /* corrupt log! */
346             }
347         }
348         if (logIsGood) {
349             if (syncFile >= 0)
350                 code = (*adbase->sync) (adbase, syncFile);
351             if (code)
352                 return code;
353         } else {
354             ubik_print("Log read error on pass 2\n");
355             return UBADLOG;
356         }
357     }
358
359     /* now truncate the log, we're done with it */
360     code = (*adbase->truncate) (adbase, LOGFILE, 0);
361     return code;
362 }
363
364 /*! \brief
365  * Called at initialization to figure out version of the dbase we really have.
366  *
367  * This routine is called after replaying the log; it reads the restored labels.
368  */
369 static int
370 InitializeDB(struct ubik_dbase *adbase)
371 {
372     afs_int32 code;
373
374     code = (*adbase->getlabel) (adbase, 0, &adbase->version);
375     if (code) {
376         /* try setting the label to a new value */
377         adbase->version.epoch = 1;      /* value for newly-initialized db */
378         adbase->version.counter = 1;
379         code = (*adbase->setlabel) (adbase, 0, &adbase->version);
380         if (code) {
381             /* failed, try to set it back */
382             adbase->version.epoch = 0;
383             adbase->version.counter = 0;
384             (*adbase->setlabel) (adbase, 0, &adbase->version);
385         }
386 #ifdef AFS_PTHREAD_ENV
387         CV_BROADCAST(&adbase->version_cond);
388 #else
389         LWP_NoYieldSignal(&adbase->version);
390 #endif
391     }
392     return 0;
393 }
394
395 /*!
396  * \brief initialize the local ubik_dbase
397  *
398  * We replay the logs and then read the resulting file to figure out what version we've really got.
399  */
400 int
401 urecovery_Initialize(struct ubik_dbase *adbase)
402 {
403     afs_int32 code;
404
405     code = ReplayLog(adbase);
406     if (code)
407         return code;
408     code = InitializeDB(adbase);
409     return code;
410 }
411
412 /*!
413  * \brief Main interaction loop for the recovery manager
414  *
415  * The recovery light-weight process only runs when you're the
416  * synchronization site.  It performs the following tasks, if and only
417  * if the prerequisite tasks have been performed successfully (it
418  * keeps track of which ones have been performed in its bit map,
419  * \p urecovery_state).
420  *
421  * First, it is responsible for probing that all servers are up.  This
422  * is the only operation that must be performed even if this is not
423  * yet the sync site, since otherwise this site may not notice that
424  * enough other machines are running to even elect this guy to be the
425  * sync site.
426  *
427  * After that, the recovery process does nothing until the beacon and
428  * voting modules manage to get this site elected sync site.
429  *
430  * After becoming sync site, recovery first attempts to find the best
431  * database available in the network (it must do this in order to
432  * ensure finding the latest committed data).  After finding the right
433  * database, it must fetch this dbase to the sync site.
434  *
435  * After fetching the dbase, it relabels it with a new version number,
436  * to ensure that everyone recognizes this dbase as the most recent
437  * dbase.
438  *
439  * One the dbase has been relabelled, this machine can start handling
440  * requests.  However, the recovery module still has one more task:
441  * propagating the dbase out to everyone who is up in the network.
442  */
443 void *
444 urecovery_Interact(void *dummy)
445 {
446     afs_int32 code, tcode;
447     struct ubik_server *bestServer = NULL;
448     struct ubik_server *ts;
449     int dbok, doingRPC, now;
450     afs_int32 lastProbeTime;
451     /* if we're the sync site, the best db version we've found yet */
452     static struct ubik_version bestDBVersion;
453     struct ubik_version tversion;
454     struct timeval tv;
455     int length, tlen, offset, file, nbytes;
456     struct rx_call *rxcall;
457     char tbuffer[1024];
458     struct ubik_stat ubikstat;
459     struct in_addr inAddr;
460     char hoststr[16];
461 #ifndef OLD_URECOVERY
462     char pbuffer[1028];
463     int fd = -1;
464     afs_int32 pass;
465 #endif
466
467     /* otherwise, begin interaction */
468     urecovery_state = 0;
469     lastProbeTime = 0;
470     while (1) {
471         /* Run through this loop every 4 seconds */
472         tv.tv_sec = 4;
473         tv.tv_usec = 0;
474 #ifdef AFS_PTHREAD_ENV
475         select(0, 0, 0, 0, &tv);
476 #else
477         IOMGR_Select(0, 0, 0, 0, &tv);
478 #endif
479
480         ubik_dprint("recovery running in state %x\n", urecovery_state);
481
482         /* Every 30 seconds, check all the down servers and mark them
483          * as up if they respond. When a server comes up or found to
484          * not be current, then re-find the the best database and
485          * propogate it.
486          */
487         if ((now = FT_ApproxTime()) > 30 + lastProbeTime) {
488
489 #ifdef AFS_PTHREAD_ENV
490             DBHOLD(ubik_dbase);
491 #endif
492
493             for (ts = ubik_servers, doingRPC = 0; ts; ts = ts->next) {
494                 if (!ts->up) {
495                     doingRPC = 1;
496                     code = DoProbe(ts);
497                     if (code == 0) {
498                         ts->up = 1;
499                         urecovery_state &= ~UBIK_RECFOUNDDB;
500                     }
501                 } else if (!ts->currentDB) {
502                     urecovery_state &= ~UBIK_RECFOUNDDB;
503                 }
504             }
505
506 #ifdef AFS_PTHREAD_ENV
507             DBRELE(ubik_dbase);
508 #endif
509
510             if (doingRPC)
511                 now = FT_ApproxTime();
512             lastProbeTime = now;
513         }
514
515         /* Mark whether we are the sync site */
516         if (!ubeacon_AmSyncSite()) {
517             urecovery_state &= ~UBIK_RECSYNCSITE;
518             continue;           /* nothing to do */
519         }
520         urecovery_state |= UBIK_RECSYNCSITE;
521
522         /* If a server has just come up or if we have not found the
523          * most current database, then go find the most current db.
524          */
525         if (!(urecovery_state & UBIK_RECFOUNDDB)) {
526             bestServer = (struct ubik_server *)0;
527             bestDBVersion.epoch = 0;
528             bestDBVersion.counter = 0;
529             for (ts = ubik_servers; ts; ts = ts->next) {
530                 if (!ts->up)
531                     continue;   /* don't bother with these guys */
532                 if (ts->isClone)
533                     continue;
534                 code = DISK_GetVersion(ts->disk_rxcid, &ts->version);
535                 if (code == 0) {
536                     /* perhaps this is the best version */
537                     if (vcmp(ts->version, bestDBVersion) > 0) {
538                         /* new best version */
539                         bestDBVersion = ts->version;
540                         bestServer = ts;
541                     }
542                 }
543             }
544             /* take into consideration our version. Remember if we,
545              * the sync site, have the best version. Also note that
546              * we may need to send the best version out.
547              */
548             if (vcmp(ubik_dbase->version, bestDBVersion) >= 0) {
549                 bestDBVersion = ubik_dbase->version;
550                 bestServer = (struct ubik_server *)0;
551                 urecovery_state |= UBIK_RECHAVEDB;
552             } else {
553                 /* Clear the flag only when we know we have to retrieve
554                  * the db. Because urecovery_AllBetter() looks at it.
555                  */
556                 urecovery_state &= ~UBIK_RECHAVEDB;
557             }
558             urecovery_state |= UBIK_RECFOUNDDB;
559             urecovery_state &= ~UBIK_RECSENTDB;
560         }
561 #if defined(UBIK_PAUSE)
562         /* it's not possible for UBIK_RECFOUNDDB not to be set here.
563          * However, we might have lost UBIK_RECSYNCSITE, and that
564          * IS important.
565          */
566         if (!(urecovery_state & UBIK_RECSYNCSITE))
567             continue;           /* lost sync */
568 #else
569         if (!(urecovery_state & UBIK_RECFOUNDDB))
570             continue;           /* not ready */
571 #endif /* UBIK_PAUSE */
572
573         /* If we, the sync site, do not have the best db version, then
574          * go and get it from the server that does.
575          */
576         if ((urecovery_state & UBIK_RECHAVEDB) || !bestServer) {
577             urecovery_state |= UBIK_RECHAVEDB;
578         } else {
579             /* we don't have the best version; we should fetch it. */
580             DBHOLD(ubik_dbase);
581             urecovery_AbortAll(ubik_dbase);
582
583             /* Rx code to do the Bulk fetch */
584             file = 0;
585             offset = 0;
586             rxcall = rx_NewCall(bestServer->disk_rxcid);
587
588             ubik_print("Ubik: Synchronize database with server %s\n",
589                        afs_inet_ntoa_r(bestServer->addr[0], hoststr));
590
591             code = StartDISK_GetFile(rxcall, file);
592             if (code) {
593                 ubik_dprint("StartDiskGetFile failed=%d\n", code);
594                 goto FetchEndCall;
595             }
596             nbytes = rx_Read(rxcall, (char *)&length, sizeof(afs_int32));
597             length = ntohl(length);
598             if (nbytes != sizeof(afs_int32)) {
599                 ubik_dprint("Rx-read length error=%d\n", code = BULK_ERROR);
600                 code = EIO;
601                 goto FetchEndCall;
602             }
603
604 #ifdef OLD_URECOVERY
605             /* Truncate the file first */
606             code = (*ubik_dbase->truncate) (ubik_dbase, file, 0);
607             if (code) {
608                 ubik_dprint("truncate io error=%d\n", code);
609                 goto FetchEndCall;
610             }
611             tversion.counter = 0;
612 #endif
613             /* give invalid label during file transit */
614             tversion.epoch = 0;
615             code = (*ubik_dbase->setlabel) (ubik_dbase, file, &tversion);
616             if (code) {
617                 ubik_dprint("setlabel io error=%d\n", code);
618                 goto FetchEndCall;
619             }
620 #ifndef OLD_URECOVERY
621             afs_snprintf(pbuffer, sizeof(pbuffer), "%s.DB%s%d.TMP", ubik_dbase->pathName, (file<0)?"SYS":"", (file<0)?-file:file);
622             fd = open(pbuffer, O_CREAT | O_RDWR | O_TRUNC, 0600);
623             if (fd < 0) {
624                 code = errno;
625                 goto FetchEndCall;
626             }
627             code = lseek(fd, HDRSIZE, 0);
628             if (code != HDRSIZE) {
629                 close(fd);
630                 goto FetchEndCall;
631             }
632 #endif
633
634             pass = 0;
635             while (length > 0) {
636                 tlen = (length > sizeof(tbuffer) ? sizeof(tbuffer) : length);
637 #ifndef AFS_PTHREAD_ENV
638                 if (pass % 4 == 0)
639                     IOMGR_Poll();
640 #endif
641                 nbytes = rx_Read(rxcall, tbuffer, tlen);
642                 if (nbytes != tlen) {
643                     ubik_dprint("Rx-read bulk error=%d\n", code = BULK_ERROR);
644                     code = EIO;
645                     close(fd);
646                     goto FetchEndCall;
647                 }
648 #ifdef OLD_URECOVERY
649                 nbytes =
650                     (*ubik_dbase->write) (ubik_dbase, file, tbuffer, offset,
651                                           tlen);
652 #else
653                 nbytes = write(fd, tbuffer, tlen);
654                 pass++;
655 #endif
656                 if (nbytes != tlen) {
657                     code = UIOERROR;
658                     close(fd);
659                     goto FetchEndCall;
660                 }
661                 offset += tlen;
662                 length -= tlen;
663             }
664 #ifndef OLD_URECOVERY
665             code = close(fd);
666             if (code)
667                 goto FetchEndCall;
668 #endif
669             code = EndDISK_GetFile(rxcall, &tversion);
670           FetchEndCall:
671             tcode = rx_EndCall(rxcall, code);
672             if (!code)
673                 code = tcode;
674             if (!code) {
675                 /* we got a new file, set up its header */
676                 urecovery_state |= UBIK_RECHAVEDB;
677                 memcpy(&ubik_dbase->version, &tversion,
678                        sizeof(struct ubik_version));
679 #ifdef OLD_URECOVERY
680                 (*ubik_dbase->sync) (ubik_dbase, 0);    /* get data out first */
681 #else
682                 afs_snprintf(tbuffer, sizeof(tbuffer), "%s.DB%s%d", ubik_dbase->pathName, (file<0)?"SYS":"", (file<0)?-file:file);
683 #ifdef AFS_NT40_ENV
684                 afs_snprintf(pbuffer, sizeof(pbuffer), "%s.DB%s%d.OLD", ubik_dbase->pathName, (file<0)?"SYS":"", (file<0)?-file:file);
685                 code = unlink(pbuffer);
686                 if (!code)
687                     code = rename(tbuffer, pbuffer);
688                 afs_snprintf(pbuffer, sizeof(pbuffer), "%s.DB%s%d.TMP", ubik_dbase->pathName, (file<0)?"SYS":"", (file<0)?-file:file);
689 #endif
690                 if (!code)
691                     code = rename(pbuffer, tbuffer);
692                 if (!code) {
693                     (*ubik_dbase->open) (ubik_dbase, file);
694 #endif
695                     /* after data is good, sync disk with correct label */
696                     code =
697                         (*ubik_dbase->setlabel) (ubik_dbase, 0,
698                                                  &ubik_dbase->version);
699 #ifndef OLD_URECOVERY
700                 }
701 #ifdef AFS_NT40_ENV
702                 afs_snprintf(pbuffer, sizeof(pbuffer), "%s.DB%s%d.OLD", ubik_dbase->pathName, (file<0)?"SYS":"", (file<0)?-file:file);
703                 unlink(pbuffer);
704 #endif
705 #endif
706             }
707             if (code) {
708 #ifndef OLD_URECOVERY
709                 unlink(pbuffer);
710                 /*
711                  * We will effectively invalidate the old data forever now.
712                  * Unclear if we *should* but we do.
713                  */
714 #endif
715                 ubik_dbase->version.epoch = 0;
716                 ubik_dbase->version.counter = 0;
717                 ubik_print("Ubik: Synchronize database failed (error = %d)\n",
718                            code);
719             } else {
720                 ubik_print("Ubik: Synchronize database completed\n");
721                 urecovery_state |= UBIK_RECHAVEDB;
722             }
723             udisk_Invalidate(ubik_dbase, 0);    /* data has changed */
724 #ifdef AFS_PTHREAD_ENV
725             CV_BROADCAST(&ubik_dbase->version_cond);
726 #else
727             LWP_NoYieldSignal(&ubik_dbase->version);
728 #endif
729             DBRELE(ubik_dbase);
730         }
731 #if defined(UBIK_PAUSE)
732         if (!(urecovery_state & UBIK_RECSYNCSITE))
733             continue;           /* lost sync */
734 #endif /* UBIK_PAUSE */
735         if (!(urecovery_state & UBIK_RECHAVEDB))
736             continue;           /* not ready */
737
738         /* If the database was newly initialized, then when we establish quorum, write
739          * a new label. This allows urecovery_AllBetter() to allow access for reads.
740          * Setting it to 2 also allows another site to come along with a newer
741          * database and overwrite this one.
742          */
743         if (ubik_dbase->version.epoch == 1) {
744             DBHOLD(ubik_dbase);
745             urecovery_AbortAll(ubik_dbase);
746             ubik_epochTime = 2;
747             ubik_dbase->version.epoch = ubik_epochTime;
748             ubik_dbase->version.counter = 1;
749             code =
750                 (*ubik_dbase->setlabel) (ubik_dbase, 0, &ubik_dbase->version);
751             udisk_Invalidate(ubik_dbase, 0);    /* data may have changed */
752 #ifdef AFS_PTHREAD_ENV
753             CV_BROADCAST(&ubik_dbase->version_cond);
754 #else
755             LWP_NoYieldSignal(&ubik_dbase->version);
756 #endif
757             DBRELE(ubik_dbase);
758         }
759
760         /* Check the other sites and send the database to them if they
761          * do not have the current db.
762          */
763         if (!(urecovery_state & UBIK_RECSENTDB)) {
764             /* now propagate out new version to everyone else */
765             dbok = 1;           /* start off assuming they all worked */
766
767             DBHOLD(ubik_dbase);
768             /*
769              * Check if a write transaction is in progress. We can't send the
770              * db when a write is in progress here because the db would be
771              * obsolete as soon as it goes there. Also, ops after the begin
772              * trans would reach the recepient and wouldn't find a transaction
773              * pending there.  Frankly, I don't think it's possible to get past
774              * the write-lock above if there is a write transaction in progress,
775              * but then, it won't hurt to check, will it?
776              */
777             if (ubik_dbase->flags & DBWRITING) {
778                 struct timeval tv;
779                 int safety = 0;
780                 tv.tv_sec = 0;
781                 tv.tv_usec = 50000;
782                 while ((ubik_dbase->flags & DBWRITING) && (safety < 500)) {
783                     DBRELE(ubik_dbase);
784                     /* sleep for a little while */
785 #ifdef AFS_PTHREAD_ENV
786                     select(0, 0, 0, 0, &tv);
787 #else
788                     IOMGR_Select(0, 0, 0, 0, &tv);
789 #endif
790                     tv.tv_usec += 10000;
791                     safety++;
792                     DBHOLD(ubik_dbase);
793                 }
794             }
795
796             for (ts = ubik_servers; ts; ts = ts->next) {
797                 inAddr.s_addr = ts->addr[0];
798                 if (!ts->up) {
799                     ubik_dprint("recovery cannot send version to %s\n",
800                                 afs_inet_ntoa_r(inAddr.s_addr, hoststr));
801                     dbok = 0;
802                     continue;
803                 }
804                 ubik_dprint("recovery sending version to %s\n",
805                             afs_inet_ntoa_r(inAddr.s_addr, hoststr));
806                 if (vcmp(ts->version, ubik_dbase->version) != 0) {
807                     ubik_dprint("recovery stating local database\n");
808
809                     /* Rx code to do the Bulk Store */
810                     code = (*ubik_dbase->stat) (ubik_dbase, 0, &ubikstat);
811                     if (!code) {
812                         length = ubikstat.size;
813                         file = offset = 0;
814                         rxcall = rx_NewCall(ts->disk_rxcid);
815                         code =
816                             StartDISK_SendFile(rxcall, file, length,
817                                                &ubik_dbase->version);
818                         if (code) {
819                             ubik_dprint("StartDiskSendFile failed=%d\n",
820                                         code);
821                             goto StoreEndCall;
822                         }
823                         while (length > 0) {
824                             tlen =
825                                 (length >
826                                  sizeof(tbuffer) ? sizeof(tbuffer) : length);
827                             nbytes =
828                                 (*ubik_dbase->read) (ubik_dbase, file,
829                                                      tbuffer, offset, tlen);
830                             if (nbytes != tlen) {
831                                 ubik_dprint("Local disk read error=%d\n",
832                                             code = UIOERROR);
833                                 goto StoreEndCall;
834                             }
835                             nbytes = rx_Write(rxcall, tbuffer, tlen);
836                             if (nbytes != tlen) {
837                                 ubik_dprint("Rx-write bulk error=%d\n", code =
838                                             BULK_ERROR);
839                                 goto StoreEndCall;
840                             }
841                             offset += tlen;
842                             length -= tlen;
843                         }
844                         code = EndDISK_SendFile(rxcall);
845                       StoreEndCall:
846                         code = rx_EndCall(rxcall, code);
847                     }
848                     if (code == 0) {
849                         /* we set a new file, process its header */
850                         ts->version = ubik_dbase->version;
851                         ts->currentDB = 1;
852                     } else
853                         dbok = 0;
854                 } else {
855                     /* mark file up to date */
856                     ts->currentDB = 1;
857                 }
858             }
859             DBRELE(ubik_dbase);
860             if (dbok)
861                 urecovery_state |= UBIK_RECSENTDB;
862         }
863     }
864     return NULL;
865 }
866
867 /*!
868  * \brief send a Probe to all the network address of this server
869  *
870  * \return 0 if success, else return 1
871  */
872 int
873 DoProbe(struct ubik_server *server)
874 {
875     struct rx_connection *conns[UBIK_MAX_INTERFACE_ADDR];
876     struct rx_connection *connSuccess = 0;
877     int i, j, success_i = -1;
878     afs_uint32 addr;
879     char buffer[32];
880     char hoststr[16];
881     extern afs_int32 ubikSecIndex;
882     extern struct rx_securityClass *ubikSecClass;
883
884     for (i = 0; (addr = server->addr[i]) && (i < UBIK_MAX_INTERFACE_ADDR);
885          i++) {
886         conns[i] =
887             rx_NewConnection(addr, ubik_callPortal, DISK_SERVICE_ID,
888                              ubikSecClass, ubikSecIndex);
889
890         /* user requirement to use only the primary interface */
891         if (ubikPrimaryAddrOnly) {
892             i = 1;
893             break;
894         }
895     }
896     osi_Assert(i);                      /* at least one interface address for this server */
897
898 #ifdef AFS_PTHREAD_ENV
899     DBRELE(ubik_dbase);
900 #endif
901
902     multi_Rx(conns, i) {
903         multi_DISK_Probe();
904         if (!multi_error) {     /* first success */
905             success_i = multi_i;
906
907             multi_Abort;
908         }
909     } multi_End_Ignore;
910
911 #ifdef AFS_PTHREAD_ENV
912     DBHOLD(ubik_dbase);
913 #endif
914
915     if (success_i >= 0) {
916         addr = server->addr[success_i]; /* successful interface addr */
917
918         if (server->disk_rxcid) /* destroy existing conn */
919             rx_DestroyConnection(server->disk_rxcid);
920         if (server->vote_rxcid)
921             rx_DestroyConnection(server->vote_rxcid);
922
923         /* make new connections */
924         server->disk_rxcid = conns[success_i];
925         server->vote_rxcid = rx_NewConnection(addr, ubik_callPortal,
926                                               VOTE_SERVICE_ID, ubikSecClass,
927                                               ubikSecIndex);
928
929         connSuccess = conns[success_i];
930         strcpy(buffer, afs_inet_ntoa_r(server->addr[0], hoststr));
931
932         ubik_print("ubik:server %s is back up: will be contacted through %s\n",
933              buffer, afs_inet_ntoa_r(addr, hoststr));
934     }
935
936     /* Destroy all connections except the one on which we succeeded */
937     for (j = 0; j < i; j++)
938         if (conns[j] != connSuccess)
939             rx_DestroyConnection(conns[j]);
940
941     if (!connSuccess)
942         ubik_dprint("ubik:server %s still down\n",
943                     afs_inet_ntoa_r(server->addr[0], hoststr));
944
945     if (connSuccess)
946         return 0;               /* success */
947     else
948         return 1;               /* failure */
949 }