]> mj.ucw.cz Git - libucw.git/blob - lib/bucket.c
Changed locking mechanism of the bucket library to fcntl() instead
[libucw.git] / lib / bucket.c
1 /*
2  *      Sherlock Library -- Object Buckets
3  *
4  *      (c) 2001 Martin Mares <mj@ucw.cz>
5  */
6
7 #include "lib/lib.h"
8 #include "lib/bucket.h"
9 #include "lib/fastbuf.h"
10 #include "lib/lfs.h"
11 #include "lib/conf.h"
12
13 #include <string.h>
14 #include <fcntl.h>
15 #include <unistd.h>
16 #include <sys/file.h>
17
18 static int obuck_fd;
19 static unsigned int obuck_remains, obuck_check_pad;
20 static struct fastbuf *obuck_fb;
21 static struct obuck_header obuck_hdr;
22 static sh_off_t bucket_start;
23
24 /*** Configuration ***/
25
26 byte *obuck_name = "not/configured";
27 static int obuck_io_buflen = 65536;
28
29 static struct cfitem obuck_config[] = {
30   { "Buckets",          CT_SECTION,     NULL },
31   { "BucketFile",       CT_STRING,      &obuck_name },
32   { "BufSize",          CT_INT,         &obuck_io_buflen },
33   { NULL,               CT_STOP,        NULL }
34 };
35
36 static void CONSTRUCTOR obuck_init_config(void)
37 {
38   cf_register(obuck_config);
39 }
40
41 /*** Internal operations ***/
42
43 static void
44 obuck_broken(char *msg)
45 {
46   die("Object pool corrupted: %s (pos=%Lx)", msg, (long long) bucket_start);
47 }
48
49 /*
50  *  Unfortunately we cannot use flock() here since it happily permits
51  *  locking a shared fd (e.g., after fork()) multiple times. The fcntl
52  *  locks are very ugly and they don't support 64-bit offsets, but we
53  *  can work around the problem by always locking the first header
54  *  in the file.
55  */
56
57 static inline void
58 obuck_do_lock(int type)
59 {
60   struct flock fl;
61
62   fl.l_type = type;
63   fl.l_whence = SEEK_SET;
64   fl.l_start = 0;
65   fl.l_len = sizeof(struct obuck_header);
66   if (fcntl(obuck_fd, F_SETLKW, &fl) < 0)
67     die("fcntl lock: %m");
68 }
69
70 static inline void
71 obuck_lock_read(void)
72 {
73   obuck_do_lock(F_RDLCK);
74 }
75
76 static inline void
77 obuck_lock_write(void)
78 {
79   obuck_do_lock(F_WRLCK);
80 }
81
82 static inline void
83 obuck_unlock(void)
84 {
85   obuck_do_lock(F_UNLCK);
86 }
87
88 /*** FastIO emulation ***/
89
90 /* We need to use pread/pwrite since we work on fd's shared between processes */
91
92 static int
93 obuck_fb_refill(struct fastbuf *f)
94 {
95   unsigned limit = (f->buflen < obuck_remains) ? f->buflen : obuck_remains;
96   unsigned size = (limit == obuck_remains) ? (limit+obuck_check_pad+4) : limit;
97   int l;
98
99   if (!limit)
100     return 0;
101   l = sh_pread(f->fd, f->buffer, size, f->fdpos);
102   if (l < 0)
103     die("Error reading bucket: %m");
104   if ((unsigned) l != size)
105     obuck_broken("Short read");
106   f->bptr = f->buffer;
107   f->bstop = f->buffer + limit;
108   f->pos = f->fdpos;
109   f->fdpos += limit;
110   obuck_remains -= limit;
111   if (!obuck_remains)   /* Should check the trailer */
112     {
113       u32 check;
114       memcpy(&check, f->buffer + size - 4, 4);
115       if (check != OBUCK_TRAILER)
116         obuck_broken("Missing trailer");
117     }
118   return limit;
119 }
120
121 static void
122 obuck_fb_spout(struct fastbuf *f)
123 {
124   int l = f->bptr - f->buffer;
125   char *c = f->buffer;
126
127   while (l)
128     {
129       int z = sh_pwrite(f->fd, c, l, f->fdpos);
130       if (z <= 0)
131         die("Error writing bucket: %m");
132       f->fdpos += z;
133       l -= z;
134       c += z;
135     }
136   f->bptr = f->buffer;
137   f->pos = f->fdpos;
138 }
139
140 static void
141 obuck_fb_close(struct fastbuf *f)
142 {
143   close(f->fd);
144 }
145
146 /*** Exported functions ***/
147
148 void
149 obuck_init(int writeable)
150 {
151   struct fastbuf *b;
152   sh_off_t size;
153
154   obuck_fd = sh_open(obuck_name, (writeable ? O_RDWR | O_CREAT : O_RDONLY), 0666);
155   if (obuck_fd < 0)
156     die("Unable to open bucket file %s: %m", obuck_name);
157   obuck_fb = b = xmalloc_zero(sizeof(struct fastbuf) + obuck_io_buflen + OBUCK_ALIGN + 4);
158   b->buflen = obuck_io_buflen;
159   b->buffer = (char *)(b+1);
160   b->bptr = b->bstop = b->buffer;
161   b->bufend = b->buffer + obuck_io_buflen;
162   b->name = "bucket";
163   b->fd = obuck_fd;
164   b->refill = obuck_fb_refill;
165   b->spout = obuck_fb_spout;
166   b->close = obuck_fb_close;
167   obuck_lock_read();
168   size = sh_seek(obuck_fd, 0, SEEK_END);
169   if (size)
170     {
171       /* If the bucket pool is not empty, check consistency of its end */
172       u32 check;
173       bucket_start = size - 4;  /* for error reporting */
174       if (sh_pread(obuck_fd, &check, 4, size-4) != 4 ||
175           check != OBUCK_TRAILER)
176         obuck_broken("Missing trailer of last object");
177     }
178   obuck_unlock();
179 }
180
181 void
182 obuck_cleanup(void)
183 {
184   bclose(obuck_fb);
185 }
186
187 void
188 obuck_sync(void)
189 {
190   bflush(obuck_fb);
191   fsync(obuck_fd);
192 }
193
194 static void
195 obuck_get(oid_t oid)
196 {
197   struct fastbuf *b = obuck_fb;
198
199   bucket_start = obuck_get_pos(oid);
200   bflush(b);
201   if (sh_pread(obuck_fd, &obuck_hdr, sizeof(obuck_hdr), bucket_start) != sizeof(obuck_hdr))
202     obuck_broken("Short header read");
203   b->fdpos = bucket_start + sizeof(obuck_hdr);
204   if (obuck_hdr.magic != OBUCK_MAGIC)
205     obuck_broken("Missing magic number");
206   if (obuck_hdr.oid == OBUCK_OID_DELETED)
207     obuck_broken("Access to deleted bucket");
208   if (obuck_hdr.oid != oid)
209     obuck_broken("Invalid backlink");
210 }
211
212 void
213 obuck_find_by_oid(struct obuck_header *hdrp)
214 {
215   oid_t oid = hdrp->oid;
216
217   ASSERT(oid < OBUCK_OID_FIRST_SPECIAL);
218   obuck_lock_read();
219   obuck_get(oid);
220   obuck_unlock();
221   memcpy(hdrp, &obuck_hdr, sizeof(obuck_hdr));
222 }
223
224 int
225 obuck_find_first(struct obuck_header *hdrp, int full)
226 {
227   bucket_start = 0;
228   obuck_hdr.magic = 0;
229   return obuck_find_next(hdrp, full);
230 }
231
232 int
233 obuck_find_next(struct obuck_header *hdrp, int full)
234 {
235   int c;
236   struct fastbuf *b = obuck_fb;
237
238   for(;;)
239     {
240       if (obuck_hdr.magic)
241         bucket_start = (bucket_start + sizeof(obuck_hdr) + obuck_hdr.length +
242                         4 + OBUCK_ALIGN - 1) & ~((sh_off_t)(OBUCK_ALIGN - 1));
243       bflush(b);
244       obuck_lock_read();
245       c = sh_pread(obuck_fd, &obuck_hdr, sizeof(obuck_hdr), bucket_start);
246       obuck_unlock();
247       if (!c)
248         return 0;
249       if (c != sizeof(obuck_hdr))
250         obuck_broken("Short header read");
251       b->fdpos = bucket_start + sizeof(obuck_hdr);
252       if (obuck_hdr.magic != OBUCK_MAGIC)
253         obuck_broken("Missing magic number");
254       if (obuck_hdr.oid != OBUCK_OID_DELETED || full)
255         {
256           memcpy(hdrp, &obuck_hdr, sizeof(obuck_hdr));
257           return 1;
258         }
259     }
260 }
261
262 struct fastbuf *
263 obuck_fetch(void)
264 {
265   obuck_remains = obuck_hdr.length;
266   obuck_check_pad = (OBUCK_ALIGN - sizeof(obuck_hdr) - obuck_hdr.length - 4) & (OBUCK_ALIGN - 1);
267   return obuck_fb;
268 }
269
270 void
271 obuck_fetch_end(struct fastbuf *b UNUSED)
272 {
273 }
274
275 struct fastbuf *
276 obuck_create(void)
277 {
278   obuck_lock_write();
279   bflush(obuck_fb);
280   bucket_start = sh_seek(obuck_fd, 0, SEEK_END);
281   if (bucket_start & (OBUCK_ALIGN - 1))
282     obuck_broken("Misaligned file");
283   obuck_hdr.magic = OBUCK_INCOMPLETE_MAGIC;
284   obuck_hdr.oid = bucket_start >> OBUCK_SHIFT;
285   obuck_hdr.length = obuck_hdr.orig_length = 0;
286   obuck_fb->fdpos = obuck_fb->pos = bucket_start;
287   bwrite(obuck_fb, &obuck_hdr, sizeof(obuck_hdr));
288   return obuck_fb;
289 }
290
291 void
292 obuck_create_end(struct fastbuf *b UNUSED, struct obuck_header *hdrp)
293 {
294   int pad;
295   obuck_hdr.magic = OBUCK_MAGIC;
296   obuck_hdr.length = obuck_hdr.orig_length = btell(obuck_fb) - bucket_start - sizeof(obuck_hdr);
297   pad = (OBUCK_ALIGN - sizeof(obuck_hdr) - obuck_hdr.length - 4) & (OBUCK_ALIGN - 1);
298   while (pad--)
299     bputc(obuck_fb, 0);
300   bputl(obuck_fb, OBUCK_TRAILER);
301   bflush(obuck_fb);
302   ASSERT(!(btell(obuck_fb) & (OBUCK_ALIGN - 1)));
303   sh_pwrite(obuck_fd, &obuck_hdr, sizeof(obuck_hdr), bucket_start);
304   obuck_unlock();
305   memcpy(hdrp, &obuck_hdr, sizeof(obuck_hdr));
306 }
307
308 void
309 obuck_delete(oid_t oid)
310 {
311   obuck_lock_write();
312   obuck_get(oid);
313   obuck_hdr.oid = OBUCK_OID_DELETED;
314   sh_pwrite(obuck_fd, &obuck_hdr, sizeof(obuck_hdr), bucket_start);
315   obuck_unlock();
316 }
317
318 /*** Testing ***/
319
320 #ifdef TEST
321
322 #define COUNT 5000
323 #define MAXLEN 10000
324 #define KILLPERC 13
325 #define LEN(i) ((259309*(i))%MAXLEN)
326
327 int main(int argc, char **argv)
328 {
329   int ids[COUNT];
330   unsigned int i, j, cnt;
331   struct obuck_header h;
332   struct fastbuf *b;
333
334   log_init(NULL);
335   if (cf_getopt(argc, argv, CF_SHORT_OPTS, CF_NO_LONG_OPTS, NULL) >= 0 ||
336       optind < argc)
337     die("This program has no command-line arguments.");
338
339   unlink(obuck_name);
340   obuck_init(1);
341   for(j=0; j<COUNT; j++)
342     {
343       b = obuck_create();
344       for(i=0; i<LEN(j); i++)
345         bputc(b, (i+j) % 256);
346       obuck_create_end(b, &h);
347       printf("Writing %08x %d -> %d\n", h.oid, h.orig_length, h.length);
348       ids[j] = h.oid;
349     }
350   for(j=0; j<COUNT; j++)
351     if (j % 100 < KILLPERC)
352       {
353         printf("Deleting %08x\n", ids[j]);
354         obuck_delete(ids[j]);
355       }
356   cnt = 0;
357   for(j=0; j<COUNT; j++)
358     if (j % 100 >= KILLPERC)
359       {
360         cnt++;
361         h.oid = ids[j];
362         obuck_find_by_oid(&h);
363         b = obuck_fetch();
364         printf("Reading %08x %d -> %d\n", h.oid, h.orig_length, h.length);
365         if (h.orig_length != LEN(j))
366           die("Invalid length");
367         for(i=0; i<h.orig_length; i++)
368           if ((unsigned) bgetc(b) != (i+j) % 256)
369             die("Contents mismatch");
370         if (bgetc(b) != EOF)
371           die("EOF mismatch");
372         obuck_fetch_end(b);
373       }
374   if (obuck_find_first(&h, 0))
375     do
376       {
377         printf("<<< %08x\t%d\n", h.oid, h.orig_length);
378         cnt--;
379       }
380     while (obuck_find_next(&h, 0));
381   if (cnt)
382     die("Walk mismatch");
383   obuck_cleanup();
384   return 0;
385 }
386
387 #endif