2 * Copyright 2012 GRNET S.A. All rights reserved.
4 * Redistribution and use in source and binary forms, with or
5 * without modification, are permitted provided that the following
8 * 1. Redistributions of source code must retain the above
9 * copyright notice, this list of conditions and the following
11 * 2. Redistributions in binary form must reproduce the above
12 * copyright notice, this list of conditions and the following
13 * disclaimer in the documentation and/or other materials
14 * provided with the distribution.
16 * THIS SOFTWARE IS PROVIDED BY GRNET S.A. ``AS IS'' AND ANY EXPRESS
17 * OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED
18 * WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR
19 * PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL GRNET S.A OR
20 * CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
21 * SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT
22 * LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF
23 * USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED
24 * AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT
25 * LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN
26 * ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE
27 * POSSIBILITY OF SUCH DAMAGE.
29 * The views and conclusions contained in the software and
30 * documentation are those of the authors and should not be
31 * interpreted as representing official policies, either expressed
32 * or implied, of GRNET S.A.
36 * The Pithos File Blocker Peer (pfiled)
42 #include <sys/types.h>
53 #include <sys/sendfile.h>
55 #include <xseg/xseg.h>
56 #include <xseg/protocol.h>
58 #include "common.h" /* FIXME: */
60 #define MAX_PATH_SIZE 1024
61 #define MAX_FILENAME_SIZE 255
63 /* default concurrency level (number of threads) */
64 #define DEFAULT_NR_OPS 16
66 /* Pithos hash for the zero block
67 * FIXME: Should it be hardcoded?
70 "e3b0c44298fc1c149afbf4c8996fb92427ae41e4649b934ca495991b7852b85"
73 * Globals, holding command-line arguments
75 long cmdline_portno = -1;
76 char *cmdline_xseg_spec = NULL;
77 char *cmdline_path = NULL;
78 char *cmdline_vpath = NULL;
79 char *cmdline_pidfile = NULL;
80 int cmdline_daemon = 0;
81 long cmdline_nr_ops = DEFAULT_NR_OPS;
82 long cmdline_verbose = 0;
83 volatile unsigned int terminated = 0;
85 static int usage(char *argv0)
88 "Usage: %s <PATH> <VPATH> [-p PORT] [-g XSEG_SPEC] [-n NR_OPS] [-v]\n\n"
90 "\tPATH: path to pithos data blocks\n"
91 "\tVPATH: path to modified volume blocks\n"
92 "\tPORT: xseg port to listen for requests on\n"
93 "\tXSEG_SPEC: xseg spec as 'type:name:nr_ports:nr_requests:"
94 "request_size:extra_size:page_shift'\n"
95 "\tNR_OPS: number of outstanding xseg requests\n"
96 "\t-v: verbose mode\n",
102 /* fdcache_node flags */
103 #define READY (1 << 1)
105 /* fdcache node info */
106 struct fdcache_node {
108 volatile unsigned int ref;
109 volatile unsigned long time;
110 volatile unsigned int flags;
112 char target[MAX_FILENAME_SIZE + 1];
118 struct xseg_port *xport;
125 struct sigevent sigevent;
128 uint64_t handled_reqs;
130 struct fdcache_node *fdcache;
132 pthread_mutex_t cache_lock;
133 char path[MAX_PATH_SIZE + 1];
134 char vpath[MAX_PATH_SIZE + 1];
138 * pfiled specific structure
139 * containing information on a pending I/O operation
142 struct pfiled *pfiled;
143 struct xseg_request *req;
148 pthread_mutex_t lock;
152 static inline int isTerminate()
154 /* ta doesn't need to be taken into account, because the main loops
155 * doesn't check the terminated flag if ta is not 0.
158 * return (!ta & terminated);
166 void signal_handler(int signal)
171 static int setup_signals()
175 sigemptyset(&sa.sa_mask);
177 sa.sa_handler = signal_handler;
178 r = sigaction(SIGTERM, &sa, NULL);
181 r = sigaction(SIGINT, &sa, NULL);
184 r = sigaction(SIGQUIT, &sa, NULL);
191 static unsigned long sigaction_count;
193 static void sigaction_handler(int sig, siginfo_t *siginfo, void *arg)
198 static void log_io(char *msg, struct io *io)
200 char target[65], data[65];
201 /* null terminate name in case of req->target is less than 63 characters,
202 * and next character after name (aka first byte of next buffer) is not
205 unsigned int end = (io->req->targetlen> 64) ? 64 : io->req->targetlen;
206 unsigned int dend = (io->req->datalen > 64) ? 64 : io->req->datalen;
207 char *req_target = xseg_get_target(io->pfiled->xseg, io->req);
208 char *req_data = xseg_get_data(io->pfiled->xseg, io->req);
209 strncpy(target, req_target, end);
211 strncpy(data, req_data, 64);
215 "%s: fd:%u, op:%u offset: %llu size: %lu retval: %lu, reqstate: %u, serviced: %lu\n"
216 "target[%u]: '%s', data[%llu]:\n%s------------------\n\n",
218 (unsigned int)io->fdcacheidx, /* this is cacheidx not fd */
219 (unsigned int)io->req->op,
220 (unsigned long long)io->req->offset,
221 (unsigned long)io->req->size,
222 (unsigned long)io->retval,
223 (unsigned int)io->req->state,
224 (unsigned long)io->req->serviced,
225 (unsigned int)io->req->targetlen, target,
226 (unsigned long long)io->req->datalen, data);
229 static struct io *alloc_io(struct pfiled *pfiled)
231 xqindex idx = xq_pop_head(&pfiled->free_ops, 1);
234 return pfiled->ios + idx;
237 static inline void free_io(struct pfiled *pfiled, struct io *io)
239 xqindex idx = io - pfiled->ios;
241 xq_append_head(&pfiled->free_ops, idx, 1);
244 static void complete(struct pfiled *pfiled, struct io *io)
246 struct xseg_request *req = io->req;
247 req->state |= XS_SERVED;
249 log_io("complete", io);
250 xport p = xseg_respond(pfiled->xseg, req, pfiled->portno, X_ALLOC);
251 xseg_signal(pfiled->xseg, p);
252 __sync_fetch_and_sub(&pfiled->fdcache[io->fdcacheidx].ref, 1);
255 static void fail(struct pfiled *pfiled, struct io *io)
257 struct xseg_request *req = io->req;
258 req->state |= XS_FAILED;
261 xport p = xseg_respond(pfiled->xseg, req, pfiled->portno, X_ALLOC);
262 xseg_signal(pfiled->xseg, p);
263 if (io->fdcacheidx >= 0) {
264 __sync_fetch_and_sub(&pfiled->fdcache[io->fdcacheidx].ref, 1);
268 static void handle_unknown(struct pfiled *pfiled, struct io *io)
270 struct xseg_request *req = io->req;
271 char *data = xseg_get_data(pfiled->xseg, req);
272 snprintf(data, req->datalen, "unknown request op");
276 static int create_path(char *buf, char *path, char *target, uint32_t targetlen, int mkdirs)
280 uint32_t pathlen = strlen(path);
282 strncpy(buf, path, pathlen);
284 for (i = 0; i < 9; i+= 3) {
285 buf[pathlen + i] = target[i - (i/3)];
286 buf[pathlen + i +1] = target[i + 1 - (i/3)];
287 buf[pathlen + i + 2] = '/';
289 buf[pathlen + i + 3] = '\0';
290 if (stat(buf, &st) < 0)
291 if (mkdir(buf, 0700) < 0) {
298 strncpy(&buf[pathlen + 9], target, targetlen);
299 buf[pathlen + 9 + targetlen] = '\0';
304 static int dir_open(struct pfiled *pfiled, struct io *io,
305 char *target, uint32_t targetlen, int mode)
308 struct fdcache_node *ce = NULL;
310 char tmp[pfiled->path_len + targetlen + 10];
313 if (targetlen> MAX_FILENAME_SIZE)
318 pthread_mutex_lock(&pfiled->cache_lock);
322 for (i = 0; i < pfiled->maxfds; i++) {
323 if (pfiled->fdcache[i].ref == 0 && min > pfiled->fdcache[i].time
324 && (pfiled->fdcache[i].flags & READY)) {
325 min = pfiled->fdcache[i].time;
330 if (!strncmp(pfiled->fdcache[i].target, target, targetlen)) {
331 if (pfiled->fdcache[i].target[targetlen] == 0) {
332 ce = &pfiled->fdcache[i];
333 /* if any other io thread is currently opening
334 * the file, block until it succeeds or fails
336 if (!(ce->flags & READY)) {
337 pthread_cond_wait(&ce->cond, &pfiled->cache_lock);
338 /* when ready, restart lookup */
341 /* if successfully opened */
343 fd = pfiled->fdcache[i].fd;
347 /* else open failed for the other io thread, so
348 * it should fail for everyone waiting on this
360 /* all cache entries are currently being used */
361 pthread_mutex_unlock(&pfiled->cache_lock);
364 if (pfiled->fdcache[lru].ref){
366 printf("lru(%ld) ref not 0 (%u)\n", lru, pfiled->fdcache[lru].ref);
369 /* make room for new file */
370 ce = &pfiled->fdcache[lru];
371 /* set name here and state to not ready, for any other requests on the
372 * same target that may follow
374 strncpy(ce->target, target, targetlen);
375 ce->target[targetlen] = 0;
377 pthread_mutex_unlock(&pfiled->cache_lock);
380 if (close(ce->fd) < 0){
385 /* try opening it from pithos blocker dir */
386 if (create_path(tmp, pfiled->path, target, targetlen, 0) < 0) {
391 fd = open(tmp, O_RDWR);
393 /* try opening it from the tmp dir */
394 if (create_path(tmp, pfiled->vpath, target, targetlen, 0) < 0)
397 fd = open(tmp, O_RDWR);
399 if (create_path(tmp, pfiled->vpath, target, targetlen, 1) < 0) {
404 fd = open(tmp, O_RDWR | O_CREAT, 0600);
410 /* insert in cache a negative fd to indicate opening error to
411 * any other ios waiting for the file to open
414 /* insert in cache */
416 pthread_mutex_lock(&pfiled->cache_lock);
420 pthread_cond_broadcast(&ce->cond);
422 io->fdcacheidx = lru;
430 pfiled->handled_reqs++;
431 ce->time = pfiled->handled_reqs;
432 __sync_fetch_and_add(&ce->ref, 1);
433 pthread_mutex_unlock(&pfiled->cache_lock);
438 pthread_mutex_unlock(&pfiled->cache_lock);
442 static void handle_read_write(struct pfiled *pfiled, struct io *io)
445 struct xseg_request *req = io->req;
446 char *target = xseg_get_target(pfiled->xseg, req);
447 char *data = xseg_get_data(pfiled->xseg, req);
449 fd = dir_open(pfiled, io, target, req->targetlen, 0);
457 printf("0.%p vs %p!\n", (void *)req, (void *)io->req);
459 if (req->flags & (XF_FLUSH | XF_FUA)) {
460 /* No FLUSH/FUA support yet (O_SYNC ?).
461 * note that with FLUSH/size == 0
462 * there will probably be a (uint64_t)-1 offset */
463 complete(pfiled, io);
466 complete(pfiled, io);
473 while (req->serviced < req->datalen) {
474 r = pread(fd, data + req->serviced,
475 req->datalen - req->serviced,
476 req->offset + req->serviced);
478 req->datalen = req->serviced;
482 /* reached end of file. zero out the rest data buffer */
483 memset(data + req->serviced, 0, req->datalen - req->serviced);
484 req->serviced = req->datalen;
492 while (req->serviced < req->datalen) {
493 r = pwrite(fd, data + req->serviced,
494 req->datalen - req->serviced,
495 req->offset + req->serviced);
497 req->datalen = req->serviced;
500 fprintf(stderr, "write returned 0\n");
501 memset(data + req->serviced, 0, req->datalen - req->serviced);
502 req->serviced = req->datalen;
511 /* if fsync fails, then no bytes serviced correctly */
516 snprintf(data, req->datalen,
517 "wtf, corrupt op %u?\n", req->op);
522 if (req->serviced > 0 ) {
523 complete(pfiled, io);
526 strerror_r(errno, data, req->datalen);
532 static void handle_info(struct pfiled *pfiled, struct io *io)
534 struct xseg_request *req = io->req;
538 char *target = xseg_get_target(pfiled->xseg, req);
539 char *data = xseg_get_data(pfiled->xseg, req);
540 struct xseg_reply_info *xinfo = (struct xseg_reply_info *)data;
542 fd = dir_open(pfiled, io, target, req->targetlen, 0);
548 r = fstat(fd, &stat);
555 size = (uint64_t)stat.st_size;
558 complete(pfiled, io);
561 static void handle_copy(struct pfiled *pfiled, struct io *io)
563 struct xseg_request *req = io->req;
564 char *target = xseg_get_target(pfiled->xseg, req);
565 char *data = xseg_get_data(pfiled->xseg, req);
566 struct xseg_request_copy *xcopy = (struct xseg_request_copy *)data;
568 //FIXME is 256 enough?
569 char *buf = malloc(256);
572 dst = dir_open(pfiled, io, target, req->targetlen, 1);
574 fprintf(stderr, "fail in dst\n");
579 if (create_path(buf, pfiled->path, xcopy->target, xcopy->targetlen, 0) < 0) {
584 src = open(buf, O_RDWR);
586 XSEGLOG("fail in src %s\n", buf);
593 n = sendfile(dst, src, 0, st.st_size);
594 if (n != st.st_size) {
595 fprintf(stderr, "fail in copy\n");
601 fprintf(stderr, "fail in cp\n");
606 complete(pfiled, io);
612 static void handle_delete(struct pfiled *pfiled, struct io *io)
614 struct xseg_request *req = io->req;
615 char *buf = malloc(255);
617 char *target = xseg_get_target(pfiled->xseg, req);
619 fd = dir_open(pfiled, io, target, req->targetlen, 0);
621 fprintf(stderr, "fail in dir_open\n");
626 /* 'invalidate' cache entry */
627 if (io->fdcacheidx >= 0) {
628 pfiled->fdcache[io->fdcacheidx].fd = -1;
633 if (create_path(buf, pfiled->vpath, target, req->targetlen, 0) < 0) {
639 complete(pfiled, io);
644 static void handle_open(struct pfiled *pfiled, struct io *io)
646 struct xseg_request *req = io->req;
647 char *buf = malloc(MAX_FILENAME_SIZE + strlen("_lock"));
648 char *pathname = malloc(MAX_PATH_SIZE + MAX_FILENAME_SIZE + strlen("_lock"));
650 char *target = xseg_get_target(pfiled->xseg, req);
652 if (!buf || !pathname) {
657 strncpy(buf, target, req->targetlen);
658 strncpy(buf+req->targetlen, "_lock", strlen("_lock"));
660 if (create_path(pathname, pfiled->vpath, buf, req->targetlen + strlen("_lock"), 1) < 0) {
664 fd = open(pathname, O_CREAT | O_EXCL, S_IRWXU | S_IRUSR);
671 complete(pfiled, io);
681 static void handle_close(struct pfiled *pfiled, struct io *io)
683 struct xseg_request *req = io->req;
684 char *buf = malloc(MAX_FILENAME_SIZE + strlen("_lock"));
685 char *pathname = malloc(MAX_PATH_SIZE + MAX_FILENAME_SIZE + strlen("_lock"));
686 char *target = xseg_get_target(pfiled->xseg, req);
688 if (!buf || !pathname) {
693 strncpy(buf, target, req->targetlen);
694 strncpy(buf+req->targetlen, "_lock", strlen("_lock"));
696 if (create_path(pathname, pfiled->vpath, buf, req->targetlen + strlen("_lock"), 1) < 0) {
702 complete(pfiled, io);
712 static void dispatch(struct pfiled *pfiled, struct io *io)
714 if (cmdline_verbose) {
715 fprintf(stderr, "io: 0x%p, req: 0x%p, op %u\n",
716 (void *)io, (void *)io->req, io->req->op);
719 switch (io->req->op) {
722 handle_read_write(pfiled, io); break;
724 handle_info(pfiled, io); break;
726 handle_copy(pfiled, io); break;
728 handle_delete(pfiled, io); break;
730 handle_open(pfiled, io); break;
732 handle_close(pfiled, io); break;
736 handle_unknown(pfiled, io);
740 static void handle_accepted(struct pfiled *pfiled, struct io *io)
742 struct xseg_request *req = io->req;
744 io->state = XS_ACCEPTED;
746 dispatch(pfiled, io);
749 static struct io* wake_up_next_iothread(struct pfiled *pfiled)
751 struct io *io = alloc_io(pfiled);
754 pthread_mutex_lock(&io->lock);
755 pthread_cond_signal(&io->cond);
756 pthread_mutex_unlock(&io->lock);
761 void *io_loop(void *arg)
763 struct io *io = (struct io *) arg;
764 struct pfiled *pfiled = io->pfiled;
765 struct xseg *xseg = pfiled->xseg;
766 uint32_t portno = pfiled->portno;
767 struct xseg_request *accepted;
772 accepted = xseg_accept(xseg, portno, 0);
775 wake_up_next_iothread(pfiled);
776 handle_accepted(pfiled, io);
779 pthread_mutex_lock(&io->lock);
781 pthread_cond_wait(&io->cond, &io->lock);
782 pthread_mutex_unlock(&io->lock);
789 static struct xseg *join_or_create(char *spec)
791 struct xseg_config config;
794 (void)xseg_parse_spec(spec, &config);
795 xseg = xseg_join(config.type, config.name, "posix", NULL);
799 (void)xseg_create(&config);
800 return xseg_join(config.type, config.name, "posix", NULL);
803 static int pfiled_loop(struct pfiled *pfiled)
805 struct xseg *xseg = pfiled->xseg;
806 uint32_t portno = pfiled->portno;
807 /* GCC + pthreads glitch? */
811 for (;!(isTerminate() && xq_count(&pfiled->free_ops) == pfiled->nr_ops);) {
812 io = wake_up_next_iothread(pfiled);
813 xseg_prepare_wait(xseg, portno);
814 xseg_wait_signal(xseg, 1000000UL);
820 static int pfiled_init(struct pfiled *pfiled)
826 pfiled->sigevent.sigev_notify = SIGEV_SIGNAL;
827 pfiled->sigevent.sigev_signo = SIGIO;
828 sa.sa_sigaction = sigaction_handler;
829 sa.sa_flags = SA_SIGINFO;
831 if ((ret = sigemptyset(&sa.sa_mask))) {
832 perr(PE, 0, "[sigemptyset]");
836 if ((ret = sigaction(SIGIO, &sa, NULL))) {
837 perr(PE, 0, "[sigaction]");
838 /* FIXME: Since this is an init routine, if it fails the program will
839 * exit and clean its own stuff (mem, sigs etc). We only have to cleanup
840 * anything xseg-related
845 pfiled->nr_ops = cmdline_nr_ops;
846 pfiled->maxfds = 2 * cmdline_nr_ops;
848 pfiled->fdcache = calloc(pfiled->maxfds, sizeof(struct fdcache_node));
849 if(!pfiled->fdcache) {
851 perr(PE, 0, "could not allocate memory [fdcache]");
856 pfiled->free_bufs = calloc(pfiled->nr_ops, sizeof(xqindex));
857 if(!pfiled->free_bufs) {
859 perr(PE, 0, "could not allocate memory [free_bufs]");
863 pfiled->iothread = calloc(pfiled->nr_ops, sizeof(pthread_t));
864 if(!pfiled->iothread) {
866 perr(PE, 0, "could not allocate memory [iothreads]");
870 pfiled->ios = calloc(pfiled->nr_ops, sizeof(struct io));
873 perr(PE, 0, "could not allocate memory [ios]");
877 for (i = 0; i < pfiled->nr_ops; i++) {
878 pfiled->ios[i].pfiled = pfiled;
879 pthread_cond_init(&pfiled->ios[i].cond, NULL);
880 pthread_mutex_init(&pfiled->ios[i].lock, NULL);
883 xq_init_seq(&pfiled->free_ops, pfiled->nr_ops, pfiled->nr_ops,
886 pfiled->handled_reqs = 0;
888 strncpy(pfiled->path, cmdline_path, MAX_PATH_SIZE);
889 pfiled->path[MAX_PATH_SIZE] = 0;
891 strncpy(pfiled->vpath, cmdline_vpath, MAX_PATH_SIZE);
892 pfiled->vpath[MAX_PATH_SIZE] = 0;
894 pfiled->path_len = strlen(pfiled->path);
895 if (pfiled->path[pfiled->path_len -1] != '/'){
896 pfiled->path[pfiled->path_len] = '/';
897 pfiled->path[++pfiled->path_len]= 0;
900 pfiled->vpath_len = strlen(pfiled->vpath);
901 if (pfiled->vpath[pfiled->vpath_len -1] != '/'){
902 pfiled->vpath[pfiled->vpath_len] = '/';
903 pfiled->vpath[++pfiled->vpath_len]= 0;
906 if (xseg_initialize()) {
908 perr(PE, 0, "could not initialize xseg library");
912 pfiled->xseg = join_or_create(cmdline_xseg_spec);
915 perr(PE, 0, "could not join xseg with spec '%s'\n",
917 goto out_with_xseginit;
920 pfiled->xport = xseg_bind_port(pfiled->xseg, cmdline_portno, NULL);
921 if (!pfiled->xport) {
923 perr(PE, 0, "could not bind to xseg port %ld", cmdline_portno);
924 goto out_with_xsegjoin;
927 pfiled->portno = xseg_portno(pfiled->xseg, pfiled->xport);
928 perr(PI, 0, "filed on port %u/%u\n",
929 pfiled->portno, pfiled->xseg->config.nr_ports);
931 if (xseg_init_local_signal(pfiled->xseg, pfiled->portno) < 0){
932 perr(PE, 0, "cannot int local signals\n");
936 for (i = 0; i < pfiled->nr_ops; i++) {
937 pthread_cond_init(&pfiled->fdcache[i].cond, NULL);
938 pfiled->fdcache[i].flags = READY;
940 for (i = 0; i < pfiled->nr_ops; i++) {
942 * TODO: error check + cond variable to stop io from starting
943 * unless all threads are created successfully
945 pthread_create(pfiled->iothread + i, NULL, io_loop, (void *) (pfiled->ios + i));
947 pthread_mutex_init(&pfiled->cache_lock, NULL);
952 xseg_leave(pfiled->xseg);
959 static int safe_atoi(char *s)
964 l = strtol(s, &endp, 10);
965 if (s != endp && *endp == '\0')
971 static void parse_cmdline(int argc, char **argv)
973 char *argv0 = argv[0];
979 c = getopt(argc, argv, "dhp:n:g:vf:");
985 perr(PFE, 0, "Unknown option: -%c", optopt);
988 perr(PFE, 0, "Option -%c requires an argument",
996 cmdline_portno = safe_atoi(optarg);
999 cmdline_nr_ops = safe_atoi(optarg);
1002 /* FIXME: Max length of spec? strdup, eww */
1003 cmdline_xseg_spec = strdup(optarg);
1004 if (!cmdline_xseg_spec)
1005 perr(PFE, 0, "out of memory");
1008 cmdline_verbose = 1;
1014 /* FIXME: Max length of spec? strdup, eww */
1015 cmdline_pidfile = strdup(optarg);
1016 if (!cmdline_pidfile)
1017 perr(PFE, 0, "out of memory");
1025 /* Sanity check for all arguments */
1026 if (cmdline_portno < 0) {
1028 perr(PFE, 0, "no or invalid port specified");
1030 if (cmdline_nr_ops < 1) {
1032 perr(PFE, 0, "specified outstanding request count is invalid");
1034 if (!cmdline_xseg_spec) {
1036 perr(PFE, 0, "xseg specification is mandatory");
1041 perr(PFE, 0, "path and vpath specification is mandatory");
1044 cmdline_path = strdup(argv[0]);
1046 perr(PFE, 0, "out of memory");
1048 cmdline_vpath = strdup(argv[1]);
1050 perr(PFE, 0, "out of memory");
1053 int pidfile_remove(char *path, int fd)
1056 return (unlink(path));
1059 int pidfile_write(int pid_fd)
1062 snprintf(buf, sizeof(buf), "%ld", syscall(SYS_gettid));
1065 lseek(pid_fd, 0, SEEK_SET);
1066 int ret = write(pid_fd, buf, strlen(buf));
1070 int pidfile_read(char *path, pid_t *pid)
1072 char buf[16], *endptr;
1075 int fd = open(path, O_RDONLY);
1078 int ret = read(fd, buf, 15);
1084 *pid = strtol(buf, &endptr, 10);
1085 if (endptr != &buf[ret]){
1093 int pidfile_open(char *path, pid_t *old_pid)
1096 int fd = open(path, O_CREAT|O_EXCL|O_WRONLY);
1098 if (errno == -EEXIST)
1099 pidfile_read(path, old_pid);
1104 int main(int argc, char **argv)
1106 int pid_fd = -1, r = 0;
1108 struct pfiled pfiled;
1110 init_perr("pfiled");
1111 parse_cmdline(argc, argv);
1113 perr(PI, 0, "p = %ld, nr_ops = %lu\n", cmdline_portno, cmdline_nr_ops);
1115 if (cmdline_pidfile){
1116 pid_fd = pidfile_open(cmdline_pidfile, &old_pid);
1119 perr(PFE, 0, "Daemon already running, pid: %d.", old_pid);
1121 perr(PFE, 0, "Cannot open or create pidfile");
1127 if (cmdline_daemon){
1128 if (daemon(0, 1) < 0){
1129 perr(PFE, 0, "Cannot daemonize");
1136 pidfile_write(pid_fd);
1139 if (pfiled_init(&pfiled) < 0){
1144 r = pfiled_loop(&pfiled);
1147 pidfile_remove(cmdline_pidfile, pid_fd);