ubik: Do not hide ReplayLog errors
[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 #include <assert.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(register struct ubik_dbase *adbase, int areadAny)
112 {
113     register 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     register 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(register struct ubik_tid *atid)
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) {
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(register struct ubik_dbase *adbase)
207 {
208     afs_int32 opcode;
209     register 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                 tpos += 4;
281                 code =
282                     (*adbase->read) (adbase, LOGFILE, (char *)buffer, tpos,
283                                      2 * sizeof(afs_int32));
284                 if (code != 2 * sizeof(afs_int32))
285                     return UBADLOG;
286                 code = (*adbase->setlabel) (adbase, 0, (ubik_version *)buffer);
287                 if (code)
288                     return code;
289                 logIsGood = 1;
290                 break;          /* all done now */
291             } else if (opcode == LOGTRUNCATE) {
292                 tpos += 4;
293                 code =
294                     (*adbase->read) (adbase, LOGFILE, (char *)buffer, tpos,
295                                      2 * sizeof(afs_int32));
296                 if (code != 2 * sizeof(afs_int32))
297                     break;      /* premature eof or io error */
298                 tpos += 2 * sizeof(afs_int32);
299                 code =
300                     (*adbase->truncate) (adbase, ntohl(buffer[0]),
301                                          ntohl(buffer[1]));
302                 if (code)
303                     return code;
304             } else if (opcode == LOGDATA) {
305                 tpos += 4;
306                 code =
307                     (*adbase->read) (adbase, LOGFILE, (char *)buffer, tpos,
308                                      3 * sizeof(afs_int32));
309                 if (code != 3 * sizeof(afs_int32))
310                     break;
311                 tpos += 3 * sizeof(afs_int32);
312                 /* otherwise, skip over the data bytes, too */
313                 len = ntohl(buffer[2]); /* total number of bytes to copy */
314                 filePos = ntohl(buffer[1]);
315                 tfile = ntohl(buffer[0]);
316                 /* try to minimize file syncs */
317                 if (syncFile != tfile) {
318                     if (syncFile >= 0)
319                         code = (*adbase->sync) (adbase, syncFile);
320                     else
321                         code = 0;
322                     syncFile = tfile;
323                     if (code)
324                         return code;
325                 }
326                 while (len > 0) {
327                     thisSize = (len > sizeof(data) ? sizeof(data) : len);
328                     /* copy sizeof(data) buffer bytes at a time */
329                     code =
330                         (*adbase->read) (adbase, LOGFILE, (char *)data, tpos,
331                                          thisSize);
332                     if (code != thisSize)
333                         return UBADLOG;
334                     code =
335                         (*adbase->write) (adbase, tfile, (char *)data, filePos,
336                                           thisSize);
337                     if (code != thisSize)
338                         return UBADLOG;
339                     filePos += thisSize;
340                     tpos += thisSize;
341                     len -= thisSize;
342                 }
343             } else {
344                 ubik_print("corrupt log opcode (%d) at position %d\n",
345                            opcode, tpos);
346                 break;          /* corrupt log! */
347             }
348         }
349         if (logIsGood) {
350             if (syncFile >= 0)
351                 code = (*adbase->sync) (adbase, syncFile);
352             if (code)
353                 return code;
354         } else {
355             ubik_print("Log read error on pass 2\n");
356             return UBADLOG;
357         }
358     }
359
360     /* now truncate the log, we're done with it */
361     code = (*adbase->truncate) (adbase, LOGFILE, 0);
362     return code;
363 }
364
365 /*! \brief
366  * Called at initialization to figure out version of the dbase we really have.
367  *
368  * This routine is called after replaying the log; it reads the restored labels.
369  */
370 static int
371 InitializeDB(register struct ubik_dbase *adbase)
372 {
373     register afs_int32 code;
374
375     code = (*adbase->getlabel) (adbase, 0, &adbase->version);
376     if (code) {
377         /* try setting the label to a new value */
378         adbase->version.epoch = 1;      /* value for newly-initialized db */
379         adbase->version.counter = 1;
380         code = (*adbase->setlabel) (adbase, 0, &adbase->version);
381         if (code) {
382             /* failed, try to set it back */
383             adbase->version.epoch = 0;
384             adbase->version.counter = 0;
385             (*adbase->setlabel) (adbase, 0, &adbase->version);
386         }
387 #ifdef AFS_PTHREAD_ENV
388         assert(pthread_cond_broadcast(&adbase->version_cond) == 0);
389 #else
390         LWP_NoYieldSignal(&adbase->version);
391 #endif
392     }
393     return 0;
394 }
395
396 /*!
397  * \brief initialize the local ubik_dbase
398  *
399  * We replay the logs and then read the resulting file to figure out what version we've really got.
400  */
401 int
402 urecovery_Initialize(register struct ubik_dbase *adbase)
403 {
404     register afs_int32 code;
405
406     code = ReplayLog(adbase);
407     if (code)
408         return code;
409     code = InitializeDB(adbase);
410     return code;
411 }
412
413 /*!
414  * \brief Main interaction loop for the recovery manager
415  *
416  * The recovery light-weight process only runs when you're the
417  * synchronization site.  It performs the following tasks, if and only
418  * if the prerequisite tasks have been performed successfully (it
419  * keeps track of which ones have been performed in its bit map,
420  * \p urecovery_state).
421  *
422  * First, it is responsible for probing that all servers are up.  This
423  * is the only operation that must be performed even if this is not
424  * yet the sync site, since otherwise this site may not notice that
425  * enough other machines are running to even elect this guy to be the
426  * sync site.
427  *
428  * After that, the recovery process does nothing until the beacon and
429  * voting modules manage to get this site elected sync site.
430  *
431  * After becoming sync site, recovery first attempts to find the best
432  * database available in the network (it must do this in order to
433  * ensure finding the latest committed data).  After finding the right
434  * database, it must fetch this dbase to the sync site.
435  *
436  * After fetching the dbase, it relabels it with a new version number,
437  * to ensure that everyone recognizes this dbase as the most recent
438  * dbase.
439  *
440  * One the dbase has been relabelled, this machine can start handling
441  * requests.  However, the recovery module still has one more task:
442  * propagating the dbase out to everyone who is up in the network.
443  */
444 void *
445 urecovery_Interact(void *dummy)
446 {
447     afs_int32 code, tcode;
448     struct ubik_server *bestServer = NULL;
449     struct ubik_server *ts;
450     int dbok, doingRPC, now;
451     afs_int32 lastProbeTime, lastDBVCheck;
452     /* if we're the sync site, the best db version we've found yet */
453     static struct ubik_version bestDBVersion;
454     struct ubik_version tversion;
455     struct timeval tv;
456     int length, tlen, offset, file, nbytes;
457     struct rx_call *rxcall;
458     char tbuffer[1024];
459     struct ubik_stat ubikstat;
460     struct in_addr inAddr;
461     char hoststr[16];
462 #ifndef OLD_URECOVERY
463     char pbuffer[1028];
464     int flen, fd = -1;
465     afs_int32 pass;
466 #endif
467
468     /* otherwise, begin interaction */
469     urecovery_state = 0;
470     lastProbeTime = 0;
471     lastDBVCheck = 0;
472     while (1) {
473         /* Run through this loop every 4 seconds */
474         tv.tv_sec = 4;
475         tv.tv_usec = 0;
476 #ifdef AFS_PTHREAD_ENV
477         select(0, 0, 0, 0, &tv);
478 #else
479         IOMGR_Select(0, 0, 0, 0, &tv);
480 #endif
481
482         ubik_dprint("recovery running in state %x\n", urecovery_state);
483
484         /* Every 30 seconds, check all the down servers and mark them
485          * as up if they respond. When a server comes up or found to
486          * not be current, then re-find the the best database and 
487          * propogate it.
488          */
489         if ((now = FT_ApproxTime()) > 30 + lastProbeTime) {
490             for (ts = ubik_servers, doingRPC = 0; ts; ts = ts->next) {
491                 if (!ts->up) {
492                     doingRPC = 1;
493                     code = DoProbe(ts);
494                     if (code == 0) {
495                         ts->up = 1;
496                         urecovery_state &= ~UBIK_RECFOUNDDB;
497                     }
498                 } else if (!ts->currentDB) {
499                     urecovery_state &= ~UBIK_RECFOUNDDB;
500                 }
501             }
502             if (doingRPC)
503                 now = FT_ApproxTime();
504             lastProbeTime = now;
505         }
506
507         /* Mark whether we are the sync site */
508         if (!ubeacon_AmSyncSite()) {
509             urecovery_state &= ~UBIK_RECSYNCSITE;
510             continue;           /* nothing to do */
511         }
512         urecovery_state |= UBIK_RECSYNCSITE;
513
514         /* If a server has just come up or if we have not found the 
515          * most current database, then go find the most current db.
516          */
517         if (!(urecovery_state & UBIK_RECFOUNDDB)) {
518             bestServer = (struct ubik_server *)0;
519             bestDBVersion.epoch = 0;
520             bestDBVersion.counter = 0;
521             for (ts = ubik_servers; ts; ts = ts->next) {
522                 if (!ts->up)
523                     continue;   /* don't bother with these guys */
524                 if (ts->isClone)
525                     continue;
526                 code = DISK_GetVersion(ts->disk_rxcid, &ts->version);
527                 if (code == 0) {
528                     /* perhaps this is the best version */
529                     if (vcmp(ts->version, bestDBVersion) > 0) {
530                         /* new best version */
531                         bestDBVersion = ts->version;
532                         bestServer = ts;
533                     }
534                 }
535             }
536             /* take into consideration our version. Remember if we,
537              * the sync site, have the best version. Also note that
538              * we may need to send the best version out.
539              */
540             if (vcmp(ubik_dbase->version, bestDBVersion) >= 0) {
541                 bestDBVersion = ubik_dbase->version;
542                 bestServer = (struct ubik_server *)0;
543                 urecovery_state |= UBIK_RECHAVEDB;
544             } else {
545                 /* Clear the flag only when we know we have to retrieve
546                  * the db. Because urecovery_AllBetter() looks at it.
547                  */
548                 urecovery_state &= ~UBIK_RECHAVEDB;
549             }
550             lastDBVCheck = FT_ApproxTime();
551             urecovery_state |= UBIK_RECFOUNDDB;
552             urecovery_state &= ~UBIK_RECSENTDB;
553         }
554 #if defined(UBIK_PAUSE)
555         /* it's not possible for UBIK_RECFOUNDDB not to be set here.
556          * However, we might have lost UBIK_RECSYNCSITE, and that
557          * IS important.
558          */
559         if (!(urecovery_state & UBIK_RECSYNCSITE))
560             continue;           /* lost sync */
561 #else
562         if (!(urecovery_state & UBIK_RECFOUNDDB))
563             continue;           /* not ready */
564 #endif /* UBIK_PAUSE */
565
566         /* If we, the sync site, do not have the best db version, then
567          * go and get it from the server that does.
568          */
569         if ((urecovery_state & UBIK_RECHAVEDB) || !bestServer) {
570             urecovery_state |= UBIK_RECHAVEDB;
571         } else {
572             /* we don't have the best version; we should fetch it. */
573             DBHOLD(ubik_dbase);
574             urecovery_AbortAll(ubik_dbase);
575
576             /* Rx code to do the Bulk fetch */
577             file = 0;
578             offset = 0;
579             rxcall = rx_NewCall(bestServer->disk_rxcid);
580
581             ubik_print("Ubik: Synchronize database with server %s\n",
582                        afs_inet_ntoa_r(bestServer->addr[0], hoststr));
583
584             code = StartDISK_GetFile(rxcall, file);
585             if (code) {
586                 ubik_dprint("StartDiskGetFile failed=%d\n", code);
587                 goto FetchEndCall;
588             }
589             nbytes = rx_Read(rxcall, (char *)&length, sizeof(afs_int32));
590             length = ntohl(length);
591             if (nbytes != sizeof(afs_int32)) {
592                 ubik_dprint("Rx-read length error=%d\n", code = BULK_ERROR);
593                 code = EIO;
594                 goto FetchEndCall;
595             }
596
597 #ifdef OLD_URECOVERY
598             /* Truncate the file first */
599             code = (*ubik_dbase->truncate) (ubik_dbase, file, 0);
600             if (code) {
601                 ubik_dprint("truncate io error=%d\n", code);
602                 goto FetchEndCall;
603             }
604             tversion.counter = 0;
605 #endif
606             /* give invalid label during file transit */
607             tversion.epoch = 0;
608             code = (*ubik_dbase->setlabel) (ubik_dbase, file, &tversion);
609             if (code) {
610                 ubik_dprint("setlabel io error=%d\n", code);
611                 goto FetchEndCall;
612             }
613 #ifndef OLD_URECOVERY
614             flen = length;
615             afs_snprintf(pbuffer, sizeof(pbuffer), "%s.DB%s%d.TMP", ubik_dbase->pathName, (file<0)?"SYS":"", (file<0)?-file:file);
616             fd = open(pbuffer, O_CREAT | O_RDWR | O_TRUNC, 0600);
617             if (fd < 0) {
618                 code = errno;
619                 goto FetchEndCall;
620             }
621             code = lseek(fd, HDRSIZE, 0);
622             if (code != HDRSIZE) {
623                 close(fd);
624                 goto FetchEndCall;
625             }
626 #endif
627
628             pass = 0;
629             while (length > 0) {
630                 tlen = (length > sizeof(tbuffer) ? sizeof(tbuffer) : length);
631 #ifndef AFS_PTHREAD_ENV
632                 if (pass % 4 == 0)
633                     IOMGR_Poll();
634 #endif
635                 nbytes = rx_Read(rxcall, tbuffer, tlen);
636                 if (nbytes != tlen) {
637                     ubik_dprint("Rx-read bulk error=%d\n", code = BULK_ERROR);
638                     code = EIO;
639                     close(fd);
640                     goto FetchEndCall;
641                 }
642 #ifdef OLD_URECOVERY
643                 nbytes =
644                     (*ubik_dbase->write) (ubik_dbase, file, tbuffer, offset,
645                                           tlen);
646 #else
647                 nbytes = write(fd, tbuffer, tlen);
648                 pass++;
649 #endif
650                 if (nbytes != tlen) {
651                     code = UIOERROR;
652                     close(fd);
653                     goto FetchEndCall;
654                 }
655                 offset += tlen;
656                 length -= tlen;
657             }
658 #ifndef OLD_URECOVERY
659             code = close(fd);
660             if (code)
661                 goto FetchEndCall;
662 #endif      
663             code = EndDISK_GetFile(rxcall, &tversion);
664           FetchEndCall:
665             tcode = rx_EndCall(rxcall, code);
666             if (!code)
667                 code = tcode;
668             if (!code) {
669                 /* we got a new file, set up its header */
670                 urecovery_state |= UBIK_RECHAVEDB;
671                 memcpy(&ubik_dbase->version, &tversion,
672                        sizeof(struct ubik_version));
673 #ifdef OLD_URECOVERY
674                 (*ubik_dbase->sync) (ubik_dbase, 0);    /* get data out first */
675 #else
676                 afs_snprintf(tbuffer, sizeof(tbuffer), "%s.DB%s%d", ubik_dbase->pathName, (file<0)?"SYS":"", (file<0)?-file:file);
677 #ifdef AFS_NT40_ENV
678                 afs_snprintf(pbuffer, sizeof(pbuffer), "%s.DB%s%d.OLD", ubik_dbase->pathName, (file<0)?"SYS":"", (file<0)?-file:file);
679                 code = unlink(pbuffer);
680                 if (!code)
681                     code = rename(tbuffer, pbuffer);
682                 afs_snprintf(pbuffer, sizeof(pbuffer), "%s.DB%s%d.TMP", ubik_dbase->pathName, (file<0)?"SYS":"", (file<0)?-file:file);
683 #endif
684                 if (!code) 
685                     code = rename(pbuffer, tbuffer);
686                 if (!code) {
687                     (*ubik_dbase->open) (ubik_dbase, 0);
688 #endif
689                     /* after data is good, sync disk with correct label */
690                     code =
691                         (*ubik_dbase->setlabel) (ubik_dbase, 0,
692                                                  &ubik_dbase->version);
693 #ifndef OLD_URECOVERY
694                 }
695 #ifdef AFS_NT40_ENV
696                 afs_snprintf(pbuffer, sizeof(pbuffer), "%s.DB%s%d.OLD", ubik_dbase->pathName, (file<0)?"SYS":"", (file<0)?-file:file);
697                 unlink(pbuffer);
698 #endif
699 #endif
700             }
701             if (code) {
702 #ifndef OLD_URECOVERY
703                 unlink(pbuffer);
704                 /* 
705                  * We will effectively invalidate the old data forever now.
706                  * Unclear if we *should* but we do.
707                  */
708 #endif
709                 ubik_dbase->version.epoch = 0;
710                 ubik_dbase->version.counter = 0;
711                 ubik_print("Ubik: Synchronize database failed (error = %d)\n",
712                            code);
713             } else {
714                 ubik_print("Ubik: Synchronize database completed\n");
715                 urecovery_state |= UBIK_RECHAVEDB;
716             }
717             udisk_Invalidate(ubik_dbase, 0);    /* data has changed */
718 #ifdef AFS_PTHREAD_ENV
719             assert(pthread_cond_broadcast(&ubik_dbase->version_cond) == 0);
720 #else
721             LWP_NoYieldSignal(&ubik_dbase->version);
722 #endif
723             DBRELE(ubik_dbase);
724         }
725 #if defined(UBIK_PAUSE)
726         if (!(urecovery_state & UBIK_RECSYNCSITE))
727             continue;           /* lost sync */
728 #endif /* UBIK_PAUSE */
729         if (!(urecovery_state & UBIK_RECHAVEDB))
730             continue;           /* not ready */
731
732         /* If the database was newly initialized, then when we establish quorum, write
733          * a new label. This allows urecovery_AllBetter() to allow access for reads.
734          * Setting it to 2 also allows another site to come along with a newer
735          * database and overwrite this one.
736          */
737         if (ubik_dbase->version.epoch == 1) {
738             DBHOLD(ubik_dbase);
739             urecovery_AbortAll(ubik_dbase);
740             ubik_epochTime = 2;
741             ubik_dbase->version.epoch = ubik_epochTime;
742             ubik_dbase->version.counter = 1;
743             code =
744                 (*ubik_dbase->setlabel) (ubik_dbase, 0, &ubik_dbase->version);
745             udisk_Invalidate(ubik_dbase, 0);    /* data may have changed */
746 #ifdef AFS_PTHREAD_ENV
747             assert(pthread_cond_broadcast(&ubik_dbase->version_cond) == 0);
748 #else
749             LWP_NoYieldSignal(&ubik_dbase->version);
750 #endif
751             DBRELE(ubik_dbase);
752         }
753
754         /* Check the other sites and send the database to them if they
755          * do not have the current db.
756          */
757         if (!(urecovery_state & UBIK_RECSENTDB)) {
758             /* now propagate out new version to everyone else */
759             dbok = 1;           /* start off assuming they all worked */
760
761             DBHOLD(ubik_dbase);
762             /*
763              * Check if a write transaction is in progress. We can't send the
764              * db when a write is in progress here because the db would be
765              * obsolete as soon as it goes there. Also, ops after the begin
766              * trans would reach the recepient and wouldn't find a transaction
767              * pending there.  Frankly, I don't think it's possible to get past
768              * the write-lock above if there is a write transaction in progress,
769              * but then, it won't hurt to check, will it?
770              */
771             if (ubik_dbase->flags & DBWRITING) {
772                 struct timeval tv;
773                 int safety = 0;
774                 tv.tv_sec = 0;
775                 tv.tv_usec = 50000;
776                 while ((ubik_dbase->flags & DBWRITING) && (safety < 500)) {
777                     DBRELE(ubik_dbase);
778                     /* sleep for a little while */
779 #ifdef AFS_PTHREAD_ENV
780                     select(0, 0, 0, 0, &tv);
781 #else
782                     IOMGR_Select(0, 0, 0, 0, &tv);
783 #endif
784                     tv.tv_usec += 10000;
785                     safety++;
786                     DBHOLD(ubik_dbase);
787                 }
788             }
789
790             for (ts = ubik_servers; ts; ts = ts->next) {
791                 inAddr.s_addr = ts->addr[0];
792                 if (!ts->up) {
793                     ubik_dprint("recovery cannot send version to %s\n",
794                                 afs_inet_ntoa_r(inAddr.s_addr, hoststr));
795                     dbok = 0;
796                     continue;
797                 }
798                 ubik_dprint("recovery sending version to %s\n",
799                             afs_inet_ntoa_r(inAddr.s_addr, hoststr));
800                 if (vcmp(ts->version, ubik_dbase->version) != 0) {
801                     ubik_dprint("recovery stating local database\n");
802
803                     /* Rx code to do the Bulk Store */
804                     code = (*ubik_dbase->stat) (ubik_dbase, 0, &ubikstat);
805                     if (!code) {
806                         length = ubikstat.size;
807                         file = offset = 0;
808                         rxcall = rx_NewCall(ts->disk_rxcid);
809                         code =
810                             StartDISK_SendFile(rxcall, file, length,
811                                                &ubik_dbase->version);
812                         if (code) {
813                             ubik_dprint("StartDiskSendFile failed=%d\n",
814                                         code);
815                             goto StoreEndCall;
816                         }
817                         while (length > 0) {
818                             tlen =
819                                 (length >
820                                  sizeof(tbuffer) ? sizeof(tbuffer) : length);
821                             nbytes =
822                                 (*ubik_dbase->read) (ubik_dbase, file,
823                                                      tbuffer, offset, tlen);
824                             if (nbytes != tlen) {
825                                 ubik_dprint("Local disk read error=%d\n",
826                                             code = UIOERROR);
827                                 goto StoreEndCall;
828                             }
829                             nbytes = rx_Write(rxcall, tbuffer, tlen);
830                             if (nbytes != tlen) {
831                                 ubik_dprint("Rx-write bulk error=%d\n", code =
832                                             BULK_ERROR);
833                                 goto StoreEndCall;
834                             }
835                             offset += tlen;
836                             length -= tlen;
837                         }
838                         code = EndDISK_SendFile(rxcall);
839                       StoreEndCall:
840                         code = rx_EndCall(rxcall, code);
841                     }
842                     if (code == 0) {
843                         /* we set a new file, process its header */
844                         ts->version = ubik_dbase->version;
845                         ts->currentDB = 1;
846                     } else
847                         dbok = 0;
848                 } else {
849                     /* mark file up to date */
850                     ts->currentDB = 1;
851                 }
852             }
853             DBRELE(ubik_dbase);
854             if (dbok)
855                 urecovery_state |= UBIK_RECSENTDB;
856         }
857     }
858     return NULL;
859 }
860
861 /*!
862  * \brief send a Probe to all the network address of this server 
863  * 
864  * \return 0 if success, else return 1
865  */
866 int
867 DoProbe(struct ubik_server *server)
868 {
869     struct rx_connection *conns[UBIK_MAX_INTERFACE_ADDR];
870     struct rx_connection *connSuccess = 0;
871     int i, j;
872     afs_uint32 addr;
873     char buffer[32];
874     char hoststr[16];
875     extern afs_int32 ubikSecIndex;
876     extern struct rx_securityClass *ubikSecClass;
877
878     for (i = 0; (addr = server->addr[i]) && (i < UBIK_MAX_INTERFACE_ADDR);
879          i++) {
880         conns[i] =
881             rx_NewConnection(addr, ubik_callPortal, DISK_SERVICE_ID,
882                              ubikSecClass, ubikSecIndex);
883
884         /* user requirement to use only the primary interface */
885         if (ubikPrimaryAddrOnly) {
886             i = 1;
887             break;
888         }
889     }
890     assert(i);                  /* at least one interface address for this server */
891
892     multi_Rx(conns, i) {
893         multi_DISK_Probe();
894         if (!multi_error) {     /* first success */
895             addr = server->addr[multi_i];       /* successful interface addr */
896
897             if (server->disk_rxcid)     /* destroy existing conn */
898                 rx_DestroyConnection(server->disk_rxcid);
899             if (server->vote_rxcid)
900                 rx_DestroyConnection(server->vote_rxcid);
901
902             /* make new connections */
903             server->disk_rxcid = conns[multi_i];
904             server->vote_rxcid = rx_NewConnection(addr, ubik_callPortal, VOTE_SERVICE_ID, ubikSecClass, ubikSecIndex);  /* for vote reqs */
905
906             connSuccess = conns[multi_i];
907             strcpy(buffer, afs_inet_ntoa_r(server->addr[0], hoststr));
908             ubik_print
909                 ("ubik:server %s is back up: will be contacted through %s\n",
910                  buffer, afs_inet_ntoa_r(addr, hoststr));
911
912             multi_Abort;
913         }
914     } multi_End_Ignore;
915
916     /* Destroy all connections except the one on which we succeeded */
917     for (j = 0; j < i; j++)
918         if (conns[j] != connSuccess)
919             rx_DestroyConnection(conns[j]);
920
921     if (!connSuccess)
922         ubik_dprint("ubik:server %s still down\n",
923                     afs_inet_ntoa_r(server->addr[0], hoststr));
924
925     if (connSuccess)
926         return 0;               /* success */
927     else
928         return 1;               /* failure */
929 }