]> mj.ucw.cz Git - libucw.git/blob - lib/bucket.c
Changed locking rules. Scans and appends can peacefully co-exist now.
[libucw.git] / lib / bucket.c
1 /*
2  *      Sherlock Library -- Object Buckets
3  *
4  *      (c) 2001--2004 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 #undef LOCAL_DEBUG
11
12 #include "lib/lib.h"
13 #include "lib/bucket.h"
14 #include "lib/fastbuf.h"
15 #include "lib/lfs.h"
16 #include "lib/conf.h"
17
18 #include <string.h>
19 #include <stdlib.h>
20 #include <fcntl.h>
21 #include <unistd.h>
22 #include <sys/file.h>
23 #include <alloca.h>
24
25 static int obuck_fd;
26 static struct obuck_header obuck_hdr, obuck_create_hdr;
27 static sh_off_t bucket_find_pos;
28 static struct fastbuf *obuck_write_fb;
29
30 /*** Configuration ***/
31
32 byte *obuck_name = "not/configured";
33 static uns obuck_io_buflen = 65536;
34 static int obuck_shake_buflen = 1048576;
35 static uns obuck_shake_security;
36 static uns obuck_slurp_buflen = 65536;
37
38 static struct cfitem obuck_config[] = {
39   { "Buckets",          CT_SECTION,     NULL },
40   { "BucketFile",       CT_STRING,      &obuck_name },
41   { "BufSize",          CT_INT,         &obuck_io_buflen },
42   { "ShakeBufSize",     CT_INT,         &obuck_shake_buflen },
43   { "ShakeSecurity",    CT_INT,         &obuck_shake_security },
44   { "SlurpBufSize",     CT_INT,         &obuck_slurp_buflen },
45   { NULL,               CT_STOP,        NULL }
46 };
47
48 static void CONSTRUCTOR obuck_init_config(void)
49 {
50   cf_register(obuck_config);
51 }
52
53 /*** Internal operations ***/
54
55 static void
56 obuck_broken(char *msg, sh_off_t pos)
57 {
58   die("Object pool corrupted: %s (pos=%Lx)", msg, (long long) pos);
59 }
60
61 /*
62  *  We need several types of locks:
63  *
64  *      Read lock       reading parts of bucket file
65  *      Write lock      any write operations
66  *      Append lock     appending to the end of the file
67  *      Scan lock       reading parts which we are certain they exist
68  *
69  *  Multiple read and scan locks can co-exist together.
70  *  Scan locks can co-exist with an append lock.
71  *  There can be at most one write/append lock at a time.
72  *
73  *  These lock types map to a pair of normal read-write locks which
74  *  we represent as fcntl() locks on the first and second byte of the
75  *  bucket file. [We cannot use flock() since it happily permits
76  *  locking a shared fd (e.g., after fork()) multiple times at it also
77  *  doesn't offer multiple locks on a single file.]
78  *
79  *                      byte0           byte1
80  *      Read            <read>          <read>
81  *      Write           <write>         <write>
82  *      Append          -               <write>
83  *      Scan            <read>          -
84  */
85
86 static inline void
87 obuck_do_lock(int type, int start, int len)
88 {
89   struct flock fl;
90
91   fl.l_type = type;
92   fl.l_whence = SEEK_SET;
93   fl.l_start = start;
94   fl.l_len = len;
95   if (fcntl(obuck_fd, F_SETLKW, &fl) < 0)
96     die("fcntl lock: %m");
97 }
98
99 inline void
100 obuck_lock_read(void)
101 {
102   obuck_do_lock(F_RDLCK, 0, 2);
103 }
104
105 inline void
106 obuck_lock_write(void)
107 {
108   obuck_do_lock(F_WRLCK, 0, 2);
109 }
110
111 static inline void
112 obuck_lock_append(void)
113 {
114   obuck_do_lock(F_WRLCK, 0, 1);
115 }
116
117 static inline void
118 obuck_lock_read_to_scan(void)
119 {
120   obuck_do_lock(F_UNLCK, 0, 1);
121 }
122
123 inline void
124 obuck_unlock(void)
125 {
126   obuck_do_lock(F_UNLCK, 0, 2);
127 }
128
129 /*** FastIO emulation ***/
130
131 struct fb_bucket {
132   struct fastbuf fb;
133   sh_off_t start_pos;
134   uns bucket_size;
135   byte buffer[0];
136 };
137 #define FB_BUCKET(f) ((struct fb_bucket *)(f)->is_fastbuf)
138
139 static int obuck_fb_count;
140
141 static void
142 obuck_fb_close(struct fastbuf *f)
143 {
144   obuck_fb_count--;
145   xfree(f);
146 }
147
148 /* We need to use pread/pwrite since we work on fd's shared between processes */
149
150 static int
151 obuck_fb_refill(struct fastbuf *f)
152 {
153   uns remains, bufsize, size, datasize;
154
155   remains = FB_BUCKET(f)->bucket_size - (uns)f->pos;
156   bufsize = f->bufend - f->buffer;
157   if (!remains)
158     return 0;
159   sh_off_t start = FB_BUCKET(f)->start_pos;
160   sh_off_t pos = start + sizeof(struct obuck_header) + f->pos;
161   if (remains <= bufsize)
162     {
163       datasize = remains;
164       size = start + obuck_bucket_size(FB_BUCKET(f)->bucket_size) - pos;
165     }
166   else
167     size = datasize = bufsize;
168   int l = sh_pread(obuck_fd, f->buffer, size, pos);
169   if (l < 0)
170     die("Error reading bucket: %m");
171   if ((unsigned) l != size)
172     obuck_broken("Short read", FB_BUCKET(f)->start_pos);
173   f->bptr = f->buffer;
174   f->bstop = f->buffer + datasize;
175   f->pos += datasize;
176   if (datasize < size)
177     {
178       if (GET_U32(f->buffer + size - 4) != OBUCK_TRAILER)
179         obuck_broken("Missing trailer", FB_BUCKET(f)->start_pos);
180     }
181   return datasize;
182 }
183
184 static void
185 obuck_fb_spout(struct fastbuf *f)
186 {
187   int l = f->bptr - f->buffer;
188   char *c = f->buffer;
189
190   while (l)
191     {
192       int z = sh_pwrite(obuck_fd, c, l, FB_BUCKET(f)->start_pos + sizeof(struct obuck_header) + f->pos);
193       if (z <= 0)
194         die("Error writing bucket: %m");
195       f->pos += z;
196       l -= z;
197       c += z;
198     }
199   f->bptr = f->buffer;
200 }
201
202 /*** Exported functions ***/
203
204 void
205 obuck_init(int writeable)
206 {
207   sh_off_t size;
208
209   obuck_fd = sh_open(obuck_name, (writeable ? O_RDWR | O_CREAT : O_RDONLY), 0666);
210   if (obuck_fd < 0)
211     die("Unable to open bucket file %s: %m", obuck_name);
212   obuck_lock_read();
213   size = sh_seek(obuck_fd, 0, SEEK_END);
214   if (size)
215     {
216       /* If the bucket pool is not empty, check consistency of its end */
217       u32 check;
218       if (sh_pread(obuck_fd, &check, 4, size-4) != 4 ||
219           check != OBUCK_TRAILER)
220         obuck_broken("Missing trailer of last object", size - 4);
221     }
222   obuck_unlock();
223 }
224
225 void
226 obuck_cleanup(void)
227 {
228   close(obuck_fd);
229   if (obuck_fb_count)
230     log(L_ERROR, "Bug: Unbalanced bucket opens/closes: %d streams remain", obuck_fb_count);
231   if (obuck_write_fb)
232     log(L_ERROR, "Bug: Forgot to close bucket write stream");
233 }
234
235 void
236 obuck_sync(void)
237 {
238   if (obuck_write_fb)
239     bflush(obuck_write_fb);
240   fsync(obuck_fd);
241 }
242
243 static void
244 obuck_get(oid_t oid)
245 {
246   bucket_find_pos = obuck_get_pos(oid);
247   if (sh_pread(obuck_fd, &obuck_hdr, sizeof(obuck_hdr), bucket_find_pos) != sizeof(obuck_hdr))
248     obuck_broken("Short header read", bucket_find_pos);
249   if (obuck_hdr.magic != OBUCK_MAGIC)
250     obuck_broken("Missing magic number", bucket_find_pos);
251   if (obuck_hdr.oid == OBUCK_OID_DELETED)
252     obuck_broken("Access to deleted bucket", bucket_find_pos);
253   if (obuck_hdr.oid != oid)
254     obuck_broken("Invalid backlink", bucket_find_pos);
255 }
256
257 void
258 obuck_find_by_oid(struct obuck_header *hdrp)
259 {
260   oid_t oid = hdrp->oid;
261
262   ASSERT(oid < OBUCK_OID_FIRST_SPECIAL);
263   obuck_lock_read();
264   obuck_get(oid);
265   obuck_unlock();
266   memcpy(hdrp, &obuck_hdr, sizeof(obuck_hdr));
267 }
268
269 int
270 obuck_find_first(struct obuck_header *hdrp, int full)
271 {
272   bucket_find_pos = 0;
273   obuck_hdr.magic = 0;
274   return obuck_find_next(hdrp, full);
275 }
276
277 int
278 obuck_find_next(struct obuck_header *hdrp, int full)
279 {
280   int c;
281
282   for(;;)
283     {
284       if (obuck_hdr.magic)
285         bucket_find_pos += obuck_bucket_size(obuck_hdr.length);
286       obuck_lock_read();
287       c = sh_pread(obuck_fd, &obuck_hdr, sizeof(obuck_hdr), bucket_find_pos);
288       obuck_unlock();
289       if (!c)
290         return 0;
291       if (c != sizeof(obuck_hdr))
292         obuck_broken("Short header read", bucket_find_pos);
293       if (obuck_hdr.magic != OBUCK_MAGIC)
294         obuck_broken("Missing magic number", bucket_find_pos);
295       if (obuck_hdr.oid != OBUCK_OID_DELETED || full)
296         {
297           memcpy(hdrp, &obuck_hdr, sizeof(obuck_hdr));
298           return 1;
299         }
300     }
301 }
302
303 struct fastbuf *
304 obuck_fetch(void)
305 {
306   struct fastbuf *b;
307   uns official_buflen = ALIGN(MIN(obuck_hdr.length, obuck_io_buflen), OBUCK_ALIGN);
308   uns real_buflen = official_buflen + OBUCK_ALIGN;
309
310   b = xmalloc(sizeof(struct fb_bucket) + real_buflen);
311   b->buffer = b->bptr = b->bstop = FB_BUCKET(b)->buffer;
312   b->bufend = b->buffer + official_buflen;
313   b->name = "bucket-read";
314   b->pos = 0;
315   b->refill = obuck_fb_refill;
316   b->spout = NULL;
317   b->seek = NULL;
318   b->close = obuck_fb_close;
319   b->config = NULL;
320   FB_BUCKET(b)->start_pos = bucket_find_pos;
321   FB_BUCKET(b)->bucket_size = obuck_hdr.length;
322   obuck_fb_count++;
323   return b;
324 }
325
326 oid_t
327 obuck_predict_last_oid(void)
328 {
329   sh_off_t size = sh_seek(obuck_fd, 0, SEEK_END);
330   return (oid_t)(size >> OBUCK_SHIFT);
331 }
332
333 struct fastbuf *
334 obuck_create(u32 type)
335 {
336   ASSERT(!obuck_write_fb);
337
338   obuck_lock_append();
339   sh_off_t start = sh_seek(obuck_fd, 0, SEEK_END);
340   if (start & (OBUCK_ALIGN - 1))
341     obuck_broken("Misaligned file", start);
342   obuck_create_hdr.magic = OBUCK_INCOMPLETE_MAGIC;
343   obuck_create_hdr.oid = start >> OBUCK_SHIFT;
344   obuck_create_hdr.length = 0;
345   obuck_create_hdr.type = type;
346
347   struct fastbuf *b = xmalloc(sizeof(struct fb_bucket) + obuck_io_buflen);
348   obuck_write_fb = b;
349   b->buffer = FB_BUCKET(b)->buffer;
350   b->bptr = b->bstop = b->buffer;
351   b->bufend = b->buffer + obuck_io_buflen;
352   b->pos = -(int)sizeof(obuck_create_hdr);
353   b->name = "bucket-write";
354   b->refill = NULL;
355   b->spout = obuck_fb_spout;
356   b->seek = NULL;
357   b->close = NULL;
358   b->config = NULL;
359   FB_BUCKET(b)->start_pos = start;
360   FB_BUCKET(b)->bucket_size = 0;
361   bwrite(b, &obuck_create_hdr, sizeof(obuck_create_hdr));
362
363   return b;
364 }
365
366 void
367 obuck_create_end(struct fastbuf *b, struct obuck_header *hdrp)
368 {
369   ASSERT(b == obuck_write_fb);
370   obuck_write_fb = NULL;
371
372   obuck_create_hdr.magic = OBUCK_MAGIC;
373   obuck_create_hdr.length = btell(b);
374   int pad = (OBUCK_ALIGN - sizeof(obuck_create_hdr) - obuck_create_hdr.length - 4) & (OBUCK_ALIGN - 1);
375   while (pad--)
376     bputc(b, 0);
377   bputl(b, OBUCK_TRAILER);
378   bflush(b);
379   ASSERT(!((FB_BUCKET(b)->start_pos + sizeof(obuck_create_hdr) + b->pos) & (OBUCK_ALIGN - 1)));
380   if (sh_pwrite(obuck_fd, &obuck_create_hdr, sizeof(obuck_create_hdr), FB_BUCKET(b)->start_pos) != sizeof(obuck_create_hdr))
381     die("Bucket header update failed: %m");
382   obuck_unlock();
383   memcpy(hdrp, &obuck_create_hdr, sizeof(obuck_create_hdr));
384   xfree(b);
385 }
386
387 void
388 obuck_delete(oid_t oid)
389 {
390   obuck_lock_write();
391   obuck_get(oid);
392   obuck_hdr.oid = OBUCK_OID_DELETED;
393   sh_pwrite(obuck_fd, &obuck_hdr, sizeof(obuck_hdr), bucket_find_pos);
394   obuck_unlock();
395 }
396
397 /*** Fast reading of the whole pool ***/
398
399 static struct fastbuf *obuck_rpf;
400 static uns slurp_remains;
401 static sh_off_t slurp_start, slurp_current, slurp_end;
402
403 static int
404 obuck_slurp_refill(struct fastbuf *f)
405 {
406   uns l;
407
408   if (!slurp_remains)
409     return 0;
410   l = bdirect_read_prepare(obuck_rpf, &f->buffer);
411   if (!l)
412     obuck_broken("Incomplete object", slurp_start);
413   l = MIN(l, slurp_remains);
414   bdirect_read_commit(obuck_rpf, f->buffer + l);
415   slurp_remains -= l;
416   f->bptr = f->buffer;
417   f->bufend = f->bstop = f->buffer + l;
418   return 1;
419 }
420
421 struct fastbuf *
422 obuck_slurp_pool(struct obuck_header *hdrp)
423 {
424   static struct fastbuf limiter;
425   uns l;
426
427   do
428     {
429       if (!obuck_rpf)
430         {
431           obuck_lock_read();
432           obuck_rpf = bopen(obuck_name, O_RDONLY, obuck_slurp_buflen);
433           bseek(obuck_rpf, 0, SEEK_END);
434           slurp_end = btell(obuck_rpf);
435           bsetpos(obuck_rpf, 0);
436           obuck_lock_read_to_scan();
437         }
438       else
439         {
440           bsetpos(obuck_rpf, slurp_current - 4);
441           if (bgetl(obuck_rpf) != OBUCK_TRAILER)
442             obuck_broken("Missing trailer", slurp_start);
443         }
444       slurp_start = btell(obuck_rpf);
445       if (slurp_start < slurp_end)
446         l = bread(obuck_rpf, hdrp, sizeof(struct obuck_header));
447       else
448         l = 0;
449       if (!l)
450         {
451           bclose(obuck_rpf);
452           obuck_rpf = NULL;
453           obuck_unlock();
454           return NULL;
455         }
456       if (l != sizeof(struct obuck_header))
457         obuck_broken("Short header read", slurp_start);
458       if (hdrp->magic != OBUCK_MAGIC)
459         obuck_broken("Missing magic number", slurp_start);
460       slurp_current = slurp_start + obuck_bucket_size(hdrp->length);
461     }
462   while (hdrp->oid == OBUCK_OID_DELETED);
463   if (obuck_get_pos(hdrp->oid) != slurp_start)
464     obuck_broken("Invalid backlink", slurp_start);
465   slurp_remains = hdrp->length;
466   limiter.bptr = limiter.bstop = limiter.buffer = limiter.bufend = NULL;
467   limiter.name = "Bucket";
468   limiter.pos = 0;
469   limiter.refill = obuck_slurp_refill;
470   return &limiter;
471 }
472
473 /*** Shakedown ***/
474
475 static inline void
476 shake_write(void *addr, int len, sh_off_t pos)
477 {
478   int l = sh_pwrite(obuck_fd, addr, len, pos);
479   if (l != len)
480     {
481       if (l < 0)
482         die("obuck_shakedown write error: %m");
483       else
484         die("obuck_shakedown write error: disk full");
485     }
486 }
487
488 static inline void
489 shake_sync(void)
490 {
491   if (obuck_shake_security > 1)
492     fdatasync(obuck_fd);
493 }
494
495 static void
496 shake_write_backup(sh_off_t bpos, byte *norm_buf, int norm_size, byte *fragment, int frag_size, sh_off_t frag_pos, int more_size)
497 {
498   struct obuck_header *bhdr;
499   int boff = 0;
500   int l;
501   oid_t old_oid;
502
503   /* First of all, the "normal" part -- everything that will be written in this pass */
504   DBG("Backing up first round of changes at position %Lx + %x", (long long) bpos, norm_size);
505   while (boff < norm_size)
506     {
507       /* This needn't be optimized for speed. */
508       bhdr = (struct obuck_header *) (norm_buf + boff);
509       ASSERT(bhdr->magic == OBUCK_MAGIC);
510       l = obuck_bucket_size(bhdr->length);
511       old_oid = bhdr->oid;
512       bhdr->oid = bpos >> OBUCK_SHIFT;
513       shake_write(bhdr, l, bpos);
514       bhdr->oid = old_oid;
515       boff += l;
516       bpos += l;
517     }
518
519   /* If we have an incomplete bucket at the end of the buffer, we must copy it as well. */
520   if (more_size)
521     {
522       DBG("Backing up fragment of size %x and %x more", frag_size, more_size);
523
524       /* First the part we already have in the buffer */
525       bhdr = (struct obuck_header *) fragment;
526       ASSERT(bhdr->magic == OBUCK_MAGIC);
527       old_oid = bhdr->oid;
528       bhdr->oid = bpos >> OBUCK_SHIFT;
529       shake_write(bhdr, frag_size, bpos);
530       bhdr->oid = old_oid;
531       bpos += frag_size;
532
533       /* And then the rest, using a small 64K buffer */
534       byte *auxbuf = alloca(65536);
535       l = 0;
536       while (l < more_size)
537         {
538           int j = MIN(more_size-l, 65536);
539           if (sh_pread(obuck_fd, auxbuf, j, frag_pos + frag_size + l) != j)
540             die("obuck_shakedown read error: %m");
541           shake_write(auxbuf, j, bpos);
542           bpos += j;
543           l += j;
544         }
545     }
546 }
547
548 static void
549 shake_erase(sh_off_t start, sh_off_t end)
550 {
551   if (start > end)
552     die("shake_erase called with negative length, that's a bug");
553   ASSERT(!(start & (OBUCK_ALIGN-1)) && !(end & (OBUCK_ALIGN-1)));
554   while (start < end)
555     {
556       u32 check = OBUCK_TRAILER;
557       obuck_hdr.magic = OBUCK_MAGIC;
558       obuck_hdr.oid = OBUCK_OID_DELETED;
559       uns len = MIN(0x40000000, end-start);
560       obuck_hdr.length = len - sizeof(obuck_hdr) - 4;
561       DBG("Erasing %08x bytes at %Lx", len, (long long) start);
562       shake_write(&obuck_hdr, sizeof(obuck_hdr), start);
563       start += len;
564       shake_write(&check, 4, start-4);
565     }
566 }
567
568 void
569 obuck_shakedown(int (*kibitz)(struct obuck_header *old, oid_t new, byte *buck))
570 {
571   byte *buf;                                            /* Shakedown buffer and its size */
572   int buflen = ALIGN(obuck_shake_buflen, OBUCK_ALIGN);
573   byte *msg;                                            /* Error message we will print */
574   sh_off_t rstart, wstart;                              /* Original and new position of buffer start */
575   sh_off_t r_bucket_start, w_bucket_start;              /* Original and new position of the current bucket */
576   int roff, woff;                                       /* Orig/new position of the current bucket relative to buffer start */
577   int rsize;                                            /* Number of original bytes in the buffer */
578   int l;                                                /* Raw size of the current bucket */
579   int changed = 0;                                      /* "Something has been altered" flag */
580   int wrote_anything = 0;                               /* We already did a write to the bucket file */
581   struct obuck_header *rhdr, *whdr;                     /* Original and new address of header of the current bucket */
582   sh_off_t r_file_size;                                 /* Original size of the bucket file */
583   int more;                                             /* How much does the last bucket overlap the buffer */
584
585   buf = xmalloc(buflen);
586   rstart = wstart = 0;
587   roff = woff = rsize = 0;
588
589   /* We need to be the only accessor, all the object ID's are becoming invalid */
590   obuck_lock_write();
591   r_file_size = sh_seek(obuck_fd, 0, SEEK_END);
592   ASSERT(!(r_file_size & (OBUCK_ALIGN - 1)));
593   if (r_file_size >= (0x100000000 << OBUCK_SHIFT) - buflen)
594     die("Bucket file is too large for safe shakedown. Shaking down with Bucket.ShakeSecurity=0 will still work.");
595
596   DBG("Starting shakedown. Buffer size is %d, original length %Lx", buflen, (long long) r_file_size);
597
598   for(;;)
599     {
600       r_bucket_start = rstart + roff;
601       w_bucket_start = wstart + woff;
602       rhdr = (struct obuck_header *)(buf + roff);
603       whdr = (struct obuck_header *)(buf + woff);
604       if (roff == rsize)
605         {
606           more = 0;
607           goto next;
608         }
609       if (rhdr->magic != OBUCK_MAGIC ||
610           rhdr->oid != OBUCK_OID_DELETED && rhdr->oid != (oid_t)(r_bucket_start >> OBUCK_SHIFT))
611         {
612           msg = "header mismatch";
613           goto broken;
614         }
615       l = obuck_bucket_size(rhdr->length);
616       if (l > buflen)
617         {
618           if (rhdr->oid != OBUCK_OID_DELETED)
619             {
620               msg = "bucket longer than ShakeBufSize";
621               goto broken;
622             }
623           /* Empty buckets are allowed to be large, but we need to handle them extra */
624           DBG("Tricking around an extra-large empty bucket at %Lx + %x", (long long)r_bucket_start, l);
625           rsize = roff + l;
626         }
627       else
628         {
629           if (rsize - roff < l)
630             {
631               more = l - (rsize - roff);
632               goto next;
633             }
634           if (GET_U32((byte *)rhdr + l - 4) != OBUCK_TRAILER)
635             {
636               msg = "missing trailer";
637               goto broken;
638             }
639         }
640       if (rhdr->oid != OBUCK_OID_DELETED)
641         {
642           int status = kibitz(rhdr, w_bucket_start >> OBUCK_SHIFT, (byte *)(rhdr+1));
643           if (status)
644             {
645               int lnew = l;
646               if (status > 1)
647                 {
648                   /* Changed! Reconstruct the trailer. */
649                   lnew = obuck_bucket_size(rhdr->length);
650                   ASSERT(lnew <= l);
651                   PUT_U32((byte *)rhdr + lnew - 4, OBUCK_TRAILER);
652                   changed = 1;
653                 }
654               whdr = (struct obuck_header *)(buf+woff);
655               if (rhdr != whdr)
656                 memmove(whdr, rhdr, lnew);
657               whdr->oid = w_bucket_start >> OBUCK_SHIFT;
658               woff += lnew;
659             }
660           else
661             changed = 1;
662         }
663       else
664         {
665           kibitz(rhdr, OBUCK_OID_DELETED, NULL);
666           changed = 1;
667         }
668       roff += l;
669       continue;
670
671     next:
672       if (changed)
673         {
674           /* Write the new contents of the bucket file */
675           if (!wrote_anything)
676             {
677               if (obuck_shake_security)
678                 {
679                   /* But first write a backup at the end of the file to ensure nothing can be lost. */
680                   shake_write_backup(r_file_size, buf, woff, buf+roff, rsize-roff, rstart+roff, more);
681                   shake_sync();
682                 }
683               wrote_anything = 1;
684             }
685           if (woff)
686             {
687               DBG("Write %Lx %x", wstart, woff);
688               shake_write(buf, woff, wstart);
689               shake_sync();
690             }
691         }
692       else
693         ASSERT(wstart == rstart);
694
695       /* In any case, update the write position */
696       wstart += woff;
697       woff = 0;
698
699       /* Skip what's been read and if there is any fragment at the end of the buffer, move it to the start */
700       rstart += roff;
701       if (more)
702         {
703           memmove(buf, buf+roff, rsize-roff);
704           rsize = rsize-roff;
705         }
706       else
707         rsize = 0;
708
709       /* And refill the buffer */
710       r_bucket_start = rstart+rsize;    /* Also needed for error messages */
711       l = sh_pread(obuck_fd, buf+rsize, MIN(buflen-rsize, r_file_size - r_bucket_start), r_bucket_start);
712       DBG("Read  %Lx %x (%x inherited)", (long long)r_bucket_start, l, rsize);
713       if (l < 0)
714         die("obuck_shakedown read error: %m");
715       if (!l)
716         {
717           if (!more)
718             break;
719           msg = "unexpected EOF";
720           goto broken;
721         }
722       if (l & (OBUCK_ALIGN-1))
723         {
724           msg = "garbage at the end of file";
725           goto broken;
726         }
727       rsize += l;
728       roff = 0;
729     }
730
731   DBG("Finished at position %Lx", (long long) wstart);
732   sh_ftruncate(obuck_fd, wstart);
733   shake_sync();
734
735   obuck_unlock();
736   xfree(buf);
737   return;
738
739  broken:
740   log(L_ERROR, "Error during object pool shakedown: %s (pos=%Ld, id=%x), gathering debris",
741       msg, (long long) r_bucket_start, (uns)(r_bucket_start >> OBUCK_SHIFT));
742   /*
743    * We can attempt to clean up the bucket file by erasing everything between the last
744    * byte written and the next byte to be read. If the secure mode is switched on, we can
745    * guarantee that no data are lost, only some might be duplicated.
746    */
747   shake_erase(wstart, rstart);
748   die("Fatal error during object pool shakedown");
749 }
750
751 /*** Testing ***/
752
753 #ifdef TEST
754
755 #define COUNT 5000
756 #define MAXLEN 10000
757 #define KILLPERC 13
758 #define LEN(i) ((259309*(i))%MAXLEN)
759
760 static int test_kibitz(struct obuck_header *h, oid_t new, byte *buck)
761 {
762   return 1;
763 }
764
765 int main(int argc, char **argv)
766 {
767   int ids[COUNT];
768   unsigned int i, j, cnt;
769   struct obuck_header h;
770   struct fastbuf *b;
771
772   log_init(NULL);
773   if (cf_getopt(argc, argv, CF_SHORT_OPTS, CF_NO_LONG_OPTS, NULL) >= 0 ||
774       optind < argc)
775   {
776     fputs("This program supports only the following command-line arguments:\n" CF_USAGE, stderr);
777     exit(1);
778   }
779
780   unlink(obuck_name);
781   obuck_init(1);
782   for(j=0; j<COUNT; j++)
783     {
784       b = obuck_create(BUCKET_TYPE_PLAIN);
785       for(i=0; i<LEN(j); i++)
786         bputc(b, (i+j) % 256);
787       obuck_create_end(b, &h);
788       printf("Writing %08x %d\n", h.oid, h.length);
789       ids[j] = h.oid;
790     }
791   for(j=0; j<COUNT; j++)
792     if (j % 100 < KILLPERC)
793       {
794         printf("Deleting %08x\n", ids[j]);
795         obuck_delete(ids[j]);
796       }
797   cnt = 0;
798   for(j=0; j<COUNT; j++)
799     if (j % 100 >= KILLPERC)
800       {
801         cnt++;
802         h.oid = ids[j];
803         obuck_find_by_oid(&h);
804         b = obuck_fetch();
805         printf("Reading %08x %d\n", h.oid, h.length);
806         if (h.length != LEN(j))
807           die("Invalid length");
808         for(i=0; i<h.length; i++)
809           if ((unsigned) bgetc(b) != (i+j) % 256)
810             die("Contents mismatch");
811         if (bgetc(b) != EOF)
812           die("EOF mismatch");
813         bclose(b);
814       }
815   obuck_shakedown(test_kibitz);
816   if (obuck_find_first(&h, 0))
817     do
818       {
819         printf("<<< %08x\t%d\n", h.oid, h.length);
820         cnt--;
821       }
822     while (obuck_find_next(&h, 0));
823   if (cnt)
824     die("Walk mismatch");
825   obuck_cleanup();
826   return 0;
827 }
828
829 #endif