code.grnet.gr Git - archipelago/blob - xseg/peers/user/mpeer.c

   1 #define _GNU_SOURCE
   2 #include <stdio.h>
   3 #include <stdlib.h>
   4 #include <sys/types.h>
   5 #include <unistd.h>
   6 #include <xseg/xseg.h>
   7 #include <pthread.h>
   8 #include <mpeer.h>
   9 #include <sys/syscall.h>
  10 #include <sys/time.h>
  11 #include <signal.h>
  12
  13 #define REARRANGE(__fun_name__, __format__, ...) __format__ "%s", __fun_name__, ##__VA_ARGS__
  14 #define LOG(level, ...)                                              \
  15                 do {                                                               \
  16                         if (level <=  verbose) {                           \
  17                                 fprintf(stderr, "%s: "  REARRANGE( __func__ , ## __VA_ARGS__, "" )); \
  18                         }                                                          \
  19                 }while (0)
  20
  21
  22 unsigned int verbose = 0;
  23
  24 struct thread {
  25         struct peerd *peer;
  26         pthread_t tid;
  27         pthread_cond_t cond;
  28         pthread_mutex_t lock;
  29         void (*func)(void *arg);
  30         void *arg;
  31 };
  32
  33
  34 inline int canDefer(struct peerd *peer)
  35 {
  36         return !(peer->defer_portno == NoPort);
  37 }
  38
  39 void print_req(struct xseg *xseg, struct xseg_request *req)
  40 {
  41         char target[64], data[64];
  42         char *req_target, *req_data;
  43         unsigned int end = (req->targetlen> 63) ? 63 : req->targetlen;
  44         req_target = xseg_get_target(xseg, req);
  45         req_data = xseg_get_data(xseg, req);
  46
  47         if (1) {
  48                 strncpy(target, req_target, end);
  49                 target[end] = 0;
  50                 strncpy(data, req_data, 63);
  51                 data[63] = 0;
  52                 printf("req id:%lu, op:%u %llu:%lu serviced: %lu, reqstate: %u\n"
  53                                 "src: %u, st: %u, dst: %u dt: %u\n"
  54                                 "target[%u]:'%s', data[%llu]:\n%s------------------\n\n",
  55                                 (unsigned long)(req),
  56                                 (unsigned int)req->op,
  57                                 (unsigned long long)req->offset,
  58                                 (unsigned long)req->size,
  59                                 (unsigned long)req->serviced,
  60                                 (unsigned int)req->state,
  61                                 (unsigned int)req->src_portno,
  62                                 (unsigned int)req->src_transit_portno,
  63                                 (unsigned int)req->dst_portno,
  64                                 (unsigned int)req->dst_transit_portno,
  65                                 (unsigned int)req->targetlen, target,
  66                                 (unsigned long long)req->datalen, data);
  67         }
  68 }
  69 void log_pr(char *msg, struct peer_req *pr)
  70 {
  71         char target[64], data[64];
  72         char *req_target, *req_data;
  73         struct peerd *peer = pr->peer;
  74         struct xseg *xseg = pr->peer->xseg;
  75         req_target = xseg_get_target(xseg, pr->req);
  76         req_data = xseg_get_data(xseg, pr->req);
  77         /* null terminate name in case of req->target is less than 63 characters,
  78          *          * and next character after name (aka first byte of next buffer) is not
  79          *                   * null
  80          *                            */
  81         unsigned int end = (pr->req->targetlen> 63) ? 63 : pr->req->targetlen;
  82         if (verbose) {
  83                 strncpy(target, req_target, end);
  84                 target[end] = 0;
  85                 strncpy(data, req_data, 63);
  86                 data[63] = 0;
  87                 printf("%s: req id:%u, op:%u %llu:%lu serviced: %lu, retval: %lu, reqstate: %u\n"
  88                                 "target[%u]:'%s', data[%llu]:\n%s------------------\n\n",
  89                                 msg,
  90                                 (unsigned int)(pr - peer->peer_reqs),
  91                                 (unsigned int)pr->req->op,
  92                                 (unsigned long long)pr->req->offset,
  93                                 (unsigned long)pr->req->size,
  94                                 (unsigned long)pr->req->serviced,
  95                                 (unsigned long)pr->retval,
  96                                 (unsigned int)pr->req->state,
  97                                 (unsigned int)pr->req->targetlen, target,
  98                                 (unsigned long long)pr->req->datalen, data);
  99         }
 100 }
 101
 102 inline struct peer_req *alloc_peer_req(struct peerd *peer)
 103 {
 104         xqindex idx = xq_pop_head(&peer->free_reqs, 1);
 105         if (idx == Noneidx)
 106                 return NULL;
 107         return peer->peer_reqs + idx;
 108 }
 109
 110 inline void free_peer_req(struct peerd *peer, struct peer_req *pr)
 111 {
 112         xqindex idx = pr - peer->peer_reqs;
 113         pr->req = NULL;
 114         xq_append_head(&peer->free_reqs, idx, 1);
 115 }
 116
 117 inline static struct thread* alloc_thread(struct peerd *peer)
 118 {
 119         xqindex idx = xq_pop_head(&peer->threads, 1);
 120         if (idx == Noneidx)
 121                 return NULL;
 122         return peer->thread + idx;
 123 }
 124
 125 inline static void free_thread(struct peerd *peer, struct thread *t)
 126 {
 127         xqindex idx = t - peer->thread;
 128         xq_append_head(&peer->threads, idx, 1);
 129 }
 130
 131
 132 inline static void __wake_up_thread(struct thread *t)
 133 {
 134         pthread_mutex_lock(&t->lock);
 135         pthread_cond_signal(&t->cond);
 136         pthread_mutex_unlock(&t->lock);
 137 }
 138
 139 inline static void wake_up_thread(struct thread* t)
 140 {
 141         if (t){
 142                 __wake_up_thread(t);
 143         }
 144 }
 145
 146 inline static int wake_up_next_thread(struct peerd *peer)
 147 {
 148         //struct thread *t = alloc_thread(peer);
 149         //wake_up_thread(t);
 150         //return t;
 151         return (xseg_signal(peer->xseg, peer->portno));
 152 }
 153
 154 struct timeval resp_start, resp_end, resp_accum = {0, 0};
 155 uint64_t responds = 0;
 156 void get_responds_stats(){
 157                 printf("Time waiting respond %lu.%06lu sec for %llu times.\n",
 158                                 //(unsigned int)(t - peer->thread),
 159                                 resp_accum.tv_sec, resp_accum.tv_usec, responds);
 160 }
 161
 162 //FIXME error check
 163 void fail(struct peerd *peer, struct peer_req *pr)
 164 {
 165         struct xseg_request *req = pr->req;
 166         uint32_t p;
 167         LOG(5, "failing req %u\n", (unsigned int) (pr - peer->peer_reqs));
 168         req->state |= XS_FAILED;
 169         //xseg_set_req_data(peer->xseg, pr->req, NULL);
 170         p = xseg_respond(peer->xseg, req, peer->portno, X_ALLOC);
 171         xseg_signal(peer->xseg, p);
 172         free_peer_req(peer, pr);
 173         wake_up_next_thread(peer);
 174 }
 175
 176 //FIXME error check
 177 void complete(struct peerd *peer, struct peer_req *pr)
 178 {
 179         struct xseg_request *req = pr->req;
 180         uint32_t p;
 181         req->state |= XS_SERVED;
 182         //xseg_set_req_data(peer->xseg, pr->req, NULL);
 183         //gettimeofday(&resp_start, NULL);
 184         p = xseg_respond(peer->xseg, req, peer->portno, X_ALLOC);
 185         //gettimeofday(&resp_end, NULL);
 186         //responds++;
 187         //timersub(&resp_end, &resp_start, &resp_end);
 188         //timeradd(&resp_end, &resp_accum, &resp_accum);
 189         //printf("xseg_signal: %u\n", p);
 190         xseg_signal(peer->xseg, p);
 191         free_peer_req(peer, pr);
 192         wake_up_next_thread(peer);
 193 }
 194
 195 void pending(struct peerd *peer, struct peer_req *pr)
 196 {
 197                 pr->req->state = XS_PENDING;
 198 }
 199
 200 static void handle_accepted(struct peerd *peer, struct peer_req *pr)
 201 {
 202         struct xseg_request *req = pr->req;
 203         LOG(4, "Handle accepted \n");
 204         req->serviced = 0;
 205         req->state = XS_ACCEPTED;
 206         pr->retval = 0;
 207         dispatch(peer, pr);
 208 }
 209
 210 static void handle_received(struct peerd *peer, struct peer_req *pr)
 211 {
 212         //struct xseg_request *req = pr->req;
 213         //assert req->state != XS_ACCEPTED;
 214         LOG(4, "Handle received \n");
 215         dispatch(peer, pr);
 216
 217 }
 218 struct timeval sub_start, sub_end, sub_accum = {0, 0};
 219 uint64_t submits = 0;
 220 void get_submits_stats(){
 221                 printf("Time waiting submit %lu.%06lu sec for %llu times.\n",
 222                                 //(unsigned int)(t - peer->thread),
 223                                 sub_accum.tv_sec, sub_accum.tv_usec, submits);
 224 }
 225
 226 int submit_peer_req(struct peerd *peer, struct peer_req *pr)
 227 {
 228         uint32_t ret;
 229         struct xseg_request *req = pr->req;
 230         // assert req->portno == peer->portno ?
 231         //TODO small function with error checking
 232         LOG (5, "submitting peer req %u\n", (unsigned int)(pr - peer->peer_reqs));
 233         ret = xseg_set_req_data(peer->xseg, req, (void *)(pr));
 234         if (ret < 0)
 235                 return -1;
 236         //printf("pr: %x , req_data: %x \n", pr, xseg_get_req_data(peer->xseg, req));
 237         //gettimeofday(&sub_start, NULL);
 238         ret = xseg_submit(peer->xseg, req, peer->portno, X_ALLOC);
 239         //gettimeofday(&sub_end, NULL);
 240         //submits++;
 241         //timersub(&sub_end, &sub_start, &sub_end);
 242         //timeradd(&sub_end, &sub_accum, &sub_accum);
 243         if (ret == NoPort)
 244                 return -1;
 245         xseg_signal(peer->xseg, ret);
 246         return 0;
 247 }
 248
 249 int thread_execute(struct peerd *peer, void (*func)(void *arg), void *arg)
 250 {
 251         struct thread *t = alloc_thread(peer);
 252         if (t) {
 253                 t->func = func;
 254                 t->arg = arg;
 255                 wake_up_thread(t);
 256                 return 0;
 257         } else
 258                 // we could hijack a thread
 259                 return -1;
 260 }
 261
 262 static void* thread_loop(void *arg)
 263 {
 264         struct thread *t = (struct thread *) arg;
 265         struct peerd *peer = t->peer;
 266         struct xseg *xseg = peer->xseg;
 267         uint32_t portno = peer->portno;
 268         struct peer_req *pr;
 269         uint64_t threshold=1000;
 270         pid_t pid =syscall(SYS_gettid);
 271         uint64_t loops;
 272         struct xseg_request *accepted, *received;
 273         int r;
 274
 275         printf("thread %u\n",  (unsigned int) (t- peer->thread));
 276
 277         LOG(0, "Thread %u has tid %u.\n", (unsigned int) (t- peer->thread), pid);
 278         xseg_init_local_signal(xseg, portno);
 279         for (;;) {
 280                 if (t->func) {
 281                         LOG(5, "Thread %u executes function\n", (unsigned int) (t- peer->thread));
 282                         xseg_cancel_wait(xseg, portno);
 283                         t->func(t->arg);
 284                         t->func = NULL;
 285                         t->arg = NULL;
 286                         continue;
 287                 }
 288
 289                 for(loops= threshold; loops > 0; loops--) {
 290                         accepted = NULL;
 291                         received = NULL;
 292                         if (loops == 1)
 293                                 xseg_prepare_wait(xseg, portno);
 294
 295 //                      if (xq_count(&peer->xport->request_queue)){
 296                                 pr = alloc_peer_req(peer);
 297                                 if (pr) {
 298                                         accepted = xseg_accept(xseg, peer->portno);
 299                                         LOG(5, "Thread %u accepted\n", (unsigned int) (t- peer->thread));
 300                                         if (accepted) {
 301                                                 pr->req = accepted;
 302                                                 xseg_cancel_wait(xseg, portno);
 303                                                 wake_up_next_thread(peer);
 304                                                 handle_accepted(peer, pr);
 305                                                 loops = threshold;
 306                                         }
 307                                         else {
 308                                                 free_peer_req(peer, pr);
 309                                         }
 310                                 }
 311 //                      }
 312 //                      if (xq_count(&peer->xport->reply_queue)){
 313                                 received = xseg_receive(xseg, peer->portno);
 314                                 if (received) {
 315                                         //printf("received req id: %u\n", received - xseg->requests);
 316                                         //print_req(peer->xseg, received);
 317                                         r =  xseg_get_req_data(xseg, received, (void **) &pr);
 318                                         if (r < 0 || !pr){
 319                                                 //FIXME what to do here ?
 320                                                 LOG(0, "Received request with no pr data\n");
 321                                                 xseg_respond(peer->xseg, received, peer->portno, X_ALLOC);
 322                                         }
 323                                         //fail(peer, received);
 324                                         //assert pr->req == received;
 325                                         xseg_cancel_wait(xseg, portno);
 326                                         wake_up_next_thread(peer);
 327                                         handle_received(peer, pr);
 328                                         loops = threshold;
 329                                 }
 330 //                      }
 331                 }
 332                 LOG(1, "Thread %u goes to sleep\n", (unsigned int) (t- peer->thread));
 333                 xseg_wait_signal(xseg, 10000000UL);
 334                 xseg_cancel_wait(xseg, portno);
 335                 LOG(1, "Thread %u woke up\n", (unsigned int) (t- peer->thread));
 336         }
 337         return NULL;
 338 }
 339
 340 void defer_request(struct peerd *peer, struct peer_req *pr)
 341 {
 342         // assert canDefer(peer);
 343 //      xseg_submit(peer->xseg, peer->defer_portno, pr->req);
 344 //      xseg_signal(peer->xseg, peer->defer_portno);
 345 //      free_peer_req(peer, pr);
 346 }
 347
 348 static int peerd_loop(struct peerd *peer)
 349 {
 350         if (peer->interactive_func)
 351                 peer->interactive_func();
 352         for (;;) {
 353                 pthread_join(peer->thread[0].tid, NULL);
 354         }
 355         return 0;
 356 }
 357
 358 static struct xseg *join(char *spec)
 359 {
 360         struct xseg_config config;
 361         struct xseg *xseg;
 362
 363         (void)xseg_parse_spec(spec, &config);
 364         xseg = xseg_join(config.type, config.name, "pthread", NULL);
 365         if (xseg)
 366                 return xseg;
 367
 368         (void)xseg_create(&config);
 369         return xseg_join(config.type, config.name, "pthread", NULL);
 370 }
 371
 372 int peerd_start_threads(struct peerd *peer)
 373 {
 374         int i;
 375         uint32_t nr_threads = peer->nr_threads;
 376         //TODO err check
 377         for (i = 0; i < nr_threads; i++) {
 378                 peer->thread[i].peer = peer;
 379                 pthread_cond_init(&peer->thread[i].cond,NULL);
 380                 pthread_mutex_init(&peer->thread[i].lock, NULL);
 381                 pthread_create(&peer->thread[i].tid, NULL, thread_loop, (void *)(peer->thread + i));
 382                 peer->thread[i].func = NULL;
 383                 peer->thread[i].arg = NULL;
 384
 385         }
 386         return 0;
 387 }
 388
 389 static struct peerd* peerd_init(uint32_t nr_ops, char* spec, long portno, uint32_t nr_threads, uint32_t defer_portno)
 390 {
 391         int i;
 392         struct peerd *peer;
 393         peer = malloc(sizeof(struct peerd));
 394         if (!peer) {
 395                 perror("malloc");
 396                 return NULL;
 397         }
 398         peer->nr_ops = nr_ops;
 399         peer->defer_portno = defer_portno;
 400         peer->nr_threads = nr_threads;
 401
 402         peer->thread = calloc(nr_threads, sizeof(struct thread));
 403         if (!peer->thread)
 404                 goto malloc_fail;
 405         peer->peer_reqs = calloc(nr_ops, sizeof(struct peer_req));
 406         if (!peer->peer_reqs){
 407 malloc_fail:
 408                 perror("malloc");
 409                 return NULL;
 410         }
 411
 412         if (!xq_alloc_seq(&peer->free_reqs, nr_ops, nr_ops))
 413                 goto malloc_fail;
 414         if (!xq_alloc_empty(&peer->threads, nr_threads))
 415                 goto malloc_fail;
 416
 417         if (xseg_initialize()){
 418                 printf("cannot initialize library\n");
 419                 return NULL;
 420         }
 421         peer->xseg = join(spec);
 422         if (!peer->xseg)
 423                 return NULL;
 424
 425         peer->xport = xseg_bind_port(peer->xseg, portno);
 426         if (!peer->xport){
 427                 printf("cannot bind to port %ld\n", portno);
 428                 return NULL;
 429         }
 430         printf("%lx\n", (unsigned long) peer->xport);
 431         peer->portno = xseg_portno(peer->xseg, peer->xport);
 432         printf("Peer on port %u/%u\n", peer->portno,
 433                         peer->xseg->config.nr_ports);
 434
 435         for (i = 0; i < nr_ops; i++) {
 436                 peer->peer_reqs[i].peer = peer;
 437                 peer->peer_reqs[i].req = NULL;
 438                 peer->peer_reqs[i].retval = 0;
 439                 peer->peer_reqs[i].priv = NULL;
 440         }
 441         peer->interactive_func = NULL;
 442         peerd_start_threads(peer);
 443         return peer;
 444 }
 445
 446
 447 int main(int argc, const char *argv[])
 448 {
 449         struct peerd *peer = NULL;
 450         //parse args
 451         char *spec = "";
 452         int i, r;
 453         long portno = -1;
 454         //set defaults here
 455         uint32_t nr_ops = 16;
 456         uint32_t nr_threads = 16 ;
 457         unsigned int debug_level = 0;
 458         uint32_t defer_portno = NoPort;
 459
 460         //capture here -g spec, -n nr_ops, -p portno, -t nr_threads -v verbose level
 461         // -dp xseg_portno to defer blocking requests
 462         //maybe -l log file ?
 463         //TODO print messages on arg parsing error
 464         LOG(5, "Main thread has tid %ld.\n", syscall(SYS_gettid));
 465
 466         for (i = 1; i < argc; i++) {
 467                 if (!strcmp(argv[i], "-g") && i + 1 < argc) {
 468                         spec = argv[i+1];
 469                         i += 1;
 470                         continue;
 471                 }
 472
 473                 if (!strcmp(argv[i], "-p") && i + 1 < argc) {
 474                         portno = strtoul(argv[i+1], NULL, 10);
 475                         i += 1;
 476                         continue;
 477                 }
 478
 479                 if (!strcmp(argv[i], "-n") && i + 1 < argc) {
 480                         nr_ops = strtoul(argv[i+1], NULL, 10);
 481                         i += 1;
 482                         continue;
 483                 }
 484                 if (!strcmp(argv[i], "-v") && i + 1 < argc ) {
 485                         debug_level = atoi(argv[i+1]);
 486                         i += 1;
 487                         continue;
 488                 }
 489                 if (!strcmp(argv[i], "-t") && i + 1 < argc ) {
 490                         nr_threads = strtoul(argv[i+1], NULL, 10);
 491                         i += 1;
 492                         continue;
 493                 }
 494                 if (!strcmp(argv[i], "-dp") && i + 1 < argc ) {
 495                         defer_portno = strtoul(argv[i+1], NULL, 10);
 496                         i += 1;
 497                         continue;
 498                 }
 499
 500         }
 501
 502         //TODO perform argument sanity checks
 503         verbose = debug_level;
 504
 505         //TODO err check
 506         peer = peerd_init(nr_ops, spec, portno, nr_threads, defer_portno);
 507         r = custom_peer_init(peer, argc, argv);
 508 //      peerd_start_threads(peer);
 509         return peerd_loop(peer);
 510 }