bluesky/log.c

   1 /* Blue Sky: File Systems in the Cloud
   2  *
   3  * Copyright (C) 2010  The Regents of the University of California
   4  * Written by Michael Vrable <mvrable@cs.ucsd.edu>
   5  *
   6  * TODO: Licensing
   7  */
   8
   9 #define _GNU_SOURCE
  10 #define _ATFILE_SOURCE
  11
  12 #include <stdio.h>
  13 #include <stdint.h>
  14 #include <stdlib.h>
  15 #include <glib.h>
  16 #include <string.h>
  17 #include <errno.h>
  18 #include <sys/types.h>
  19 #include <sys/stat.h>
  20 #include <fcntl.h>
  21 #include <unistd.h>
  22 #include <sys/mman.h>
  23
  24 #include "bluesky-private.h"
  25
  26 /* The logging layer for BlueSky.  This is used to write filesystem changes
  27  * durably to disk so that they can be recovered in the event of a system
  28  * crash. */
  29
  30 /* The logging layer takes care out writing out a sequence of log records to
  31  * disk.  On disk, each record consists of a header, a data payload, and a
  32  * footer.  The footer contains a checksum of the record, meant to help with
  33  * identifying corrupt log records (we would assume because the log record was
  34  * only incompletely written out before a crash, which should only happen for
  35  * log records that were not considered committed). */
  36
  37 // Rough size limit for a log segment.  This is not a firm limit and there are
  38 // no absolute guarantees on the size of a log segment.
  39 #define LOG_SEGMENT_SIZE (1 << 23)
  40
  41 #define HEADER_MAGIC 0x676f4c0a
  42 #define FOOTER_MAGIC 0x2e435243
  43
  44 struct log_header {
  45     uint32_t magic;             // HEADER_MAGIC
  46     uint64_t offset;            // Starting byte offset of the log header
  47     uint32_t size;              // Size of the data item (bytes)
  48     BlueSkyCloudID id;          // Object identifier
  49 } __attribute__((packed));
  50
  51 struct log_footer {
  52     uint32_t magic;             // FOOTER_MAGIC
  53     uint32_t crc;               // Computed from log_header to log_footer.magic
  54 } __attribute__((packed));
  55
  56 static void writebuf(int fd, const char *buf, size_t len)
  57 {
  58     while (len > 0) {
  59         ssize_t written;
  60         written = write(fd, buf, len);
  61         if (written < 0 && errno == EINTR)
  62             continue;
  63         g_assert(written >= 0);
  64         buf += written;
  65         len -= written;
  66     }
  67 }
  68
  69 /* All log writes (at least for a single log) are made by one thread, so we
  70  * don't need to worry about concurrent access to the log file.  Log items to
  71  * write are pulled off a queue (and so may be posted by any thread).
  72  * fdatasync() is used to ensure the log items are stable on disk.
  73  *
  74  * The log is broken up into separate files, roughly of size LOG_SEGMENT_SIZE
  75  * each.  If a log segment is not currently open (log->fd is negative), a new
  76  * one is created.  Log segment filenames are assigned sequentially.
  77  *
  78  * Log replay ought to be implemented later, and ought to set the initial
  79  * sequence number appropriately.
  80  */
  81 static gpointer log_thread(gpointer d)
  82 {
  83     BlueSkyLog *log = (BlueSkyLog *)d;
  84
  85     /* If there are multiple log items to write, we may write more than one
  86      * before calling fsync().  The committed list is used to track all the
  87      * items that should be marked as committed once that final fsync() is
  88      * done. */
  89     GSList *committed = NULL;
  90
  91     int dirfd = open(log->log_directory, O_DIRECTORY);
  92     if (dirfd < 0) {
  93         fprintf(stderr, "Unable to open logging directory: %m\n");
  94         return NULL;
  95     }
  96
  97     while (TRUE) {
  98         if (log->fd < 0) {
  99             char logfile[64];
 100             g_snprintf(logfile, sizeof(logfile), "log-%08d", log->seq_num);
 101             log->fd = openat(dirfd, logfile, O_CREAT|O_WRONLY|O_EXCL, 0600);
 102             if (log->fd < 0 && errno == EEXIST) {
 103                 fprintf(stderr, "Log file %s already exists...\n", logfile);
 104                 log->seq_num++;
 105                 continue;
 106             } else if (log->fd < 0) {
 107                 fprintf(stderr, "Error opening logfile %s: %m\n", logfile);
 108                 return NULL;
 109             }
 110             fsync(log->fd);
 111             fsync(dirfd);
 112         }
 113
 114         BlueSkyCloudLog *item
 115             = (BlueSkyCloudLog *)g_async_queue_pop(log->queue);
 116         g_mutex_lock(item->lock);
 117         g_assert(item->data != NULL);
 118
 119         if ((item->location_flags | item->pending_write) & CLOUDLOG_JOURNAL) {
 120             g_mutex_unlock(item->lock);
 121             bluesky_cloudlog_unref(item);
 122             continue;
 123         }
 124
 125         item->pending_write |= CLOUDLOG_JOURNAL;
 126
 127         off_t logsize = lseek(log->fd, 0, SEEK_CUR);
 128         struct log_header header;
 129         struct log_footer footer;
 130
 131         header.magic = GUINT32_TO_LE(HEADER_MAGIC);
 132         header.offset = GUINT64_TO_LE(logsize);
 133         header.size = GUINT32_TO_LE(item->data->len);
 134         header.id = item->id;
 135         footer.magic = GUINT32_TO_LE(FOOTER_MAGIC);
 136
 137         uint32_t crc = BLUESKY_CRC32C_SEED;
 138
 139         writebuf(log->fd, (const char *)&header, sizeof(header));
 140         crc = crc32c(crc, (const char *)&header, sizeof(header));
 141
 142         writebuf(log->fd, item->data->data, item->data->len);
 143         crc = crc32c(crc, item->data->data, item->data->len);
 144
 145         crc = crc32c(crc, (const char *)&footer,
 146                      sizeof(footer) - sizeof(uint32_t));
 147         footer.crc = crc32c_finalize(crc);
 148         writebuf(log->fd, (const char *)&footer, sizeof(footer));
 149
 150         item->log_seq = log->seq_num;
 151         item->log_offset = logsize + sizeof(header);
 152         item->log_size = item->data->len;
 153
 154         logsize += sizeof(header) + sizeof(footer) + item->data->len;
 155
 156         committed  = g_slist_prepend(committed, item);
 157         g_mutex_unlock(item->lock);
 158
 159         /* Force an fsync either if we will be closing this log segment and
 160          * opening a new file, or if there are no other log items currently
 161          * waiting to be written. */
 162
 163         if (logsize >= LOG_SEGMENT_SIZE
 164             || g_async_queue_length(log->queue) <= 0)
 165         {
 166             int batchsize = 0;
 167             fdatasync(log->fd);
 168             while (committed != NULL) {
 169                 item = (BlueSkyCloudLog *)committed->data;
 170                 g_mutex_lock(item->lock);
 171                 item->pending_write &= ~CLOUDLOG_JOURNAL;
 172                 item->location_flags |= CLOUDLOG_JOURNAL;
 173                 g_cond_signal(item->cond);
 174                 g_mutex_unlock(item->lock);
 175                 committed = g_slist_delete_link(committed, committed);
 176                 batchsize++;
 177             }
 178             /* if (batchsize > 1)
 179                 g_print("Log batch size: %d\n", batchsize); */
 180         }
 181
 182         if (logsize < 0 || logsize >= LOG_SEGMENT_SIZE) {
 183             close(log->fd);
 184             log->fd = -1;
 185             log->seq_num++;
 186         }
 187     }
 188
 189     return NULL;
 190 }
 191
 192 BlueSkyLog *bluesky_log_new(const char *log_directory)
 193 {
 194     BlueSkyLog *log = g_new0(BlueSkyLog, 1);
 195
 196     log->log_directory = g_strdup(log_directory);
 197     log->fd = -1;
 198     log->seq_num = 0;
 199     log->queue = g_async_queue_new();
 200
 201     g_thread_create(log_thread, log, FALSE, NULL);
 202
 203     return log;
 204 }
 205
 206 void bluesky_log_item_submit(BlueSkyCloudLog *item, BlueSkyLog *log)
 207 {
 208     bluesky_cloudlog_ref(item);
 209     g_async_queue_push(log->queue, item);
 210 }
 211
 212 void bluesky_log_finish_all(GList *log_items)
 213 {
 214     while (log_items != NULL) {
 215         BlueSkyCloudLog *item = (BlueSkyCloudLog *)log_items->data;
 216
 217         g_mutex_lock(item->lock);
 218         while ((item->pending_write & CLOUDLOG_JOURNAL))
 219             g_cond_wait(item->cond, item->lock);
 220         g_mutex_unlock(item->lock);
 221         bluesky_cloudlog_unref(item);
 222
 223         log_items = g_list_delete_link(log_items, log_items);
 224     }
 225 }
 226
 227 /* Memory-map the given log object into memory (read-only) and return a pointer
 228  * to it.  FIXME: Use some type of cache, map entire log segments, and use
 229  * reference counting? */
 230 static int page_size = 0;
 231
 232 BlueSkyRCStr *bluesky_log_map_object(BlueSkyLog *log,
 233                                      int log_seq, int log_offset, int log_size)
 234 {
 235     if (page_size == 0) {
 236         page_size = getpagesize();
 237     }
 238
 239     int dirfd = open(log->log_directory, O_DIRECTORY);
 240     if (dirfd < 0) {
 241         fprintf(stderr, "Unable to open logging directory: %m\n");
 242         return NULL;
 243     }
 244
 245     char logfile[64];
 246     g_snprintf(logfile, sizeof(logfile), "log-%08d", log_seq);
 247     int fd = openat(dirfd, logfile, O_RDONLY);
 248     close(dirfd);
 249
 250     if (fd < 0) {
 251         fprintf(stderr, "Error opening logfile %s: %m\n", logfile);
 252         return NULL;
 253     }
 254
 255     off_t off_start, off_end;
 256     off_start = log_offset;
 257     off_end = off_start + log_size;
 258     off_start &= ~(page_size - 1);
 259     off_end = (off_end + (page_size - 1)) & (page_size - 1);
 260
 261     const char *ptr = (const char *)mmap(NULL, off_end - off_start, PROT_READ,
 262                                          MAP_SHARED, fd, off_start);
 263
 264     if (ptr == NULL) {
 265         fprintf(stderr, "Error mapping logfile: %m\n");
 266         close(fd);
 267         return NULL;
 268     }
 269
 270     close(fd);
 271
 272     BlueSkyMmap *mmap = g_new0(BlueSkyMmap, 1);
 273     mmap->addr = ptr;
 274     mmap->len = off_end - off_start;
 275     g_atomic_int_set(&mmap->refcount, 1);
 276
 277     return bluesky_string_new_from_mmap(mmap,
 278                                         log_offset - off_start, log_size);
 279 }