Add write throttling based on the size of the uncommitted journal
[bluesky.git] / bluesky / bluesky-private.h
index fccc7a3..2595367 100644 (file)
@@ -14,6 +14,7 @@
 #define _BLUESKY_PRIVATE_H
 
 #include "bluesky.h"
+#include <stdlib.h>
 
 #ifdef __cplusplus
 extern "C" {
@@ -58,6 +59,34 @@ void bluesky_serialize_cloudlog(BlueSkyCloudLog *log,
                                 GString *authenticated,
                                 GString *writable);
 
+/* Cryptographic operations. */
+#define CRYPTO_BLOCK_SIZE 16        /* 128-bit AES */
+#define CRYPTO_KEY_SIZE   16
+#define CRYPTO_HASH_SIZE  32        /* SHA-256 */
+
+typedef struct BlueSkyCryptKeys {
+    uint8_t encryption_key[CRYPTO_KEY_SIZE];
+    uint8_t authentication_key[CRYPTO_HASH_SIZE];
+} BlueSkyCryptKeys;
+
+void bluesky_crypt_init();
+void bluesky_crypt_hash_key(const char *keystr, uint8_t *out);
+void bluesky_crypt_random_bytes(guchar *buf, gint len);
+void bluesky_crypt_derive_keys(BlueSkyCryptKeys *keys, const gchar *master);
+BlueSkyRCStr *bluesky_crypt_encrypt(BlueSkyRCStr *in, const uint8_t *key);
+BlueSkyRCStr *bluesky_crypt_decrypt(BlueSkyRCStr *in, const uint8_t *key);
+
+void bluesky_crypt_block_encrypt(gchar *cloud_block, size_t len,
+                                 BlueSkyCryptKeys *keys);
+gboolean bluesky_crypt_block_decrypt(gchar *cloud_block, size_t len,
+                                     BlueSkyCryptKeys *keys,
+                                     gboolean allow_unauth);
+void bluesky_cloudlog_encrypt(GString *segment, BlueSkyCryptKeys *keys);
+void bluesky_cloudlog_decrypt(char *segment, size_t len,
+                              BlueSkyCryptKeys *keys,
+                              BlueSkyRangeset *items,
+                              gboolean allow_unauth);
+
 /* Storage layer.  Requests can be performed asynchronously, so these objects
  * help keep track of operations in progress. */
 typedef enum {
@@ -76,8 +105,8 @@ typedef enum {
 } BlueSkyAsyncStatus;
 
 struct BlueSkyNotifierList;
-typedef struct _BlueSkyStoreAsync BlueSkyStoreAsync;
-struct _BlueSkyStoreAsync {
+typedef struct BlueSkyStoreAsync BlueSkyStoreAsync;
+struct BlueSkyStoreAsync {
     BlueSkyStore *store;
 
     GMutex *lock;
@@ -91,6 +120,14 @@ struct _BlueSkyStoreAsync {
     gchar *key;                 /* Key to read/write */
     BlueSkyRCStr *data;         /* Data read/to write */
 
+    /* For range requests on reads: starting byte offset and length; len 0
+     * implies reading to the end of the object.  At completion, the backend
+     * should set range_done if a range read was made; if not set the entire
+     * object was read and the storage layer will select out just the
+     * appropriate bytes. */
+    size_t start, len;
+    gboolean range_done;
+
     int result;                 /* Result code; 0 for success. */
     struct BlueSkyNotifierList *notifiers;
     gint notifier_count;
@@ -104,6 +141,10 @@ struct _BlueSkyStoreAsync {
     bluesky_time_hires exec_time;   /* Time processing started on operation. */
 
     gpointer store_private;     /* For use by the storage implementation */
+
+    /* If storage operations should be charged to any particular profile, which
+     * one? */
+    BlueSkyProfile *profile;
 };
 
 /* Support for notification lists.  These are lists of one-shot functions which
@@ -157,7 +198,7 @@ void bluesky_store_add_barrier(BlueSkyStoreAsync *barrier,
 
 void bluesky_inode_start_sync(BlueSkyInode *inode);
 
-void bluesky_block_touch(BlueSkyInode *inode, uint64_t i);
+void bluesky_block_touch(BlueSkyInode *inode, uint64_t i, gboolean preserve);
 void bluesky_block_fetch(BlueSkyInode *inode, BlueSkyBlock *block,
                          BlueSkyStoreAsync *barrier);
 void bluesky_block_flush(BlueSkyInode *inode, BlueSkyBlock *block,
@@ -169,6 +210,12 @@ void bluesky_file_drop_cached(BlueSkyInode *inode);
  * various pieces of data (both where in the cloud and where cached locally).
  * */
 
+/* Eventually we'll want to support multiple writers.  But for now, hard-code
+ * separate namespaces in the cloud for the proxy and the cleaner to write to.
+ * */
+#define BLUESKY_CLOUD_DIR_PRIMARY 0
+#define BLUESKY_CLOUD_DIR_CLEANER 1
+
 typedef struct {
     char bytes[16];
 } BlueSkyCloudID;
@@ -221,15 +268,21 @@ struct cloudlog_header {
     uint32_t size1, size2, size3;
 } __attribute__((packed));
 
+// 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 << 22)
+
 #define JOURNAL_MAGIC "\nLog"
 #define CLOUDLOG_MAGIC "AgI-"
+#define CLOUDLOG_MAGIC_ENCRYPTED "AgI="     // CLOUDLOG_MAGIC[3] ^= 0x10
 
 /* A record which tracks an object which has been written to a local log,
  * cached, locally, and/or written to the cloud. */
 #define CLOUDLOG_JOURNAL    0x01
 #define CLOUDLOG_CLOUD      0x02
 #define CLOUDLOG_CACHE      0x04
-struct _BlueSkyCloudLog {
+#define CLOUDLOG_UNCOMMITTED 0x10
+struct BlueSkyCloudLog {
     gint refcount;
     GMutex *lock;
     GCond *cond;
@@ -273,12 +326,22 @@ struct _BlueSkyCloudLog {
 };
 
 /* Serialize objects into a log segment to be written to the cloud. */
-struct _BlueSkyCloudLogState {
+struct BlueSkyCloudLogState {
     GString *data;
     BlueSkyCloudPointer location;
     GList *inode_list;
     GSList *writeback_list;     // Items which are being serialized right now
     GList *pending_segments;    // Segments which are being uploaded now
+
+    int uploads_pending;        // Count of uploads in progress, not completed
+    GMutex *uploads_pending_lock;
+    GCond *uploads_pending_cond;
+
+    /* What is the most recent sequence number written by the cleaner which we
+     * have processed and incorporated into our own log?  This gets
+     * incorporated into the version vector written out with our checkpoint
+     * records. */
+    int latest_cleaner_seq_seen;
 };
 
 gboolean bluesky_cloudlog_equal(gconstpointer a, gconstpointer b);
@@ -286,15 +349,19 @@ guint bluesky_cloudlog_hash(gconstpointer a);
 BlueSkyCloudLog *bluesky_cloudlog_new(BlueSkyFS *fs, const BlueSkyCloudID *id);
 gchar *bluesky_cloudlog_id_to_string(BlueSkyCloudID id);
 BlueSkyCloudID bluesky_cloudlog_id_from_string(const gchar *idstr);
+void bluesky_cloudlog_threads_init(BlueSkyFS *fs);
 void bluesky_cloudlog_ref(BlueSkyCloudLog *log);
 void bluesky_cloudlog_unref(BlueSkyCloudLog *log);
+void bluesky_cloudlog_unref_delayed(BlueSkyCloudLog *log);
 void bluesky_cloudlog_erase(BlueSkyCloudLog *log);
 void bluesky_cloudlog_stats_update(BlueSkyCloudLog *log, int type);
 void bluesky_cloudlog_sync(BlueSkyCloudLog *log);
 void bluesky_cloudlog_insert(BlueSkyCloudLog *log);
 void bluesky_cloudlog_insert_locked(BlueSkyCloudLog *log);
 BlueSkyCloudLog *bluesky_cloudlog_get(BlueSkyFS *fs, BlueSkyCloudID id);
+void bluesky_cloudlog_prefetch(BlueSkyCloudLog *log);
 void bluesky_cloudlog_fetch(BlueSkyCloudLog *log);
+void bluesky_cloudlog_background_fetch(BlueSkyCloudLog *item);
 BlueSkyCloudPointer bluesky_cloudlog_serialize(BlueSkyCloudLog *log,
                                                BlueSkyFS *fs);
 void bluesky_cloudlog_flush(BlueSkyFS *fs);
@@ -306,7 +373,7 @@ void bluesky_cloudlog_flush(BlueSkyFS *fs);
 uint32_t crc32c(uint32_t crc, const char *buf, unsigned int length);
 uint32_t crc32c_finalize(uint32_t crc);
 
-struct _BlueSkyLog {
+struct BlueSkyLog {
     BlueSkyFS *fs;
     char *log_directory;
     GAsyncQueue *queue;
@@ -333,7 +400,7 @@ struct _BlueSkyLog {
 /* An object for tracking log files which are stored locally--either the
  * journal for filesystem consistency or log segments which have been fetched
  * back from cloud storage. */
-struct _BlueSkyCacheFile {
+struct BlueSkyCacheFile {
     GMutex *lock;
     GCond *cond;
     gint refcount;
@@ -344,10 +411,14 @@ struct _BlueSkyCacheFile {
     gint mapcount;              // References to the mmaped data
     const char *addr;           // May be null if data is not mapped in memory
     size_t len;
+    int disk_used;
     BlueSkyFS *fs;
     BlueSkyLog *log;
-    gboolean fetching, ready;   // Cloud data: downloading or ready for use
+    gboolean fetching;          // Cloud data: downloading or ready for use
+    gboolean complete;          // Complete file has been fetched from cloud
     int64_t atime;              // Access time, for cache management
+    BlueSkyRangeset *items;     // Locations of valid items
+    BlueSkyRangeset *prefetches;// Locations we have been requested to prefetch
 };
 
 BlueSkyLog *bluesky_log_new(const char *log_directory);
@@ -356,20 +427,25 @@ void bluesky_log_finish_all(GList *log_items);
 BlueSkyCloudLog *bluesky_log_get_commit_point(BlueSkyFS *fs);
 void bluesky_log_write_commit_point(BlueSkyFS *fs, BlueSkyCloudLog *marker);
 
-BlueSkyRCStr *bluesky_log_map_object(BlueSkyFS *fs, int log_dir, int log_seq,
-                                     int log_offset, int log_size);
+BlueSkyRCStr *bluesky_cachefile_map_raw(BlueSkyCacheFile *cachefile,
+                                        off_t offset, size_t size);
+BlueSkyRCStr *bluesky_log_map_object(BlueSkyCloudLog *item, gboolean map_data);
 void bluesky_mmap_unref(BlueSkyCacheFile *mmap);
 void bluesky_cachefile_unref(BlueSkyCacheFile *cachefile);
 
 BlueSkyCacheFile *bluesky_cachefile_lookup(BlueSkyFS *fs,
-                                           int clouddir, int log_seq);
+                                           int clouddir, int log_seq,
+                                           gboolean start_fetch);
 void bluesky_cachefile_gc(BlueSkyFS *fs);
 
 void bluesky_replay(BlueSkyFS *fs);
 
 /* Used to track log segments that are being written to the cloud. */
 typedef struct {
-    BlueSkyRCStr *data;
+    BlueSkyFS *fs;
+    char *key;                  /* File name for log segment in backend */
+    GString *raw_data;          /* Data before encryption */
+    BlueSkyRCStr *data;         /* Data after encryption */
     GSList *items;
     GMutex *lock;
     GCond *cond;
@@ -383,14 +459,9 @@ typedef struct {
 typedef struct {
     uint64_t inum;
 
-    /* The ID of the most recent version of the inode. */
-    BlueSkyCloudID id;
-
-    /* The location where that version is written in the cloud. */
-    BlueSkyCloudPointer location;
-
-    /* If the cloud log entry exists in memory, then a pointer to it, otherwise
-     * NULL. */
+    /* A pointer to the cloud log entry for this inode.  This may or may not
+     * actually have data loaded (it might just contain pointers to the data
+     * location, and in fact this will likely often be the case). */
     BlueSkyCloudLog *item;
 } InodeMapEntry;
 
@@ -412,9 +483,14 @@ typedef struct {
 InodeMapEntry *bluesky_inode_map_lookup(GSequence *inode_map, uint64_t inum,
                                         int action);
 BlueSkyCloudLog *bluesky_inode_map_serialize(BlueSkyFS *fs);
+void bluesky_inode_map_minimize(BlueSkyFS *fs);
 
 gboolean bluesky_checkpoint_load(BlueSkyFS *fs);
 
+/* Merging of log state with the work of the cleaner. */
+void bluesky_cleaner_merge(BlueSkyFS *fs);
+void bluesky_cleaner_thread_launch(BlueSkyFS *fs);
+
 #ifdef __cplusplus
 }
 #endif