11 #include <xseg/xseg.h>
16 #include <sys/syscall.h>
18 /* maybe add this to store struct */
19 #define objsize (4*1024*1024)
20 #define MAX_VOL_NAME 242
22 static int usage(void)
24 printf("Usage: ./sosd <path_to_disk_image> [options]\n"
25 "Options: [-p portno]\n"
26 " [-g type:name:nr_ports:nr_requests:request_size:extra_size:page_shift]\n"
27 " [-n nr_parallel_ops]\n");
32 struct xseg_request *req;
34 struct sos_request sos_req;
35 char objname[MAX_VOL_NAME +1 + 12 + 1];
41 struct xseg_port *xport;
48 struct xq resubmit_ops;
58 static void sigaction_handler(int sig, siginfo_t *siginfo, void *arg)
63 static void signal_self(struct store *store)
65 union sigval sigval = {0};
66 pid_t me = store->pid;
67 if (sigqueue(me, SIGIO, sigval) < 0)
71 static int wait_signal(struct store *store)
76 uint32_t usec_timeout = 5000;
78 ts.tv_sec = usec_timeout / 1000000;
79 ts.tv_nsec = 1000 * (usec_timeout - ts.tv_sec * 1000000);
81 r = sigtimedwait(&store->signal_set, &siginfo, &ts);
85 return siginfo.si_signo;
89 static struct io *alloc_io(struct store *store)
91 xqindex idx = xq_pop_head(&store->free_ops);
94 return store->ios + idx;
97 static inline void free_io(struct store *store, struct io *io)
99 xqindex idx = io - store->ios;
101 xq_append_head(&store->free_ops, idx);
105 static void resubmit_io(struct store *store, struct io *io)
107 xqindex idx = io - store->ios;
108 xq_append_tail(&store->resubmit_ops, idx);
111 static struct io* get_resubmitted_io(struct store *store)
113 xqindex idx = xq_pop_head(&store->resubmit_ops);
116 return store->ios + idx;
119 static void log_io(char *msg, struct io *io)
121 char name[64], data[64];
122 /* null terminate name in case of req->name is less than 63 characters,
123 * and next character after name (aka first byte of next buffer) is not
126 unsigned int end = (io->req->namesize > 63) ? 63 : io->req->namesize;
127 strncpy(name, io->req->name, end);
129 strncpy(data, io->req->data, 63);
131 printf("%s: sos req id:%u, op:%u %llu:%lu serviced: %lu, retval: %lu, reqstate: %u\n"
132 "name[%u]:'%s', data[%llu]:\n%s------------------\n\n",
134 (unsigned int)io->sos_req.id,
135 (unsigned int)io->req->op,
136 (unsigned long long)io->sos_req.offset,
137 (unsigned long)io->sos_req.size,
138 (unsigned long)io->req->serviced,
139 (unsigned long)io->retval,
140 (unsigned int)io->req->state,
141 (unsigned int)io->req->namesize, name,
142 (unsigned long long)io->req->datasize, data);
145 static void complete(struct store *store, struct io *io)
147 struct xseg_request *req = io->req;
151 gettimeofday(&end, NULL);
152 timersub(&end, &io->start, &end);
153 us = end.tv_sec*1000000 +end.tv_usec;
154 printf("sosd: Request %lu completed after %lu us\n", io->sos_req.id, us);
157 req->state |= XS_SERVED;
158 log_io("complete", io);
159 xseg_respond(store->xseg, req->portno, req);
160 xseg_signal(store->xseg, req->portno);
164 static void fail(struct store *store, struct io *io)
166 struct xseg_request *req = io->req;
167 req->state |= XS_ERROR;
169 xseg_respond(store->xseg, req->portno, req);
170 xseg_signal(store->xseg, req->portno);
174 static void pending(struct store *store, struct io *io)
176 io->req->state = XS_PENDING;
179 static void handle_unknown(struct store *store, struct io *io)
181 struct xseg_request *req = io->req;
182 snprintf(req->data, req->datasize, "unknown request op");
186 static int32_t get_sos_op(uint32_t xseg_op)
198 static uint32_t get_sos_flags(uint32_t xseg_flags)
201 if (xseg_flags & XF_FLUSH) {
204 if (xseg_flags & XF_FUA) {
210 static int calculate_sosreq(struct xseg_request *xseg_req, struct sos_request *sos_req)
216 /* get object name from offset in volume */
218 suffix = (unsigned int) ((xseg_req->offset+xseg_req->serviced) / (uint64_t)objsize) ;
219 // printf("suffix: %u\n", suffix);
220 if (xseg_req->namesize > MAX_VOL_NAME){
221 printf("xseg_req namesize > MAX_VOL_NAME\n");
224 strncpy(buf, xseg_req->name, xseg_req->namesize);
225 buf[xseg_req->namesize] = '_';
226 r = snprintf(buf+xseg_req->namesize+1, 13, "%012u", suffix);
230 //sos_req->name = buf;
231 sos_req->namesize = xseg_req->namesize+1+12;
233 /* offset should be set to offset in object */
234 sos_req->offset = (xseg_req->offset + xseg_req->serviced) % objsize;
235 /* sos_req offset + sos_req size < objsize always
236 * request data up to the end of object.
238 sos_req->size = (xseg_req->datasize - xseg_req->serviced) ; /* should this be xseg_req->size ? */
239 if (sos_req->size > objsize - sos_req->offset)
240 sos_req->size = objsize - sos_req->offset;
241 /* this should have been checked before this call */
242 if (xseg_req->serviced < xseg_req->datasize)
243 sos_req->data = xseg_req->data + xseg_req->serviced;
246 // printf("name: %s, size: %lu, offset: %lu, data:%s\n", sos_req->name,
247 // sos_req->size, sos_req->offset, sos_req->data);
251 static void prepare_sosreq(struct store *store, struct io *io)
253 struct xseg_request *xseg_req = io->req;
254 struct sos_request *sos_req = &io->sos_req;
255 sos_req->flags = get_sos_flags(xseg_req->flags);
256 sos_req->state = S_PENDING;
258 sos_req->op = get_sos_op(xseg_req->op);
259 sos_req->priv = store;
260 sos_req->name = io->objname;
263 static inline void prepare_io(struct store *store, struct io *io)
265 prepare_sosreq(store, io);
266 /* Assign io id to sos_req id. This can be done as an initialization of
267 * ios, to avoid reseting it every time */
268 io->sos_req.id = io - store->ios;
272 static void handle_resubmit(struct store *store, struct io *io);
274 static void complete_rw(struct store *store, struct io *io)
277 struct xseg_request *req = io->req;
278 struct sos_request *sos_req = &io->sos_req;
279 if (req->state == XS_ACCEPTED) {
280 /* should not happen */
285 req->serviced += io->retval;
286 else if (io->retval == 0) {
287 /* reached end of object. zero out rest of data
288 * requested from this object
290 memset(sos_req->data, 0, sos_req->size);
291 req->serviced += sos_req->size;
293 else if (io->retval == -2) {
294 /* object not found. return zeros instead */
295 memset(sos_req->data, 0, sos_req->size);
296 req->serviced += sos_req->size;
303 /* request completed ? */
304 if (req->serviced >= req->datasize) {
310 printf("0.%p vs %p!\n", (void *)req, (void *)io->req);
312 /* should not happen */
320 log_io("resubmitting", io);
321 resubmit_io(store, io);
325 snprintf(req->data, req->datasize,
326 "wtf, corrupt op %u?\n", req->op);
332 static void handle_read_write(struct store *store, struct io *io)
335 struct xseg_request *req = io->req;
336 struct sos_request *sos_req = &io->sos_req;
337 struct io *resubmit_io;
340 printf("0.%p vs %p!\n", (void *)req, (void *)io->req);
342 prepare_io(store, io);
344 if (req->flags & XF_FLUSH) {
345 /* note that with FLUSH/size == 0
346 * there will probably be a (uint64_t)-1 offset */
348 /* size must be zero */
350 /* all these should be irrelevant on a flush request */
352 sos_req->namesize = 0;
356 * make sure all pending requests are completed and then
357 * perform flush request to flush them to disk.
359 while (xq_size(&store->free_ops) != store->nr_ops){
361 /* handle any possible resubmissions */
362 resubmit_io = get_resubmitted_io(store);
364 handle_resubmit(store, resubmit_io);
365 resubmit_io = get_resubmitted_io(store);
368 r = sos_submit(store->sos, sos_req);
380 r = calculate_sosreq(req, sos_req);
389 //log_io("submit", io);
391 r = sos_submit(store->sos, sos_req);
394 snprintf(req->data, req->datasize,
395 "wtf, corrupt op %u?\n", req->op);
401 strerror_r(errno, req->data, req->datasize);
407 static void handle_returned(struct store *store, struct io *io)
409 io->retval = io->sos_req.retval;
410 switch (io->req->op){
413 complete_rw(store, io);
416 if (io->sos_req.state & S_FAILED)
423 /* this is safe for now, as long as callback is only called once.
424 * if callback gets called, then sos_request has been completed and no race
427 static int sos_cb(struct sos_request *sos_req, unsigned long event)
429 struct store *store = (struct store *) sos_req->priv;
430 struct io *io = (struct io*) store->ios + sos_req->id;
432 if (event == S_NOTIFY_FAIL){
433 sos_req->state = S_FAILED;
435 else if (event == S_NOTIFY_ACK) {
436 sos_req->state = S_ACKED;
438 else if (event == S_NOTIFY_COMMIT){
439 sos_req->state = S_COMMITED;
441 handle_returned(store, io);
446 static void dispatch(struct store *store, struct io *io)
448 switch (io->req->op) {
451 handle_read_write(store, io); break;
454 handle_unknown(store, io);
458 static void handle_resubmit(struct store *store, struct io *io)
463 static void handle_accepted(struct store *store, struct io *io)
465 struct xseg_request *req = io->req;
467 req->state = XS_ACCEPTED;
469 //log_io("accepted", io);
470 gettimeofday(&io->start, NULL);
474 static int sosd_loop(struct store *store)
476 struct xseg *xseg = store->xseg;
477 uint32_t portno = store->portno;
478 struct io *io, *resubmit_io;
479 struct xseg_request *accepted;
483 xseg_prepare_wait(xseg, portno);
484 io = alloc_io(store);
486 accepted = xseg_accept(xseg, portno);
488 xseg_cancel_wait(xseg, portno);
490 handle_accepted(store, io);
494 resubmit_io = get_resubmitted_io(store);
496 xseg_cancel_wait(xseg, portno);
497 handle_resubmit(store, resubmit_io);
499 if (!accepted && !resubmit_io)
500 xseg_wait_signal(xseg, portno, 10000);
506 static struct xseg *join(char *spec)
508 struct xseg_config config;
511 (void)xseg_parse_spec(spec, &config);
512 xseg = xseg_join(config.type, config.name);
516 (void)xseg_create(&config);
517 return xseg_join(config.type, config.name);
520 static int sosd(char *path, unsigned long size, uint32_t nr_ops,
521 char *spec, long portno)
525 store = malloc(sizeof(struct store));
531 store->sos = sos_init(sos_cb);
533 fprintf(stderr, "SOS init failed\n");
543 store->pid = syscall(SYS_gettid);
545 if (sigemptyset(&store->signal_set))
546 perror("sigemptyset");
548 if (sigaddset(&store->signal_set, SIGIO))
552 store->nr_ops = nr_ops;
553 store->free_bufs = calloc(nr_ops, sizeof(xqindex));
554 if (!store->free_bufs)
557 store->resubmit_bufs = calloc(nr_ops, sizeof(xqindex));
558 if (!store->resubmit_bufs)
561 store->ios = calloc(nr_ops, sizeof(struct io));
568 xq_init_seq(&store->free_ops, nr_ops, nr_ops, store->free_bufs);
569 xq_init_empty(&store->resubmit_ops, nr_ops, store->resubmit_bufs);
572 if (xseg_initialize("posix")) {
573 printf("cannot initialize library\n");
576 store->xseg = join(spec);
580 store->xport = xseg_bind_port(store->xseg, portno);
582 printf("cannot bind to port %ld\n", portno);
586 store->portno = xseg_portno(store->xseg, store->xport);
587 printf("sosd on port %u/%u\n",
588 store->portno, store->xseg->config.nr_ports);
590 return sosd_loop(store);
593 int main(int argc, char **argv)
595 char *path, *spec = "";
600 unsigned int debug_level = 0;
610 for (i = 2; i < argc; i++) {
611 if (!strcmp(argv[i], "-g") && i + 1 < argc) {
617 if (!strcmp(argv[i], "-p") && i + 1 < argc) {
618 portno = strtoul(argv[i+1], NULL, 10);
623 if (!strcmp(argv[i], "-p") && i + 1 < argc) {
624 nr_ops = strtoul(argv[i+1], NULL, 10);
628 if (!strcmp(argv[i], "-v") ) {
634 sos_set_debug_level(debug_level);
639 return sosd(path, size, nr_ops, spec, portno);