Plots for the CIFS SPECsfs results.
[bluesky.git] / bluesky / store-simple.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 /* Interface to the simple BlueSky test storage server. */
10
11 #include <stdint.h>
12 #include <stdlib.h>
13 #include <errno.h>
14 #include <glib.h>
15 #include <string.h>
16 #include <sys/types.h>
17 #include <sys/socket.h>
18 #include <netdb.h>
19 #include <unistd.h>
20
21 #include "bluesky-private.h"
22
23 #define MAX_IDLE_CONNECTIONS 8
24
25 typedef struct {
26     GThreadPool *thread_pool;
27     struct sockaddr_in server_addr;
28
29     /* A pool of open file connections to the server which are not currently in
30      * use. */
31     GQueue *fd_pool;
32     GMutex *fd_pool_lock;
33 } SimpleStore;
34
35 static int get_connection(SimpleStore *store)
36 {
37     int fd = -1;
38
39     g_mutex_lock(store->fd_pool_lock);
40     if (!g_queue_is_empty(store->fd_pool)) {
41         fd = GPOINTER_TO_INT(g_queue_pop_head(store->fd_pool));
42     }
43     g_mutex_unlock(store->fd_pool_lock);
44     if (fd != -1)
45         return fd;
46
47     fd = socket(PF_INET, SOCK_STREAM, 0);
48     if (fd < 0) {
49         g_warning("Error creating simplestore socket: %m");
50         return -1;
51     }
52
53     if (connect(fd, (struct sockaddr *)&store->server_addr,
54                 sizeof(store->server_addr)) < 0) {
55         g_warning("Error connecting to simplestore server: %m");
56         return -1;
57     }
58
59     return fd;
60 }
61
62 static void put_connection(SimpleStore *store, int fd)
63 {
64     g_mutex_lock(store->fd_pool_lock);
65     g_queue_push_head(store->fd_pool, GINT_TO_POINTER(fd));
66     while (g_queue_get_length(store->fd_pool) > MAX_IDLE_CONNECTIONS) {
67         fd = GPOINTER_TO_INT(g_queue_pop_tail(store->fd_pool));
68         close(fd);
69     }
70     g_mutex_unlock(store->fd_pool_lock);
71 }
72
73 static gboolean write_data(int fd, const char *buf, size_t len)
74 {
75     while (len > 0) {
76         ssize_t bytes = write(fd, buf, len);
77         if (bytes < 0) {
78             if (errno == EINTR)
79                 continue;
80             return FALSE;
81         }
82         buf += bytes;
83         len -= bytes;
84     }
85     return TRUE;
86 }
87
88 static ssize_t read_line(int fd, char *buf, size_t maxlen)
89 {
90     size_t buflen = 0;
91
92     /* Leave room for a null byte at the end */
93     maxlen--;
94
95     while (buflen == 0 || memchr(buf, '\n', buflen) == NULL) {
96         ssize_t bytes;
97
98         if (buflen == maxlen) {
99             return -1;
100         }
101
102         bytes = read(fd, buf + buflen, maxlen - buflen);
103         if (bytes < 0) {
104             if (errno == EINTR)
105                 continue;
106             else
107                 return -1;
108         }
109         if (bytes == 0)
110             break;
111
112         g_assert(bytes <= maxlen - buflen);
113         buflen += bytes;
114     }
115
116     buf[buflen] = '\0';
117     return buflen;
118 }
119
120 static gboolean read_bytes(int fd, char *buf, size_t len)
121 {
122     while (len > 0) {
123         ssize_t bytes;
124
125         bytes = read(fd, buf, len);
126         if (bytes < 0) {
127             if (errno == EINTR)
128                 continue;
129             else
130                 return FALSE;
131         }
132         if (bytes == 0)
133             return FALSE;
134
135         g_assert(bytes <= len);
136         len -= bytes;
137         buf += bytes;
138     }
139
140     return TRUE;
141 }
142
143 /* TODO: Re-use a connection for multiple requests, and support pipelining. */
144 static void simplestore_task(gpointer a, gpointer b)
145 {
146     BlueSkyStoreAsync *async = (BlueSkyStoreAsync *)a;
147     SimpleStore *server = (SimpleStore *)bluesky_store_async_get_handle(async);
148
149     async->status = ASYNC_RUNNING;
150
151     int fd = get_connection(server);
152     if (fd < 0) {
153         bluesky_store_async_mark_complete(async);
154         bluesky_store_async_unref(async);
155         return;
156     }
157
158     switch (async->op) {
159     case STORE_OP_GET:
160     {
161         async->result = -1;
162         char *cmd = g_strdup_printf("GET %s %zd %zd\n",
163                                     async->key, async->start, async->len);
164         char result_buf[256];
165
166         if (!write_data(fd, cmd, strlen(cmd))) {
167             g_free(cmd);
168             break;
169         }
170
171         g_free(cmd);
172
173         int bytes = read_line(fd, result_buf, sizeof(result_buf));
174         if (bytes < 0)
175             break;
176         int result = atoi(result_buf);
177         if (result < 0)
178             break;
179
180         char *data = g_malloc(result);
181         if (strchr(result_buf, '\n') != NULL) {
182             int header_size = strchr(result_buf, '\n') - result_buf + 1;
183             int data_bytes = bytes - header_size;
184             if (data_bytes > result)
185                 data_bytes = result;
186             memcpy(data, result_buf + header_size, data_bytes);
187             if (!read_bytes(fd, data + data_bytes, result - data_bytes))
188                 break;
189         } else {
190             if (!read_bytes(fd, data, result))
191                 break;
192         }
193
194         async->data = bluesky_string_new(data, result);
195         async->result = 0;
196         async->range_done = TRUE;
197         break;
198     }
199
200     case STORE_OP_PUT:
201     {
202         async->result = -1;
203         char *cmd = g_strdup_printf("PUT %s %zd\n",
204                                     async->key, async->data->len);
205         char result_buf[256];
206
207         if (!write_data(fd, cmd, strlen(cmd))) {
208             g_free(cmd);
209             break;
210         }
211
212         g_free(cmd);
213         if (!write_data(fd, async->data->data, async->data->len)) {
214             break;
215         }
216
217         if (read_line(fd, result_buf, sizeof(result_buf)) < 0)
218             break;
219         if (atoi(result_buf) != 0)
220             break;
221
222         async->result = 0;
223         break;
224     }
225
226     default:
227         break;
228     }
229
230     int success = (async->result == 0);
231     bluesky_store_async_mark_complete(async);
232     bluesky_store_async_unref(async);
233
234     if (success) {
235         put_connection(server, fd);
236     } else {
237         close(fd);
238     }
239 }
240
241 static char *simplestore_lookup_last(gpointer s, const char *prefix)
242 {
243     return NULL;
244 }
245
246 static gpointer simplestore_new(const gchar *path)
247 {
248     SimpleStore *store = g_new0(SimpleStore, 1);
249
250     /* TODO: Right now we leak this memory.  We should probably clean up in
251      * simplestore_destroy, but it's not a big deal. */
252     const gchar *host = "127.0.0.1", *port = "9541";
253     if (path != NULL) {
254         gchar **target = g_strsplit(path, ":", 0);
255         if (target[0] != NULL) {
256             host = target[0];
257             if (target[1] != NULL) {
258                 port = target[1];
259             }
260         }
261     }
262
263     g_print("simplestore: %s port %s\n", host, port);
264
265     struct addrinfo hints;
266     struct addrinfo *lookup_result = NULL;
267     memset(&hints, 0, sizeof(hints));
268     hints.ai_family = AF_INET;
269     hints.ai_socktype = SOCK_STREAM;
270     int res = getaddrinfo(host, port, &hints, &lookup_result);
271     if (res != 0) {
272         fprintf(stderr, "simplestore: cannot resolve target name: %s\n",
273                 gai_strerror(res));
274         return NULL;
275     }
276     for (struct addrinfo *ai = lookup_result; ai != NULL; ai = ai->ai_next) {
277         printf("flags=%d family=%d socktype=%d proto=%d\n",
278                ai->ai_flags,
279                ai->ai_family,
280                ai->ai_socktype,
281                ai->ai_protocol);
282         if (ai->ai_addrlen == sizeof(struct sockaddr_in)) {
283             memcpy(&store->server_addr, ai->ai_addr,
284                    sizeof(struct sockaddr_in));
285         } else {
286             fprintf(stderr, "Warning: Bad address record size!\n");
287         }
288     }
289     freeaddrinfo(lookup_result);
290
291     store->fd_pool = g_queue_new();
292     store->fd_pool_lock = g_mutex_new();
293
294     store->thread_pool = g_thread_pool_new(simplestore_task, NULL,
295                                            bluesky_max_threads, FALSE, NULL);
296
297     return store;
298 }
299
300 static void simplestore_destroy(gpointer store)
301 {
302 }
303
304 static void simplestore_submit(gpointer store, BlueSkyStoreAsync *async)
305 {
306     SimpleStore *server = (SimpleStore *)store;
307
308     g_return_if_fail(async->status == ASYNC_NEW);
309     g_return_if_fail(async->op != STORE_OP_NONE);
310
311     switch (async->op) {
312     case STORE_OP_GET:
313     case STORE_OP_PUT:
314         async->status = ASYNC_PENDING;
315         bluesky_store_async_ref(async);
316         g_thread_pool_push(server->thread_pool, async, NULL);
317         break;
318
319     default:
320         g_warning("Uknown operation type for simplestore: %d\n", async->op);
321         bluesky_store_async_mark_complete(async);
322         break;
323     }
324 }
325
326 static void simplestore_cleanup(gpointer store, BlueSkyStoreAsync *async)
327 {
328 }
329
330 static BlueSkyStoreImplementation store_impl = {
331     .create = simplestore_new,
332     .destroy = simplestore_destroy,
333     .submit = simplestore_submit,
334     .cleanup = simplestore_cleanup,
335     .lookup_last = simplestore_lookup_last,
336 };
337
338 void bluesky_store_init_simple(void)
339 {
340     bluesky_store_register(&store_impl, "simple");
341 }