More aggressively use memory-mapped data for cloud log items.

[bluesky.git] / bluesky / log.c

diff --git a/bluesky/log.c b/bluesky/log.c
index 9b4b62b..99b5ae5 100644 (file)
--- a/bluesky/log.c
+++ b/bluesky/log.c
@@ -36,7 +36,7 @@
 
 // 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
@@ -77,11 +77,14 @@ static void log_commit(BlueSkyLog *log)
     while (log->committed != NULL) {
         BlueSkyCloudLog *item = (BlueSkyCloudLog *)log->committed->data;
         g_mutex_lock(item->lock);
+        bluesky_cloudlog_stats_update(item, -1);
         item->pending_write &= ~CLOUDLOG_JOURNAL;
         item->location_flags |= CLOUDLOG_JOURNAL;
+        bluesky_cloudlog_stats_update(item, 1);
         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++;
     }
 
@@ -153,10 +156,13 @@ static gpointer log_thread(gpointer d)
         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;
         }
 
+        bluesky_cloudlog_stats_update(item, -1);
         item->pending_write |= CLOUDLOG_JOURNAL;
+        bluesky_cloudlog_stats_update(item, 1);
 
         struct log_header header;
         struct log_footer footer;
@@ -198,7 +204,16 @@ static gpointer log_thread(gpointer d)
 
         offset += sizeof(header) + sizeof(footer) + item->data->len;
 
+        /* Replace the log item's string data with a memory-mapped copy of the
+         * data, now that it has been written to the log file.  (Even if it
+         * isn't yet on disk, it should at least be in the page cache and so
+         * available to memory map.) */
+        bluesky_string_unref(item->data);
+        item->data = NULL;
+        bluesky_cloudlog_fetch(item);
+
         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
@@ -235,6 +250,7 @@ BlueSkyLog *bluesky_log_new(const char *log_directory)
 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);
 }
 
@@ -282,13 +298,17 @@ BlueSkyRCStr *bluesky_log_map_object(BlueSkyLog *log,
         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);
     }
 
@@ -296,3 +316,31 @@ BlueSkyRCStr *bluesky_log_map_object(BlueSkyLog *log,
 
     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);
+    }
+}
+