database/database.c

   1 /* Blerg is (C) 2011 The Dominion of Awesome, and is distributed under a
   2  * BSD-style license.  Please see the COPYING file for details.
   3  */
   4 #include <stdint.h>
   5 #include <stdlib.h>
   6 #include <string.h>
   7 #include <stdio.h>
   8 #include <time.h>
   9 #include <unistd.h>
  10 #include <sys/stat.h>
  11 #include <sys/types.h>
  12 #include <sys/mman.h>
  13 #include <sys/file.h>
  14 #include <fcntl.h>
  15 #include "database.h"
  16 #include "subscription.h"
  17 #include "util.h"
  18 #include "config.h"
  19
  20 uint64_t blerg_get_record_count(struct blerg *blerg) {
  21         uint64_t count;
  22         flock(blerg->meta_fd, LOCK_SH);
  23         count = blerg->meta->sequence;
  24         flock(blerg->meta_fd, LOCK_UN);
  25         return count;
  26 }
  27
  28 /* Returns last usable record */
  29 uint64_t blerg_increment_record_count(struct blerg *blerg) {
  30         uint64_t count;
  31         flock(blerg->meta_fd, LOCK_EX);
  32         count = blerg->meta->sequence++;
  33         flock(blerg->meta_fd, LOCK_UN);
  34         return count;
  35 }
  36
  37 void blerg_segment_close(struct blerg *blerg) {
  38         if (blerg->data != NULL)
  39                 munmap((void *)blerg->data, blerg->data_size);
  40         if (blerg->data_fd != -1)
  41                 close(blerg->data_fd);
  42         if (blerg->index != NULL)
  43                 munmap((void *)blerg->index, RECORDS_PER_SEGMENT * sizeof(struct record));
  44         if (blerg->index_fd != -1)
  45                 close(blerg->index_fd);
  46 }
  47
  48 int blerg_segment_switch(struct blerg *blerg, int new_segment) {
  49         char filename[512];
  50         uint64_t max_sequence = blerg_get_record_count(blerg);
  51         struct stat st;
  52
  53         if (new_segment > max_sequence / RECORDS_PER_SEGMENT) {
  54                 fprintf(stderr, "Cannot switch to sequence beyond last record\n");
  55                 return 0;
  56         }
  57
  58         blerg_segment_close(blerg);
  59
  60         /* Load and map the index */
  61         snprintf(filename, 512, "%s/index%d", blerg->base_path, new_segment);
  62         blerg->index_fd = open(filename, O_RDWR | O_CREAT, 0600);
  63         if (blerg->index_fd == -1) {
  64                 perror("Could not open index");
  65                 goto open_failed_index_open;
  66         }
  67         flock(blerg->index_fd, LOCK_EX);
  68         fstat(blerg->index_fd, &st);
  69         if (st.st_size == 0) {
  70                 int i;
  71                 struct record r;
  72                 memset((void *)&r, 0, sizeof(struct record));
  73                 for (i = 0; i < RECORDS_PER_SEGMENT; i++) {
  74                         write(blerg->index_fd, &r, sizeof(struct record));
  75                 }
  76         }
  77         flock(blerg->index_fd, LOCK_UN);
  78
  79         blerg->index = (struct record *) mmap(NULL, RECORDS_PER_SEGMENT * sizeof(struct record), PROT_READ | PROT_WRITE, MAP_SHARED, blerg->index_fd, 0);
  80         if (blerg->index == MAP_FAILED) {
  81                 perror("Could not mmap index");
  82                 goto open_failed_index_mmap;
  83         }
  84
  85         /* Load data file */
  86         sprintf(filename, "%s/data%d", blerg->base_path, new_segment);
  87         blerg->data_fd = open(filename, O_RDWR | O_APPEND | O_CREAT, 0600);
  88         fstat(blerg->data_fd, &st);
  89         blerg->data_size = st.st_size;
  90         if (blerg->data_fd == -1) {
  91                 perror("Could not open data");
  92                 goto open_failed_data_open;
  93         }
  94
  95         if (blerg->data_size > 0) {
  96                 blerg->data = (char *) mmap(NULL, blerg->data_size, PROT_READ, MAP_SHARED, blerg->data_fd, 0);
  97                 if (blerg->data == MAP_FAILED) {
  98                         perror("Could not mmap data");
  99                         goto open_failed_data_mmap;
 100                 }
 101         }
 102
 103         return 1;
 104
 105 open_failed_data_mmap:
 106         close(blerg->data_fd);
 107 open_failed_data_open:
 108         munmap((void *)blerg->index, sizeof(RECORDS_PER_SEGMENT * sizeof(struct record)));
 109 open_failed_index_mmap:
 110         close(blerg->index_fd);
 111 open_failed_index_open:
 112         return 0;
 113 }
 114
 115 int blerg_exists(const char *name) {
 116         int namelen = strlen(name);
 117         char filename[512];
 118
 119         if (!valid_name(name)) {
 120                 fprintf(stderr, "Invalid name\n");
 121                 return 0;
 122         }
 123
 124         snprintf(filename, 512, "%s/%s", DATA_PATH, name);
 125         if (access(filename, F_OK) == -1)
 126                 return 0;
 127         else
 128                 return 1;
 129 }
 130
 131 struct blerg *blerg_open(const char *name) {
 132         int namelen = strlen(name);
 133         char filename[512];
 134         struct stat st;
 135         uint64_t sequence;
 136
 137         if (!valid_name(name)) {
 138                 fprintf(stderr, "Invalid name\n");
 139                 return NULL;
 140         }
 141         struct blerg *blerg = malloc(sizeof(struct blerg));
 142         if (!blerg) {
 143                 perror("Cannot allocate memory for blerg");
 144                 goto open_failed_blerg_malloc;
 145         }
 146         blerg->name = malloc(namelen + 1);
 147         memcpy(blerg->name, name, namelen + 1);
 148         blerg->meta_fd = blerg->index_fd = blerg->data_fd = -1;
 149         blerg->meta = NULL;
 150         blerg->index = NULL;
 151         blerg->data = NULL;
 152
 153         /* Make the directory if it doesn't exist */
 154         blerg->base_path = malloc(512);
 155         snprintf(blerg->base_path, 512, "%s/%s", DATA_PATH, name);
 156         if (access(blerg->base_path, F_OK) == -1)
 157                 mkdir(blerg->base_path, 0755);
 158
 159         /* Open and map metadata */
 160         snprintf(filename, 512, "%s/meta", blerg->base_path);
 161         blerg->meta_fd = open(filename, O_RDWR | O_CREAT, 0600);
 162         if (blerg->meta_fd == -1) {
 163                 perror("Could not open metadata");
 164                 goto open_failed_meta_open;
 165         }
 166         fstat(blerg->meta_fd, &st);
 167         if (st.st_size == 0) {
 168                 char *buf = (char *) malloc(sizeof(struct meta));
 169                 memset(buf, 0, sizeof(struct meta));
 170                 write(blerg->meta_fd, buf, sizeof(struct meta));
 171                 free(buf);
 172         }
 173         blerg->meta = (struct meta *) mmap(NULL, sizeof(struct meta), PROT_READ | PROT_WRITE, MAP_SHARED, blerg->meta_fd, 0);
 174         if (blerg->meta == MAP_FAILED) {
 175                 perror("Could not map metadata");
 176                 goto open_failed_meta_mmap;
 177         }
 178
 179         /* Open and map index and data for the current segment */
 180         blerg->current_segment = blerg_get_record_count(blerg) / RECORDS_PER_SEGMENT;
 181         if (!blerg_segment_switch(blerg, blerg->current_segment)) {
 182                 fprintf(stderr, "Could not switch segment\n");
 183                 goto open_failed_segment_switch;
 184         }
 185
 186         return blerg;
 187
 188 open_failed_segment_switch:
 189         munmap((void *)blerg->meta, sizeof(struct meta));
 190 open_failed_meta_mmap:
 191         close(blerg->meta_fd);
 192 open_failed_meta_open:
 193         free(blerg->name);
 194         free(blerg);
 195 open_failed_blerg_malloc:
 196         return NULL;
 197 }
 198
 199 int blerg_close(struct blerg *blerg) {
 200         blerg_segment_close(blerg);
 201         munmap((void *)blerg->meta, sizeof(struct meta));
 202         close(blerg->meta_fd);
 203         free(blerg->base_path);
 204         free(blerg->name);
 205         free(blerg);
 206         return 1;
 207 }
 208
 209 int blerg_store(struct blerg *blerg, const char *data, int len) {
 210         if (len > MAX_RECORD_SIZE) {
 211                 fprintf(stderr, "len > 64K\n");
 212                 return -1;
 213         }
 214
 215         flock(blerg->index_fd, LOCK_EX);
 216         flock(blerg->data_fd, LOCK_EX);
 217
 218         uint64_t record = blerg_increment_record_count(blerg);
 219         if (record == -1) {
 220                 fprintf(stderr, "Could not find free record\n");
 221                 return -1;
 222         }
 223         int segment = record / RECORDS_PER_SEGMENT;
 224         if (segment != blerg->current_segment)
 225                 blerg_segment_switch(blerg, segment);
 226         int seg_rec = record % RECORDS_PER_SEGMENT;
 227
 228         /* Get the position for the new data */
 229         FILE *datafile = fdopen(dup(blerg->data_fd), "a");
 230         fseek(datafile, 0, SEEK_END);
 231         int curpos = ftell(datafile);
 232         fclose(datafile);
 233
 234         int bytes = 0;
 235         do {
 236                 int n = write(blerg->data_fd, data + bytes, len);
 237                 if (n == -1) {
 238                         perror("Could not write data");
 239                         /* Truncate anything we may have written */
 240                         ftruncate(blerg->data_fd, curpos);
 241                         return -1;
 242                 }
 243                 bytes += n;
 244         } while (bytes < len);
 245         blerg->index[seg_rec].flags = 0x0001;
 246         blerg->index[seg_rec].offset = curpos;
 247         blerg->index[seg_rec].length = len;
 248         blerg->index[seg_rec].timestamp = time(NULL);
 249
 250         flock(blerg->data_fd, LOCK_UN);
 251         flock(blerg->index_fd, LOCK_UN);
 252
 253         tag_scan(blerg->name, data, len, record);
 254         subscription_notify(blerg->name, record);
 255
 256         return record;
 257 }
 258
 259 int blerg_fetch(struct blerg *blerg, int record, char **data, int *length) {
 260         if (record < 0) {
 261                 fprintf(stderr, "Invalid record\n");
 262                 return 0;
 263         }
 264
 265         int segment = record / RECORDS_PER_SEGMENT;
 266         if (segment != blerg->current_segment)
 267                 blerg_segment_switch(blerg, segment);
 268         int seg_rec = record % RECORDS_PER_SEGMENT;
 269
 270         if ((blerg->index[seg_rec].flags & 0x1) == 0) {
 271                 fprintf(stderr, "Invalid record\n");
 272                 return 0;
 273         }
 274
 275         int rec_offset = blerg->index[seg_rec].offset;
 276         int rec_length = blerg->index[seg_rec].length;
 277         if (rec_offset >= blerg->data_size) {
 278                 /* We're accessing an out-of-bounds record in our mmap.
 279                    Recheck size and remap. */
 280                 struct stat st;
 281                 fstat(blerg->data_fd, &st);
 282                 blerg->data_size = st.st_size;
 283                 if (rec_offset > blerg->data_size) {
 284                         fprintf(stderr, "Record offset outside of data!?");
 285                         return 0;
 286                 }
 287
 288                 munmap(blerg->data, blerg->data_size);
 289                 blerg->data = (char *) mmap(NULL, blerg->data_size, PROT_READ, MAP_SHARED, blerg->data_fd, 0);
 290                 if (blerg->data == MAP_FAILED) {
 291                         perror("Could not remap data");
 292                         return 0;
 293                 }
 294         }
 295
 296         *data = malloc(rec_length);
 297         if (*data == NULL) {
 298                 perror("Could not allocate string in fetch");
 299                 return 0;
 300         }
 301
 302         memcpy(*data, blerg->data + rec_offset, rec_length);
 303
 304         *length = rec_length;
 305
 306         return 1;
 307 }
 308
 309 time_t blerg_get_timestamp(struct blerg *blerg, int record) {
 310         if (record < 0) {
 311                 fprintf(stderr, "Invalid record\n");
 312                 return 0;
 313         }
 314
 315         int segment = record / RECORDS_PER_SEGMENT;
 316         if (segment != blerg->current_segment)
 317                 blerg_segment_switch(blerg, segment);
 318         int seg_rec = record % RECORDS_PER_SEGMENT;
 319
 320         if ((blerg->index[seg_rec].flags & 0x1) == 0) {
 321                 fprintf(stderr, "Invalid record\n");
 322                 return 0;
 323         }
 324
 325         return blerg->index[seg_rec].timestamp;
 326 }