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