X-Git-Url: http://git.efficios.com/?p=lttng-tools.git;a=blobdiff_plain;f=src%2Fbin%2Flttng-sessiond%2Fconsumer.c;h=d79dee3cafd99c1232f2458a985561fb0e7a1875;hp=ee6f3c3ff8e9ddb5c614d6b9c60d5f2d592e21ff;hb=9363801e2d2069022a05e67066d8f527538946d0;hpb=fdadac0853f699c7dc829162955cc8f459b6d2f2 diff --git a/src/bin/lttng-sessiond/consumer.c b/src/bin/lttng-sessiond/consumer.c index ee6f3c3ff..d79dee3ca 100644 --- a/src/bin/lttng-sessiond/consumer.c +++ b/src/bin/lttng-sessiond/consumer.c @@ -48,11 +48,11 @@ int consumer_socket_send(struct consumer_socket *socket, void *msg, size_t len) ssize_t size; assert(socket); - assert(socket->fd); + assert(socket->fd_ptr); assert(msg); /* Consumer socket is invalid. Stopping. */ - fd = *socket->fd; + fd = *socket->fd_ptr; if (fd < 0) { goto error; } @@ -68,7 +68,7 @@ int consumer_socket_send(struct consumer_socket *socket, void *msg, size_t len) /* This call will PERROR on error. */ (void) lttcomm_close_unix_sock(fd); - *socket->fd = -1; + *socket->fd_ptr = -1; goto error; } @@ -92,11 +92,11 @@ int consumer_socket_recv(struct consumer_socket *socket, void *msg, size_t len) ssize_t size; assert(socket); - assert(socket->fd); + assert(socket->fd_ptr); assert(msg); /* Consumer socket is invalid. Stopping. */ - fd = *socket->fd; + fd = *socket->fd_ptr; if (fd < 0) { goto error; } @@ -112,7 +112,7 @@ int consumer_socket_recv(struct consumer_socket *socket, void *msg, size_t len) /* This call will PERROR on error. */ (void) lttcomm_close_unix_sock(fd); - *socket->fd = -1; + *socket->fd_ptr = -1; goto error; } @@ -203,7 +203,7 @@ int consumer_send_destroy_relayd(struct consumer_socket *sock, assert(consumer); assert(sock); - DBG2("Sending destroy relayd command to consumer sock %d", *sock->fd); + DBG2("Sending destroy relayd command to consumer sock %d", *sock->fd_ptr); msg.cmd_type = LTTNG_CONSUMER_DESTROY_RELAYD; msg.u.destroy_relayd.net_seq_idx = consumer->net_seq_index; @@ -378,7 +378,7 @@ struct consumer_socket *consumer_allocate_socket(int *fd) goto error; } - socket->fd = fd; + socket->fd_ptr = fd; lttng_ht_node_init_ulong(&socket->node, *fd); error: @@ -443,8 +443,8 @@ void consumer_destroy_socket(struct consumer_socket *sock) * consumer was registered, */ if (sock->registered) { - DBG3("Consumer socket was registered. Closing fd %d", *sock->fd); - lttcomm_close_unix_sock(*sock->fd); + DBG3("Consumer socket was registered. Closing fd %d", *sock->fd_ptr); + lttcomm_close_unix_sock(*sock->fd_ptr); } call_rcu(&sock->node.head, destroy_socket_rcu); @@ -575,13 +575,13 @@ int consumer_copy_sockets(struct consumer_output *dst, rcu_read_lock(); cds_lfht_for_each_entry(src->socks->ht, &iter.iter, socket, node.node) { /* Ignore socket that are already there. */ - copy_sock = consumer_find_socket(*socket->fd, dst); + copy_sock = consumer_find_socket(*socket->fd_ptr, dst); if (copy_sock) { continue; } /* Create new socket object. */ - copy_sock = consumer_allocate_socket(socket->fd); + copy_sock = consumer_allocate_socket(socket->fd_ptr); if (copy_sock == NULL) { rcu_read_unlock(); ret = -ENOMEM; @@ -714,13 +714,12 @@ int consumer_send_fds(struct consumer_socket *sock, int *fds, size_t nb_fd) assert(fds); assert(sock); - assert(sock->fd); assert(nb_fd > 0); - ret = lttcomm_send_fds_unix_sock(*sock->fd, fds, nb_fd); + ret = lttcomm_send_fds_unix_sock(*sock->fd_ptr, fds, nb_fd); if (ret < 0) { /* The above call will print a PERROR on error. */ - DBG("Error when sending consumer fds on sock %d", *sock->fd); + DBG("Error when sending consumer fds on sock %d", *sock->fd_ptr); goto error; } @@ -740,7 +739,6 @@ int consumer_send_msg(struct consumer_socket *sock, assert(msg); assert(sock); - assert(sock->fd); ret = consumer_socket_send(sock, msg, sizeof(struct lttcomm_consumer_msg)); if (ret < 0) { @@ -763,7 +761,6 @@ int consumer_send_channel(struct consumer_socket *sock, assert(msg); assert(sock); - assert(sock->fd); ret = consumer_send_msg(sock, msg); if (ret < 0) { @@ -915,7 +912,6 @@ int consumer_send_stream(struct consumer_socket *sock, assert(msg); assert(dst); assert(sock); - assert(sock->fd); assert(fds); ret = consumer_send_msg(sock, msg); @@ -948,7 +944,6 @@ int consumer_send_relayd_socket(struct consumer_socket *consumer_sock, assert(rsock); assert(consumer); assert(consumer_sock); - assert(consumer_sock->fd); /* Bail out if consumer is disabled */ if (!consumer->enabled) { @@ -967,7 +962,7 @@ int consumer_send_relayd_socket(struct consumer_socket *consumer_sock, msg.u.relayd_sock.session_id = session_id; memcpy(&msg.u.relayd_sock.sock, rsock, sizeof(msg.u.relayd_sock.sock)); - DBG3("Sending relayd sock info to consumer on %d", *consumer_sock->fd); + DBG3("Sending relayd sock info to consumer on %d", *consumer_sock->fd_ptr); ret = consumer_send_msg(consumer_sock, &msg); if (ret < 0) { goto error; @@ -1061,9 +1056,6 @@ int consumer_is_data_pending(uint64_t session_id, rcu_read_lock(); cds_lfht_for_each_entry(consumer->socks->ht, &iter.iter, socket, node.node) { - /* Code flow error */ - assert(socket->fd); - pthread_mutex_lock(socket->lock); ret = consumer_socket_send(socket, &msg, sizeof(msg)); if (ret < 0) { @@ -1109,7 +1101,6 @@ int consumer_flush_channel(struct consumer_socket *socket, uint64_t key) struct lttcomm_consumer_msg msg; assert(socket); - assert(socket->fd); DBG2("Consumer flush channel key %" PRIu64, key); @@ -1142,7 +1133,6 @@ int consumer_close_metadata(struct consumer_socket *socket, struct lttcomm_consumer_msg msg; assert(socket); - assert(socket->fd); DBG2("Consumer close metadata channel key %" PRIu64, metadata_key); @@ -1175,7 +1165,6 @@ int consumer_setup_metadata(struct consumer_socket *socket, struct lttcomm_consumer_msg msg; assert(socket); - assert(socket->fd); DBG2("Consumer setup metadata channel key %" PRIu64, metadata_key); @@ -1209,9 +1198,8 @@ int consumer_push_metadata(struct consumer_socket *socket, struct lttcomm_consumer_msg msg; assert(socket); - assert(socket->fd); - DBG2("Consumer push metadata to consumer socket %d", *socket->fd); + DBG2("Consumer push metadata to consumer socket %d", *socket->fd_ptr); msg.cmd_type = LTTNG_CONSUMER_PUSH_METADATA; msg.u.push_metadata.key = metadata_key; @@ -1224,7 +1212,8 @@ int consumer_push_metadata(struct consumer_socket *socket, goto end; } - DBG3("Consumer pushing metadata on sock %d of len %zu", *socket->fd, len); + DBG3("Consumer pushing metadata on sock %d of len %zu", *socket->fd_ptr, + len); ret = consumer_socket_send(socket, metadata_str, len); if (ret < 0) { @@ -1255,7 +1244,6 @@ int consumer_snapshot_channel(struct consumer_socket *socket, uint64_t key, struct lttcomm_consumer_msg msg; assert(socket); - assert(socket->fd); assert(output); assert(output->consumer);