X-Git-Url: http://git.lttng.org/?a=blobdiff_plain;f=libustcomm%2Fustcomm.c;h=cb72db65733eb5353c7d10a74018ff856874c648;hb=69c37f741b8bd8aea95019e1debf5e3bfa9d7fff;hp=ba98098b2d3a76387d57a40d3f811f16a4b1e0c8;hpb=c555b1339182be95b625dc4c1d6ac7217d3f471b;p=ust.git diff --git a/libustcomm/ustcomm.c b/libustcomm/ustcomm.c index ba98098..cb72db6 100644 --- a/libustcomm/ustcomm.c +++ b/libustcomm/ustcomm.c @@ -120,6 +120,13 @@ static int signal_process(pid_t pid) return 0; } +void ustcomm_init_connection(struct ustcomm_connection *conn) +{ + conn->recv_buf = NULL; + conn->recv_buf_size = 0; + conn->recv_buf_alloc = 0; +} + int pid_is_online(pid_t pid) { return 1; } @@ -210,65 +217,88 @@ int ustcomm_request_consumer(pid_t pid, const char *channel) * returns -1 to indicate an error */ -#define RECV_INCREMENT 1 +#define RECV_INCREMENT 1000 #define RECV_INITIAL_BUF_SIZE 10 -static int recv_message_fd(int fd, char **msg) +static int recv_message_fd(int fd, char **recv_buf, int *recv_buf_size, int *recv_buf_alloc, char **msg) { int result; - int buf_alloc_size = 0; - char *buf = NULL; - int buf_used_size = 0; - buf = malloc(RECV_INITIAL_BUF_SIZE); - buf_alloc_size = RECV_INITIAL_BUF_SIZE; + /* 1. Check if there is a message in the buf */ + /* 2. If not, do: + 2.1 receive chunk and put it in buffer + 2.2 process full message if there is one + -- while no message arrived + */ for(;;) { - if(buf_used_size + RECV_INCREMENT > buf_alloc_size) { - char *new_buf; - buf_alloc_size *= 2; - new_buf = (char *) realloc(buf, buf_alloc_size); - if(new_buf == NULL) { - ERR("realloc returned NULL"); - free(buf); - return -1; + int i; + int nulfound = 0; + + /* Search for full message in buffer */ + for(i=0; i<*recv_buf_size; i++) { + if((*recv_buf)[i] == '\0') { + nulfound = 1; + break; + } + } + + /* Process found message */ + if(nulfound == 1) { + char *newbuf; + + if(i == 0) { + /* problem */ + WARN("received empty message"); } - buf = new_buf; + *msg = strndup(*recv_buf, i); + + /* Remove processed message from buffer */ + newbuf = (char *) malloc(*recv_buf_size - (i+1)); + memcpy(newbuf, *recv_buf + (i+1), *recv_buf_size - (i+1)); + free(*recv_buf); + *recv_buf = newbuf; + *recv_buf_size -= (i+1); + *recv_buf_alloc -= (i+1); + + return 1; + } + + /* Receive a chunk from the fd */ + if(*recv_buf_alloc - *recv_buf_size < RECV_INCREMENT) { + *recv_buf_alloc += RECV_INCREMENT - (*recv_buf_alloc - *recv_buf_size); + *recv_buf = (char *) realloc(*recv_buf, *recv_buf_alloc); } - /* FIXME: this is really inefficient; but with count>1 we would - * need a buffering mechanism */ - result = recv(fd, buf+buf_used_size, RECV_INCREMENT, 0); + result = recv(fd, *recv_buf+*recv_buf_size, RECV_INCREMENT, 0); if(result == -1) { - free(buf); - if(errno != ECONNRESET) - PERROR("recv"); + if(errno == ECONNRESET) { + *recv_buf_size = 0; + return 0; + } + /* real error */ + PERROR("recv"); return -1; } if(result == 0) { - if(buf_used_size) - goto ret; - else { - free(buf); - return 0; - } + return 0; } + *recv_buf_size += result; - buf_used_size += result; - - if(buf[buf_used_size-1] == 0) { - goto ret; - } + /* Go back to the beginning to check if there is a full message in the buffer */ } -ret: - *msg = buf; - DBG("received message \"%s\"", buf); + DBG("received message \"%s\"", *recv_buf); return 1; } +static int recv_message_conn(struct ustcomm_connection *conn, char **msg) +{ + return recv_message_fd(conn->fd, &conn->recv_buf, &conn->recv_buf_size, &conn->recv_buf_alloc, msg); +} + int ustcomm_send_reply(struct ustcomm_server *server, char *msg, struct ustcomm_source *src) { int result; @@ -292,7 +322,7 @@ int ustcomm_close_all_connections(struct ustcomm_server *server) list_for_each_entry(conn, &server->connections, list) { free(deletable_conn); deletable_conn = conn; - close(conn->fd); + ustcomm_close_app(conn); list_del(&conn->list); } @@ -309,6 +339,7 @@ int ustcomm_close_all_connections(struct ustcomm_server *server) int ustcomm_recv_message(struct ustcomm_server *server, char **msg, struct ustcomm_source *src, int timeout) { struct pollfd *fds; + struct ustcomm_connection **conn_table; struct ustcomm_connection *conn; int result; int retval; @@ -327,6 +358,13 @@ int ustcomm_recv_message(struct ustcomm_server *server, char **msg, struct ustco return -1; } + conn_table = (struct ustcomm_connection **) malloc(n_fds * sizeof(struct ustcomm_connection *)); + if(conn_table == NULL) { + ERR("malloc returned NULL"); + retval = -1; + goto free_fds_return; + } + /* special idx 0 is for listening socket */ fds[idx].fd = server->listen_fd; fds[idx].events = POLLIN; @@ -335,6 +373,7 @@ int ustcomm_recv_message(struct ustcomm_server *server, char **msg, struct ustco list_for_each_entry(conn, &server->connections, list) { fds[idx].fd = conn->fd; fds[idx].events = POLLIN; + conn_table[idx] = conn; idx++; } @@ -342,11 +381,14 @@ int ustcomm_recv_message(struct ustcomm_server *server, char **msg, struct ustco /* nothing */; if(result == -1) { PERROR("poll"); - return -1; + retval = -1; + goto free_conn_table_return; } - if(result == 0) - return 0; + if(result == 0) { + retval = 0; + goto free_conn_table_return; + } if(fds[0].revents) { struct ustcomm_connection *newconn; @@ -355,7 +397,8 @@ int ustcomm_recv_message(struct ustcomm_server *server, char **msg, struct ustco result = newfd = accept(server->listen_fd, NULL, NULL); if(result == -1) { PERROR("accept"); - return -1; + retval = -1; + goto free_conn_table_return; } newconn = (struct ustcomm_connection *) malloc(sizeof(struct ustcomm_connection)); @@ -364,6 +407,7 @@ int ustcomm_recv_message(struct ustcomm_server *server, char **msg, struct ustco return -1; } + ustcomm_init_connection(newconn); newconn->fd = newfd; list_add(&newconn->list, &server->connections); @@ -371,30 +415,33 @@ int ustcomm_recv_message(struct ustcomm_server *server, char **msg, struct ustco for(idx=1; idxfd = fds[idx].fd; if(retval == 0) { /* connection finished */ - close(fds[idx].fd); - list_for_each_entry(conn, &server->connections, list) { if(conn->fd == fds[idx].fd) { + ustcomm_close_app(conn); list_del(&conn->list); + free(conn); break; } } } else { - goto free_fds_return; + goto free_conn_table_return; } } } free(fds); + free(conn_table); } +free_conn_table_return: + free(conn_table); free_fds_return: free(fds); return retval; @@ -456,7 +503,7 @@ static int init_named_socket(const char *name, char **path_out) PERROR("unlink of socket file"); goto close_sock; } - WARN("socket already exists; overwriting"); + DBG("socket already exists; overwriting"); } result = bind(fd, (struct sockaddr *)&addr, sizeof(addr)); @@ -505,7 +552,7 @@ int ustcomm_send_request(struct ustcomm_connection *conn, const char *req, char if(!reply) return 1; - result = recv_message_fd(conn->fd, reply); + result = recv_message_conn(conn, reply); if(result == -1) { return -1; } @@ -516,12 +563,19 @@ int ustcomm_send_request(struct ustcomm_connection *conn, const char *req, char return 1; } +/* Return value: + * 0: success + * -1: error + */ + int ustcomm_connect_path(const char *path, struct ustcomm_connection *conn, pid_t signalpid) { int fd; int result; struct sockaddr_un addr; + ustcomm_init_connection(conn); + result = fd = socket(PF_UNIX, SOCK_STREAM, 0); if(result == -1) { PERROR("socket"); @@ -560,6 +614,13 @@ int ustcomm_disconnect(struct ustcomm_connection *conn) return close(conn->fd); } +/* Open a connection to a traceable app. + * + * Return value: + * 0: success + * -1: error + */ + int ustcomm_connect_app(pid_t pid, struct ustcomm_connection *conn) { int result; @@ -575,6 +636,19 @@ int ustcomm_connect_app(pid_t pid, struct ustcomm_connection *conn) return ustcomm_connect_path(path, conn, pid); } +/* Close a connection to a traceable app. It frees the + * resources. It however does not free the + * ustcomm_connection itself. + */ + +int ustcomm_close_app(struct ustcomm_connection *conn) +{ + close(conn->fd); + free(conn->recv_buf); + + return 0; +} + static int ensure_dir_exists(const char *dir) { struct stat st; @@ -678,31 +752,55 @@ free_name: return retval; } -void ustcomm_fini_app(struct ustcomm_app *handle) +static void ustcomm_fini_server(struct ustcomm_server *server, int keep_socket_file) { int result; struct stat st; - /* Destroy socket */ - result = stat(handle->server.socketpath, &st); - if(result == -1) { - PERROR("stat (%s)", handle->server.socketpath); - return; - } + if(!keep_socket_file) { + /* Destroy socket */ + result = stat(server->socketpath, &st); + if(result == -1) { + PERROR("stat (%s)", server->socketpath); + return; + } - /* Paranoid check before deleting. */ - result = S_ISSOCK(st.st_mode); - if(!result) { - ERR("The socket we are about to delete is not a socket."); - return; + /* Paranoid check before deleting. */ + result = S_ISSOCK(st.st_mode); + if(!result) { + ERR("The socket we are about to delete is not a socket."); + return; + } + + result = unlink(server->socketpath); + if(result == -1) { + PERROR("unlink"); + } } - result = unlink(handle->server.socketpath); + free(server->socketpath); + + result = close(server->listen_fd); if(result == -1) { - PERROR("unlink"); + PERROR("close"); + return; } } +/* Free a traceable application server */ + +void ustcomm_fini_app(struct ustcomm_app *handle, int keep_socket_file) +{ + ustcomm_fini_server(&handle->server, keep_socket_file); +} + +/* Free a ustd server */ + +void ustcomm_fini_ustd(struct ustcomm_ustd *handle) +{ + ustcomm_fini_server(&handle->server, 0); +} + static const char *find_tok(const char *str) { while(*str == ' ') {