Fix some memory leaks.
[bluesky.git] / bluesky / cloudlog.c
1 /* Blue Sky: File Systems in the Cloud
2  *
3  * Copyright (C) 2009  The Regents of the University of California
4  * Written by Michael Vrable <mvrable@cs.ucsd.edu>
5  *
6  * TODO: Licensing
7  */
8
9 #include <stdio.h>
10 #include <stdint.h>
11 #include <glib.h>
12 #include <string.h>
13
14 #include "bluesky-private.h"
15
16 /* The locations hash table in the file system is used to map objects to their locations.  Objects are named using 128- */
17
18 typedef struct {
19     BlueSkyCloudID id;
20
21     BlueSkyCloudPointer *cloud_loc;
22 } BlueSkyLocationEntry;
23
24 BlueSkyCloudID bluesky_cloudlog_new_id()
25 {
26     BlueSkyCloudID id;
27     bluesky_crypt_random_bytes((uint8_t *)&id.bytes, sizeof(id));
28     return id;
29 }
30
31 gchar *bluesky_cloudlog_id_to_string(BlueSkyCloudID id)
32 {
33     char buf[sizeof(BlueSkyCloudID) * 2 + 1];
34     buf[0] = '\0';
35
36     for (int i = 0; i < sizeof(BlueSkyCloudID); i++) {
37         sprintf(&buf[2*i], "%02x", (uint8_t)(id.bytes[i]));
38     }
39
40     return g_strdup(buf);
41 }
42
43 BlueSkyCloudID bluesky_cloudlog_id_from_string(const gchar *idstr)
44 {
45     BlueSkyCloudID id;
46     memset(&id, 0, sizeof(id));
47     for (int i = 0; i < 2*sizeof(BlueSkyCloudID); i++) {
48         char c = idstr[i];
49         if (c == '\0') {
50             g_warning("Short cloud id: %s\n", idstr);
51             break;
52         }
53         int val = 0;
54         if (c >= '0' && c <= '9')
55             val = c - '0';
56         else if (c >= 'a' && c <= 'f')
57             val = c - 'a' + 10;
58         else
59             g_warning("Bad character in cloud id: %s\n", idstr);
60         id.bytes[i / 2] += val << (i % 2 ? 0 : 4);
61     }
62     return id;
63 }
64
65 gboolean bluesky_cloudlog_equal(gconstpointer a, gconstpointer b)
66 {
67     BlueSkyCloudID *id1 = (BlueSkyCloudID *)a, *id2 = (BlueSkyCloudID *)b;
68
69     return memcmp(id1, id2, sizeof(BlueSkyCloudID)) == 0;
70 }
71
72 guint bluesky_cloudlog_hash(gconstpointer a)
73 {
74     BlueSkyCloudID *id = (BlueSkyCloudID *)a;
75
76     // Assume that bits in the ID are randomly chosen so that any subset of the
77     // bits can be used as a hash key.
78     return *(guint *)(&id->bytes);
79 }
80
81 /* Formatting of cloud log segments.  This handles grouping items together
82  * before writing a batch to the cloud, handling indirection through items like
83  * the inode map, etc. */
84
85 BlueSkyCloudLog *bluesky_cloudlog_new(BlueSkyFS *fs)
86 {
87     BlueSkyCloudLog *log = g_new0(BlueSkyCloudLog, 1);
88
89     log->lock = g_mutex_new();
90     log->cond = g_cond_new();
91     log->fs = fs;
92     log->type = LOGTYPE_UNKNOWN;
93     log->id = bluesky_cloudlog_new_id();
94     log->links = g_array_new(FALSE, TRUE, sizeof(BlueSkyCloudLog *));
95     g_atomic_int_set(&log->refcount, 1);
96
97     return log;
98 }
99
100 /* The reference held by the hash table does not count towards the reference
101  * count.  When a new object is created, it initially has a reference count of
102  * 1 for the creator, and similarly fetching an item from the hash table will
103  * also create a reference.  If the reference count drops to zero,
104  * bluesky_cloudlog_unref attempts to remove the object from the hash
105  * table--but there is a potential race since another thread might read the
106  * object from the hash table at the same time.  So an object with a reference
107  * count of zero may still be resurrected, in which case we need to abort the
108  * destruction.  Once the object is gone from the hash table, and if the
109  * reference count is still zero, it can actually be deleted. */
110 void bluesky_cloudlog_ref(BlueSkyCloudLog *log)
111 {
112     if (log == NULL)
113         return;
114
115     g_atomic_int_inc(&log->refcount);
116 }
117
118 void bluesky_cloudlog_unref(BlueSkyCloudLog *log)
119 {
120     if (log == NULL)
121         return;
122
123     if (g_atomic_int_dec_and_test(&log->refcount)) {
124         BlueSkyFS *fs = log->fs;
125
126         g_mutex_lock(fs->lock);
127         if (g_atomic_int_get(&log->refcount) > 0) {
128             g_mutex_unlock(fs->lock);
129             return;
130         }
131
132         g_hash_table_remove(fs->locations, &log->id);
133         g_mutex_unlock(fs->lock);
134
135         log->type = LOGTYPE_INVALID;
136         g_mutex_free(log->lock);
137         g_cond_free(log->cond);
138         for (int i = 0; i < log->links->len; i++) {
139             BlueSkyCloudLog *c = g_array_index(log->links,
140                                                BlueSkyCloudLog *, i);
141             bluesky_cloudlog_unref(c);
142         }
143         g_array_unref(log->links);
144         bluesky_string_unref(log->data);
145         g_free(log);
146     }
147 }
148
149 /* Start a write of the object to the local log. */
150 void bluesky_cloudlog_sync(BlueSkyCloudLog *log)
151 {
152     bluesky_log_item_submit(log, log->fs->log);
153 }
154
155 /* Add the given entry to the global hash table containing cloud log entries.
156  * Takes ownership of the caller's reference. */
157 void bluesky_cloudlog_insert(BlueSkyCloudLog *log)
158 {
159     g_mutex_lock(log->fs->lock);
160     g_hash_table_insert(log->fs->locations, &log->id, log);
161     g_mutex_unlock(log->fs->lock);
162 }
163
164 struct log_header {
165     char magic[4];
166     uint32_t size;
167     BlueSkyCloudID id;
168     uint32_t pointer_count;
169 } __attribute__((packed));
170
171 struct logref {
172     BlueSkyCloudID id;
173     BlueSkyCloudPointer location;
174 } __attribute__((packed));
175
176 struct log_footer {
177     char refmagic[4];
178     struct logref refs[0];
179 };
180
181 BlueSkyCloudPointer bluesky_cloudlog_serialize(BlueSkyCloudLog *log,
182                                                BlueSkyCloudLogState *state)
183 {
184     if (log->location_flags & CLOUDLOG_CLOUD) {
185         return log->location;
186     }
187
188     for (int i = 0; i < log->links->len; i++) {
189         BlueSkyCloudLog *ref = g_array_index(log->links,
190                                              BlueSkyCloudLog *, i);
191         if (ref != NULL)
192             bluesky_cloudlog_serialize(ref, state);
193     }
194
195     g_mutex_lock(log->lock);
196     bluesky_cloudlog_fetch(log);
197     g_assert(log->data != NULL);
198
199     log->location = state->location;
200     log->location.offset = state->data->len;
201     log->location.size
202         = sizeof(struct log_header) + sizeof(BlueSkyCloudID) * 0
203            + log->data->len;
204
205     struct log_header header;
206     memcpy(header.magic, "AgI ", 4);
207     header.size = GUINT32_TO_LE(log->location.size);
208     header.id = log->id;
209     header.pointer_count = GUINT32_TO_LE(0);
210
211     g_string_append_len(state->data, (const char *)&header, sizeof(header));
212     g_string_append_len(state->data, log->data->data, log->data->len);
213
214     log->location_flags |= CLOUDLOG_CLOUD;
215     g_mutex_unlock(log->lock);
216
217     return log->location;
218 }
219
220 static void find_inodes(gpointer key, gpointer value, gpointer user_data)
221 {
222     BlueSkyCloudLogState *state = (BlueSkyCloudLogState *)user_data;
223     BlueSkyCloudLog *item = (BlueSkyCloudLog *)value;
224
225     if (item->type != LOGTYPE_INODE)
226         return;
227
228     bluesky_cloudlog_ref(item);
229     state->inode_list = g_list_prepend(state->inode_list, item);
230 }
231
232 void bluesky_cloudlog_write_log(BlueSkyFS *fs)
233 {
234     BlueSkyCloudLogState *state = fs->log_state;
235     if (state->data == NULL)
236         state->data = g_string_new("");
237
238     g_mutex_lock(fs->lock);
239     g_hash_table_foreach(fs->locations, find_inodes, state);
240     g_mutex_unlock(fs->lock);
241
242     while (state->inode_list != NULL) {
243         BlueSkyCloudLog *log = (BlueSkyCloudLog *)state->inode_list->data;
244         bluesky_cloudlog_serialize(log, state);
245         bluesky_cloudlog_unref(log);
246         state->inode_list = g_list_delete_link(state->inode_list,
247                                                state->inode_list);
248     }
249
250     if (state->data->len > 0) {
251         g_print("Serialized %zd bytes of data to cloud\n", state->data->len);
252
253         BlueSkyStoreAsync *async = bluesky_store_async_new(fs->store);
254         async->op = STORE_OP_PUT;
255         async->key = g_strdup_printf("log-%08d-%08d",
256                                      state->location.directory,
257                                      state->location.sequence);
258         async->data = bluesky_string_new_from_gstring(state->data);
259         bluesky_store_async_submit(async);
260         bluesky_store_async_wait(async);
261         bluesky_store_async_unref(async);
262
263         state->location.sequence++;
264         state->location.offset = 0;
265     } else {
266         g_string_free(state->data, TRUE);
267     }
268
269     state->data = NULL;
270 }
271
272 /* Ensure that a cloud log item is loaded in memory, and if not read it in.
273  * TODO: Make asynchronous, and make this also fetch from the cloud.  Right now
274  * we only read from the log.  Log item must be locked. */
275 void bluesky_cloudlog_fetch(BlueSkyCloudLog *log)
276 {
277     if (log->data != NULL)
278         return;
279
280     g_print("Re-mapping log entry %d/%d/%d...\n",
281             log->log_seq, log->log_offset, log->log_size);
282
283     g_assert(log->location_flags & CLOUDLOG_JOURNAL);
284
285     log->data = bluesky_log_map_object(log->fs->log, log->log_seq,
286                                        log->log_offset, log->log_size);
287
288     g_cond_broadcast(log->cond);
289 }