libustcomm/ustcomm.c

   1 /* Copyright (C) 2009  Pierre-Marc Fournier
   2  *
   3  * This library is free software; you can redistribute it and/or
   4  * modify it under the terms of the GNU Lesser General Public
   5  * License as published by the Free Software Foundation; either
   6  * version 2.1 of the License, or (at your option) any later version.
   7  *
   8  * This library is distributed in the hope that it will be useful,
   9  * but WITHOUT ANY WARRANTY; without even the implied warranty of
  10  * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
  11  * Lesser General Public License for more details.
  12  *
  13  * You should have received a copy of the GNU Lesser General Public
  14  * License along with this library; if not, write to the Free Software
  15  * Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA  02110-1301 USA
  16  */
  17
  18 /* API used by UST components to communicate with each other via sockets. */
  19
  20 #define _GNU_SOURCE
  21 #include <sys/types.h>
  22 #include <signal.h>
  23 #include <errno.h>
  24 #include <sys/socket.h>
  25 #include <sys/un.h>
  26 #include <unistd.h>
  27 #include <poll.h>
  28 #include <sys/stat.h>
  29
  30 #include <stdio.h>
  31 #include <stdlib.h>
  32 #include <string.h>
  33 #include <execinfo.h>
  34
  35 #include "ustcomm.h"
  36 #include "usterr.h"
  37 #include "share.h"
  38 #include "multipoll.h"
  39
  40 #define UNIX_PATH_MAX 108
  41
  42 static int mkdir_p(const char *path, mode_t mode)
  43 {
  44         const char *path_p;
  45         char *tmp;
  46
  47         int retval = 0;
  48         int result;
  49
  50         tmp = malloc(strlen(path) + 1);
  51         if (tmp == NULL)
  52                 return -1;
  53
  54         /* skip first / */
  55         path_p = path+1;
  56
  57         for(;;) {
  58                 while (*path_p != '/') {
  59                         if(*path_p == 0)
  60                                 break;
  61                         ++path_p;
  62                 }
  63                 if (*path_p == '/') {
  64                         strncpy(tmp, path, path_p - path);
  65                         tmp[path_p-path] = '\0';
  66                         if (tmp[path_p - path - 1] != '/') {
  67                                 result = mkdir(tmp, mode);
  68                                 if(result == -1) {
  69                                         if (!(errno == EEXIST || errno == EACCES || errno == EROFS)) {
  70                                                 /* Then this is a real error */
  71                                                 retval = -1;
  72                                                 break;
  73                                         }
  74                                 }
  75                         }
  76                         /* pass / */
  77                         path_p++;
  78                 } else {
  79                         /* last component */
  80                         result = mkdir(path, mode);
  81                         if (result == -1)
  82                                 retval = -1;
  83                         break;
  84                 }
  85         }
  86
  87         free(tmp);
  88         return retval;
  89 }
  90
  91 static int signal_process(pid_t pid)
  92 {
  93         return 0;
  94 }
  95
  96 void ustcomm_init_connection(struct ustcomm_connection *conn)
  97 {
  98         conn->recv_buf = NULL;
  99         conn->recv_buf_size = 0;
 100         conn->recv_buf_alloc = 0;
 101 }
 102
 103 int pid_is_online(pid_t pid) {
 104         return 1;
 105 }
 106
 107 /* Send a message
 108  *
 109  * @fd: file descriptor to send to
 110  * @msg: a null-terminated string containing the message to send
 111  *
 112  * Return value:
 113  * -1: error
 114  * 0: connection closed
 115  * 1: success
 116  */
 117
 118 static int send_message_fd(int fd, const char *msg)
 119 {
 120         int result;
 121
 122         /* Send including the final \0 */
 123         result = patient_send(fd, msg, strlen(msg)+1, MSG_NOSIGNAL);
 124         if(result == -1) {
 125                 if(errno != EPIPE)
 126                         PERROR("send");
 127                 return -1;
 128         }
 129         else if(result == 0) {
 130                 return 0;
 131         }
 132
 133         DBG("sent message \"%s\"", msg);
 134         return 1;
 135 }
 136
 137 /* Called by an app to ask the consumer daemon to connect to it. */
 138
 139 int ustcomm_request_consumer(pid_t pid, const char *channel)
 140 {
 141         char path[UNIX_PATH_MAX];
 142         int result;
 143         char *msg=NULL;
 144         int retval = 0;
 145         struct ustcomm_connection conn;
 146         char *explicit_daemon_socket_path;
 147
 148         explicit_daemon_socket_path = getenv("UST_DAEMON_SOCKET");
 149         if(explicit_daemon_socket_path) {
 150                 /* user specified explicitly a socket path */
 151                 result = snprintf(path, UNIX_PATH_MAX, "%s", explicit_daemon_socket_path);
 152         }
 153         else {
 154                 /* just use the default path */
 155                 result = snprintf(path, UNIX_PATH_MAX, "%s/ustd", SOCK_DIR);
 156         }
 157
 158         if(result >= UNIX_PATH_MAX) {
 159                 ERR("string overflow allocating socket name");
 160                 return -1;
 161         }
 162
 163         asprintf(&msg, "collect %d %s", pid, channel);
 164
 165         /* don't signal it because it's the daemon */
 166         result = ustcomm_connect_path(path, &conn, -1);
 167         if(result == -1) {
 168                 WARN("ustcomm_connect_path failed");
 169                 retval = -1;
 170                 goto del_string;
 171         }
 172
 173         result = ustcomm_send_request(&conn, msg, NULL);
 174         if(result == -1) {
 175                 WARN("ustcomm_send_request failed");
 176                 retval = -1;
 177                 goto disconnect;
 178         }
 179
 180         disconnect:
 181         ustcomm_disconnect(&conn);
 182         del_string:
 183         free(msg);
 184
 185         return retval;
 186 }
 187
 188 /* returns 1 to indicate a message was received
 189  * returns 0 to indicate no message was received (end of stream)
 190  * returns -1 to indicate an error
 191  */
 192
 193 #define RECV_INCREMENT 1000
 194 #define RECV_INITIAL_BUF_SIZE 10
 195
 196 static int recv_message_fd(int fd, char **recv_buf, int *recv_buf_size, int *recv_buf_alloc, char **msg)
 197 {
 198         int result;
 199
 200         /* 1. Check if there is a message in the buf */
 201         /* 2. If not, do:
 202            2.1 receive chunk and put it in buffer
 203            2.2 process full message if there is one
 204            -- while no message arrived
 205         */
 206
 207         for(;;) {
 208                 int i;
 209                 int nulfound = 0;
 210
 211                 /* Search for full message in buffer */
 212                 for(i=0; i<*recv_buf_size; i++) {
 213                         if((*recv_buf)[i] == '\0') {
 214                                 nulfound = 1;
 215                                 break;
 216                         }
 217                 }
 218
 219                 /* Process found message */
 220                 if(nulfound == 1) {
 221                         char *newbuf;
 222
 223                         if(i == 0) {
 224                                 /* problem */
 225                                 WARN("received empty message");
 226                         }
 227                         *msg = strndup(*recv_buf, i);
 228
 229                         /* Remove processed message from buffer */
 230                         newbuf = (char *) malloc(*recv_buf_size - (i+1));
 231                         memcpy(newbuf, *recv_buf + (i+1), *recv_buf_size - (i+1));
 232                         free(*recv_buf);
 233                         *recv_buf = newbuf;
 234                         *recv_buf_size -= (i+1);
 235                         *recv_buf_alloc -= (i+1);
 236
 237                         return 1;
 238                 }
 239
 240                 /* Receive a chunk from the fd */
 241                 if(*recv_buf_alloc - *recv_buf_size < RECV_INCREMENT) {
 242                         *recv_buf_alloc += RECV_INCREMENT - (*recv_buf_alloc - *recv_buf_size);
 243                         *recv_buf = (char *) realloc(*recv_buf, *recv_buf_alloc);
 244                 }
 245
 246                 result = recv(fd, *recv_buf+*recv_buf_size, RECV_INCREMENT, 0);
 247                 if(result == -1) {
 248                         if(errno == ECONNRESET) {
 249                                 *recv_buf_size = 0;
 250                                 return 0;
 251                         }
 252                         else if(errno == EINTR) {
 253                                 return -1;
 254                         }
 255                         else {
 256                                 PERROR("recv");
 257                                 return -1;
 258                         }
 259                 }
 260                 if(result == 0) {
 261                         return 0;
 262                 }
 263                 *recv_buf_size += result;
 264
 265                 /* Go back to the beginning to check if there is a full message in the buffer */
 266         }
 267
 268         DBG("received message \"%s\"", *recv_buf);
 269
 270         return 1;
 271
 272 }
 273
 274 static int recv_message_conn(struct ustcomm_connection *conn, char **msg)
 275 {
 276         return recv_message_fd(conn->fd, &conn->recv_buf, &conn->recv_buf_size, &conn->recv_buf_alloc, msg);
 277 }
 278
 279 int ustcomm_send_reply(struct ustcomm_server *server, char *msg, struct ustcomm_source *src)
 280 {
 281         int result;
 282
 283         result = send_message_fd(src->fd, msg);
 284         if(result < 0) {
 285                 ERR("error in send_message_fd");
 286                 return -1;
 287         }
 288
 289         return 0;
 290 }
 291
 292 /* Called after a fork. */
 293
 294 int ustcomm_close_all_connections(struct ustcomm_server *server)
 295 {
 296         struct ustcomm_connection *conn;
 297         struct ustcomm_connection *deletable_conn = NULL;
 298
 299         list_for_each_entry(conn, &server->connections, list) {
 300                 free(deletable_conn);
 301                 deletable_conn = conn;
 302                 ustcomm_close_app(conn);
 303                 list_del(&conn->list);
 304         }
 305
 306         return 0;
 307 }
 308
 309 /* @timeout: max blocking time in milliseconds, -1 means infinity
 310  *
 311  * returns 1 to indicate a message was received
 312  * returns 0 to indicate no message was received
 313  * returns -1 to indicate an error
 314  */
 315
 316 int ustcomm_recv_message(struct ustcomm_server *server, char **msg, struct ustcomm_source *src, int timeout)
 317 {
 318         struct pollfd *fds;
 319         struct ustcomm_connection **conn_table;
 320         struct ustcomm_connection *conn;
 321         int result;
 322         int retval;
 323
 324         for(;;) {
 325                 int idx = 0;
 326                 int n_fds = 1;
 327
 328                 list_for_each_entry(conn, &server->connections, list) {
 329                         n_fds++;
 330                 }
 331
 332                 fds = (struct pollfd *) malloc(n_fds * sizeof(struct pollfd));
 333                 if(fds == NULL) {
 334                         ERR("malloc returned NULL");
 335                         return -1;
 336                 }
 337
 338                 conn_table = (struct ustcomm_connection **) malloc(n_fds * sizeof(struct ustcomm_connection *));
 339                 if(conn_table == NULL) {
 340                         ERR("malloc returned NULL");
 341                         retval = -1;
 342                         goto free_fds_return;
 343                 }
 344
 345                 /* special idx 0 is for listening socket */
 346                 fds[idx].fd = server->listen_fd;
 347                 fds[idx].events = POLLIN;
 348                 idx++;
 349
 350                 list_for_each_entry(conn, &server->connections, list) {
 351                         fds[idx].fd = conn->fd;
 352                         fds[idx].events = POLLIN;
 353                         conn_table[idx] = conn;
 354                         idx++;
 355                 }
 356
 357                 result = poll(fds, n_fds, timeout);
 358                 if(result == -1 && errno == EINTR) {
 359                         /* That's ok. ustd receives signals to notify it must shutdown. */
 360                         retval = -1;
 361                         goto free_conn_table_return;
 362                 }
 363                 else if(result == -1) {
 364                         PERROR("poll");
 365                         retval = -1;
 366                         goto free_conn_table_return;
 367                 }
 368                 else if(result == 0) {
 369                         retval = 0;
 370                         goto free_conn_table_return;
 371                 }
 372
 373                 if(fds[0].revents) {
 374                         struct ustcomm_connection *newconn;
 375                         int newfd;
 376
 377                         result = newfd = accept(server->listen_fd, NULL, NULL);
 378                         if(result == -1) {
 379                                 PERROR("accept");
 380                                 retval = -1;
 381                                 goto free_conn_table_return;
 382                         }
 383
 384                         newconn = (struct ustcomm_connection *) malloc(sizeof(struct ustcomm_connection));
 385                         if(newconn == NULL) {
 386                                 ERR("malloc returned NULL");
 387                                 return -1;
 388                         }
 389
 390                         ustcomm_init_connection(newconn);
 391                         newconn->fd = newfd;
 392
 393                         list_add(&newconn->list, &server->connections);
 394                 }
 395
 396                 for(idx=1; idx<n_fds; idx++) {
 397                         if(fds[idx].revents) {
 398                                 retval = recv_message_conn(conn_table[idx], msg);
 399                                 if(src)
 400                                         src->fd = fds[idx].fd;
 401
 402                                 if(retval == 0) {
 403                                         /* connection finished */
 404                                         list_for_each_entry(conn, &server->connections, list) {
 405                                                 if(conn->fd == fds[idx].fd) {
 406                                                         ustcomm_close_app(conn);
 407                                                         list_del(&conn->list);
 408                                                         free(conn);
 409                                                         break;
 410                                                 }
 411                                         }
 412                                 }
 413                                 else {
 414                                         goto free_conn_table_return;
 415                                 }
 416                         }
 417                 }
 418
 419                 free(fds);
 420                 free(conn_table);
 421         }
 422
 423 free_conn_table_return:
 424         free(conn_table);
 425 free_fds_return:
 426         free(fds);
 427         return retval;
 428 }
 429
 430 int ustcomm_ustd_recv_message(struct ustcomm_ustd *ustd, char **msg, struct ustcomm_source *src, int timeout)
 431 {
 432         return ustcomm_recv_message(&ustd->server, msg, src, timeout);
 433 }
 434
 435 int ustcomm_app_recv_message(struct ustcomm_app *app, char **msg, struct ustcomm_source *src, int timeout)
 436 {
 437         return ustcomm_recv_message(&app->server, msg, src, timeout);
 438 }
 439
 440 /* This removes src from the list of active connections of app.
 441  */
 442
 443 int ustcomm_app_detach_client(struct ustcomm_app *app, struct ustcomm_source *src)
 444 {
 445         struct ustcomm_server *server = (struct ustcomm_server *)app;
 446         struct ustcomm_connection *conn;
 447
 448         list_for_each_entry(conn, &server->connections, list) {
 449                 if(conn->fd == src->fd) {
 450                         list_del(&conn->list);
 451                         goto found;
 452                 }
 453         }
 454
 455         return -1;
 456 found:
 457         return src->fd;
 458 }
 459
 460 static int init_named_socket(const char *name, char **path_out)
 461 {
 462         int result;
 463         int fd;
 464
 465         struct sockaddr_un addr;
 466
 467         result = fd = socket(PF_UNIX, SOCK_STREAM, 0);
 468         if(result == -1) {
 469                 PERROR("socket");
 470                 return -1;
 471         }
 472
 473         addr.sun_family = AF_UNIX;
 474
 475         strncpy(addr.sun_path, name, UNIX_PATH_MAX);
 476         addr.sun_path[UNIX_PATH_MAX-1] = '\0';
 477
 478         result = access(name, F_OK);
 479         if(result == 0) {
 480                 /* file exists */
 481                 result = unlink(name);
 482                 if(result == -1) {
 483                         PERROR("unlink of socket file");
 484                         goto close_sock;
 485                 }
 486                 DBG("socket already exists; overwriting");
 487         }
 488
 489         result = bind(fd, (struct sockaddr *)&addr, sizeof(addr));
 490         if(result == -1) {
 491                 PERROR("bind");
 492                 goto close_sock;
 493         }
 494
 495         result = listen(fd, 1);
 496         if(result == -1) {
 497                 PERROR("listen");
 498                 goto close_sock;
 499         }
 500
 501         if(path_out) {
 502                 *path_out = strdup(addr.sun_path);
 503         }
 504
 505         return fd;
 506
 507         close_sock:
 508         close(fd);
 509
 510         return -1;
 511 }
 512
 513 /*
 514  * Return value:
 515  *   0: Success, but no reply because recv() returned 0
 516  *   1: Success
 517  *   -1: Error
 518  *
 519  * On error, the error message is printed, except on
 520  * ECONNRESET, which is normal when the application dies.
 521  */
 522
 523 int ustcomm_send_request(struct ustcomm_connection *conn, const char *req, char **reply)
 524 {
 525         int result;
 526
 527         /* Send including the final \0 */
 528         result = send_message_fd(conn->fd, req);
 529         if(result != 1)
 530                 return result;
 531
 532         if(!reply)
 533                 return 1;
 534
 535         result = recv_message_conn(conn, reply);
 536         if(result == -1) {
 537                 return -1;
 538         }
 539         else if(result == 0) {
 540                 return 0;
 541         }
 542
 543         return 1;
 544 }
 545
 546 /* Return value:
 547  *  0: success
 548  * -1: error
 549  */
 550
 551 int ustcomm_connect_path(const char *path, struct ustcomm_connection *conn, pid_t signalpid)
 552 {
 553         int fd;
 554         int result;
 555         struct sockaddr_un addr;
 556
 557         ustcomm_init_connection(conn);
 558
 559         result = fd = socket(PF_UNIX, SOCK_STREAM, 0);
 560         if(result == -1) {
 561                 PERROR("socket");
 562                 return -1;
 563         }
 564
 565         addr.sun_family = AF_UNIX;
 566
 567         result = snprintf(addr.sun_path, UNIX_PATH_MAX, "%s", path);
 568         if(result >= UNIX_PATH_MAX) {
 569                 ERR("string overflow allocating socket name");
 570                 return -1;
 571         }
 572
 573         if(signalpid >= 0) {
 574                 result = signal_process(signalpid);
 575                 if(result == -1) {
 576                         ERR("could not signal process");
 577                         return -1;
 578                 }
 579         }
 580
 581         result = connect(fd, (struct sockaddr *)&addr, sizeof(addr));
 582         if(result == -1) {
 583                 PERROR("connect (path=%s)", path);
 584                 return -1;
 585         }
 586
 587         conn->fd = fd;
 588
 589         return 0;
 590 }
 591
 592 int ustcomm_disconnect(struct ustcomm_connection *conn)
 593 {
 594         return close(conn->fd);
 595 }
 596
 597 /* Open a connection to a traceable app.
 598  *
 599  * Return value:
 600  *  0: success
 601  * -1: error
 602  */
 603
 604 int ustcomm_connect_app(pid_t pid, struct ustcomm_connection *conn)
 605 {
 606         int result;
 607         char path[UNIX_PATH_MAX];
 608
 609
 610         result = snprintf(path, UNIX_PATH_MAX, "%s/%d", SOCK_DIR, pid);
 611         if(result >= UNIX_PATH_MAX) {
 612                 ERR("string overflow allocating socket name");
 613                 return -1;
 614         }
 615
 616         return ustcomm_connect_path(path, conn, pid);
 617 }
 618
 619 /* Close a connection to a traceable app. It frees the
 620  * resources. It however does not free the
 621  * ustcomm_connection itself.
 622  */
 623
 624 int ustcomm_close_app(struct ustcomm_connection *conn)
 625 {
 626         close(conn->fd);
 627         free(conn->recv_buf);
 628
 629         return 0;
 630 }
 631
 632 static int ensure_dir_exists(const char *dir)
 633 {
 634         struct stat st;
 635         int result;
 636
 637         if(!strcmp(dir, ""))
 638                 return -1;
 639
 640         result = stat(dir, &st);
 641         if(result == -1 && errno != ENOENT) {
 642                 return -1;
 643         }
 644         else if(result == -1) {
 645                 /* ENOENT */
 646                 int result;
 647
 648                 result = mkdir_p(dir, 0777);
 649                 if(result != 0) {
 650                         ERR("executing in recursive creation of directory %s", dir);
 651                         return -1;
 652                 }
 653         }
 654
 655         return 0;
 656 }
 657
 658 /* Called by an application to initialize its server so daemons can
 659  * connect to it.
 660  */
 661
 662 int ustcomm_init_app(pid_t pid, struct ustcomm_app *handle)
 663 {
 664         int result;
 665         char *name;
 666
 667         result = asprintf(&name, "%s/%d", SOCK_DIR, (int)pid);
 668         if(result >= UNIX_PATH_MAX) {
 669                 ERR("string overflow allocating socket name");
 670                 return -1;
 671         }
 672
 673         result = ensure_dir_exists(SOCK_DIR);
 674         if(result == -1) {
 675                 ERR("Unable to create socket directory %s", SOCK_DIR);
 676                 return -1;
 677         }
 678
 679         handle->server.listen_fd = init_named_socket(name, &(handle->server.socketpath));
 680         if(handle->server.listen_fd < 0) {
 681                 ERR("Error initializing named socket (%s). Check that directory exists and that it is writable.", name);
 682                 goto free_name;
 683         }
 684         free(name);
 685
 686         INIT_LIST_HEAD(&handle->server.connections);
 687
 688         return 0;
 689
 690 free_name:
 691         free(name);
 692         return -1;
 693 }
 694
 695 /* Used by the daemon to initialize its server so applications
 696  * can connect to it.
 697  */
 698
 699 int ustcomm_init_ustd(struct ustcomm_ustd *handle, const char *sock_path)
 700 {
 701         char *name;
 702         int retval = 0;
 703
 704         if(sock_path) {
 705                 asprintf(&name, "%s", sock_path);
 706         }
 707         else {
 708                 int result;
 709
 710                 /* Only check if socket dir exists if we are using the default directory */
 711                 result = ensure_dir_exists(SOCK_DIR);
 712                 if(result == -1) {
 713                         ERR("Unable to create socket directory %s", SOCK_DIR);
 714                         return -1;
 715                 }
 716
 717                 asprintf(&name, "%s/%s", SOCK_DIR, "ustd");
 718         }
 719
 720         handle->server.listen_fd = init_named_socket(name, &handle->server.socketpath);
 721         if(handle->server.listen_fd < 0) {
 722                 ERR("error initializing named socket at %s", name);
 723                 retval = -1;
 724                 goto free_name;
 725         }
 726
 727         INIT_LIST_HEAD(&handle->server.connections);
 728
 729 free_name:
 730         free(name);
 731
 732         return retval;
 733 }
 734
 735 static void ustcomm_fini_server(struct ustcomm_server *server, int keep_socket_file)
 736 {
 737         int result;
 738         struct stat st;
 739
 740         if(!keep_socket_file) {
 741                 /* Destroy socket */
 742                 result = stat(server->socketpath, &st);
 743                 if(result == -1) {
 744                         PERROR("stat (%s)", server->socketpath);
 745                         return;
 746                 }
 747
 748                 /* Paranoid check before deleting. */
 749                 result = S_ISSOCK(st.st_mode);
 750                 if(!result) {
 751                         ERR("The socket we are about to delete is not a socket.");
 752                         return;
 753                 }
 754
 755                 result = unlink(server->socketpath);
 756                 if(result == -1) {
 757                         PERROR("unlink");
 758                 }
 759         }
 760
 761         free(server->socketpath);
 762
 763         result = close(server->listen_fd);
 764         if(result == -1) {
 765                 PERROR("close");
 766                 return;
 767         }
 768 }
 769
 770 /* Free a traceable application server */
 771
 772 void ustcomm_fini_app(struct ustcomm_app *handle, int keep_socket_file)
 773 {
 774         ustcomm_fini_server(&handle->server, keep_socket_file);
 775 }
 776
 777 /* Free a ustd server */
 778
 779 void ustcomm_fini_ustd(struct ustcomm_ustd *handle)
 780 {
 781         ustcomm_fini_server(&handle->server, 0);
 782 }
 783
 784 static const char *find_tok(const char *str)
 785 {
 786         while(*str == ' ') {
 787                 str++;
 788
 789                 if(*str == 0)
 790                         return NULL;
 791         }
 792
 793         return str;
 794 }
 795
 796 static const char *find_sep(const char *str)
 797 {
 798         while(*str != ' ') {
 799                 str++;
 800
 801                 if(*str == 0)
 802                         break;
 803         }
 804
 805         return str;
 806 }
 807
 808 int nth_token_is(const char *str, const char *token, int tok_no)
 809 {
 810         int i;
 811         const char *start;
 812         const char *end;
 813
 814         for(i=0; i<=tok_no; i++) {
 815                 str = find_tok(str);
 816                 if(str == NULL)
 817                         return -1;
 818
 819                 start = str;
 820
 821                 str = find_sep(str);
 822                 if(str == NULL)
 823                         return -1;
 824
 825                 end = str;
 826         }
 827
 828         if(end-start != strlen(token))
 829                 return 0;
 830
 831         if(strncmp(start, token, end-start))
 832                 return 0;
 833
 834         return 1;
 835 }
 836
 837 char *nth_token(const char *str, int tok_no)
 838 {
 839         static char *retval = NULL;
 840         int i;
 841         const char *start;
 842         const char *end;
 843
 844         for(i=0; i<=tok_no; i++) {
 845                 str = find_tok(str);
 846                 if(str == NULL)
 847                         return NULL;
 848
 849                 start = str;
 850
 851                 str = find_sep(str);
 852                 if(str == NULL)
 853                         return NULL;
 854
 855                 end = str;
 856         }
 857
 858         if(retval) {
 859                 free(retval);
 860                 retval = NULL;
 861         }
 862
 863         asprintf(&retval, "%.*s", (int)(end-start), start);
 864
 865         return retval;
 866 }
 867
 868 /* Callback from multipoll.
 869  * Receive a new connection on the listening socket.
 870  */
 871
 872 static int process_mp_incoming_conn(void *priv, int fd, short events)
 873 {
 874         struct ustcomm_connection *newconn;
 875         struct ustcomm_server *server = (struct ustcomm_server *) priv;
 876         int newfd;
 877         int result;
 878
 879         result = newfd = accept(server->listen_fd, NULL, NULL);
 880         if(result == -1) {
 881                 PERROR("accept");
 882                 return -1;
 883         }
 884
 885         newconn = (struct ustcomm_connection *) malloc(sizeof(struct ustcomm_connection));
 886         if(newconn == NULL) {
 887                 ERR("malloc returned NULL");
 888                 return -1;
 889         }
 890
 891         ustcomm_init_connection(newconn);
 892         newconn->fd = newfd;
 893
 894         list_add(&newconn->list, &server->connections);
 895
 896         return 0;
 897 }
 898
 899 /* Callback from multipoll.
 900  * Receive a message on an existing connection.
 901  */
 902
 903 static int process_mp_conn_msg(void *priv, int fd, short revents)
 904 {
 905         struct ustcomm_multipoll_conn_info *mpinfo = (struct ustcomm_multipoll_conn_info *) priv;
 906         int result;
 907         char *msg;
 908         struct ustcomm_source src;
 909
 910         if(revents) {
 911                 src.fd = fd;
 912
 913                 result = recv_message_conn(mpinfo->conn, &msg);
 914                 if(result == -1) {
 915                         ERR("error in recv_message_conn");
 916                 }
 917
 918                 else if(result == 0) {
 919                         /* connection finished */
 920                         ustcomm_close_app(mpinfo->conn);
 921                         list_del(&mpinfo->conn->list);
 922                         free(mpinfo->conn);
 923                 }
 924                 else {
 925                         mpinfo->cb(msg, &src);
 926                         free(msg);
 927                 }
 928         }
 929
 930         return 0;
 931 }
 932
 933 int free_ustcomm_client_poll(void *data)
 934 {
 935         free(data);
 936         return 0;
 937 }
 938
 939 void ustcomm_mp_add_app_clients(struct mpentries *ent, struct ustcomm_app *app, int (*cb)(char *recvbuf, struct ustcomm_source *src))
 940 {
 941         struct ustcomm_connection *conn;
 942
 943         /* add listener socket */
 944         multipoll_add(ent, app->server.listen_fd, POLLIN, process_mp_incoming_conn, &app->server, NULL);
 945
 946         list_for_each_entry(conn, &app->server.connections, list) {
 947                 struct ustcomm_multipoll_conn_info *mpinfo = (struct ustcomm_multipoll_conn_info *) malloc(sizeof(struct ustcomm_multipoll_conn_info));
 948                 mpinfo->conn = conn;
 949                 mpinfo->cb = cb;
 950                 multipoll_add(ent, conn->fd, POLLIN, process_mp_conn_msg, mpinfo, free_ustcomm_client_poll);
 951         }
 952 }