ubik recovery and remote use correct file number
[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(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                 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(struct ubik_dbase *adbase)
372 {
373     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(struct ubik_dbase *adbase)
403 {
404     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
491 #ifdef AFS_PTHREAD_ENV
492             DBHOLD(ubik_dbase);
493 #endif
494
495             for (ts = ubik_servers, doingRPC = 0; ts; ts = ts->next) {
496                 if (!ts->up) {
497                     doingRPC = 1;
498                     code = DoProbe(ts);
499                     if (code == 0) {
500                         ts->up = 1;
501                         urecovery_state &= ~UBIK_RECFOUNDDB;
502                     }
503                 } else if (!ts->currentDB) {
504                     urecovery_state &= ~UBIK_RECFOUNDDB;
505                 }
506             }
507
508 #ifdef AFS_PTHREAD_ENV
509             DBRELE(ubik_dbase);
510 #endif
511
512             if (doingRPC)
513                 now = FT_ApproxTime();
514             lastProbeTime = now;
515         }
516
517         /* Mark whether we are the sync site */
518         if (!ubeacon_AmSyncSite()) {
519             urecovery_state &= ~UBIK_RECSYNCSITE;
520             continue;           /* nothing to do */
521         }
522         urecovery_state |= UBIK_RECSYNCSITE;
523
524         /* If a server has just come up or if we have not found the
525          * most current database, then go find the most current db.
526          */
527         if (!(urecovery_state & UBIK_RECFOUNDDB)) {
528             bestServer = (struct ubik_server *)0;
529             bestDBVersion.epoch = 0;
530             bestDBVersion.counter = 0;
531             for (ts = ubik_servers; ts; ts = ts->next) {
532                 if (!ts->up)
533                     continue;   /* don't bother with these guys */
534                 if (ts->isClone)
535                     continue;
536                 code = DISK_GetVersion(ts->disk_rxcid, &ts->version);
537                 if (code == 0) {
538                     /* perhaps this is the best version */
539                     if (vcmp(ts->version, bestDBVersion) > 0) {
540                         /* new best version */
541                         bestDBVersion = ts->version;
542                         bestServer = ts;
543                     }
544                 }
545             }
546             /* take into consideration our version. Remember if we,
547              * the sync site, have the best version. Also note that
548              * we may need to send the best version out.
549              */
550             if (vcmp(ubik_dbase->version, bestDBVersion) >= 0) {
551                 bestDBVersion = ubik_dbase->version;
552                 bestServer = (struct ubik_server *)0;
553                 urecovery_state |= UBIK_RECHAVEDB;
554             } else {
555                 /* Clear the flag only when we know we have to retrieve
556                  * the db. Because urecovery_AllBetter() looks at it.
557                  */
558                 urecovery_state &= ~UBIK_RECHAVEDB;
559             }
560             lastDBVCheck = FT_ApproxTime();
561             urecovery_state |= UBIK_RECFOUNDDB;
562             urecovery_state &= ~UBIK_RECSENTDB;
563         }
564 #if defined(UBIK_PAUSE)
565         /* it's not possible for UBIK_RECFOUNDDB not to be set here.
566          * However, we might have lost UBIK_RECSYNCSITE, and that
567          * IS important.
568          */
569         if (!(urecovery_state & UBIK_RECSYNCSITE))
570             continue;           /* lost sync */
571 #else
572         if (!(urecovery_state & UBIK_RECFOUNDDB))
573             continue;           /* not ready */
574 #endif /* UBIK_PAUSE */
575
576         /* If we, the sync site, do not have the best db version, then
577          * go and get it from the server that does.
578          */
579         if ((urecovery_state & UBIK_RECHAVEDB) || !bestServer) {
580             urecovery_state |= UBIK_RECHAVEDB;
581         } else {
582             /* we don't have the best version; we should fetch it. */
583             DBHOLD(ubik_dbase);
584             urecovery_AbortAll(ubik_dbase);
585
586             /* Rx code to do the Bulk fetch */
587             file = 0;
588             offset = 0;
589             rxcall = rx_NewCall(bestServer->disk_rxcid);
590
591             ubik_print("Ubik: Synchronize database with server %s\n",
592                        afs_inet_ntoa_r(bestServer->addr[0], hoststr));
593
594             code = StartDISK_GetFile(rxcall, file);
595             if (code) {
596                 ubik_dprint("StartDiskGetFile failed=%d\n", code);
597                 goto FetchEndCall;
598             }
599             nbytes = rx_Read(rxcall, (char *)&length, sizeof(afs_int32));
600             length = ntohl(length);
601             if (nbytes != sizeof(afs_int32)) {
602                 ubik_dprint("Rx-read length error=%d\n", code = BULK_ERROR);
603                 code = EIO;
604                 goto FetchEndCall;
605             }
606
607 #ifdef OLD_URECOVERY
608             /* Truncate the file first */
609             code = (*ubik_dbase->truncate) (ubik_dbase, file, 0);
610             if (code) {
611                 ubik_dprint("truncate io error=%d\n", code);
612                 goto FetchEndCall;
613             }
614             tversion.counter = 0;
615 #endif
616             /* give invalid label during file transit */
617             tversion.epoch = 0;
618             code = (*ubik_dbase->setlabel) (ubik_dbase, file, &tversion);
619             if (code) {
620                 ubik_dprint("setlabel io error=%d\n", code);
621                 goto FetchEndCall;
622             }
623 #ifndef OLD_URECOVERY
624             flen = length;
625             afs_snprintf(pbuffer, sizeof(pbuffer), "%s.DB%s%d.TMP", ubik_dbase->pathName, (file<0)?"SYS":"", (file<0)?-file:file);
626             fd = open(pbuffer, O_CREAT | O_RDWR | O_TRUNC, 0600);
627             if (fd < 0) {
628                 code = errno;
629                 goto FetchEndCall;
630             }
631             code = lseek(fd, HDRSIZE, 0);
632             if (code != HDRSIZE) {
633                 close(fd);
634                 goto FetchEndCall;
635             }
636 #endif
637
638             pass = 0;
639             while (length > 0) {
640                 tlen = (length > sizeof(tbuffer) ? sizeof(tbuffer) : length);
641 #ifndef AFS_PTHREAD_ENV
642                 if (pass % 4 == 0)
643                     IOMGR_Poll();
644 #endif
645                 nbytes = rx_Read(rxcall, tbuffer, tlen);
646                 if (nbytes != tlen) {
647                     ubik_dprint("Rx-read bulk error=%d\n", code = BULK_ERROR);
648                     code = EIO;
649                     close(fd);
650                     goto FetchEndCall;
651                 }
652 #ifdef OLD_URECOVERY
653                 nbytes =
654                     (*ubik_dbase->write) (ubik_dbase, file, tbuffer, offset,
655                                           tlen);
656 #else
657                 nbytes = write(fd, tbuffer, tlen);
658                 pass++;
659 #endif
660                 if (nbytes != tlen) {
661                     code = UIOERROR;
662                     close(fd);
663                     goto FetchEndCall;
664                 }
665                 offset += tlen;
666                 length -= tlen;
667             }
668 #ifndef OLD_URECOVERY
669             code = close(fd);
670             if (code)
671                 goto FetchEndCall;
672 #endif
673             code = EndDISK_GetFile(rxcall, &tversion);
674           FetchEndCall:
675             tcode = rx_EndCall(rxcall, code);
676             if (!code)
677                 code = tcode;
678             if (!code) {
679                 /* we got a new file, set up its header */
680                 urecovery_state |= UBIK_RECHAVEDB;
681                 memcpy(&ubik_dbase->version, &tversion,
682                        sizeof(struct ubik_version));
683 #ifdef OLD_URECOVERY
684                 (*ubik_dbase->sync) (ubik_dbase, 0);    /* get data out first */
685 #else
686                 afs_snprintf(tbuffer, sizeof(tbuffer), "%s.DB%s%d", ubik_dbase->pathName, (file<0)?"SYS":"", (file<0)?-file:file);
687 #ifdef AFS_NT40_ENV
688                 afs_snprintf(pbuffer, sizeof(pbuffer), "%s.DB%s%d.OLD", ubik_dbase->pathName, (file<0)?"SYS":"", (file<0)?-file:file);
689                 code = unlink(pbuffer);
690                 if (!code)
691                     code = rename(tbuffer, pbuffer);
692                 afs_snprintf(pbuffer, sizeof(pbuffer), "%s.DB%s%d.TMP", ubik_dbase->pathName, (file<0)?"SYS":"", (file<0)?-file:file);
693 #endif
694                 if (!code)
695                     code = rename(pbuffer, tbuffer);
696                 if (!code) {
697                     (*ubik_dbase->open) (ubik_dbase, file);
698 #endif
699                     /* after data is good, sync disk with correct label */
700                     code =
701                         (*ubik_dbase->setlabel) (ubik_dbase, 0,
702                                                  &ubik_dbase->version);
703 #ifndef OLD_URECOVERY
704                 }
705 #ifdef AFS_NT40_ENV
706                 afs_snprintf(pbuffer, sizeof(pbuffer), "%s.DB%s%d.OLD", ubik_dbase->pathName, (file<0)?"SYS":"", (file<0)?-file:file);
707                 unlink(pbuffer);
708 #endif
709 #endif
710             }
711             if (code) {
712 #ifndef OLD_URECOVERY
713                 unlink(pbuffer);
714                 /*
715                  * We will effectively invalidate the old data forever now.
716                  * Unclear if we *should* but we do.
717                  */
718 #endif
719                 ubik_dbase->version.epoch = 0;
720                 ubik_dbase->version.counter = 0;
721                 ubik_print("Ubik: Synchronize database failed (error = %d)\n",
722                            code);
723             } else {
724                 ubik_print("Ubik: Synchronize database completed\n");
725                 urecovery_state |= UBIK_RECHAVEDB;
726             }
727             udisk_Invalidate(ubik_dbase, 0);    /* data has changed */
728 #ifdef AFS_PTHREAD_ENV
729             assert(pthread_cond_broadcast(&ubik_dbase->version_cond) == 0);
730 #else
731             LWP_NoYieldSignal(&ubik_dbase->version);
732 #endif
733             DBRELE(ubik_dbase);
734         }
735 #if defined(UBIK_PAUSE)
736         if (!(urecovery_state & UBIK_RECSYNCSITE))
737             continue;           /* lost sync */
738 #endif /* UBIK_PAUSE */
739         if (!(urecovery_state & UBIK_RECHAVEDB))
740             continue;           /* not ready */
741
742         /* If the database was newly initialized, then when we establish quorum, write
743          * a new label. This allows urecovery_AllBetter() to allow access for reads.
744          * Setting it to 2 also allows another site to come along with a newer
745          * database and overwrite this one.
746          */
747         if (ubik_dbase->version.epoch == 1) {
748             DBHOLD(ubik_dbase);
749             urecovery_AbortAll(ubik_dbase);
750             ubik_epochTime = 2;
751             ubik_dbase->version.epoch = ubik_epochTime;
752             ubik_dbase->version.counter = 1;
753             code =
754                 (*ubik_dbase->setlabel) (ubik_dbase, 0, &ubik_dbase->version);
755             udisk_Invalidate(ubik_dbase, 0);    /* data may have changed */
756 #ifdef AFS_PTHREAD_ENV
757             assert(pthread_cond_broadcast(&ubik_dbase->version_cond) == 0);
758 #else
759             LWP_NoYieldSignal(&ubik_dbase->version);
760 #endif
761             DBRELE(ubik_dbase);
762         }
763
764         /* Check the other sites and send the database to them if they
765          * do not have the current db.
766          */
767         if (!(urecovery_state & UBIK_RECSENTDB)) {
768             /* now propagate out new version to everyone else */
769             dbok = 1;           /* start off assuming they all worked */
770
771             DBHOLD(ubik_dbase);
772             /*
773              * Check if a write transaction is in progress. We can't send the
774              * db when a write is in progress here because the db would be
775              * obsolete as soon as it goes there. Also, ops after the begin
776              * trans would reach the recepient and wouldn't find a transaction
777              * pending there.  Frankly, I don't think it's possible to get past
778              * the write-lock above if there is a write transaction in progress,
779              * but then, it won't hurt to check, will it?
780              */
781             if (ubik_dbase->flags & DBWRITING) {
782                 struct timeval tv;
783                 int safety = 0;
784                 tv.tv_sec = 0;
785                 tv.tv_usec = 50000;
786                 while ((ubik_dbase->flags & DBWRITING) && (safety < 500)) {
787                     DBRELE(ubik_dbase);
788                     /* sleep for a little while */
789 #ifdef AFS_PTHREAD_ENV
790                     select(0, 0, 0, 0, &tv);
791 #else
792                     IOMGR_Select(0, 0, 0, 0, &tv);
793 #endif
794                     tv.tv_usec += 10000;
795                     safety++;
796                     DBHOLD(ubik_dbase);
797                 }
798             }
799
800             for (ts = ubik_servers; ts; ts = ts->next) {
801                 inAddr.s_addr = ts->addr[0];
802                 if (!ts->up) {
803                     ubik_dprint("recovery cannot send version to %s\n",
804                                 afs_inet_ntoa_r(inAddr.s_addr, hoststr));
805                     dbok = 0;
806                     continue;
807                 }
808                 ubik_dprint("recovery sending version to %s\n",
809                             afs_inet_ntoa_r(inAddr.s_addr, hoststr));
810                 if (vcmp(ts->version, ubik_dbase->version) != 0) {
811                     ubik_dprint("recovery stating local database\n");
812
813                     /* Rx code to do the Bulk Store */
814                     code = (*ubik_dbase->stat) (ubik_dbase, 0, &ubikstat);
815                     if (!code) {
816                         length = ubikstat.size;
817                         file = offset = 0;
818                         rxcall = rx_NewCall(ts->disk_rxcid);
819                         code =
820                             StartDISK_SendFile(rxcall, file, length,
821                                                &ubik_dbase->version);
822                         if (code) {
823                             ubik_dprint("StartDiskSendFile failed=%d\n",
824                                         code);
825                             goto StoreEndCall;
826                         }
827                         while (length > 0) {
828                             tlen =
829                                 (length >
830                                  sizeof(tbuffer) ? sizeof(tbuffer) : length);
831                             nbytes =
832                                 (*ubik_dbase->read) (ubik_dbase, file,
833                                                      tbuffer, offset, tlen);
834                             if (nbytes != tlen) {
835                                 ubik_dprint("Local disk read error=%d\n",
836                                             code = UIOERROR);
837                                 goto StoreEndCall;
838                             }
839                             nbytes = rx_Write(rxcall, tbuffer, tlen);
840                             if (nbytes != tlen) {
841                                 ubik_dprint("Rx-write bulk error=%d\n", code =
842                                             BULK_ERROR);
843                                 goto StoreEndCall;
844                             }
845                             offset += tlen;
846                             length -= tlen;
847                         }
848                         code = EndDISK_SendFile(rxcall);
849                       StoreEndCall:
850                         code = rx_EndCall(rxcall, code);
851                     }
852                     if (code == 0) {
853                         /* we set a new file, process its header */
854                         ts->version = ubik_dbase->version;
855                         ts->currentDB = 1;
856                     } else
857                         dbok = 0;
858                 } else {
859                     /* mark file up to date */
860                     ts->currentDB = 1;
861                 }
862             }
863             DBRELE(ubik_dbase);
864             if (dbok)
865                 urecovery_state |= UBIK_RECSENTDB;
866         }
867     }
868     return NULL;
869 }
870
871 /*!
872  * \brief send a Probe to all the network address of this server
873  *
874  * \return 0 if success, else return 1
875  */
876 int
877 DoProbe(struct ubik_server *server)
878 {
879     struct rx_connection *conns[UBIK_MAX_INTERFACE_ADDR];
880     struct rx_connection *connSuccess = 0;
881     int i, j, success_i = -1;
882     afs_uint32 addr;
883     char buffer[32];
884     char hoststr[16];
885     extern afs_int32 ubikSecIndex;
886     extern struct rx_securityClass *ubikSecClass;
887
888     for (i = 0; (addr = server->addr[i]) && (i < UBIK_MAX_INTERFACE_ADDR);
889          i++) {
890         conns[i] =
891             rx_NewConnection(addr, ubik_callPortal, DISK_SERVICE_ID,
892                              ubikSecClass, ubikSecIndex);
893
894         /* user requirement to use only the primary interface */
895         if (ubikPrimaryAddrOnly) {
896             i = 1;
897             break;
898         }
899     }
900     assert(i);                  /* at least one interface address for this server */
901
902 #ifdef AFS_PTHREAD_ENV
903     DBRELE(ubik_dbase);
904 #endif
905
906     multi_Rx(conns, i) {
907         multi_DISK_Probe();
908         if (!multi_error) {     /* first success */
909             success_i = multi_i;
910
911             multi_Abort;
912         }
913     } multi_End_Ignore;
914
915 #ifdef AFS_PTHREAD_ENV
916     DBHOLD(ubik_dbase);
917 #endif
918
919     if (success_i >= 0) {
920         addr = server->addr[success_i]; /* successful interface addr */
921
922         if (server->disk_rxcid) /* destroy existing conn */
923             rx_DestroyConnection(server->disk_rxcid);
924         if (server->vote_rxcid)
925             rx_DestroyConnection(server->vote_rxcid);
926
927         /* make new connections */
928         server->disk_rxcid = conns[success_i];
929         server->vote_rxcid = rx_NewConnection(addr, ubik_callPortal,
930                                               VOTE_SERVICE_ID, ubikSecClass,
931                                               ubikSecIndex);
932
933         connSuccess = conns[success_i];
934         strcpy(buffer, afs_inet_ntoa_r(server->addr[0], hoststr));
935
936         ubik_print("ubik:server %s is back up: will be contacted through %s\n",
937              buffer, afs_inet_ntoa_r(addr, hoststr));
938     }
939
940     /* Destroy all connections except the one on which we succeeded */
941     for (j = 0; j < i; j++)
942         if (conns[j] != connSuccess)
943             rx_DestroyConnection(conns[j]);
944
945     if (!connSuccess)
946         ubik_dprint("ubik:server %s still down\n",
947                     afs_inet_ntoa_r(server->addr[0], hoststr));
948
949     if (connSuccess)
950         return 0;               /* success */
951     else
952         return 1;               /* failure */
953 }