X-Git-Url: https://git.liburcu.org/?a=blobdiff_plain;f=libustcomm%2Fustcomm.c;h=4fe46c537d880d7f9cb27dd13d73c1a660343a3e;hb=3a7b90de71f2a82f73f06fb14a7b77805aea1064;hp=0275a736b99a22b7f4b82304f5c289ca2544446d;hpb=b0540e11c3e9a3596b4b3865b0960d693fa66a88;p=ust.git diff --git a/libustcomm/ustcomm.c b/libustcomm/ustcomm.c index 0275a73..4fe46c5 100644 --- a/libustcomm/ustcomm.c +++ b/libustcomm/ustcomm.c @@ -5,6 +5,7 @@ #include #include #include +#include #include #include @@ -20,13 +21,32 @@ #define MSG_MAX 1000 -static void bt(void) +/* FIXME: ustcomm blocks on message sending, which might be problematic in + * some cases. Fix the poll() usage so sends are buffered until they don't + * block. + */ + +//static void bt(void) +//{ +// void *buffer[100]; +// int result; +// +// result = backtrace(&buffer, 100); +// backtrace_symbols_fd(buffer, result, STDERR_FILENO); +//} + +char *strdup_malloc(const char *s) { - void *buffer[100]; - int result; + char *retval; - result = backtrace(&buffer, 100); - backtrace_symbols_fd(buffer, result, STDERR_FILENO); + if(s == NULL) + return NULL; + + retval = (char *) malloc(strlen(s)+1); + + strcpy(retval, s); + + return retval; } static void signal_process(pid_t pid) @@ -42,13 +62,44 @@ static void signal_process(pid_t pid) sleep(1); } +int send_message_fd(int fd, const char *msg, char **reply) +{ + int result; + + result = send(fd, msg, strlen(msg), 0); + if(result == -1) { + PERROR("send"); + return -1; + } + else if(result == 0) { + return 0; + } + + if(!reply) + return 1; + + *reply = (char *) malloc(MSG_MAX+1); + result = recv(fd, *reply, MSG_MAX, 0); + if(result == -1) { + PERROR("recv"); + return -1; + } + else if(result == 0) { + return 0; + } + + (*reply)[result] = '\0'; + + return 1; +} + int send_message_path(const char *path, const char *msg, char **reply, int signalpid) { int fd; int result; struct sockaddr_un addr; - result = fd = socket(PF_UNIX, SOCK_DGRAM, 0); + result = fd = socket(PF_UNIX, SOCK_STREAM, 0); if(result == -1) { PERROR("socket"); return -1; @@ -65,25 +116,13 @@ int send_message_path(const char *path, const char *msg, char **reply, int signa if(signalpid >= 0) signal_process(signalpid); - result = sendto(fd, msg, strlen(msg), 0, (struct sockaddr *)&addr, sizeof(addr)); - if(result == -1) { - PERROR("sendto"); - return -1; - } - - if(!reply) - return 0; - - *reply = (char *) malloc(MSG_MAX+1); - result = recvfrom(fd, *reply, MSG_MAX, 0, NULL, NULL); + result = connect(fd, (struct sockaddr *)&addr, sizeof(addr)); if(result == -1) { - PERROR("recvfrom"); + PERROR("connect"); return -1; } - - (*reply)[result] = '\0'; - return 0; + return send_message_fd(fd, msg, reply); } /* pid: the pid of the trace process that must receive the msg @@ -124,21 +163,26 @@ int ustcomm_request_consumer(pid_t pid, const char *channel) asprintf(&msg, "collect %d %s", pid, channel); - send_message_path(path, msg, NULL, pid); + send_message_path(path, msg, NULL, -1); free(msg); return 0; } -static int recv_message_fd(int fd, char **msg) +/* returns 1 to indicate a message was received + * returns 0 to indicate no message was received (cannot happen) + * returns -1 to indicate an error + */ + +static int recv_message_fd(int fd, char **msg, struct ustcomm_source *src) { int result; - struct sockaddr_un addr; *msg = (char *) malloc(MSG_MAX+1); - result = recvfrom(fd, *msg, MSG_MAX, 0, NULL, NULL); + + result = recv(fd, *msg, MSG_MAX, 0); if(result == -1) { - PERROR("recvfrom"); + PERROR("recv"); return -1; } @@ -146,19 +190,150 @@ static int recv_message_fd(int fd, char **msg) DBG("ustcomm_app_recv_message: result is %d, message is %s", result, (*msg)); - bt(); + if(src) + src->fd = fd; + + return 1; +} + +int ustcomm_send_reply(struct ustcomm_server *server, char *msg, struct ustcomm_source *src) +{ + int result; + + result = send_message_fd(src->fd, msg, NULL); + if(result < 0) { + ERR("error in send_message_fd"); + return -1; + } return 0; +} + +/* @timeout: max blocking time in milliseconds, -1 means infinity + * + * returns 1 to indicate a message was received + * returns 0 to indicate no message was received + * returns -1 to indicate an error + */ + +int ustcomm_recv_message(struct ustcomm_server *server, char **msg, struct ustcomm_source *src, int timeout) +{ + struct pollfd *fds; + struct ustcomm_connection *conn; + int result; + int retval; + + for(;;) { + int idx = 0; + int n_fds = 1; + + list_for_each_entry(conn, &server->connections, list) { + n_fds++; + } + + fds = (struct pollfd *) malloc(n_fds * sizeof(struct pollfd)); + if(fds == NULL) { + ERR("malloc returned NULL"); + return -1; + } + + /* special idx 0 is for listening socket */ + fds[idx].fd = server->listen_fd; + fds[idx].events = POLLIN; + idx++; + + list_for_each_entry(conn, &server->connections, list) { + fds[idx].fd = conn->fd; + fds[idx].events = POLLIN; + idx++; + } + + result = poll(fds, n_fds, timeout); + if(result == -1) { + PERROR("poll"); + return -1; + } + + if(result == 0) + return 0; + + if(fds[0].revents) { + struct ustcomm_connection *newconn; + int newfd; + + result = newfd = accept(server->listen_fd, NULL, NULL); + if(result == -1) { + PERROR("accept"); + return -1; + } + + newconn = (struct ustcomm_connection *) malloc(sizeof(struct ustcomm_connection)); + if(newconn == NULL) { + ERR("malloc returned NULL"); + return -1; + } + + newconn->fd = newfd; + + list_add(&newconn->list, &server->connections); + } + + for(idx=1; idxconnections, list) { + if(conn->fd == fds[idx].fd) { + list_del(&conn->list); + break; + } + } + } + else { + goto free_fds_return; + } + } + } + + free(fds); + } + +free_fds_return: + free(fds); + return retval; +} + +int ustcomm_ustd_recv_message(struct ustcomm_ustd *ustd, char **msg, struct ustcomm_source *src, int timeout) +{ + return ustcomm_recv_message(&ustd->server, msg, src, timeout); } -int ustcomm_ustd_recv_message(struct ustcomm_ustd *ustd, char **msg) +int ustcomm_app_recv_message(struct ustcomm_app *app, char **msg, struct ustcomm_source *src, int timeout) { - return recv_message_fd(ustd->fd, msg); + return ustcomm_recv_message(&app->server, msg, src, timeout); } -int ustcomm_app_recv_message(struct ustcomm_app *app, char **msg) +/* This removes src from the list of active connections of app. + */ + +int ustcomm_app_detach_client(struct ustcomm_app *app, struct ustcomm_source *src) { - return recv_message_fd(app->fd, msg); + struct ustcomm_server *server = (struct ustcomm_server *)app; + struct ustcomm_connection *conn; + + list_for_each_entry(conn, &server->connections, list) { + if(conn->fd == src->fd) { + list_del(&conn->list); + goto found; + } + } + + return -1; +found: + return src->fd; } static int init_named_socket(char *name, char **path_out) @@ -168,7 +343,7 @@ static int init_named_socket(char *name, char **path_out) struct sockaddr_un addr; - result = fd = socket(PF_UNIX, SOCK_DGRAM, 0); + result = fd = socket(PF_UNIX, SOCK_STREAM, 0); if(result == -1) { PERROR("socket"); return -1; @@ -179,12 +354,29 @@ static int init_named_socket(char *name, char **path_out) strncpy(addr.sun_path, name, UNIX_PATH_MAX); addr.sun_path[UNIX_PATH_MAX-1] = '\0'; + result = access(name, F_OK); + if(result == 0) { + /* file exists */ + result = unlink(name); + if(result == -1) { + PERROR("unlink of socket file"); + goto close_sock; + } + WARN("socket already exists; overwriting"); + } + result = bind(fd, (struct sockaddr *)&addr, sizeof(addr)); if(result == -1) { PERROR("bind"); goto close_sock; } + result = listen(fd, 1); + if(result == -1) { + PERROR("listen"); + goto close_sock; + } + if(path_out) { *path_out = ""; *path_out = strdupa(addr.sun_path); @@ -209,12 +401,15 @@ int ustcomm_init_app(pid_t pid, struct ustcomm_app *handle) return -1; } - handle->fd = init_named_socket(name, &(handle->socketpath)); - if(handle->fd < 0) { + handle->server.listen_fd = init_named_socket(name, &(handle->server.socketpath)); + if(handle->server.listen_fd < 0) { + ERR("error initializing named socket"); goto free_name; } free(name); + INIT_LIST_HEAD(&handle->server.connections); + return 0; free_name: @@ -224,9 +419,112 @@ free_name: int ustcomm_init_ustd(struct ustcomm_ustd *handle) { - handle->fd = init_named_socket("ustd", &handle->socketpath); - if(handle->fd < 0) - return handle->fd; + int result; + char *name; + + result = asprintf(&name, "%s/%s", SOCK_DIR, "ustd"); + if(result >= UNIX_PATH_MAX) { + ERR("string overflow allocating socket name"); + return -1; + } + + handle->server.listen_fd = init_named_socket(name, &handle->server.socketpath); + if(handle->server.listen_fd < 0) { + ERR("error initializing named socket"); + goto free_name; + } + free(name); + + INIT_LIST_HEAD(&handle->server.connections); return 0; + +free_name: + free(name); + return -1; +} + +static char *find_tok(char *str) +{ + while(*str == ' ') { + str++; + + if(*str == 0) + return NULL; + } + + return str; } + +static char *find_sep(char *str) +{ + while(*str != ' ') { + str++; + + if(*str == 0) + break; + } + + return str; +} + +int nth_token_is(char *str, char *token, int tok_no) +{ + int i; + char *start; + char *end; + + for(i=0; i<=tok_no; i++) { + str = find_tok(str); + if(str == NULL) + return -1; + + start = str; + + str = find_sep(str); + if(str == NULL) + return -1; + + end = str; + } + + if(end-start != strlen(token)) + return 0; + + if(strncmp(start, token, end-start)) + return 0; + + return 1; +} + +char *nth_token(char *str, int tok_no) +{ + static char *retval = NULL; + int i; + char *start; + char *end; + + for(i=0; i<=tok_no; i++) { + str = find_tok(str); + if(str == NULL) + return NULL; + + start = str; + + str = find_sep(str); + if(str == NULL) + return NULL; + + end = str; + } + + if(retval) { + free(retval); + retval = NULL; + } + + asprintf(&retval, "%.*s", (int)(end-start), start); + + return retval; +} +