13 #include <xseg/xseg.h>
16 #define MAX_PATH_SIZE 255
17 #define MAX_FILENAME_SIZE 255
19 static int usage(void)
21 printf("Usage: ./filed <path_to_directory> [options]\n"
22 "Options: [-p portno]\n"
23 " [-g type:name:nr_ports:nr_requests:request_size:extra_size:page_shift]\n"
24 " [-n nr_parallel_ops]\n"
30 unsigned long cacheidx;
37 struct xseg_request *req;
44 #define READY (1 << 1)
48 volatile unsigned int ref;
49 volatile unsigned long time;
50 volatile unsigned int flags;
52 char target[MAX_FILENAME_SIZE + 1];
57 struct xseg_port *xport;
64 struct sigevent sigevent;
67 uint64_t handled_reqs;
69 struct fdcache_node *fdcache;
71 pthread_mutex_t cache_lock;
72 char path[MAX_PATH_SIZE + 1];
75 static unsigned verbose;
77 static unsigned long sigaction_count;
79 static void sigaction_handler(int sig, siginfo_t *siginfo, void *arg)
84 static void log_io(char *msg, struct io *io)
86 char target[64], data[64];
87 /* null terminate name in case of req->target is less than 63 characters,
88 * and next character after name (aka first byte of next buffer) is not
91 struct store* store = io->store;
92 struct xseg *xseg = store->xseg;
93 struct xseg_request *req = io->req;
94 char *req_target = xseg_get_target(xseg, req);
95 char *req_data = xseg_get_data(xseg, req);
97 unsigned int end = (req->targetlen> 63) ? 63 : req->targetlen;
99 strncpy(target, req_target, end);
101 strncpy(data, req_data, 63);
105 "%s: fd:%u, op:%u offset: %llu size: %lu retval: %lu, reqstate: %u\n"
106 "target[%u]: '%s', data[%llu]:\n%s------------------\n\n",
108 (unsigned int)io->fdcacheidx, //this is cacheidx not fd
109 (unsigned int)req->op,
110 (unsigned long long)req->offset,
111 (unsigned long)req->size,
112 (unsigned long)io->retval,
113 (unsigned int)req->state,
114 (unsigned int)req->targetlen, target,
115 (unsigned long long)req->datalen, data);
118 static struct io *alloc_io(struct store *store)
120 xqindex idx = xq_pop_head(&store->free_ops, 1);
123 return store->ios + idx;
126 static inline void free_io(struct store *store, struct io *io)
128 xqindex idx = io - store->ios;
130 xq_append_head(&store->free_ops, idx, 1);
134 static void complete(struct store *store, struct io *io)
136 struct xseg_request *req = io->req;
137 req->state |= XS_SERVED;
139 log_io("complete", io);
140 xseg_respond(store->xseg, req->portno, req);
141 xseg_signal(store->xseg, req->portno);
142 __sync_fetch_and_sub(&store->fdcache[io->fdcacheidx].ref, 1);
145 static void fail(struct store *store, struct io *io)
147 struct xseg_request *req = io->req;
148 req->state |= XS_FAILED;
151 xseg_respond(store->xseg, req->portno, req);
152 xseg_signal(store->xseg, req->portno);
153 if (io->fdcacheidx >= 0) {
154 __sync_fetch_and_sub(&store->fdcache[io->fdcacheidx].ref, 1);
158 static void pending(struct store *store, struct io *io)
160 io->req->state = XS_PENDING;
163 static void handle_unknown(struct store *store, struct io *io)
165 struct xseg *xseg = store->xseg;
166 struct xseg_request *req = io->req;
167 char *data = xseg_get_data(xseg, req);
168 snprintf(data, req->datalen, "unknown request op");
172 static inline void prepare_io(struct store *store, struct io *io)
177 static int dir_open( struct store *store, struct io *io,
178 char *target, uint32_t targetlen, int mode )
181 struct fdcache_node *ce = NULL;
185 if (targetlen> MAX_FILENAME_SIZE)
190 pthread_mutex_lock(&store->cache_lock);
194 for (i = 0; i < store->maxfds; i++) {
195 if (store->fdcache[i].ref == 0 && min > store->fdcache[i].time
196 && (store->fdcache[i].flags & READY)) {
197 min = store->fdcache[i].time;
201 if (!strncmp(store->fdcache[i].target, target, targetlen)) {
202 if (store->fdcache[i].target[targetlen] == 0) {
203 ce = &store->fdcache[i];
204 /* if any other io thread is currently opening
205 * the file, block until it succeeds or fails
207 if (!(ce->flags & READY)) {
208 pthread_cond_wait(&ce->cond, &store->cache_lock);
209 /* when ready, restart lookup */
212 /* if successfully opened */
214 fd = store->fdcache[i].fd;
218 /* else open failed for the other io thread, so
219 * it should fail for everyone waiting on this
231 /* all cache entries are currently being used */
232 pthread_mutex_unlock(&store->cache_lock);
235 if (store->fdcache[lru].ref){
237 printf("lru(%ld) ref not 0 (%u)\n", lru, store->fdcache[lru].ref);
240 /* make room for new file */
241 ce = &store->fdcache[lru];
242 /* set name here and state to not ready, for any other requests on the
243 * same target that may follow
245 strncpy(ce->target, target, targetlen);
246 ce->target[targetlen] = 0;
248 pthread_mutex_unlock(&store->cache_lock);
251 if (close(ce->fd) < 0){
255 fd = openat(store->dirfd, ce->target, O_RDWR);
257 if (errno == ENOENT){
258 fd = openat(store->dirfd, ce->target,
259 O_RDWR | O_CREAT, 0600);
264 /* insert in cache a negative fd to indicate opening error to
265 * any other ios waiting for the file to open
268 /* insert in cache */
270 pthread_mutex_lock(&store->cache_lock);
274 pthread_cond_broadcast(&ce->cond);
276 io->fdcacheidx = lru;
284 store->handled_reqs++;
285 ce->time = store->handled_reqs;
286 __sync_fetch_and_add(&ce->ref, 1);
287 pthread_mutex_unlock(&store->cache_lock);
292 pthread_mutex_unlock(&store->cache_lock);
296 static void handle_read_write(struct store *store, struct io *io)
299 struct xseg *xseg = store->xseg;
300 struct xseg_request *req = io->req;
301 char *target = xseg_get_target(xseg, req);
302 char *data = xseg_get_data(xseg, req);
304 if (req->op == X_WRITE)
308 fd = dir_open(store, io, target, req->targetlen, mode);
316 printf("0.%p vs %p!\n", (void *)req, (void *)io->req);
318 if (req->flags & (XF_FLUSH | XF_FUA)) {
319 /* No FLUSH/FUA support yet (O_SYNC ?).
320 * note that with FLUSH/size == 0
321 * there will probably be a (uint64_t)-1 offset */
331 prepare_io(store, io);
335 while (req->serviced < req->datalen) {
336 r = pread(fd, data + req->serviced,
337 req->datalen - req->serviced,
338 req->offset + req->serviced);
340 req->datalen = req->serviced;
344 /* reached end of file. zero out the rest data buffer */
345 memset(data + req->serviced, 0, req->datalen - req->serviced);
346 req->serviced = req->datalen;
354 while (req->serviced < req->datalen) {
355 r = pwrite(fd, data + req->serviced,
356 req->datalen - req->serviced,
357 req->offset + req->serviced);
359 req->datalen = req->serviced;
362 /* reached end of file. zero out the rest data buffer */
363 memset(data + req->serviced, 0, req->datalen - req->serviced);
364 req->serviced = req->datalen;
373 /* if fsync fails, then no bytes serviced correctly */
378 snprintf(data, req->datalen,
379 "wtf, corrupt op %u?\n", req->op);
384 if (req->serviced > 0 ) {
388 strerror_r(errno, data, req->datalen);
394 static void handle_info(struct store *store, struct io *io)
396 struct xseg *xseg = store->xseg;
397 struct xseg_request *req = io->req;
398 char *target = xseg_get_target(xseg, req);
399 char *data = xseg_get_data(xseg, req);
404 fd = dir_open(store, io, target, req->targetlen, 0);
409 r = fstat(fd, &stat);
416 *((off_t *) data) = size;
417 req->datalen = sizeof(size);
422 static void dispatch(struct store *store, struct io *io)
425 printf("io: 0x%p, req: 0x%p, op %u\n",
426 (void *)io, (void *)io->req, io->req->op);
427 switch (io->req->op) {
430 handle_read_write(store, io); break;
432 handle_info(store, io); break;
435 handle_unknown(store, io);
439 static void handle_accepted(struct store *store, struct io *io)
441 struct xseg_request *req = io->req;
443 req->state = XS_ACCEPTED;
448 static struct io* wake_up_next_iothread(struct store *store)
450 struct io *io = alloc_io(store);
453 pthread_mutex_lock(&io->lock);
454 pthread_cond_signal(&io->cond);
455 pthread_mutex_unlock(&io->lock);
460 void *io_loop(void *arg)
462 struct io *io = (struct io *) arg;
463 struct store *store = io->store;
464 struct xseg *xseg = store->xseg;
465 uint32_t portno = store->portno;
466 struct xseg_request *accepted;
470 accepted = xseg_accept(xseg, portno);
473 wake_up_next_iothread(store);
474 handle_accepted(store, io);
477 pthread_mutex_lock(&io->lock);
479 pthread_cond_wait(&io->cond, &io->lock);
480 pthread_mutex_unlock(&io->lock);
487 static struct xseg *join(char *spec)
489 struct xseg_config config;
492 (void)xseg_parse_spec(spec, &config);
493 xseg = xseg_join(config.type, config.name, "posix", NULL);
497 fprintf(stderr, "Failed to join xseg, creating it...\n");
498 (void)xseg_create(&config);
499 return xseg_join(config.type, config.name, "posix", NULL);
502 static int filed_loop(struct store *store)
504 struct xseg *xseg = store->xseg;
505 uint32_t portno = store->portno;
509 io = wake_up_next_iothread(store);
510 xseg_prepare_wait(xseg, portno);
511 xseg_wait_signal(xseg, 1000000UL);
516 static int filed( char *path, unsigned long size, uint32_t nr_ops,
517 char *spec, long portno )
523 store = malloc(sizeof(struct store));
536 store->sigevent.sigev_notify = SIGEV_SIGNAL;
537 store->sigevent.sigev_signo = SIGIO;
538 sa.sa_sigaction = sigaction_handler;
539 sa.sa_flags = SA_SIGINFO;
540 if (sigemptyset(&sa.sa_mask))
541 perror("sigemptyset");
543 if (sigaction(SIGIO, &sa, NULL)) {
548 store->nr_ops = nr_ops;
549 store->maxfds = 2 * nr_ops;
551 store->fdcache = calloc(store->maxfds, sizeof(struct fdcache_node));
555 store->free_bufs = calloc(store->nr_ops, sizeof(xqindex));
556 if(!store->free_bufs)
559 store->iothread = calloc(store->nr_ops, sizeof(pthread_t));
563 store->ios = calloc(nr_ops, sizeof(struct io));
570 for (i = 0; i < nr_ops; i++) {
571 store->ios[i].store = store;
572 pthread_cond_init(&store->ios[i].cond, NULL);
573 pthread_mutex_init(&store->ios[i].lock, NULL);
576 xq_init_seq(&store->free_ops, store->nr_ops, store->nr_ops, store->free_bufs);
578 store->handled_reqs = 0;
579 strncpy(store->path, path, MAX_PATH_SIZE);
580 store->path[MAX_PATH_SIZE] = 0;
582 store->path_len = strlen(store->path);
583 if (store->path[store->path_len -1] != '/'){
584 store->path[store->path_len] = '/';
585 store->path[++store->path_len]= 0;
587 store->dirfd = open(store->path, O_RDWR);
588 if (!(store->dirfd < 0 && errno == EISDIR)){
589 fprintf(stderr, "%s is not a directory\n", store->path);
593 store->dirfd = open(store->path, O_RDONLY);
594 if (store->dirfd < 0){
595 perror("Directory open");
600 int fd = dir_open(store, ".__tmp", 6, 1);
602 perror("Directory check");
606 if (xseg_initialize()) {
607 printf("cannot initialize library\n");
610 store->xseg = join(spec);
614 store->xport = xseg_bind_port(store->xseg, portno);
616 printf("cannot bind to port %ld\n", portno);
620 store->portno = xseg_portno(store->xseg, store->xport);
621 printf("filed on port %u/%u\n",
622 store->portno, store->xseg->config.nr_ports);
624 for (i = 0; i < nr_ops; i++) {
625 pthread_cond_init(&store->fdcache[i].cond, NULL);
626 store->fdcache[i].flags = READY;
628 for (i = 0; i < nr_ops; i++) {
629 //TODO error check + cond variable to stop io from starting
630 //unless all threads are created successfully
631 pthread_create(store->iothread + i, NULL, io_loop, (void *) (store->ios + i));
633 pthread_mutex_init(&store->cache_lock, NULL);
634 return filed_loop(store);
637 int main(int argc, char **argv)
639 char *path, *spec = "";
653 for (i = 2; i < argc; i++) {
654 if (!strcmp(argv[i], "-g") && i + 1 < argc) {
660 if (!strcmp(argv[i], "-p") && i + 1 < argc) {
661 portno = strtoul(argv[i+1], NULL, 10);
666 if (!strcmp(argv[i], "-n") && i + 1 < argc) {
667 nr_ops = strtoul(argv[i+1], NULL, 10);
671 if (!strcmp(argv[i], "-v")) {
680 return filed(path, size, nr_ops, spec, portno);