/* Event on the registration socket */
if (pollfd == sock) {
- if (revents & (LPOLLERR | LPOLLHUP | LPOLLRDHUP)) {
+ if (revents & (LPOLLERR | LPOLLHUP | LPOLLRDHUP)
+ && !(revents & LPOLLIN)) {
ERR("Health socket poll error");
goto error;
}
/* Event on the registration socket */
if (pollfd == sock) {
- if (revents & (LPOLLERR | LPOLLHUP | LPOLLRDHUP)) {
+ if (revents & LPOLLIN) {
+ continue;
+ } else if (revents & (LPOLLERR | LPOLLHUP | LPOLLRDHUP)) {
ERR("Health socket poll error");
goto error;
+ } else {
+ ERR("Unexpected poll events %u for sock %d", revents, pollfd);
+ goto error;
}
}
}
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
* viewer connection is allocated in this
* exchange in cds_wfcq_enqueue.
*/
futex_nto1_wake(&viewer_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;
}
}
}
/* Inspect the relay conn pipe for new connection. */
if (pollfd == live_conn_pipe[0]) {
- if (revents & (LPOLLERR | LPOLLHUP | LPOLLRDHUP)) {
- ERR("Relay live pipe error");
- goto error;
- } else if (revents & LPOLLIN) {
+ if (revents & LPOLLIN) {
struct relay_connection *conn;
ret = lttng_read(live_conn_pipe[0],
LPOLLIN | LPOLLRDHUP);
connection_ht_add(viewer_connections_ht, conn);
DBG("Connection socket %d added to poll", conn->sock->fd);
+ } else if (revents & (LPOLLERR | LPOLLHUP | LPOLLRDHUP)) {
+ ERR("Relay live pipe error");
+ goto error;
+ } else {
+ ERR("Unexpected poll events %u for sock %d", revents, pollfd);
+ goto error;
}
} else {
/* Connection activity. */
continue;
}
- if (revents & (LPOLLERR | LPOLLHUP | LPOLLRDHUP)) {
- cleanup_connection_pollfd(&events, pollfd);
- /* Put "create" ownership reference. */
- connection_put(conn);
- } else if (revents & LPOLLIN) {
+ if (revents & LPOLLIN) {
ret = conn->sock->ops->recvmsg(conn->sock, &recv_hdr,
sizeof(recv_hdr), 0);
if (ret <= 0) {
DBG("Viewer connection closed with %d", pollfd);
}
}
+ } else if (revents & (LPOLLERR | LPOLLHUP | LPOLLRDHUP)) {
+ cleanup_connection_pollfd(&events, pollfd);
+ /* Put "create" ownership reference. */
+ connection_put(conn);
+ } else {
+ ERR("Unexpected poll events %u for sock %d", revents, pollfd);
+ connection_put(conn);
+ goto error;
}
/* Put local "get_by_sock" reference. */
connection_put(conn);
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
* 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;
}
}
}
/* 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));
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;
/* 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
*/
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);
}
}
/* 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
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;
goto exit;
}
- /*
- * Check first if this is a POLLERR since POLLIN is also included
- * in an error value thus checking first.
- */
- if (revents & (LPOLLERR | LPOLLHUP | LPOLLRDHUP)) {
- /* Removing from the poll set */
- ret = lttng_poll_del(&events, pollfd);
- if (ret < 0) {
- goto error;
- }
-
- agent_destroy_app_by_sock(pollfd);
- } else if (revents & (LPOLLIN)) {
+ if (revents & LPOLLIN) {
int new_fd;
struct agent_app *app = NULL;
- /* Pollin event of agent app socket should NEVER happen. */
assert(pollfd == reg_sock->fd);
-
new_fd = handle_registration(reg_sock, &app);
if (new_fd < 0) {
- WARN("[agent-thread] agent registration failed. Ignoring.");
- /* Somehow the communication failed. Just continue. */
continue;
}
/* Should not have a NULL app on success. */
assert(app);
- /* Only add poll error event to only detect shutdown. */
+ /*
+ * Since this is a command socket (write then read),
+ * only add poll error event to only detect shutdown.
+ */
ret = lttng_poll_add(&events, new_fd,
LPOLLERR | LPOLLHUP | LPOLLRDHUP);
if (ret < 0) {
update_agent_app(app);
/* On failure, the poll will detect it and clean it up. */
- (void) agent_send_registration_done(app);
+ ret = agent_send_registration_done(app);
+ if (ret < 0) {
+ /* Removing from the poll set */
+ ret = lttng_poll_del(&events, new_fd);
+ if (ret < 0) {
+ goto error;
+ }
+ agent_destroy_app_by_sock(new_fd);
+ continue;
+ }
+ } else if (revents & (LPOLLERR | LPOLLHUP | LPOLLRDHUP)) {
+ /* Removing from the poll set */
+ ret = lttng_poll_del(&events, pollfd);
+ if (ret < 0) {
+ goto error;
+ }
+ agent_destroy_app_by_sock(pollfd);
} else {
- ERR("Unknown poll events %u for sock %d", revents, pollfd);
- continue;
+ ERR("Unexpected poll events %u for sock %d", revents, pollfd);
+ goto error;
}
}
}
continue;
}
- if (revents & (LPOLLERR | LPOLLHUP | LPOLLRDHUP)) {
+ if (revents & LPOLLIN) {
+ /* Get socket from dispatch thread. */
+ size_ret = lttng_read(ht_cleanup_pipe[0], &ht,
+ sizeof(ht));
+ if (size_ret < sizeof(ht)) {
+ PERROR("ht cleanup notify pipe");
+ goto error;
+ }
+ health_code_update();
+ /*
+ * The whole point of this thread is to call
+ * lttng_ht_destroy from a context that is NOT:
+ * 1) a read-side RCU lock,
+ * 2) a call_rcu thread.
+ */
+ lttng_ht_destroy(ht);
+
+ health_code_update();
+ } else if (revents & (LPOLLERR | LPOLLHUP | LPOLLRDHUP)) {
ERR("ht cleanup pipe error");
goto error;
- } else if (!(revents & LPOLLIN)) {
- /* No POLLIN and not a catched error, stop the thread. */
- ERR("ht cleanup failed. revent: %u", revents);
+ } else {
+ ERR("Unexpected poll events %u for sock %d", revents, pollfd);
goto error;
}
-
- /* Get socket from dispatch thread. */
- size_ret = lttng_read(ht_cleanup_pipe[0], &ht,
- sizeof(ht));
- if (size_ret < sizeof(ht)) {
- PERROR("ht cleanup notify pipe");
- goto error;
- }
- health_code_update();
- /*
- * The whole point of this thread is to call
- * lttng_ht_destroy from a context that is NOT:
- * 1) a read-side RCU lock,
- * 2) a call_rcu thread.
- */
- lttng_ht_destroy(ht);
-
- health_code_update();
}
for (i = 0; i < nb_fd; i++) {
}
/* Check for data on kernel pipe */
- if (pollfd == kernel_poll_pipe[0] && (revents & LPOLLIN)) {
- (void) lttng_read(kernel_poll_pipe[0],
- &tmp, 1);
- /*
- * Ret value is useless here, if this pipe gets any actions an
- * update is required anyway.
- */
- update_poll_flag = 1;
- continue;
- } else {
- /*
- * New CPU detected by the kernel. Adding kernel stream to
- * kernel session and updating the kernel consumer
- */
- if (revents & LPOLLIN) {
+ if (revents & LPOLLIN) {
+ if (pollfd == kernel_poll_pipe[0]) {
+ (void) lttng_read(kernel_poll_pipe[0],
+ &tmp, 1);
+ /*
+ * Ret value is useless here, if this pipe gets any actions an
+ * update is required anyway.
+ */
+ update_poll_flag = 1;
+ continue;
+ } else {
+ /*
+ * New CPU detected by the kernel. Adding kernel stream to
+ * kernel session and updating the kernel consumer
+ */
ret = update_kernel_stream(&kconsumer_data, pollfd);
if (ret < 0) {
continue;
}
break;
- /*
- * TODO: We might want to handle the LPOLLERR | LPOLLHUP
- * and unregister kernel stream at this point.
- */
}
+ } else if (revents & (LPOLLERR | LPOLLHUP | LPOLLRDHUP)) {
+ update_poll_flag = 1;
+ continue;
+ } else {
+ ERR("Unexpected poll events %u for sock %d", revents, pollfd);
+ goto error;
}
}
}
/* Event on the registration socket */
if (pollfd == consumer_data->err_sock) {
- if (revents & (LPOLLERR | LPOLLHUP | LPOLLRDHUP)) {
+ if (revents & LPOLLIN) {
+ continue;
+ } else if (revents & (LPOLLERR | LPOLLHUP | LPOLLRDHUP)) {
ERR("consumer err socket poll error");
goto error;
+ } else {
+ ERR("Unexpected poll events %u for sock %d", revents, pollfd);
+ goto error;
}
}
}
if (pollfd == sock) {
/* Event on the consumerd socket */
- if (revents & (LPOLLERR | LPOLLHUP | LPOLLRDHUP)) {
+ if (revents & (LPOLLERR | LPOLLHUP | LPOLLRDHUP)
+ && !(revents & LPOLLIN)) {
ERR("consumer err socket second poll error");
goto error;
}
goto exit;
} else if (pollfd == consumer_data->metadata_fd) {
+ if (revents & (LPOLLERR | LPOLLHUP | LPOLLRDHUP)
+ && !(revents & LPOLLIN)) {
+ ERR("consumer err metadata socket second poll error");
+ goto error;
+ }
/* UST metadata requests */
ret = ust_consumer_metadata_request(
&consumer_data->metadata_sock);
/* Inspect the apps cmd pipe */
if (pollfd == apps_cmd_pipe[0]) {
- if (revents & (LPOLLERR | LPOLLHUP | LPOLLRDHUP)) {
- ERR("Apps command pipe error");
- goto error;
- } else if (revents & LPOLLIN) {
+ if (revents & LPOLLIN) {
int sock;
/* Empty pipe */
health_code_update();
/*
- * We only monitor the error events of the socket. This
- * thread does not handle any incoming data from UST
- * (POLLIN).
+ * Since this is a command socket (write then read),
+ * we only monitor the error events of the socket.
*/
ret = lttng_poll_add(&events, sock,
LPOLLERR | LPOLLHUP | LPOLLRDHUP);
}
DBG("Apps with sock %d added to poll set", sock);
+ } else if (revents & (LPOLLERR | LPOLLHUP | LPOLLRDHUP)) {
+ ERR("Apps command pipe error");
+ goto error;
+ } else {
+ ERR("Unknown poll events %u for sock %d", revents, pollfd);
+ goto error;
}
} else {
/*
/* Socket closed on remote end. */
ust_app_unregister(pollfd);
+ } else {
+ ERR("Unexpected poll events %u for sock %d", revents, pollfd);
+ goto error;
}
}
ust_app_destroy(wait_node->app);
free(wait_node);
break;
+ } else {
+ ERR("Unexpected poll events %u for sock %d", revents, pollfd);
+ goto error;
}
}
}
/* Event on the registration socket */
if (pollfd == apps_sock) {
- if (revents & (LPOLLERR | LPOLLHUP | LPOLLRDHUP)) {
- ERR("Register apps socket poll error");
- goto error;
- } else if (revents & LPOLLIN) {
+ if (revents & LPOLLIN) {
sock = lttcomm_accept_unix_sock(apps_sock);
if (sock < 0) {
goto error;
* barrier with the exchange in cds_wfcq_enqueue.
*/
futex_nto1_wake(&ust_cmd_queue.futex);
+ } else if (revents & (LPOLLERR | LPOLLHUP | LPOLLRDHUP)) {
+ ERR("Register apps socket poll error");
+ goto error;
+ } else {
+ ERR("Unexpected poll events %u for sock %d", revents, pollfd);
+ goto error;
}
}
}
/* Event on the registration socket */
if (pollfd == sock) {
- if (revents & (LPOLLERR | LPOLLHUP | LPOLLRDHUP)) {
+ if (revents & LPOLLIN) {
+ continue;
+ } else if (revents & (LPOLLERR | LPOLLHUP | LPOLLRDHUP)) {
ERR("Health socket poll error");
goto error;
+ } else {
+ ERR("Unexpected poll events %u for sock %d", revents, pollfd);
+ goto error;
}
}
}
/* Event on the registration socket */
if (pollfd == client_sock) {
- if (revents & (LPOLLERR | LPOLLHUP | LPOLLRDHUP)) {
+ if (revents & LPOLLIN) {
+ continue;
+ } else if (revents & (LPOLLERR | LPOLLHUP | LPOLLRDHUP)) {
ERR("Client socket poll error");
goto error;
+ } else {
+ ERR("Unexpected poll events %u for sock %d", revents, pollfd);
+ goto error;
}
}
}
}
/* Add notify pipe to the pollset. */
- ret = lttng_poll_add(&events, apps_cmd_notify_pipe[0], LPOLLIN | LPOLLERR);
+ ret = lttng_poll_add(&events, apps_cmd_notify_pipe[0],
+ LPOLLIN | LPOLLERR | LPOLLHUP | LPOLLRDHUP);
if (ret < 0) {
goto error;
}
if (pollfd == apps_cmd_notify_pipe[0]) {
int sock;
- if (revents & (LPOLLERR | LPOLLHUP | LPOLLRDHUP)) {
+ if (revents & LPOLLIN) {
+ /* Get socket from dispatch thread. */
+ size_ret = lttng_read(apps_cmd_notify_pipe[0],
+ &sock, sizeof(sock));
+ if (size_ret < sizeof(sock)) {
+ PERROR("read apps notify pipe");
+ goto error;
+ }
+ health_code_update();
+
+ ret = lttng_poll_add(&events, sock,
+ LPOLLIN | LPOLLERR | LPOLLHUP | LPOLLRDHUP);
+ if (ret < 0) {
+ /*
+ * It's possible we've reached the max poll fd allowed.
+ * Let's close the socket but continue normal execution.
+ */
+ ret = close(sock);
+ if (ret) {
+ PERROR("close notify socket %d", sock);
+ }
+ lttng_fd_put(LTTNG_FD_APPS, 1);
+ continue;
+ }
+ DBG3("UST thread notify added sock %d to pollset", sock);
+ } else if (revents & (LPOLLERR | LPOLLHUP | LPOLLRDHUP)) {
ERR("Apps notify command pipe error");
goto error;
- } else if (!(revents & LPOLLIN)) {
- /* No POLLIN and not a catched error, stop the thread. */
- ERR("Notify command pipe failed. revent: %u", revents);
- goto error;
- }
-
- /* Get socket from dispatch thread. */
- size_ret = lttng_read(apps_cmd_notify_pipe[0],
- &sock, sizeof(sock));
- if (size_ret < sizeof(sock)) {
- PERROR("read apps notify pipe");
+ } else {
+ ERR("Unexpected poll events %u for sock %d", revents, pollfd);
goto error;
}
- health_code_update();
-
- ret = lttng_poll_add(&events, sock,
- LPOLLIN | LPOLLERR | LPOLLHUP | LPOLLRDHUP);
- if (ret < 0) {
- /*
- * It's possible we've reached the max poll fd allowed.
- * Let's close the socket but continue normal execution.
- */
- ret = close(sock);
- if (ret) {
- PERROR("close notify socket %d", sock);
- }
- lttng_fd_put(LTTNG_FD_APPS, 1);
- continue;
- }
- DBG3("UST thread notify added sock %d to pollset", sock);
} else {
/*
* At this point, we know that a registered application
* triggered the event.
*/
- if (revents & (LPOLLERR | LPOLLHUP | LPOLLRDHUP)) {
+ if (revents & (LPOLLIN | LPOLLPRI)) {
+ ret = ust_app_recv_notify(pollfd);
+ if (ret < 0) {
+ /* Removing from the poll set */
+ ret = lttng_poll_del(&events, pollfd);
+ if (ret < 0) {
+ goto error;
+ }
+
+ /* The socket is closed after a grace period here. */
+ ust_app_notify_sock_unregister(pollfd);
+ }
+ } else if (revents & (LPOLLERR | LPOLLHUP | LPOLLRDHUP)) {
/* Removing from the poll set */
ret = lttng_poll_del(&events, pollfd);
if (ret < 0) {
/* The socket is closed after a grace period here. */
ust_app_notify_sock_unregister(pollfd);
- } else if (revents & (LPOLLIN | LPOLLPRI)) {
- ret = ust_app_recv_notify(pollfd);
- if (ret < 0) {
- /*
- * If the notification failed either the application is
- * dead or an internal error happened. In both cases,
- * we can only continue here. If the application is
- * dead, an unregistration will follow or else the
- * application will notice that we are not responding
- * on that socket and will close it.
- */
- continue;
- }
} else {
- ERR("Unknown poll events %u for sock %d", revents, pollfd);
- continue;
+ ERR("Unexpected poll events %u for sock %d", revents, pollfd);
+ goto error;
}
health_code_update();
}
}
if (pollfd == lttng_pipe_get_readfd(ctx->consumer_metadata_pipe)) {
- if (revents & (LPOLLERR | LPOLLHUP )) {
- DBG("Metadata thread pipe hung up");
- /*
- * Remove the pipe from the poll set and continue the loop
- * since their might be data to consume.
- */
- lttng_poll_del(&events,
- lttng_pipe_get_readfd(ctx->consumer_metadata_pipe));
- lttng_pipe_read_close(ctx->consumer_metadata_pipe);
- continue;
- } else if (revents & LPOLLIN) {
+ if (revents & LPOLLIN) {
ssize_t pipe_len;
pipe_len = lttng_pipe_read(ctx->consumer_metadata_pipe,
&stream, sizeof(stream));
if (pipe_len < sizeof(stream)) {
- PERROR("read metadata stream");
+ if (pipe_len < 0) {
+ PERROR("read metadata stream");
+ }
/*
- * Continue here to handle the rest of the streams.
+ * Remove the pipe from the poll set and continue the loop
+ * since their might be data to consume.
*/
+ lttng_poll_del(&events,
+ lttng_pipe_get_readfd(ctx->consumer_metadata_pipe));
+ lttng_pipe_read_close(ctx->consumer_metadata_pipe);
continue;
}
/* Add metadata stream to the global poll events list */
lttng_poll_add(&events, stream->wait_fd,
LPOLLIN | LPOLLPRI | LPOLLHUP);
+ } else if (revents & (LPOLLERR | LPOLLHUP)) {
+ DBG("Metadata thread pipe hung up");
+ /*
+ * Remove the pipe from the poll set and continue the loop
+ * since their might be data to consume.
+ */
+ lttng_poll_del(&events,
+ lttng_pipe_get_readfd(ctx->consumer_metadata_pipe));
+ lttng_pipe_read_close(ctx->consumer_metadata_pipe);
+ continue;
+ } else {
+ ERR("Unexpected poll events %u for sock %d", revents, pollfd);
+ goto end;
}
/* Handle other stream */
stream = caa_container_of(node, struct lttng_consumer_stream,
node);
- /* Check for error event */
- if (revents & (LPOLLERR | LPOLLHUP)) {
+ if (revents & (LPOLLIN | LPOLLPRI)) {
+ /* Get the data out of the metadata file descriptor */
+ DBG("Metadata available on fd %d", pollfd);
+ assert(stream->wait_fd == pollfd);
+
+ do {
+ health_code_update();
+
+ len = ctx->on_buffer_ready(stream, ctx);
+ /*
+ * We don't check the return value here since if we get
+ * a negative len, it means an error occured thus we
+ * simply remove it from the poll set and free the
+ * stream.
+ */
+ } while (len > 0);
+
+ /* It's ok to have an unavailable sub-buffer */
+ if (len < 0 && len != -EAGAIN && len != -ENODATA) {
+ /* Clean up stream from consumer and free it. */
+ lttng_poll_del(&events, stream->wait_fd);
+ consumer_del_metadata_stream(stream, metadata_ht);
+ }
+ } else if (revents & (LPOLLERR | LPOLLHUP)) {
DBG("Metadata fd %d is hup|err.", pollfd);
if (!stream->hangup_flush_done
&& (consumer_data.type == LTTNG_CONSUMER32_UST
* and securely free the stream.
*/
consumer_del_metadata_stream(stream, metadata_ht);
- } else if (revents & (LPOLLIN | LPOLLPRI)) {
- /* Get the data out of the metadata file descriptor */
- DBG("Metadata available on fd %d", pollfd);
- assert(stream->wait_fd == pollfd);
-
- do {
- health_code_update();
-
- len = ctx->on_buffer_ready(stream, ctx);
- /*
- * We don't check the return value here since if we get
- * a negative len, it means an error occured thus we
- * simply remove it from the poll set and free the
- * stream.
- */
- } while (len > 0);
-
- /* It's ok to have an unavailable sub-buffer */
- if (len < 0 && len != -EAGAIN && len != -ENODATA) {
- /* Clean up stream from consumer and free it. */
- lttng_poll_del(&events, stream->wait_fd);
- consumer_del_metadata_stream(stream, metadata_ht);
- }
+ } else {
+ ERR("Unexpected poll events %u for sock %d", revents, pollfd);
+ rcu_read_unlock;
+ goto end;
}
-
/* Release RCU lock for the stream looked up */
rcu_read_unlock();
}
}
if (pollfd == ctx->consumer_channel_pipe[0]) {
- if (revents & (LPOLLERR | LPOLLHUP)) {
- DBG("Channel thread pipe hung up");
- /*
- * Remove the pipe from the poll set and continue the loop
- * since their might be data to consume.
- */
- lttng_poll_del(&events, ctx->consumer_channel_pipe[0]);
- continue;
- } else if (revents & LPOLLIN) {
+ if (revents & LPOLLIN) {
enum consumer_channel_action action;
uint64_t key;
ret = read_channel_pipe(ctx, &chan, &key, &action);
if (ret <= 0) {
- ERR("Error reading channel pipe");
+ if (ret < 0) {
+ ERR("Error reading channel pipe");
+ }
+ lttng_poll_del(&events, ctx->consumer_channel_pipe[0]);
continue;
}
rcu_read_unlock();
/* Add channel to the global poll events list */
lttng_poll_add(&events, chan->wait_fd,
- LPOLLIN | LPOLLPRI);
+ LPOLLERR | LPOLLHUP);
break;
case CONSUMER_CHANNEL_DEL:
{
ERR("Unknown action");
break;
}
+ } else if (revents & (LPOLLERR | LPOLLHUP)) {
+ DBG("Channel thread pipe hung up");
+ /*
+ * Remove the pipe from the poll set and continue the loop
+ * since their might be data to consume.
+ */
+ lttng_poll_del(&events, ctx->consumer_channel_pipe[0]);
+ continue;
+ } else {
+ ERR("Unexpected poll events %u for sock %d", revents, pollfd);
+ goto end;
}
/* Handle other stream */
&& !uatomic_read(&chan->nb_init_stream_left)) {
consumer_del_channel(chan);
}
+ } else {
+ ERR("Unexpected poll events %u for sock %d", revents, pollfd);
+ rcu_read_unlock();
+ goto end;
}
/* Release RCU lock for the channel looked up */
projects / lttng-tools.git / commitdiff
