// Rough size limit for a log segment. This is not a firm limit and there are
// no absolute guarantees on the size of a log segment.
-#define LOG_SEGMENT_SIZE (1 << 23)
+#define LOG_SEGMENT_SIZE (1 << 24)
#define HEADER_MAGIC 0x676f4c0a
#define FOOTER_MAGIC 0x2e435243
g_cond_signal(item->cond);
g_mutex_unlock(item->lock);
log->committed = g_slist_delete_link(log->committed, log->committed);
+ bluesky_cloudlog_unref(item);
batchsize++;
}
if ((item->location_flags | item->pending_write) & CLOUDLOG_JOURNAL) {
g_mutex_unlock(item->lock);
bluesky_cloudlog_unref(item);
+ g_atomic_int_add(&item->data_lock_count, -1);
continue;
}
offset += sizeof(header) + sizeof(footer) + item->data->len;
log->committed = g_slist_prepend(log->committed, item);
+ g_atomic_int_add(&item->data_lock_count, -1);
g_mutex_unlock(item->lock);
/* Force an if there are no other log items currently waiting to be
void bluesky_log_item_submit(BlueSkyCloudLog *item, BlueSkyLog *log)
{
bluesky_cloudlog_ref(item);
+ g_atomic_int_add(&item->data_lock_count, 1);
g_async_queue_push(log->queue, item);
}
map = g_new0(BlueSkyMmap, 1);
off_t length = lseek(fd, 0, SEEK_END);
+ map->log_seq = log_seq;
map->addr = (const char *)mmap(NULL, length, PROT_READ, MAP_SHARED,
fd, 0);
map->len = length;
- g_atomic_int_set(&map->refcount, 1);
+ map->log = log;
+ g_atomic_int_set(&map->refcount, 0);
g_hash_table_insert(log->mmap_cache, GINT_TO_POINTER(log_seq), map);
+ g_print("Mapped log segment %d...\n", log_seq);
+
close(fd);
}
return bluesky_string_new_from_mmap(map, log_offset, log_size);
}
+
+void bluesky_mmap_unref(BlueSkyMmap *mmap)
+{
+ if (mmap == NULL)
+ return;
+
+ if (g_atomic_int_dec_and_test(&mmap->refcount)) {
+ /* There is a potential race condition here: the BlueSkyLog contains a
+ * hash table of currently-existing BlueSkyMmap objects, which does not
+ * hold a reference. Some other thread might grab a new reference to
+ * this object after reading it from the hash table. So, before
+ * destruction we need to grab the lock for the hash table, then check
+ * the reference count again. If it is still zero, we can proceed with
+ * object destruction. */
+ BlueSkyLog *log = mmap->log;
+ g_mutex_lock(log->mmap_lock);
+ if (g_atomic_int_get(&mmap->refcount) > 0) {
+ g_mutex_unlock(log->mmap_lock);
+ return;
+ }
+
+ g_hash_table_remove(log->mmap_cache, GINT_TO_POINTER(mmap->log_seq));
+ munmap((void *)mmap->addr, mmap->len);
+ g_free(mmap);
+ g_mutex_unlock(log->mmap_lock);
+ }
+}
+