X-Git-Url: http://git.liburcu.org/?a=blobdiff_plain;f=libustcomm%2Fustcomm.c;h=4fe46c537d880d7f9cb27dd13d73c1a660343a3e;hb=3a7b90de71f2a82f73f06fb14a7b77805aea1064;hp=c944214ae2ea93aec91febf836e6b8018bc873d5;hpb=b02e31e53039229b50d2f54ee31d68709aba1412;p=ust.git diff --git a/libustcomm/ustcomm.c b/libustcomm/ustcomm.c index c944214..4fe46c5 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]; @@ -29,6 +35,20 @@ // backtrace_symbols_fd(buffer, result, STDERR_FILENO); //} +char *strdup_malloc(const char *s) +{ + char *retval; + + if(s == NULL) + return NULL; + + retval = (char *) malloc(strlen(s)+1); + + strcpy(retval, s); + + return retval; +} + static void signal_process(pid_t pid) { int result; @@ -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)); + result = connect(fd, (struct sockaddr *)&addr, sizeof(addr)); if(result == -1) { - PERROR("sendto"); + PERROR("connect"); return -1; } - if(!reply) - return 0; - - *reply = (char *) malloc(MSG_MAX+1); - result = recvfrom(fd, *reply, MSG_MAX, 0, NULL, NULL); - if(result == -1) { - PERROR("recvfrom"); - 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,28 +169,20 @@ int ustcomm_request_consumer(pid_t pid, const char *channel) return 0; } +/* 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; - size_t initial_addrlen,addrlen; *msg = (char *) malloc(MSG_MAX+1); - if(src) { - initial_addrlen = addrlen = sizeof(src->addr); - - result = recvfrom(fd, *msg, MSG_MAX, 0, &src->addr, &addrlen); - if(initial_addrlen != addrlen) { - ERR("recvfrom: unexpected address length"); - return -1; - } - } - else { - 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; } @@ -159,17 +190,150 @@ static int recv_message_fd(int fd, char **msg, struct ustcomm_source *src) DBG("ustcomm_app_recv_message: result is %d, message is %s", result, (*msg)); + 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 ustcomm_ustd_recv_message(struct ustcomm_ustd *ustd, char **msg, struct ustcomm_source *src, int timeout) { - return recv_message_fd(ustd->fd, msg, src); + return ustcomm_recv_message(&ustd->server, msg, src, timeout); } -int ustcomm_app_recv_message(struct ustcomm_app *app, char **msg, struct ustcomm_source *src) +int ustcomm_app_recv_message(struct ustcomm_app *app, char **msg, struct ustcomm_source *src, int timeout) { - return recv_message_fd(app->fd, msg, src); + return ustcomm_recv_message(&app->server, msg, src, timeout); +} + +/* This removes src from the list of active connections of app. + */ + +int ustcomm_app_detach_client(struct ustcomm_app *app, struct ustcomm_source *src) +{ + 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) @@ -179,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; @@ -190,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); @@ -220,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: @@ -244,15 +428,23 @@ 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; } -char *find_tok(const char *str) +static char *find_tok(char *str) { while(*str == ' ') { str++; @@ -331,7 +523,7 @@ char *nth_token(char *str, int tok_no) retval = NULL; } - retval = strndupa(start, end-start); + asprintf(&retval, "%.*s", (int)(end-start), start); return retval; }