#include <unistd.h>
#include <assert.h>
#include <inttypes.h>
+#include <fcntl.h>
#define CLIENT_POLL_MASK_IN (LPOLLIN | LPOLLERR | LPOLLHUP | LPOLLRDHUP)
#define CLIENT_POLL_MASK_IN_OUT (CLIENT_POLL_MASK_IN | LPOLLOUT)
struct cds_lfht_node client_socket_ht_node;
struct {
struct {
+ /*
+ * During the reception of a message, the reception
+ * buffers' "size" is set to contain the current
+ * message's complete payload.
+ */
struct lttng_dynamic_buffer buffer;
/* Bytes left to receive for the current message. */
size_t bytes_to_receive;
* from the client.
*/
bool creds_received;
+ /* Only used during credentials reception. */
lttng_sock_cred creds;
} inbound;
struct {
hash ^= hash_key_str(condition->session_name, lttng_ht_seed);
}
if (condition->channel_name) {
- hash ^= hash_key_str(condition->session_name, lttng_ht_seed);
+ hash ^= hash_key_str(condition->channel_name, lttng_ht_seed);
}
if (condition->domain.set) {
hash ^= hash_key_ulong(
&iter);
node = cds_lfht_iter_get_node(&iter);
if (!node) {
+ free(client_list_element);
goto end_unlock;
}
return ret;
}
+static
int handle_notification_thread_command_unregister_trigger(
struct notification_thread_state *state,
struct lttng_trigger *trigger,
}
end:
cds_list_del(&cmd->cmd_list_node);
- futex_nto1_wake(&cmd->reply_futex);
+ lttng_waiter_wake_up(&cmd->reply_waiter);
pthread_mutex_unlock(&handle->cmd_queue.lock);
return ret;
error_unlock:
/* Wake-up and return a fatal error to the calling thread. */
- futex_nto1_wake(&cmd->reply_futex);
+ lttng_waiter_wake_up(&cmd->reply_waiter);
pthread_mutex_unlock(&handle->cmd_queue.lock);
cmd->reply_code = LTTNG_ERR_FATAL;
error:
}
static
-void client_reset_inbound_state(struct notification_client *client)
+int client_reset_inbound_state(struct notification_client *client)
{
int ret;
client->communication.inbound.receive_creds = false;
LTTNG_SOCK_SET_UID_CRED(&client->communication.inbound.creds, -1);
LTTNG_SOCK_SET_GID_CRED(&client->communication.inbound.creds, -1);
+ ret = lttng_dynamic_buffer_set_size(
+ &client->communication.inbound.buffer,
+ client->communication.inbound.bytes_to_receive);
+ return ret;
}
int handle_notification_thread_client_connect(
CDS_INIT_LIST_HEAD(&client->condition_list);
lttng_dynamic_buffer_init(&client->communication.inbound.buffer);
lttng_dynamic_buffer_init(&client->communication.outbound.buffer);
- client_reset_inbound_state(client);
+ ret = client_reset_inbound_state(client);
+ if (ret) {
+ ERR("[notification-thread] Failed to reset client communication's inbound state");
+ ret = 0;
+ goto error;
+ }
ret = lttcomm_accept_unix_sock(state->notification_channel_socket);
if (ret < 0) {
DBG("[notification-thread] Added new notification channel client socket (%i) to poll set",
client->socket);
- /* Add to ht. */
rcu_read_lock();
cds_lfht_add(state->client_socket_ht,
hash_client_socket(client->socket),
client->communication.inbound.receive_creds = true;
}
ret = lttng_dynamic_buffer_set_size(
- &client->communication.inbound.buffer, 0);
+ &client->communication.inbound.buffer, msg->size);
if (ret) {
goto end;
}
handshake_client =
(struct lttng_notification_channel_command_handshake *)
client->communication.inbound.buffer.data;
- client->major = handshake_client->major;
+ client->major = handshake_client->major;
client->minor = handshake_client->minor;
if (!client->communication.inbound.creds_received) {
ERR("[notification-thread] No credentials received from client");
}
/* Set reception state to receive the next message header. */
- client_reset_inbound_state(client);
+ ret = client_reset_inbound_state(client);
+ if (ret) {
+ ERR("[notification-thread] Failed to reset client communication's inbound state");
+ goto end;
+ }
client->validated = true;
break;
}
}
/* Set reception state to receive the next message header. */
- client_reset_inbound_state(client);
+ ret = client_reset_inbound_state(client);
+ if (ret) {
+ ERR("[notification-thread] Failed to reset client communication's inbound state");
+ goto end;
+ }
break;
}
default:
int handle_notification_thread_client_in(
struct notification_thread_state *state, int socket)
{
- int ret;
+ int ret = 0;
struct notification_client *client;
ssize_t recv_ret;
size_t offset;
goto end;
}
- offset = client->communication.inbound.buffer.size;
- ret = lttng_dynamic_buffer_set_size(
- &client->communication.inbound.buffer,
- client->communication.inbound.bytes_to_receive);
- if (ret) {
- goto end;
- }
-
+ offset = client->communication.inbound.buffer.size -
+ client->communication.inbound.bytes_to_receive;
if (client->communication.inbound.receive_creds) {
recv_ret = lttcomm_recv_creds_unix_sock(socket,
client->communication.inbound.buffer.data + offset,
}
client->communication.inbound.bytes_to_receive -= recv_ret;
- ret = lttng_dynamic_buffer_set_size(
- &client->communication.inbound.buffer,
- client->communication.inbound.buffer.size -
- client->communication.inbound.bytes_to_receive);
- if (ret) {
- goto end;
- }
-
if (client->communication.inbound.bytes_to_receive == 0) {
ret = client_dispatch_message(client, state);
if (ret) {
* The monitoring pipe only holds messages smaller than PIPE_BUF,
* ensuring that read/write of sampling messages are atomic.
*/
- do {
- ret = read(pipe, &sample_msg, sizeof(sample_msg));
- } while (ret == -1 && errno == EINTR);
+ ret = lttng_read(pipe, &sample_msg, sizeof(sample_msg));
if (ret != sizeof(sample_msg)) {
ERR("[notification-thread] Failed to read from monitoring pipe (fd = %i)",
pipe);