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 "configuration.h"
  17 #include "subscription.h"
  18 #include "tags.h"
  19 #include "util.h"
  20 #include "config.h"
  21
  22 #define CHECK_VALID_BLERG(r)                               \
  23         if (blerg == NULL) {                               \
  24                 fprintf(stderr, "Invalid struct blerg\n"); \
  25                 return r;                                  \
  26         }
  27
  28 int blerg_init() {
  29         if (!blerg_configuration_init()) {
  30                 return 0;
  31         }
  32         return 1;
  33 }
  34
  35 uint64_t blerg_get_record_count(struct blerg *blerg) {
  36         uint64_t count;
  37         flock(blerg->meta_fd, LOCK_SH);
  38         count = blerg->meta->sequence;
  39         flock(blerg->meta_fd, LOCK_UN);
  40         return count;
  41 }
  42
  43 /* Returns last usable record */
  44 uint64_t blerg_increment_record_count(struct blerg *blerg) {
  45         uint64_t count;
  46         flock(blerg->meta_fd, LOCK_EX);
  47         count = blerg->meta->sequence++;
  48         flock(blerg->meta_fd, LOCK_UN);
  49         return count;
  50 }
  51
  52 void blerg_segment_close(struct blerg *blerg) {
  53         if (blerg->data != NULL)
  54                 munmap((void *)blerg->data, blerg->data_size);
  55         if (blerg->data_fd != -1)
  56                 close(blerg->data_fd);
  57         if (blerg->index != NULL)
  58                 munmap((void *)blerg->index, RECORDS_PER_SEGMENT * sizeof(struct record));
  59         if (blerg->index_fd != -1)
  60                 close(blerg->index_fd);
  61 }
  62
  63 int blerg_remap_data(struct blerg *blerg) {
  64         struct stat st;
  65
  66         if (blerg->data != NULL)
  67                 munmap(blerg->data, blerg->data_size);
  68         fstat(blerg->data_fd, &st);
  69         blerg->data_size = st.st_size;
  70         if (blerg->data_size == 0) {
  71                 /* Can't map an empty data file. */
  72                 return 1;
  73         }
  74         blerg->data = (char *) mmap(NULL, blerg->data_size, PROT_READ, MAP_SHARED, blerg->data_fd, 0);
  75         if (blerg->data == MAP_FAILED) {
  76                 perror("Could not remap data");
  77                 return 0;
  78         }
  79         return 1;
  80 }
  81
  82 int blerg_segment_switch(struct blerg *blerg, int new_segment) {
  83         char filename[512];
  84         uint64_t max_sequence_no = blerg_get_record_count(blerg);
  85         struct stat st;
  86
  87         if (blerg->index != NULL && blerg->data != NULL && new_segment == blerg->current_segment) {
  88                 return 1;
  89         }
  90
  91         if (new_segment > max_sequence_no / RECORDS_PER_SEGMENT) {
  92                 fprintf(stderr, "Cannot switch to sequence beyond last record\n");
  93                 return 0;
  94         }
  95         if (new_segment < 0) {
  96                 fprintf(stderr, "Cannot switch to negative segment\n");
  97                 return 0;
  98         }
  99
 100         blerg_segment_close(blerg);
 101
 102         /* Load and map the index */
 103         snprintf(filename, 512, "%s/index%d", blerg->base_path, new_segment);
 104         blerg->index_fd = open(filename, O_RDWR | O_CREAT, 0600);
 105         if (blerg->index_fd == -1) {
 106                 perror("Could not open index");
 107                 goto open_failed_index_open;
 108         }
 109         flock(blerg->index_fd, LOCK_EX);
 110         fstat(blerg->index_fd, &st);
 111         if (st.st_size == 0) {
 112                 /* ftruncate() means never having to say you're sorry.  Sorry
 113                    in this case meaning "allocating disk space for a 1MB file
 114                    full or zeroes". */
 115                 ftruncate(blerg->index_fd, RECORDS_PER_SEGMENT * sizeof(struct record));
 116         }
 117         flock(blerg->index_fd, LOCK_UN);
 118
 119         blerg->index = (struct record *) mmap(NULL, RECORDS_PER_SEGMENT * sizeof(struct record), PROT_READ | PROT_WRITE, MAP_SHARED, blerg->index_fd, 0);
 120         if (blerg->index == MAP_FAILED) {
 121                 perror("Could not mmap index");
 122                 goto open_failed_index_mmap;
 123         }
 124
 125         /* Load data file */
 126         sprintf(filename, "%s/data%d", blerg->base_path, new_segment);
 127         blerg->data_fd = open(filename, O_RDWR | O_APPEND | O_CREAT, 0600);
 128         if (blerg->data_fd == -1) {
 129                 perror("Could not open data");
 130                 goto open_failed_data_open;
 131         }
 132
 133         if (!blerg_remap_data(blerg)) {
 134                 goto open_failed_data_mmap;
 135         }
 136
 137         blerg->current_segment = new_segment;
 138
 139         return 1;
 140
 141 open_failed_data_mmap:
 142         close(blerg->data_fd);
 143 open_failed_data_open:
 144         munmap((void *)blerg->index, RECORDS_PER_SEGMENT * sizeof(struct record));
 145 open_failed_index_mmap:
 146         close(blerg->index_fd);
 147 open_failed_index_open:
 148         return 0;
 149 }
 150
 151 int blerg_exists(const char *name) {
 152         int namelen = strlen(name);
 153         char filename[512];
 154
 155         if (!valid_name(name)) {
 156                 fprintf(stderr, "Invalid name\n");
 157                 return 0;
 158         }
 159
 160         snprintf(filename, 512, "%s/%s", blergconf.data_path, name);
 161         if (access(filename, F_OK) == -1)
 162                 return 0;
 163         else
 164                 return 1;
 165 }
 166
 167 struct blerg *blerg_open(const char *name) {
 168         int namelen = strlen(name);
 169         char filename[512];
 170         struct stat st;
 171         uint64_t sequence;
 172
 173         if (!valid_name(name)) {
 174                 fprintf(stderr, "Invalid name\n");
 175                 return NULL;
 176         }
 177         struct blerg *blerg = malloc(sizeof(struct blerg));
 178         if (!blerg) {
 179                 perror("Cannot allocate memory for blerg");
 180                 goto open_failed_blerg_malloc;
 181         }
 182         blerg->name = malloc(namelen + 1);
 183         memcpy(blerg->name, name, namelen + 1);
 184         blerg->meta_fd = blerg->index_fd = blerg->data_fd = -1;
 185         blerg->meta = NULL;
 186         blerg->index = NULL;
 187         blerg->data = NULL;
 188
 189         /* Make the directory if it doesn't exist */
 190         blerg->base_path = malloc(512);
 191         snprintf(blerg->base_path, 512, "%s/%s", blergconf.data_path, name);
 192         if (access(blerg->base_path, F_OK) == -1)
 193                 mkdir(blerg->base_path, 0755);
 194
 195         /* Open and map metadata */
 196         snprintf(filename, 512, "%s/meta", blerg->base_path);
 197         blerg->meta_fd = open(filename, O_RDWR | O_CREAT, 0600);
 198         if (blerg->meta_fd == -1) {
 199                 perror("Could not open metadata");
 200                 goto open_failed_meta_open;
 201         }
 202         fstat(blerg->meta_fd, &st);
 203         if (st.st_size < sizeof(struct meta)) {
 204                 /* Extend the file if sizeof(struct meta) is larger than the
 205                    file. This allows seamless upgrades as long as struct meta
 206                    only adds members. */
 207                 posix_fallocate(blerg->meta_fd, 0, sizeof(struct meta));
 208         }
 209         blerg->meta = (struct meta *) mmap(NULL, sizeof(struct meta), PROT_READ | PROT_WRITE, MAP_SHARED, blerg->meta_fd, 0);
 210         if (blerg->meta == MAP_FAILED) {
 211                 perror("Could not map metadata");
 212                 goto open_failed_meta_mmap;
 213         }
 214
 215         /* Open and map index and data for the current segment */
 216         blerg->current_segment = blerg_get_record_count(blerg) / RECORDS_PER_SEGMENT;
 217         if (!blerg_segment_switch(blerg, blerg->current_segment)) {
 218                 fprintf(stderr, "Could not switch segment\n");
 219                 goto open_failed_segment_switch;
 220         }
 221
 222         return blerg;
 223
 224 open_failed_segment_switch:
 225         munmap((void *)blerg->meta, sizeof(struct meta));
 226 open_failed_meta_mmap:
 227         close(blerg->meta_fd);
 228 open_failed_meta_open:
 229         free(blerg->name);
 230         free(blerg);
 231 open_failed_blerg_malloc:
 232         return NULL;
 233 }
 234
 235 int blerg_close(struct blerg *blerg) {
 236         CHECK_VALID_BLERG(0)
 237         blerg_segment_close(blerg);
 238         munmap((void *)blerg->meta, sizeof(struct meta));
 239         close(blerg->meta_fd);
 240         free(blerg->base_path);
 241         free(blerg->name);
 242         free(blerg);
 243         return 1;
 244 }
 245
 246 uint64_t blerg_store(struct blerg *blerg, const char *data, int length) {
 247         struct stat st;
 248         int n;
 249
 250         CHECK_VALID_BLERG(BLERG_INVALID_RECORD)
 251
 252         if (length > MAX_RECORD_SIZE || length <= 0) {
 253                 fprintf(stderr, "length out of bounds\n");
 254                 return BLERG_INVALID_RECORD;
 255         }
 256
 257         flock(blerg->index_fd, LOCK_EX);
 258         flock(blerg->data_fd, LOCK_EX);
 259
 260         uint64_t record = blerg_get_record_count(blerg);
 261         if (record == -1) {  /* Intentional signed-unsigned coercion */
 262                 fprintf(stderr, "Could not find free record\n");
 263                 return BLERG_INVALID_RECORD;
 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         /* Get the position for the new data */
 271         fstat(blerg->data_fd, &st);
 272         int curpos = st.st_size;
 273
 274         /* Write data to the data log */
 275         n = write(blerg->data_fd, data, length);
 276         if (n < length) {
 277                 perror("Could not write data");
 278                 /* Truncate anything we may have written */
 279                 ftruncate(blerg->data_fd, curpos);
 280                 return BLERG_INVALID_RECORD;
 281         }
 282
 283         /* Update the index */
 284         blerg->index[seg_rec].flags = 0x0001;
 285         blerg->index[seg_rec].offset = curpos;
 286         blerg->index[seg_rec].length = length;
 287         blerg->index[seg_rec].timestamp = time(NULL);
 288
 289         /* And finally increment the record count */
 290         blerg_increment_record_count(blerg);
 291
 292         flock(blerg->data_fd, LOCK_UN);
 293         flock(blerg->index_fd, LOCK_UN);
 294
 295         if (!blerg_get_mute(blerg)) {
 296                 /* Now do those dirty microblogging deeds */
 297                 tag_scan(blerg->name, data, length, record);
 298                 subscription_notify(blerg->name, record);
 299         }
 300
 301         return record;
 302 }
 303
 304 int blerg_fetch(struct blerg *blerg, uint64_t record, char **data, int *length) {
 305         CHECK_VALID_BLERG(0)
 306         if (record == BLERG_INVALID_RECORD || record >= blerg_get_record_count(blerg)) {
 307                 fprintf(stderr, "Invalid record\n");
 308                 return 0;
 309         }
 310         if (data == NULL || length == NULL) {
 311                 fprintf(stderr, "data or length is null\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         int rec_offset = blerg->index[seg_rec].offset;
 326         int rec_length = blerg->index[seg_rec].length;
 327         if (rec_offset >= blerg->data_size) {
 328                 /* We're accessing an out-of-bounds record in our mmap.  Remap
 329                    and recheck. */
 330                 if (!blerg_remap_data(blerg)) {
 331                         return 0;
 332                 }
 333                 if (rec_offset >= blerg->data_size) {
 334                         fprintf(stderr, "Record offset outside of data!?");
 335                         return 0;
 336                 }
 337         }
 338
 339         *data = malloc(rec_length);
 340         if (*data == NULL) {
 341                 perror("Could not allocate string in fetch");
 342                 return 0;
 343         }
 344
 345         memcpy(*data, blerg->data + rec_offset, rec_length);
 346
 347         *length = rec_length;
 348
 349         return 1;
 350 }
 351
 352 time_t blerg_get_timestamp(struct blerg *blerg, uint64_t record) {
 353         CHECK_VALID_BLERG(0)
 354         if (record == BLERG_INVALID_RECORD || record >= blerg_get_record_count(blerg)) {
 355                 fprintf(stderr, "Invalid record\n");
 356                 return 0;
 357         }
 358
 359         int segment = record / RECORDS_PER_SEGMENT;
 360         if (segment != blerg->current_segment)
 361                 blerg_segment_switch(blerg, segment);
 362         int seg_rec = record % RECORDS_PER_SEGMENT;
 363
 364         if ((blerg->index[seg_rec].flags & 0x1) == 0) {
 365                 fprintf(stderr, "Invalid record\n");
 366                 return 0;
 367         }
 368
 369         return blerg->index[seg_rec].timestamp;
 370 }
 371
 372 int blerg_set_subscription_mark(struct blerg *blerg) {
 373         CHECK_VALID_BLERG(0)
 374         blerg->meta->subscription_mark = subscription_count_items(blerg->name);
 375         return 1;
 376 }
 377
 378 uint64_t blerg_get_subscription_mark(struct blerg *blerg) {
 379         CHECK_VALID_BLERG(0)
 380         return blerg->meta->subscription_mark;
 381 }
 382
 383 int blerg_set_mute(struct blerg *blerg, int v) {
 384         CHECK_VALID_BLERG(0)
 385         if (v) {
 386                 blerg->meta->status |= BLERGMETA_MUTED;
 387         } else {
 388                 blerg->meta->status &= ~BLERGMETA_MUTED;
 389         }
 390         return 1;
 391 }
 392
 393 int blerg_get_mute(struct blerg *blerg) {
 394         CHECK_VALID_BLERG(0)
 395         return (blerg->meta->status & BLERGMETA_MUTED) > 0;
 396 }