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