X-Git-Url: http://git.efficios.com/?p=lttng-tools.git;a=blobdiff_plain;f=src%2Fbin%2Flttng-relayd%2Fmain.c;h=28e52696e568760c898cf7c49a8a79068f5a3add;hp=2ce9bf048bdd17e0c0402a7ef9ac467683885392;hb=36d2e35df61339e4394e84ad9790b984d259e0f0;hpb=32d1569c14b4e1efce9099a1e04c338a9c42f1f7 diff --git a/src/bin/lttng-relayd/main.c b/src/bin/lttng-relayd/main.c index 2ce9bf048..28e52696e 100644 --- a/src/bin/lttng-relayd/main.c +++ b/src/bin/lttng-relayd/main.c @@ -864,10 +864,7 @@ restart: goto exit; } - if (revents & (LPOLLERR | LPOLLHUP | LPOLLRDHUP)) { - ERR("socket poll error"); - goto error; - } else if (revents & LPOLLIN) { + if (revents & LPOLLIN) { /* * A new connection is requested, therefore a * sessiond/consumerd connection is allocated in @@ -919,6 +916,12 @@ restart: * exchange in cds_wfcq_enqueue. */ futex_nto1_wake(&relay_conn_queue.futex); + } else if (revents & (LPOLLERR | LPOLLHUP | LPOLLRDHUP)) { + ERR("socket poll error"); + goto error; + } else { + ERR("Unexpected poll events %u for sock %d", revents, pollfd); + goto error; } } } @@ -1065,12 +1068,12 @@ static int relay_create_session(struct lttcomm_relayd_hdr *recv_hdr, int ret = 0, send_ret; struct relay_session *session; struct lttcomm_relayd_status_session reply; - char session_name[NAME_MAX]; + char session_name[LTTNG_NAME_MAX]; char hostname[HOST_NAME_MAX]; uint32_t live_timer = 0; bool snapshot = false; - memset(session_name, 0, NAME_MAX); + memset(session_name, 0, LTTNG_NAME_MAX); memset(hostname, 0, HOST_NAME_MAX); memset(&reply, 0, sizeof(reply)); @@ -2081,8 +2084,8 @@ static int handle_index_data(struct relay_stream *stream, uint64_t net_seq_num, /* Get data offset because we are about to update the index. */ data_offset = htobe64(stream->tracefile_size_current); - DBG("handle_index_data: stream %" PRIu64 " data offset %" PRIu64, - stream->stream_handle, stream->tracefile_size_current); + DBG("handle_index_data: stream %" PRIu64 " net_seq_num %" PRIu64 " data offset %" PRIu64, + stream->stream_handle, net_seq_num, stream->tracefile_size_current); /* * Lookup for an existing index for that stream id/sequence @@ -2180,6 +2183,7 @@ static int relay_process_data(struct relay_connection *conn) stream_id = be64toh(data_hdr.stream_id); stream = stream_get_by_id(stream_id); if (!stream) { + ERR("relay_process_data: Cannot find stream %" PRIu64, stream_id); ret = -1; goto end; } @@ -2209,6 +2213,8 @@ static int relay_process_data(struct relay_connection *conn) if (ret == 0) { /* Orderly shutdown. Not necessary to print an error. */ DBG("Socket %d did an orderly shutdown", conn->sock->fd); + } else { + ERR("Socket %d error %d", conn->sock->fd, ret); } ret = -1; goto end_stream_put; @@ -2252,6 +2258,8 @@ static int relay_process_data(struct relay_connection *conn) if (session->minor >= 4 && !session->snapshot) { ret = handle_index_data(stream, net_seq_num, rotate_index); if (ret < 0) { + ERR("handle_index_data: fail stream %" PRIu64 " net_seq_num %" PRIu64 " ret %d", + stream->stream_handle, net_seq_num, ret); goto end_stream_unlock; } } @@ -2270,6 +2278,8 @@ static int relay_process_data(struct relay_connection *conn) ret = write_padding_to_file(stream->stream_fd->fd, be32toh(data_hdr.padding_size)); if (ret < 0) { + ERR("write_padding_to_file: fail stream %" PRIu64 " net_seq_num %" PRIu64 " ret %d", + stream->stream_handle, net_seq_num, ret); goto end_stream_unlock; } stream->tracefile_size_current += @@ -2429,10 +2439,7 @@ restart: /* Inspect the relay conn pipe for new connection */ if (pollfd == relay_conn_pipe[0]) { - if (revents & (LPOLLERR | LPOLLHUP | LPOLLRDHUP)) { - ERR("Relay connection pipe error"); - goto error; - } else if (revents & LPOLLIN) { + if (revents & LPOLLIN) { struct relay_connection *conn; ret = lttng_read(relay_conn_pipe[0], &conn, sizeof(conn)); @@ -2443,6 +2450,12 @@ restart: LPOLLIN | LPOLLRDHUP); connection_ht_add(relay_connections_ht, conn); DBG("Connection socket %d added", conn->sock->fd); + } else if (revents & (LPOLLERR | LPOLLHUP | LPOLLRDHUP)) { + ERR("Relay connection pipe error"); + goto error; + } else { + ERR("Unexpected poll events %u for sock %d", revents, pollfd); + goto error; } } else { struct relay_connection *ctrl_conn; @@ -2451,29 +2464,8 @@ restart: /* If not found, there is a synchronization issue. */ assert(ctrl_conn); - if (revents & (LPOLLERR | LPOLLHUP | LPOLLRDHUP)) { - relay_thread_close_connection(&events, pollfd, ctrl_conn); - if (last_seen_data_fd == pollfd) { - last_seen_data_fd = last_notdel_data_fd; - } - } else if (revents & LPOLLIN) { - if (ctrl_conn->type == RELAY_CONTROL) { - ret = ctrl_conn->sock->ops->recvmsg(ctrl_conn->sock, &recv_hdr, - sizeof(recv_hdr), 0); - if (ret <= 0) { - /* Connection closed */ - relay_thread_close_connection(&events, pollfd, - ctrl_conn); - } else { - ret = relay_process_control(&recv_hdr, ctrl_conn); - if (ret < 0) { - /* Clear the session on error. */ - relay_thread_close_connection(&events, pollfd, - ctrl_conn); - } - seen_control = 1; - } - } else { + if (ctrl_conn->type == RELAY_DATA) { + if (revents & LPOLLIN) { /* * Flag the last seen data fd not deleted. It will be * used as the last seen fd if any fd gets deleted in @@ -2481,9 +2473,39 @@ restart: */ last_notdel_data_fd = pollfd; } + goto put_ctrl_connection; + } + assert(ctrl_conn->type == RELAY_CONTROL); + + if (revents & LPOLLIN) { + ret = ctrl_conn->sock->ops->recvmsg(ctrl_conn->sock, + &recv_hdr, sizeof(recv_hdr), 0); + if (ret <= 0) { + /* Connection closed */ + relay_thread_close_connection(&events, pollfd, + ctrl_conn); + } else { + ret = relay_process_control(&recv_hdr, ctrl_conn); + if (ret < 0) { + /* Clear the session on error. */ + relay_thread_close_connection(&events, + pollfd, ctrl_conn); + } + seen_control = 1; + } + } else if (revents & (LPOLLERR | LPOLLHUP | LPOLLRDHUP)) { + relay_thread_close_connection(&events, + pollfd, ctrl_conn); + if (last_seen_data_fd == pollfd) { + last_seen_data_fd = last_notdel_data_fd; + } } else { - ERR("Unknown poll events %u for sock %d", revents, pollfd); + ERR("Unexpected poll events %u for control sock %d", + revents, pollfd); + connection_put(ctrl_conn); + goto error; } + put_ctrl_connection: connection_put(ctrl_conn); } } @@ -2533,17 +2555,17 @@ restart: /* Skip it. Might be removed before. */ continue; } + if (data_conn->type == RELAY_CONTROL) { + goto put_data_connection; + } + assert(data_conn->type == RELAY_DATA); if (revents & LPOLLIN) { - if (data_conn->type != RELAY_DATA) { - goto put_connection; - } - ret = relay_process_data(data_conn); /* Connection closed */ if (ret < 0) { relay_thread_close_connection(&events, pollfd, - data_conn); + data_conn); /* * Every goto restart call sets the last seen fd where * here we don't really care since we gracefully @@ -2555,8 +2577,14 @@ restart: connection_put(data_conn); goto restart; } + } else if (revents & (LPOLLERR | LPOLLHUP | LPOLLRDHUP)) { + relay_thread_close_connection(&events, pollfd, + data_conn); + } else { + ERR("Unknown poll events %u for data sock %d", + revents, pollfd); } - put_connection: + put_data_connection: connection_put(data_conn); } last_seen_data_fd = -1;