ubik-doxygen-20081121
[openafs.git] / src / ubik / disk.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 RCSID
14     ("$Header$");
15
16 #include <sys/types.h>
17 #ifdef AFS_NT40_ENV
18 #include <winsock2.h>
19 #else
20 #include <sys/file.h>
21 #include <netinet/in.h>
22 #endif
23 #include <errno.h>
24 #include <string.h>
25 #include <lock.h>
26 #include <rx/xdr.h>
27
28
29
30 #define UBIK_INTERNALS
31 #include "ubik.h"
32 #include "ubik_int.h"
33
34 #define PHSIZE  128
35 static struct buffer {
36     struct ubik_dbase *dbase;   /*!< dbase within which the buffer resides */
37     afs_int32 file;             /*!< Unique cache key */
38     afs_int32 page;             /*!< page number */
39     struct buffer *lru_next;
40     struct buffer *lru_prev;
41     struct buffer *hashNext;    /*!< next dude in hash table */
42     char *data;                 /*!< ptr to the data */
43     char lockers;               /*!< usage ref count */
44     char dirty;                 /*!< is buffer modified */
45     char hashIndex;             /*!< back ptr to hash table */
46 } *Buffers;
47
48 #define pHash(page) ((page) & (PHSIZE-1))
49
50 afs_int32 ubik_nBuffers = NBUFFERS;
51 static struct buffer *phTable[PHSIZE];  /*!< page hash table */
52 static struct buffer *LruBuffer;
53 static int nbuffers;
54 static int calls = 0, ios = 0, lastb = 0;
55 static char *BufferData;
56 static struct buffer *newslot(struct ubik_dbase *adbase, afs_int32 afid,
57                               afs_int32 apage);
58 static initd = 0;
59 #define BADFID      0xffffffff
60
61 static DTrunc(struct ubik_dbase *dbase, afs_int32 fid, afs_int32 length);
62
63 static struct ubik_trunc *freeTruncList = 0;
64
65 /*!
66  * \brief Remove a transaction from the database's active transaction list.  Don't free it.
67  */
68 static int
69 unthread(struct ubik_trans *atrans)
70 {
71     struct ubik_trans **lt, *tt;
72     lt = &atrans->dbase->activeTrans;
73     for (tt = *lt; tt; lt = &tt->next, tt = *lt) {
74         if (tt == atrans) {
75             /* found it */
76             *lt = tt->next;
77             return 0;
78         }
79     }
80     return 2;                   /* no entry */
81 }
82
83 /*!
84  * \brief some debugging assistance
85  */
86 int
87 udisk_Debug(struct ubik_debug *aparm)
88 {
89     struct buffer *tb;
90     int i;
91
92     memcpy(&aparm->localVersion, &ubik_dbase->version,
93            sizeof(struct ubik_version));
94     aparm->lockedPages = 0;
95     aparm->writeLockedPages = 0;
96     tb = Buffers;
97     for (i = 0; i < nbuffers; i++, tb++) {
98         if (tb->lockers) {
99             aparm->lockedPages++;
100             if (tb->dirty)
101                 aparm->writeLockedPages++;
102         }
103     }
104     return 0;
105 }
106
107 /*!
108  * \brief Write an opcode to the log.
109  *
110  * log format is defined here, and implicitly in recovery.c
111  *
112  * 4 byte opcode, followed by parameters, each 4 bytes long.  All integers
113  * are in logged in network standard byte order, in case we want to move logs
114  * from machine-to-machine someday.
115  *
116  * Begin transaction: opcode \n
117  * Commit transaction: opcode, version (8 bytes) \n
118  * Truncate file: opcode, file number, length \n
119  * Abort transaction: opcode \n
120  * Write data: opcode, file, position, length, <length> data bytes \n
121  */
122 int
123 udisk_LogOpcode(struct ubik_dbase *adbase, afs_int32 aopcode, int async)
124 {
125     struct ubik_stat ustat;
126     afs_int32 code;
127
128     /* figure out where to write */
129     code = (*adbase->stat) (adbase, LOGFILE, &ustat);
130     if (code < 0)
131         return code;
132
133     /* setup data and do write */
134     aopcode = htonl(aopcode);
135     code =
136         (*adbase->write) (adbase, LOGFILE, (char *)&aopcode, ustat.size,
137                           sizeof(afs_int32));
138     if (code != sizeof(afs_int32))
139         return UIOERROR;
140
141     /* optionally sync data */
142     if (async)
143         code = (*adbase->sync) (adbase, LOGFILE);
144     else
145         code = 0;
146     return code;
147 }
148
149 /*!
150  * \brief Log a commit, never syncing.
151  */
152 int
153 udisk_LogEnd(struct ubik_dbase *adbase, struct ubik_version *aversion)
154 {
155     afs_int32 code;
156     afs_int32 data[3];
157     struct ubik_stat ustat;
158
159     /* figure out where to write */
160     code = (*adbase->stat) (adbase, LOGFILE, &ustat);
161     if (code)
162         return code;
163
164     /* setup data */
165     data[0] = htonl(LOGEND);
166     data[1] = htonl(aversion->epoch);
167     data[2] = htonl(aversion->counter);
168
169     /* do write */
170     code =
171         (*adbase->write) (adbase, LOGFILE, (char *)data, ustat.size,
172                           3 * sizeof(afs_int32));
173     if (code != 3 * sizeof(afs_int32))
174         return UIOERROR;
175
176     /* finally sync the log */
177     code = (*adbase->sync) (adbase, LOGFILE);
178     return code;
179 }
180
181 /*!
182  * \brief Log a truncate operation, never syncing.
183  */
184 int
185 udisk_LogTruncate(struct ubik_dbase *adbase, afs_int32 afile,
186                   afs_int32 alength)
187 {
188     afs_int32 code;
189     afs_int32 data[3];
190     struct ubik_stat ustat;
191
192     /* figure out where to write */
193     code = (*adbase->stat) (adbase, LOGFILE, &ustat);
194     if (code < 0)
195         return code;
196
197     /* setup data */
198     data[0] = htonl(LOGTRUNCATE);
199     data[1] = htonl(afile);
200     data[2] = htonl(alength);
201
202     /* do write */
203     code =
204         (*adbase->write) (adbase, LOGFILE, (char *)data, ustat.size,
205                           3 * sizeof(afs_int32));
206     if (code != 3 * sizeof(afs_int32))
207         return UIOERROR;
208     return 0;
209 }
210
211 /*!
212  * \brief Write some data to the log, never syncing.
213  */
214 int
215 udisk_LogWriteData(struct ubik_dbase *adbase, afs_int32 afile, char *abuffer,
216                    afs_int32 apos, afs_int32 alen)
217 {
218     struct ubik_stat ustat;
219     afs_int32 code;
220     afs_int32 data[4];
221     afs_int32 lpos;
222
223     /* find end of log */
224     code = (*adbase->stat) (adbase, LOGFILE, &ustat);
225     lpos = ustat.size;
226     if (code < 0)
227         return code;
228
229     /* setup header */
230     data[0] = htonl(LOGDATA);
231     data[1] = htonl(afile);
232     data[2] = htonl(apos);
233     data[3] = htonl(alen);
234
235     /* write header */
236     code =
237         (*adbase->write) (adbase, LOGFILE, (char *)data, lpos, 4 * sizeof(afs_int32));
238     if (code != 4 * sizeof(afs_int32))
239         return UIOERROR;
240     lpos += 4 * sizeof(afs_int32);
241
242     /* write data */
243     code = (*adbase->write) (adbase, LOGFILE, abuffer, lpos, alen);
244     if (code != alen)
245         return UIOERROR;
246     return 0;
247 }
248
249 static int
250 DInit(int abuffers)
251 {
252     /* Initialize the venus buffer system. */
253     int i;
254     struct buffer *tb;
255     Buffers = (struct buffer *)malloc(abuffers * sizeof(struct buffer));
256     memset(Buffers, 0, abuffers * sizeof(struct buffer));
257     BufferData = (char *)malloc(abuffers * UBIK_PAGESIZE);
258     nbuffers = abuffers;
259     for (i = 0; i < PHSIZE; i++)
260         phTable[i] = 0;
261     for (i = 0; i < abuffers; i++) {
262         /* Fill in each buffer with an empty indication. */
263         tb = &Buffers[i];
264         tb->lru_next = &(Buffers[i + 1]);
265         tb->lru_prev = &(Buffers[i - 1]);
266         tb->data = &BufferData[UBIK_PAGESIZE * i];
267         tb->file = BADFID;
268     }
269     Buffers[0].lru_prev = &(Buffers[abuffers - 1]);
270     Buffers[abuffers - 1].lru_next = &(Buffers[0]);
271     LruBuffer = &(Buffers[0]);
272     return 0;
273 }
274
275 /*!
276  * \brief Take a buffer and mark it as the least recently used buffer.
277  */
278 static void
279 Dlru(struct buffer *abuf)
280 {
281     if (LruBuffer == abuf)
282         return;
283
284     /* Unthread from where it is in the list */
285     abuf->lru_next->lru_prev = abuf->lru_prev;
286     abuf->lru_prev->lru_next = abuf->lru_next;
287
288     /* Thread onto beginning of LRU list */
289     abuf->lru_next = LruBuffer;
290     abuf->lru_prev = LruBuffer->lru_prev;
291
292     LruBuffer->lru_prev->lru_next = abuf;
293     LruBuffer->lru_prev = abuf;
294     LruBuffer = abuf;
295 }
296
297 /*!
298  * \brief Take a buffer and mark it as the most recently used buffer.
299  */
300 static void
301 Dmru(struct buffer *abuf)
302 {
303     if (LruBuffer == abuf) {
304         LruBuffer = LruBuffer->lru_next;
305         return;
306     }
307
308     /* Unthread from where it is in the list */
309     abuf->lru_next->lru_prev = abuf->lru_prev;
310     abuf->lru_prev->lru_next = abuf->lru_next;
311
312     /* Thread onto end of LRU list - making it the MRU buffer */
313     abuf->lru_next = LruBuffer;
314     abuf->lru_prev = LruBuffer->lru_prev;
315     LruBuffer->lru_prev->lru_next = abuf;
316     LruBuffer->lru_prev = abuf;
317 }
318
319 /*!
320  * \brief Get a pointer to a particular buffer.
321  */
322 static char *
323 DRead(struct ubik_dbase *dbase, afs_int32 fid, int page)
324 {
325     /* Read a page from the disk. */
326     struct buffer *tb, *lastbuffer;
327     afs_int32 code;
328
329     calls++;
330     lastbuffer = LruBuffer->lru_prev;
331
332     if ((lastbuffer->page == page) && (lastbuffer->file == fid)
333         && (lastbuffer->dbase == dbase)) {
334         tb = lastbuffer;
335         tb->lockers++;
336         lastb++;
337         return tb->data;
338     }
339     for (tb = phTable[pHash(page)]; tb; tb = tb->hashNext) {
340         if (tb->page == page && tb->file == fid && tb->dbase == dbase) {
341             Dmru(tb);
342             tb->lockers++;
343             return tb->data;
344         }
345     }
346     /* can't find it */
347     tb = newslot(dbase, fid, page);
348     if (!tb)
349         return 0;
350     memset(tb->data, 0, UBIK_PAGESIZE);
351
352     tb->lockers++;
353     code =
354         (*dbase->read) (dbase, fid, tb->data, page * UBIK_PAGESIZE,
355                         UBIK_PAGESIZE);
356     if (code < 0) {
357         tb->file = BADFID;
358         Dlru(tb);
359         tb->lockers--;
360         ubik_print("Ubik: Error reading database file: errno=%d\n", errno);
361         return 0;
362     }
363     ios++;
364
365     /* Note that findslot sets the page field in the buffer equal to
366      * what it is searching for.
367      */
368     return tb->data;
369 }
370
371 /*!
372  * \brief Zap truncated pages.
373  */
374 static int
375 DTrunc(struct ubik_dbase *dbase, afs_int32 fid, afs_int32 length)
376 {
377     afs_int32 maxPage;
378     struct buffer *tb;
379     int i;
380
381     maxPage = (length + UBIK_PAGESIZE - 1) >> UBIK_LOGPAGESIZE; /* first invalid page now in file */
382     for (i = 0, tb = Buffers; i < nbuffers; i++, tb++) {
383         if (tb->page >= maxPage && tb->file == fid && tb->dbase == dbase) {
384             tb->file = BADFID;
385             Dlru(tb);
386         }
387     }
388     return 0;
389 }
390
391 /*!
392  * \brief Allocate a truncation entry.
393  *
394  * We allocate special entries representing truncations, rather than
395  * performing them immediately, so that we can abort a transaction easily by simply purging
396  * the in-core memory buffers and discarding these truncation entries.
397  */
398 static struct ubik_trunc *
399 GetTrunc(void)
400 {
401     struct ubik_trunc *tt;
402     if (!freeTruncList) {
403         freeTruncList =
404             (struct ubik_trunc *)malloc(sizeof(struct ubik_trunc));
405         freeTruncList->next = (struct ubik_trunc *)0;
406     }
407     tt = freeTruncList;
408     freeTruncList = tt->next;
409     return tt;
410 }
411
412 /*!
413  * \brief Free a truncation entry.
414  */
415 static int
416 PutTrunc(struct ubik_trunc *at)
417 {
418     at->next = freeTruncList;
419     freeTruncList = at;
420     return 0;
421 }
422
423 /*!
424  * \brief Find a truncation entry for a file, if any.
425  */
426 static struct ubik_trunc *
427 FindTrunc(struct ubik_trans *atrans, afs_int32 afile)
428 {
429     struct ubik_trunc *tt;
430     for (tt = atrans->activeTruncs; tt; tt = tt->next) {
431         if (tt->file == afile)
432             return tt;
433     }
434     return (struct ubik_trunc *)0;
435 }
436
437 /*!
438  * \brief Do truncates associated with \p atrans, and free them.
439  */
440 static int
441 DoTruncs(struct ubik_trans *atrans)
442 {
443     struct ubik_trunc *tt, *nt;
444     int (*tproc) ();
445     afs_int32 rcode = 0, code;
446
447     tproc = atrans->dbase->truncate;
448     for (tt = atrans->activeTruncs; tt; tt = nt) {
449         nt = tt->next;
450         DTrunc(atrans->dbase, tt->file, tt->length);    /* zap pages from buffer cache */
451         code = (*tproc) (atrans->dbase, tt->file, tt->length);
452         if (code)
453             rcode = code;
454         PutTrunc(tt);
455     }
456     /* don't unthread, because we do the entire list's worth here */
457     atrans->activeTruncs = (struct ubik_trunc *)0;
458     return (rcode);
459 }
460
461 /*!
462  * \brief Mark an \p fid as invalid.
463  */
464 int
465 udisk_Invalidate(struct ubik_dbase *adbase, afs_int32 afid)
466 {
467     struct buffer *tb;
468     int i;
469
470     for (i = 0, tb = Buffers; i < nbuffers; i++, tb++) {
471         if (tb->file == afid) {
472             tb->file = BADFID;
473             Dlru(tb);
474         }
475     }
476     return 0;
477 }
478
479 /*!
480  * \brief Move this page into the correct hash bucket.
481  */
482 static int
483 FixupBucket(struct buffer *ap)
484 {
485     struct buffer **lp, *tp;
486     int i;
487     /* first try to get it out of its current hash bucket, in which it might not be */
488     i = ap->hashIndex;
489     lp = &phTable[i];
490     for (tp = *lp; tp; tp = tp->hashNext) {
491         if (tp == ap) {
492             *lp = tp->hashNext;
493             break;
494         }
495         lp = &tp->hashNext;
496     }
497     /* now figure the new hash bucket */
498     i = pHash(ap->page);
499     ap->hashIndex = i;          /* remember where we are for deletion */
500     ap->hashNext = phTable[i];  /* add us to the list */
501     phTable[i] = ap;
502     return 0;
503 }
504
505 /*!
506  * \brief Create a new slot for a particular dbase page.
507  */
508 static struct buffer *
509 newslot(struct ubik_dbase *adbase, afs_int32 afid, afs_int32 apage)
510 {
511     /* Find a usable buffer slot */
512     afs_int32 i;
513     struct buffer *pp, *tp;
514
515     pp = 0;                     /* last pure */
516     for (i = 0, tp = LruBuffer; i < nbuffers; i++, tp = tp->lru_next) {
517         if (!tp->lockers && !tp->dirty) {
518             pp = tp;
519             break;
520         }
521     }
522
523     if (pp == 0) {
524         /* There are no unlocked buffers that don't need to be written to the disk. */
525         ubik_print
526             ("Ubik: Internal Error: Unable to find free buffer in ubik cache\n");
527         return NULL;
528     }
529
530     /* Now fill in the header. */
531     pp->dbase = adbase;
532     pp->file = afid;
533     pp->page = apage;
534
535     FixupBucket(pp);            /* move to the right hash bucket */
536     Dmru(pp);
537     return pp;
538 }
539
540 /*!
541  * \brief Release a buffer, specifying whether or not the buffer has been modified by the locker.
542  */
543 static void
544 DRelease(char *ap, int flag)
545 {
546     int index;
547     struct buffer *bp;
548
549     if (!ap)
550         return;
551     index = (int)(ap - (char *)BufferData) >> UBIK_LOGPAGESIZE;
552     bp = &(Buffers[index]);
553     bp->lockers--;
554     if (flag)
555         bp->dirty = 1;
556     return;
557 }
558
559 /*!
560  * \brief Flush all modified buffers, leaves dirty bits set (they're cleared
561  * by DSync()).  
562  *
563  * \note Note interaction with DSync(): you call this thing first,
564  * writing the buffers to the disk.  Then you call DSync() to sync all the
565  * files that were written, and to clear the dirty bits.  You should
566  * always call DFlush/DSync as a pair.
567  */
568 static int
569 DFlush(struct ubik_dbase *adbase)
570 {
571     int i;
572     afs_int32 code;
573     struct buffer *tb;
574
575     tb = Buffers;
576     for (i = 0; i < nbuffers; i++, tb++) {
577         if (tb->dirty) {
578             code = tb->page * UBIK_PAGESIZE;    /* offset within file */
579             code =
580                 (*adbase->write) (adbase, tb->file, tb->data, code,
581                                   UBIK_PAGESIZE);
582             if (code != UBIK_PAGESIZE)
583                 return UIOERROR;
584         }
585     }
586     return 0;
587 }
588
589 /*!
590  * \brief Flush all modified buffers.
591  */
592 static int
593 DAbort(struct ubik_dbase *adbase)
594 {
595     int i;
596     struct buffer *tb;
597
598     tb = Buffers;
599     for (i = 0; i < nbuffers; i++, tb++) {
600         if (tb->dirty) {
601             tb->dirty = 0;
602             tb->file = BADFID;
603             Dlru(tb);
604         }
605     }
606     return 0;
607 }
608
609 /*!
610  * \attention DSync() must only be called after DFlush(), due to its interpretation of dirty flag.
611  */
612 static int
613 DSync(struct ubik_dbase *adbase)
614 {
615     int i;
616     afs_int32 code;
617     struct buffer *tb;
618     afs_int32 file;
619     afs_int32 rCode;
620
621     rCode = 0;
622     while (1) {
623         file = BADFID;
624         for (i = 0, tb = Buffers; i < nbuffers; i++, tb++) {
625             if (tb->dirty == 1) {
626                 if (file == BADFID)
627                     file = tb->file;
628                 if (file != BADFID && tb->file == file)
629                     tb->dirty = 0;
630             }
631         }
632         if (file == BADFID)
633             break;
634         /* otherwise we have a file to sync */
635         code = (*adbase->sync) (adbase, file);
636         if (code)
637             rCode = code;
638     }
639     return rCode;
640 }
641
642 /*!
643  * \brief Same as DRead(), only do not even try to read the page.
644  */
645 static char *
646 DNew(struct ubik_dbase *dbase, afs_int32 fid, int page)
647 {
648     struct buffer *tb;
649
650     if ((tb = newslot(dbase, fid, page)) == 0)
651         return NULL;
652     tb->lockers++;
653     memset(tb->data, 0, UBIK_PAGESIZE);
654     return tb->data;
655 }
656
657 /*!
658  * \brief Read data from database.
659  */
660 int
661 udisk_read(struct ubik_trans *atrans, afs_int32 afile, char *abuffer,
662            afs_int32 apos, afs_int32 alen)
663 {
664     char *bp;
665     afs_int32 offset, len, totalLen;
666     struct ubik_dbase *dbase;
667
668     if (atrans->flags & TRDONE)
669         return UDONE;
670     totalLen = 0;
671     dbase = atrans->dbase;
672     while (alen > 0) {
673         bp = DRead(dbase, afile, apos >> UBIK_LOGPAGESIZE);
674         if (!bp)
675             return UEOF;
676         /* otherwise, min of remaining bytes and end of buffer to user mode */
677         offset = apos & (UBIK_PAGESIZE - 1);
678         len = UBIK_PAGESIZE - offset;
679         if (len > alen)
680             len = alen;
681         memcpy(abuffer, bp + offset, len);
682         abuffer += len;
683         apos += len;
684         alen -= len;
685         totalLen += len;
686         DRelease(bp, 0);
687     }
688     return 0;
689 }
690
691 /*!
692  * \brief Truncate file.
693  */
694 int
695 udisk_truncate(struct ubik_trans *atrans, afs_int32 afile, afs_int32 alength)
696 {
697     afs_int32 code;
698     struct ubik_trunc *tt;
699
700     if (atrans->flags & TRDONE)
701         return UDONE;
702     if (atrans->type != UBIK_WRITETRANS)
703         return UBADTYPE;
704
705     /* write a truncate log record */
706     code = udisk_LogTruncate(atrans->dbase, afile, alength);
707
708     /* don't truncate until commit time */
709     tt = FindTrunc(atrans, afile);
710     if (!tt) {
711         /* this file not truncated yet */
712         tt = GetTrunc();
713         tt->next = atrans->activeTruncs;
714         atrans->activeTruncs = tt;
715         tt->file = afile;
716         tt->length = alength;
717     } else {
718         /* already truncated to a certain length */
719         if (tt->length > alength)
720             tt->length = alength;
721     }
722     return code;
723 }
724
725 /*!
726  * \brief Write data to database, using logs.
727  */
728 int
729 udisk_write(struct ubik_trans *atrans, afs_int32 afile, char *abuffer,
730             afs_int32 apos, afs_int32 alen)
731 {
732     char *bp;
733     afs_int32 offset, len, totalLen;
734     struct ubik_dbase *dbase;
735     struct ubik_trunc *tt;
736     afs_int32 code;
737
738     if (atrans->flags & TRDONE)
739         return UDONE;
740     if (atrans->type != UBIK_WRITETRANS)
741         return UBADTYPE;
742
743     dbase = atrans->dbase;
744     /* first write the data to the log */
745     code = udisk_LogWriteData(dbase, afile, abuffer, apos, alen);
746     if (code)
747         return code;
748
749     /* expand any truncations of this file */
750     tt = FindTrunc(atrans, afile);
751     if (tt) {
752         if (tt->length < apos + alen) {
753             tt->length = apos + alen;
754         }
755     }
756
757     /* now update vm */
758     totalLen = 0;
759     while (alen > 0) {
760         bp = DRead(dbase, afile, apos >> UBIK_LOGPAGESIZE);
761         if (!bp) {
762             bp = DNew(dbase, afile, apos >> UBIK_LOGPAGESIZE);
763             if (!bp)
764                 return UIOERROR;
765             memset(bp, 0, UBIK_PAGESIZE);
766         }
767         /* otherwise, min of remaining bytes and end of buffer to user mode */
768         offset = apos & (UBIK_PAGESIZE - 1);
769         len = UBIK_PAGESIZE - offset;
770         if (len > alen)
771             len = alen;
772         memcpy(bp + offset, abuffer, len);
773         abuffer += len;
774         apos += len;
775         alen -= len;
776         totalLen += len;
777         DRelease(bp, 1);        /* buffer modified */
778     }
779     return 0;
780 }
781
782 /*!
783  * \brief Begin a new local transaction.
784  */
785 int
786 udisk_begin(struct ubik_dbase *adbase, int atype, struct ubik_trans **atrans)
787 {
788     afs_int32 code;
789     struct ubik_trans *tt;
790
791     *atrans = (struct ubik_trans *)NULL;
792     /* Make sure system is initialized before doing anything */
793     if (!initd) {
794         initd = 1;
795         DInit(ubik_nBuffers);
796     }
797     if (atype == UBIK_WRITETRANS) {
798         if (adbase->flags & DBWRITING)
799             return USYNC;
800         code = udisk_LogOpcode(adbase, LOGNEW, 0);
801         if (code)
802             return code;
803     }
804     tt = (struct ubik_trans *)malloc(sizeof(struct ubik_trans));
805     memset(tt, 0, sizeof(struct ubik_trans));
806     tt->dbase = adbase;
807     tt->next = adbase->activeTrans;
808     adbase->activeTrans = tt;
809     tt->type = atype;
810     if (atype == UBIK_READTRANS)
811         adbase->readers++;
812     else if (atype == UBIK_WRITETRANS)
813         adbase->flags |= DBWRITING;
814     *atrans = tt;
815     return 0;
816 }
817
818 /*!
819  * \brief Commit transaction.
820  */
821 int
822 udisk_commit(struct ubik_trans *atrans)
823 {
824     struct ubik_dbase *dbase;
825     afs_int32 code = 0;
826     struct ubik_version oldversion, newversion;
827
828     if (atrans->flags & TRDONE)
829         return (UTWOENDS);
830
831     if (atrans->type == UBIK_WRITETRANS) {
832         dbase = atrans->dbase;
833
834         /* On the first write to the database. We update the versions */
835         if (ubeacon_AmSyncSite() && !(urecovery_state & UBIK_RECLABELDB)) {
836             oldversion = dbase->version;
837             newversion.epoch = FT_ApproxTime();;
838             newversion.counter = 1;
839
840             code = (*dbase->setlabel) (dbase, 0, &newversion);
841             if (code)
842                 return (code);
843             ubik_epochTime = newversion.epoch;
844             dbase->version = newversion;
845
846             /* Ignore the error here. If the call fails, the site is
847              * marked down and when we detect it is up again, we will 
848              * send the entire database to it.
849              */
850             ContactQuorum(DISK_SetVersion, atrans, 1 /*CStampVersion */ ,
851                           &oldversion, &newversion);
852             urecovery_state |= UBIK_RECLABELDB;
853         }
854
855         dbase->version.counter++;       /* bump commit count */
856 #ifdef AFS_PTHREAD_ENV
857         assert(pthread_cond_broadcast(&dbase->version_cond) == 0);
858 #else
859         LWP_NoYieldSignal(&dbase->version);
860 #endif
861         code = udisk_LogEnd(dbase, &dbase->version);
862         if (code) {
863             dbase->version.counter--;
864             return (code);
865         }
866
867         /* If we fail anytime after this, then panic and let the
868          * recovery replay the log. 
869          */
870         code = DFlush(dbase);   /* write dirty pages to respective files */
871         if (code)
872             panic("Writing Ubik DB modifications\n");
873         code = DSync(dbase);    /* sync the files and mark pages not dirty */
874         if (code)
875             panic("Synchronizing Ubik DB modifications\n");
876
877         code = DoTruncs(atrans);        /* Perform requested truncations */
878         if (code)
879             panic("Truncating Ubik DB\n");
880
881         /* label the committed dbase */
882         code = (*dbase->setlabel) (dbase, 0, &dbase->version);
883         if (code)
884             panic("Truncating Ubik DB\n");
885
886         code = (*dbase->truncate) (dbase, LOGFILE, 0);  /* discard log (optional) */
887         if (code)
888             panic("Truncating Ubik logfile\n");
889
890     }
891
892     /* When the transaction is marked done, it also means the logfile
893      * has been truncated.
894      */
895     atrans->flags |= TRDONE;
896     return code;
897 }
898
899 /*!
900  * \brief Abort transaction.
901  */
902 int
903 udisk_abort(struct ubik_trans *atrans)
904 {
905     struct ubik_dbase *dbase;
906     afs_int32 code;
907
908     if (atrans->flags & TRDONE)
909         return UTWOENDS;
910
911     /* Check if we are the write trans before logging abort, lest we
912      * abort a good write trans in progress. 
913      * We don't really care if the LOGABORT gets to the log because we 
914      * truncate the log next. If the truncate fails, we panic; for 
915      * otherwise, the log entries remain. On restart, replay of the log
916      * will do nothing because the abort is there or no LogEnd opcode.
917      */
918     dbase = atrans->dbase;
919     if (atrans->type == UBIK_WRITETRANS && dbase->flags & DBWRITING) {
920         udisk_LogOpcode(dbase, LOGABORT, 1);
921         code = (*dbase->truncate) (dbase, LOGFILE, 0);
922         if (code)
923             panic("Truncating Ubik logfile during an abort\n");
924         DAbort(dbase);          /* remove all dirty pages */
925     }
926
927     /* When the transaction is marked done, it also means the logfile
928      * has been truncated.
929      */
930     atrans->flags |= (TRABORT | TRDONE);
931     return 0;
932 }
933
934 /*!
935  * \brief Destroy a transaction after it has been committed or aborted.
936  *
937  * If it hasn't committed before you call this routine, we'll abort the
938  * transaction for you.
939  */
940 int
941 udisk_end(struct ubik_trans *atrans)
942 {
943     struct ubik_dbase *dbase;
944
945 #if defined(UBIK_PAUSE)
946     /* Another thread is trying to lock this transaction.
947      * That can only be an RPC doing SDISK_Lock.
948      * Unlock the transaction, 'cause otherwise the other
949      * thread will never wake up.  Don't free it because
950      * the caller will do that already.
951      */
952     if (atrans->flags & TRSETLOCK) {
953         atrans->flags |= TRSTALE;
954         ulock_relLock(atrans);
955         return;
956     }
957 #endif /* UBIK_PAUSE */
958     if (!(atrans->flags & TRDONE))
959         udisk_abort(atrans);
960     dbase = atrans->dbase;
961
962     ulock_relLock(atrans);
963     unthread(atrans);
964
965     /* check if we are the write trans before unsetting the DBWRITING bit, else
966      * we could be unsetting someone else's bit.
967      */
968     if (atrans->type == UBIK_WRITETRANS && dbase->flags & DBWRITING) {
969         dbase->flags &= ~DBWRITING;
970     } else {
971         dbase->readers--;
972     }
973     if (atrans->iovec_info.iovec_wrt_val)
974         free(atrans->iovec_info.iovec_wrt_val);
975     if (atrans->iovec_data.iovec_buf_val)
976         free(atrans->iovec_data.iovec_buf_val);
977     free(atrans);
978
979     /* Wakeup any writers waiting in BeginTrans() */
980 #ifdef AFS_PTHREAD_ENV
981         assert(pthread_cond_broadcast(&dbase->flags_cond) == 0);
982 #else
983     LWP_NoYieldSignal(&dbase->flags);
984 #endif
985     return 0;
986 }