ucw/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 #include "ucw/lib.h"
  11 #include "ucw/fastbuf.h"
  12 #include "ucw/lfs.h"
  13 #include "ucw/conf.h"
  14
  15 #include <string.h>
  16 #include <fcntl.h>
  17 #include <unistd.h>
  18
  19 static uns trace;
  20
  21 static struct cf_section fbatomic_config = {
  22   CF_ITEMS {
  23     CF_UNS("Trace", &trace)
  24   }
  25 };
  26
  27 static void CONSTRUCTOR fbatomic_init_config(void)
  28 {
  29   cf_declare_section("FBAtomic", &fbatomic_config, 1);
  30 }
  31
  32 #define TRACE(m...) do { if(trace) msg(L_DEBUG, "FB_ATOMIC: " m); } while(0)
  33
  34 struct fb_atomic_file {
  35   int fd;
  36   int use_count;
  37   int record_len;
  38   uns locked;
  39   byte name[1];
  40 };
  41
  42 void
  43 fbatomic_internal_write(struct fastbuf *f)
  44 {
  45   struct fb_atomic_file *af = FB_ATOMIC(f)->af;
  46   int size = f->bptr - f->buffer;
  47   if (size)
  48     {
  49       ASSERT(af->record_len < 0 || !(size % af->record_len));
  50       int res = write(af->fd, f->buffer, size);
  51       if (res < 0)
  52         die("Error writing %s: %m", f->name);
  53       if (res != size)
  54         die("Unexpected partial write to %s: written only %d bytes of %d", f->name, res, size);
  55       f->bptr = f->buffer;
  56     }
  57 }
  58
  59 static void
  60 fbatomic_spout(struct fastbuf *f)
  61 {
  62   if (f->bptr < f->bufend)              /* Explicit flushes should be ignored */
  63     return;
  64
  65   struct fb_atomic *F = FB_ATOMIC(f);
  66   if (F->af->locked)
  67     {
  68       uns written = f->bptr - f->buffer;
  69       uns size = f->bufend - f->buffer + F->slack_size;
  70       F->slack_size *= 2;
  71       TRACE("Reallocating buffer for atomic file %s with slack %d", f->name, F->slack_size);
  72       f->buffer = xrealloc(f->buffer, size);
  73       f->bufend = f->buffer + size;
  74       f->bptr = f->buffer + written;
  75       F->expected_max_bptr = f->bufend - F->slack_size;
  76     }
  77   else
  78     fbatomic_internal_write(f);
  79 }
  80
  81 static void
  82 fbatomic_close(struct fastbuf *f)
  83 {
  84   struct fb_atomic_file *af = FB_ATOMIC(f)->af;
  85   fbatomic_internal_write(f);   /* Need to flush explicitly, because the file can be locked */
  86   if (!--af->use_count)
  87     {
  88       close(af->fd);
  89       xfree(af);
  90     }
  91   xfree(f);
  92 }
  93
  94 struct fastbuf *
  95 fbatomic_open(const char *name, struct fastbuf *master, uns bufsize, int record_len)
  96 {
  97   struct fb_atomic *F = xmalloc_zero(sizeof(*F));
  98   struct fastbuf *f = &F->fb;
  99   struct fb_atomic_file *af;
 100   if (master)
 101     {
 102       af = FB_ATOMIC(master)->af;
 103       af->use_count++;
 104       ASSERT(af->record_len == record_len);
 105     }
 106   else
 107     {
 108       af = xmalloc_zero(sizeof(*af) + strlen(name));
 109       if ((af->fd = ucw_open(name, O_WRONLY | O_CREAT | O_TRUNC | O_APPEND, 0666)) < 0)
 110         die("Cannot create %s: %m", name);
 111       af->use_count = 1;
 112       af->record_len = record_len;
 113       af->locked = (record_len < 0);
 114       strcpy(af->name, name);
 115     }
 116   F->af = af;
 117   if (record_len > 0 && bufsize % record_len)
 118     bufsize += record_len - (bufsize % record_len);
 119   f->buffer = xmalloc(bufsize);
 120   f->bufend = f->buffer + bufsize;
 121   F->slack_size = (record_len < 0) ? -record_len : 0;
 122   ASSERT(bufsize > F->slack_size);
 123   F->expected_max_bptr = f->bufend - F->slack_size;
 124   f->bptr = f->bstop = f->buffer;
 125   f->name = af->name;
 126   f->spout = fbatomic_spout;
 127   f->close = fbatomic_close;
 128   return f;
 129 }
 130
 131 #ifdef TEST
 132
 133 int main(int argc UNUSED, char **argv UNUSED)
 134 {
 135   struct fastbuf *f, *g;
 136
 137   // Always trace in the test
 138   trace = 1;
 139
 140   msg(L_INFO, "Testing block writes");
 141   f = fbatomic_open("test", NULL, 16, 4);
 142   for (u32 i=0; i<17; i++)
 143     bwrite(f, &i, 4);
 144   bclose(f);
 145
 146   msg(L_INFO, "Testing interleaved var-size writes");
 147   f = fbatomic_open("test2", NULL, 23, -5);
 148   g = fbatomic_open("test2", f, 23, -5);
 149   for (int i=0; i<100; i++)
 150     {
 151       struct fastbuf *x = (i%2) ? g : f;
 152       bprintf(x, "%c<%d>\n", "fg"[i%2], ((259309*i) % 1000000) >> (i % 8));
 153       fbatomic_commit(x);
 154     }
 155   bclose(f);
 156   bclose(g);
 157
 158   return 0;
 159 }
 160
 161 #endif