]> mj.ucw.cz Git - libucw.git/blob - lib/fb-atomic.c
Merge with git+ssh://cvs.ucw.cz/projects/sherlock/GIT/sherlock.git
[libucw.git] / lib / fb-atomic.c
1 /*
2  *      UCW Library -- Atomic Buffered Write to Files
3  *
4  *      (c) 2006 Martin Mares <mj@ucw.cz>
5  *
6  *      This software may be freely distributed and used according to the terms
7  *      of the GNU Lesser General Public License.
8  */
9
10 /*
11  *      This fastbuf backend is intended for cases where several threads
12  *      of a single program append records to a single file and while the
13  *      record can mix in an arbitrary way, the bytes inside a single
14  *      record must remain uninterrupted.
15  *
16  *      In case of files with fixed record size, we just allocate the
17  *      buffer to hold a whole number of records and take advantage
18  *      of the atomicity of the write() system call.
19  *
20  *      With variable-sized records, we need another solution: when
21  *      writing a record, we keep the fastbuf in a locked state, which
22  *      prevents buffer flushing (and if the buffer becomes full, we extend it),
23  *      and we wait for an explicit commit operation which write()s the buffer
24  *      if the free space in the buffer falls below the expected maximum record
25  *      length.
26  *
27  *      fbatomic_open() is called with the following parameters:
28  *          name - name of the file to open
29  *          master - fbatomic for the master thread or NULL if it's the first open
30  *          bufsize - initial buffer size
31  *          record_len - record length for fixed-size records;
32  *              or -(expected maximum record length) for variable-sized ones.
33  */
34
35 #define LOCAL_DEBUG
36
37 #include "lib/lib.h"
38 #include "lib/fastbuf.h"
39 #include "lib/lfs.h"
40
41 #include <string.h>
42 #include <fcntl.h>
43 #include <unistd.h>
44
45 struct fb_atomic_file {
46   int fd;
47   int use_count;
48   int record_len;
49   uns locked;
50   byte name[1];
51 };
52
53 void
54 fbatomic_internal_write(struct fastbuf *f)
55 {
56   struct fb_atomic_file *af = FB_ATOMIC(f)->af;
57   int size = f->bptr - f->buffer;
58   if (size)
59     {
60       ASSERT(af->record_len < 0 || !(size % af->record_len));
61       int res = write(af->fd, f->buffer, size);
62       if (res < 0)
63         die("Error writing %s: %m", f->name);
64       if (res != size)
65         die("Unexpected partial write to %s: written only %d bytes of %d", f->name, res, size);
66       f->bptr = f->buffer;
67     }
68 }
69
70 static void
71 fbatomic_spout(struct fastbuf *f)
72 {
73   if (f->bptr < f->bufend)              /* Explicit flushes should be ignored */
74     return;
75
76   struct fb_atomic *F = FB_ATOMIC(f);
77   if (F->af->locked)
78     {
79       uns written = f->bptr - f->buffer;
80       uns size = f->bufend - f->buffer + F->slack_size;
81       F->slack_size *= 2;
82       DBG("Reallocating buffer for atomic file %s with slack %d", f->name, F->slack_size);
83       f->buffer = xrealloc(f->buffer, size);
84       f->bufend = f->buffer + size;
85       f->bptr = f->buffer + written;
86       F->expected_max_bptr = f->bufend - F->slack_size;
87     }
88   else
89     fbatomic_internal_write(f);
90 }
91
92 static void
93 fbatomic_close(struct fastbuf *f)
94 {
95   struct fb_atomic_file *af = FB_ATOMIC(f)->af;
96   fbatomic_internal_write(f);   /* Need to flush explicitly, because the file can be locked */
97   if (!--af->use_count)
98     {
99       close(af->fd);
100       xfree(af);
101     }
102   xfree(f);
103 }
104
105 struct fastbuf *
106 fbatomic_open(byte *name, struct fastbuf *master, uns bufsize, int record_len)
107 {
108   struct fb_atomic *F = xmalloc_zero(sizeof(*F));
109   struct fastbuf *f = &F->fb;
110   struct fb_atomic_file *af;
111   if (master)
112     {
113       af = FB_ATOMIC(master)->af;
114       af->use_count++;
115       ASSERT(af->record_len == record_len);
116     }
117   else
118     {
119       af = xmalloc_zero(sizeof(*af) + strlen(name));
120       if ((af->fd = sh_open(name, O_WRONLY | O_CREAT | O_TRUNC | O_APPEND, 0666)) < 0)
121         die("Cannot create %s: %m", name);
122       af->use_count = 1;
123       af->record_len = record_len;
124       af->locked = (record_len < 0);
125       strcpy(af->name, name);
126     }
127   F->af = af;
128   if (record_len > 0 && bufsize % record_len)
129     bufsize += record_len - (bufsize % record_len);
130   f->buffer = xmalloc(bufsize);
131   f->bufend = f->buffer + bufsize;
132   F->slack_size = (record_len < 0) ? -record_len : 0;
133   ASSERT(bufsize > F->slack_size);
134   F->expected_max_bptr = f->bufend - F->slack_size;
135   f->bptr = f->bstop = f->buffer;
136   f->name = af->name;
137   f->spout = fbatomic_spout;
138   f->close = fbatomic_close;
139   return f;
140 }
141
142 #ifdef TEST
143
144 int main(int argc UNUSED, char **argv UNUSED)
145 {
146   struct fastbuf *f, *g;
147
148   log(L_INFO, "Testing block writes");
149   f = fbatomic_open("test", NULL, 16, 4);
150   for (u32 i=0; i<17; i++)
151     bwrite(f, &i, 4);
152   bclose(f);
153
154   log(L_INFO, "Testing interleaved var-size writes");
155   f = fbatomic_open("test2", NULL, 23, -5);
156   g = fbatomic_open("test2", f, 23, -5);
157   for (int i=0; i<100; i++)
158     {
159       struct fastbuf *x = (i%2) ? g : f;
160       bprintf(x, "%c<%d>\n", "fg"[i%2], ((259309*i) % 1000000) >> (i % 8));
161       fbatomic_commit(x);
162     }
163   bclose(f);
164   bclose(g);
165
166   return 0;
167 }
168
169 #endif