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.
11 #include <sys/types.h>
16 #include "configuration.h"
17 #include "subscription.h"
22 #define CHECK_VALID_BLERG(r) \
23 if (blerg == NULL) { \
24 fprintf(stderr, "Invalid struct blerg\n"); \
29 if (!blerg_configuration_init()) {
35 uint64_t blerg_get_record_count(struct blerg *blerg) {
37 flock(blerg->meta_fd, LOCK_SH);
38 count = blerg->meta->sequence;
39 flock(blerg->meta_fd, LOCK_UN);
43 /* Returns last usable record */
44 uint64_t blerg_increment_record_count(struct blerg *blerg) {
46 flock(blerg->meta_fd, LOCK_EX);
47 count = blerg->meta->sequence++;
48 flock(blerg->meta_fd, LOCK_UN);
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);
63 int blerg_remap_data(struct blerg *blerg) {
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. */
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");
82 int blerg_segment_switch(struct blerg *blerg, int new_segment) {
84 uint64_t max_sequence_no = blerg_get_record_count(blerg);
87 if (blerg->index != NULL && blerg->data != NULL && new_segment == blerg->current_segment) {
91 if (new_segment > max_sequence_no / RECORDS_PER_SEGMENT) {
92 fprintf(stderr, "Cannot switch to sequence beyond last record\n");
95 if (new_segment < 0) {
96 fprintf(stderr, "Cannot switch to negative segment\n");
100 blerg_segment_close(blerg);
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;
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
115 ftruncate(blerg->index_fd, RECORDS_PER_SEGMENT * sizeof(struct record));
117 flock(blerg->index_fd, LOCK_UN);
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;
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;
133 if (!blerg_remap_data(blerg)) {
134 goto open_failed_data_mmap;
137 blerg->current_segment = new_segment;
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:
151 int blerg_exists(const char *name) {
152 int namelen = strlen(name);
155 if (!valid_name(name)) {
156 fprintf(stderr, "Invalid name\n");
160 snprintf(filename, 512, "%s/%s", blergconf.data_path, name);
161 if (access(filename, F_OK) == -1)
167 struct blerg *blerg_open(const char *name) {
168 int namelen = strlen(name);
173 if (!valid_name(name)) {
174 fprintf(stderr, "Invalid name\n");
177 struct blerg *blerg = malloc(sizeof(struct blerg));
179 perror("Cannot allocate memory for blerg");
180 goto open_failed_blerg_malloc;
182 blerg->name = malloc(namelen + 1);
183 memcpy(blerg->name, name, namelen + 1);
184 blerg->meta_fd = blerg->index_fd = blerg->data_fd = -1;
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);
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;
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));
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;
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;
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:
231 open_failed_blerg_malloc:
235 int blerg_close(struct blerg *blerg) {
237 blerg_segment_close(blerg);
238 munmap((void *)blerg->meta, sizeof(struct meta));
239 close(blerg->meta_fd);
240 free(blerg->base_path);
246 uint64_t blerg_store(struct blerg *blerg, const char *data, int length) {
250 CHECK_VALID_BLERG(BLERG_INVALID_RECORD)
252 if (length > MAX_RECORD_SIZE || length <= 0) {
253 fprintf(stderr, "length out of bounds\n");
254 return BLERG_INVALID_RECORD;
257 flock(blerg->index_fd, LOCK_EX);
258 flock(blerg->data_fd, LOCK_EX);
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;
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;
270 /* Get the position for the new data */
271 fstat(blerg->data_fd, &st);
272 int curpos = st.st_size;
274 /* Write data to the data log */
275 n = write(blerg->data_fd, data, 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;
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);
289 /* And finally increment the record count */
290 blerg_increment_record_count(blerg);
292 flock(blerg->data_fd, LOCK_UN);
293 flock(blerg->index_fd, LOCK_UN);
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);
304 int blerg_fetch(struct blerg *blerg, uint64_t record, char **data, int *length) {
306 if (record == BLERG_INVALID_RECORD || record >= blerg_get_record_count(blerg)) {
307 fprintf(stderr, "Invalid record\n");
310 if (data == NULL || length == NULL) {
311 fprintf(stderr, "data or length is null\n");
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;
320 if ((blerg->index[seg_rec].flags & 0x1) == 0) {
321 fprintf(stderr, "Invalid record\n");
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
330 if (!blerg_remap_data(blerg)) {
333 if (rec_offset >= blerg->data_size) {
334 fprintf(stderr, "Record offset outside of data!?");
339 *data = malloc(rec_length);
341 perror("Could not allocate string in fetch");
345 memcpy(*data, blerg->data + rec_offset, rec_length);
347 *length = rec_length;
352 time_t blerg_get_timestamp(struct blerg *blerg, uint64_t record) {
354 if (record == BLERG_INVALID_RECORD || record >= blerg_get_record_count(blerg)) {
355 fprintf(stderr, "Invalid record\n");
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;
364 if ((blerg->index[seg_rec].flags & 0x1) == 0) {
365 fprintf(stderr, "Invalid record\n");
369 return blerg->index[seg_rec].timestamp;
372 int blerg_set_subscription_mark(struct blerg *blerg) {
374 blerg->meta->subscription_mark = subscription_count_items(blerg->name);
378 uint64_t blerg_get_subscription_mark(struct blerg *blerg) {
380 return blerg->meta->subscription_mark;
383 int blerg_set_mute(struct blerg *blerg, int v) {
386 blerg->meta->status |= BLERGMETA_MUTED;
388 blerg->meta->status &= ~BLERGMETA_MUTED;
393 int blerg_get_mute(struct blerg *blerg) {
395 return (blerg->meta->status & BLERGMETA_MUTED) > 0;