X-Git-Url: http://git.lttng.org./?a=blobdiff_plain;f=libustcomm%2Fustcomm.c;h=e445ed96b55339a52f98bd3c98238bd4bc490206;hb=811e4b93ba6acb72226a9243ae9a525f444e7e80;hp=d9c795633b238ac103a4315a7045fbe5e0c5afea;hpb=3847c3bab100bfb6b01b5654c2429a5d0d162ff5;p=lttng-ust.git diff --git a/libustcomm/ustcomm.c b/libustcomm/ustcomm.c index d9c79563..e445ed96 100644 --- a/libustcomm/ustcomm.c +++ b/libustcomm/ustcomm.c @@ -5,6 +5,7 @@ #include #include #include +#include #include #include @@ -20,6 +21,11 @@ #define MSG_MAX 1000 +/* 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]; @@ -42,13 +48,38 @@ 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; + } + + if(!reply) + return 0; + + *reply = (char *) malloc(MSG_MAX+1); + result = recv(fd, *reply, MSG_MAX, 0); + if(result == -1) { + PERROR("recv"); + return -1; + } + + (*reply)[result] = '\0'; + + return 0; +} + 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 +96,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 @@ -130,12 +149,15 @@ int ustcomm_request_consumer(pid_t pid, const char *channel) return 0; } -static int recv_message_fd(int fd, char **msg) + + +static int recv_message_fd(int fd, char **msg, struct ustcomm_source *src) { int result; *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"); return -1; @@ -145,17 +167,120 @@ static int recv_message_fd(int fd, char **msg) DBG("ustcomm_app_recv_message: result is %d, message is %s", result, (*msg)); + if(src) + src->fd = fd; + + return 0; +} + +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) { + ERR("error in send_message_fd"); + return -1; + } + return 0; +} + +int ustcomm_recv_message(struct ustcomm_server *server, char **msg, struct ustcomm_source *src) +{ + 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, -1); + if(result == -1) { + PERROR("poll"); + return -1; + } + + 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) +int ustcomm_ustd_recv_message(struct ustcomm_ustd *ustd, char **msg, struct ustcomm_source *src) { - return recv_message_fd(ustd->fd, msg); + return ustcomm_recv_message(&ustd->server, msg, src); } -int ustcomm_app_recv_message(struct ustcomm_app *app, char **msg) +int ustcomm_app_recv_message(struct ustcomm_app *app, char **msg, struct ustcomm_source *src) { - return recv_message_fd(app->fd, msg); + return ustcomm_recv_message(&app->server, msg, src); } static int init_named_socket(char *name, char **path_out) @@ -165,7 +290,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; @@ -176,12 +301,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); @@ -206,12 +348,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: @@ -230,10 +375,103 @@ int ustcomm_init_ustd(struct ustcomm_ustd *handle) return -1; } - handle->fd = init_named_socket(name, &handle->socketpath); - if(handle->fd < 0) - return handle->fd; + 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; +} +