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");
29 unsigned long cacheidx;
36 struct xseg_request *req;
43 #define READY (1 << 1)
47 volatile unsigned int ref;
48 volatile unsigned long time;
49 volatile unsigned int flags;
51 char name[MAX_FILENAME_SIZE + 1];
56 struct xseg_port *xport;
63 struct sigevent sigevent;
66 uint64_t handled_reqs;
68 struct fdcache_node *fdcache;
70 pthread_mutex_t cache_lock;
71 char path[MAX_PATH_SIZE + 1];
74 static unsigned long sigaction_count;
76 static void sigaction_handler(int sig, siginfo_t *siginfo, void *arg)
81 static void log_io(char *msg, struct io *io)
83 char name[64], data[64];
84 /* null terminate name in case of req->name is less than 63 characters,
85 * and next character after name (aka first byte of next buffer) is not
88 unsigned int end = (io->req->namesize > 63) ? 63 : io->req->namesize;
89 strncpy(name, io->req->name, end);
91 strncpy(data, io->req->data, 63);
96 "%s: fd:%u, op:%u offset: %llu size: %lu retval: %lu, reqstate: %u\n"
97 "name[%u]: '%s', data[%llu]:\n%s------------------\n\n",
99 (unsigned int)io->fdcacheidx, //this is cacheidx not fd
100 (unsigned int)io->req->op,
101 (unsigned long long)io->req->offset,
102 (unsigned long)io->req->size,
103 (unsigned long)io->retval,
104 (unsigned int)io->req->state,
105 (unsigned int)io->req->namesize, name,
106 (unsigned long long)io->req->datasize, data);
110 static struct io *alloc_io(struct store *store)
112 xqindex idx = xq_pop_head(&store->free_ops);
115 return store->ios + idx;
118 static inline void free_io(struct store *store, struct io *io)
120 xqindex idx = io - store->ios;
122 xq_append_head(&store->free_ops, idx);
126 static void complete(struct store *store, struct io *io)
128 struct xseg_request *req = io->req;
129 req->state |= XS_SERVED;
130 log_io("complete", io);
131 xseg_respond(store->xseg, req->portno, req);
132 xseg_signal(store->xseg, req->portno);
133 __sync_fetch_and_sub(&store->fdcache[io->fdcacheidx].ref, 1);
136 static void fail(struct store *store, struct io *io)
138 struct xseg_request *req = io->req;
139 req->state |= XS_FAILED;
141 xseg_respond(store->xseg, req->portno, req);
142 xseg_signal(store->xseg, req->portno);
143 if (io->fdcacheidx >= 0) {
144 __sync_fetch_and_sub(&store->fdcache[io->fdcacheidx].ref, 1);
148 static void pending(struct store *store, struct io *io)
150 io->req->state = XS_PENDING;
153 static void handle_unknown(struct store *store, struct io *io)
155 struct xseg_request *req = io->req;
156 snprintf(req->data, req->datasize, "unknown request op");
160 static inline void prepare_io(struct store *store, struct io *io)
165 static int dir_open( struct store *store, struct io *io,
166 char *name, uint32_t namesize, int mode )
169 struct fdcache_node *ce = NULL;
173 if (namesize > MAX_FILENAME_SIZE)
178 pthread_mutex_lock(&store->cache_lock);
182 for (i = 0; i < store->maxfds; i++) {
183 if (store->fdcache[i].ref == 0 && min > store->fdcache[i].time
184 && (store->fdcache[i].flags & READY)) {
185 min = store->fdcache[i].time;
189 if (!strncmp(store->fdcache[i].name, name, namesize)) {
190 if (store->fdcache[i].name[namesize] == 0) {
191 ce = &store->fdcache[i];
192 /* if any other io thread is currently opening
193 * the file, block until it succeeds or fails
195 if (!(ce->flags & READY)) {
196 pthread_cond_wait(&ce->cond, &store->cache_lock);
197 /* when ready, restart lookup */
200 /* if successfully opened */
202 fd = store->fdcache[i].fd;
206 /* else open failed for the other io thread, so
207 * it should fail for everyone waiting on this
219 /* all cache entries are currently being used */
220 pthread_mutex_unlock(&store->cache_lock);
223 if (store->fdcache[lru].ref){
225 printf("lru(%ld) ref not 0 (%u)\n", lru, store->fdcache[lru].ref);
228 /* make room for new file */
229 ce = &store->fdcache[lru];
230 /* set name here and state to not ready, for any other requests on the
231 * same target that may follow
233 strncpy(ce->name, name, namesize);
234 ce->name[namesize] = 0;
236 pthread_mutex_unlock(&store->cache_lock);
239 if (close(ce->fd) < 0){
243 fd = openat(store->dirfd, ce->name, O_RDWR);
245 if (errno == ENOENT){
246 fd = openat(store->dirfd, ce->name,
247 O_RDWR | O_CREAT, 0600);
252 /* insert in cache a negative fd to indicate opening error to
253 * any other ios waiting for the file to open
256 /* insert in cache */
258 pthread_mutex_lock(&store->cache_lock);
262 pthread_cond_broadcast(&ce->cond);
264 io->fdcacheidx = lru;
272 store->handled_reqs++;
273 ce->time = store->handled_reqs;
274 __sync_fetch_and_add(&ce->ref, 1);
275 pthread_mutex_unlock(&store->cache_lock);
280 pthread_mutex_unlock(&store->cache_lock);
284 static void handle_read_write(struct store *store, struct io *io)
287 struct xseg_request *req = io->req;
289 if (req->op == X_WRITE)
293 fd = dir_open(store, io, req->name, req->namesize, mode);
301 printf("0.%p vs %p!\n", (void *)req, (void *)io->req);
303 if (req->flags & (XF_FLUSH | XF_FUA)) {
304 /* No FLUSH/FUA support yet (O_SYNC ?).
305 * note that with FLUSH/size == 0
306 * there will probably be a (uint64_t)-1 offset */
316 prepare_io(store, io);
320 while (req->serviced < req->datasize) {
321 r = pread(fd, req->data + req->serviced,
322 req->datasize - req->serviced,
323 req->offset + req->serviced);
325 req->datasize = req->serviced;
329 /* reached end of file. zero out the rest data buffer */
330 memset(req->data + req->serviced, 0, req->datasize - req->serviced);
331 req->serviced = req->datasize;
339 while (req->serviced < req->datasize) {
340 r = pwrite(fd, req->data + req->serviced,
341 req->datasize - req->serviced,
342 req->offset + req->serviced);
344 req->datasize = req->serviced;
347 /* reached end of file. zero out the rest data buffer */
348 memset(req->data + req->serviced, 0, req->datasize - req->serviced);
349 req->serviced = req->datasize;
358 /* if fsync fails, then no bytes serviced correctly */
363 snprintf(req->data, req->datasize,
364 "wtf, corrupt op %u?\n", req->op);
369 if (req->serviced > 0 ) {
373 strerror_r(errno, req->data, req->datasize);
379 static void handle_info(struct store *store, struct io *io)
381 struct xseg_request *req = io->req;
386 fd = dir_open(store, io, req->name, req->namesize, 0);
391 r = fstat(fd, &stat);
398 *((off_t *) req->data) = size;
399 req->datasize = sizeof(size);
404 static void dispatch(struct store *store, struct io *io)
406 printf("io: 0x%p, req: 0x%p, op %u\n",
407 (void *)io, (void *)io->req, io->req->op);
408 switch (io->req->op) {
411 handle_read_write(store, io); break;
413 handle_info(store, io); break;
416 handle_unknown(store, io);
420 static void handle_accepted(struct store *store, struct io *io)
422 struct xseg_request *req = io->req;
424 req->state = XS_ACCEPTED;
429 static struct io* wake_up_next_iothread(struct store *store)
431 struct io *io = alloc_io(store);
433 pthread_cond_signal(&io->cond);
438 void *io_loop(void *arg)
440 struct io *io = (struct io *) arg;
441 struct store *store = io->store;
442 struct xseg *xseg = store->xseg;
443 uint32_t portno = store->portno;
444 struct xseg_request *accepted;
448 accepted = xseg_accept(xseg, portno);
451 wake_up_next_iothread(store);
452 handle_accepted(store, io);
456 pthread_mutex_lock(&io->lock);
457 pthread_cond_wait(&io->cond, &io->lock);
458 pthread_mutex_unlock(&io->lock);
465 static struct xseg *join(char *spec)
467 struct xseg_config config;
470 (void)xseg_parse_spec(spec, &config);
471 xseg = xseg_join(config.type, config.name);
475 fprintf(stderr, "Failed to join xseg, creating it...\n");
476 (void)xseg_create(&config);
477 return xseg_join(config.type, config.name);
480 static int filed_loop(struct store *store)
482 struct xseg *xseg = store->xseg;
483 uint32_t portno = store->portno;
487 io = wake_up_next_iothread(store);
488 xseg_prepare_wait(xseg, portno);
489 xseg_wait_signal(xseg, portno, 10000);
494 static int filed( char *path, unsigned long size, uint32_t nr_ops,
495 char *spec, long portno )
503 store = malloc(sizeof(struct store));
516 store->sigevent.sigev_notify = SIGEV_SIGNAL;
517 store->sigevent.sigev_signo = SIGIO;
518 sa.sa_sigaction = sigaction_handler;
519 sa.sa_flags = SA_SIGINFO;
520 if (sigemptyset(&sa.sa_mask))
521 perror("sigemptyset");
523 if (sigaction(SIGIO, &sa, NULL)) {
528 store->nr_ops = nr_ops;
529 store->maxfds = 2 * nr_ops;
531 store->fdcache = calloc(store->maxfds, sizeof(struct fdcache_node));
535 store->free_bufs = calloc(store->nr_ops, sizeof(xqindex));
536 if(!store->free_bufs)
539 store->iothread = calloc(store->nr_ops, sizeof(pthread_t));
543 store->ios = calloc(nr_ops, sizeof(struct io));
550 for (i = 0; i < nr_ops; i++) {
551 store->ios[i].store = store;
552 pthread_cond_init(&store->ios[i].cond, NULL);
553 pthread_mutex_init(&store->ios[i].lock, NULL);
556 xq_init_seq(&store->free_ops, store->nr_ops, store->nr_ops, store->free_bufs);
558 store->handled_reqs = 0;
559 strncpy(store->path, path, MAX_PATH_SIZE);
560 store->path[MAX_PATH_SIZE] = 0;
562 store->path_len = strlen(store->path);
563 if (store->path[store->path_len -1] != '/'){
564 store->path[store->path_len] = '/';
565 store->path[++store->path_len]= 0;
567 store->dirfd = open(store->path, O_RDWR);
568 if (!(store->dirfd < 0 && errno == EISDIR)){
569 fprintf(stderr, "%s is not a directory\n", store->path);
573 store->dirfd = open(store->path, O_RDONLY);
574 if (store->dirfd < 0){
575 perror("Directory open");
580 int fd = dir_open(store, ".__tmp", 6, 1);
582 perror("Directory check");
586 if (xseg_initialize("posix")) {
587 printf("cannot initialize library\n");
590 store->xseg = join(spec);
594 store->xport = xseg_bind_port(store->xseg, portno);
596 printf("cannot bind to port %ld\n", portno);
600 store->portno = xseg_portno(store->xseg, store->xport);
601 printf("filed on port %u/%u\n",
602 store->portno, store->xseg->config.nr_ports);
604 for (i = 0; i < nr_ops; i++) {
605 pthread_cond_init(&store->fdcache[i].cond, NULL);
606 store->fdcache[i].flags = READY;
608 for (i = 0; i < nr_ops; i++) {
609 //TODO error check + cond variable to stop io from starting
610 //unless all threads are created successfully
611 pthread_create(store->iothread + i, NULL, io_loop, (void *) (store->ios + i));
613 pthread_mutex_init(&store->cache_lock, NULL);
614 return filed_loop(store);
617 int main(int argc, char **argv)
619 char *path, *spec = "";
633 for (i = 2; i < argc; i++) {
634 if (!strcmp(argv[i], "-g") && i + 1 < argc) {
640 if (!strcmp(argv[i], "-p") && i + 1 < argc) {
641 portno = strtoul(argv[i+1], NULL, 10);
646 if (!strcmp(argv[i], "-n") && i + 1 < argc) {
647 nr_ops = strtoul(argv[i+1], NULL, 10);
656 return filed(path, size, nr_ops, spec, portno);