*/
#define _GNU_SOURCE
+#define _LGPL_SOURCE
#include <assert.h>
#include <common/common.h>
#include "fd-limit.h"
#include "agent-thread.h"
+#include "agent.h"
#include "lttng-sessiond.h"
#include "session.h"
#include "utils.h"
if (session->ust_session) {
struct agent *agt;
+ rcu_read_lock();
agt = trace_ust_find_agent(session->ust_session, app->domain);
if (agt) {
agent_update(agt, app->sock->fd);
}
+ rcu_read_unlock();
}
session_unlock(session);
}
session_unlock_list();
}
-/*
- * Destroy a agent application by socket.
- */
-static void destroy_agent_app(int sock)
-{
- struct agent_app *app;
-
- assert(sock >= 0);
-
- /*
- * Not finding an application is a very important error that should NEVER
- * happen. The hash table deletion is ONLY done through this call even on
- * thread cleanup.
- */
- rcu_read_lock();
- app = agent_find_app_by_sock(sock);
- assert(app);
- rcu_read_unlock();
-
- /* RCU read side lock is taken in this function call. */
- agent_delete_app(app);
-
- /* The application is freed in a RCU call but the socket is closed here. */
- agent_destroy_app(app);
-}
-
-/*
- * Cleanup remaining agent apps in the hash table. This should only be called in
- * the exit path of the thread.
- */
-static void clean_agent_apps_ht(void)
-{
- struct lttng_ht_node_ulong *node;
- struct lttng_ht_iter iter;
-
- DBG3("[agent-thread] Cleaning agent apps ht");
-
- rcu_read_lock();
- cds_lfht_for_each_entry(agent_apps_ht_by_sock->ht, &iter.iter, node, node) {
- struct agent_app *app;
-
- app = caa_container_of(node, struct agent_app, node);
- destroy_agent_app(app->sock->fd);
- }
- rcu_read_unlock();
-}
-
/*
* Create and init socket from uri.
*/
{
int ret;
pid_t pid;
+ uint32_t major_version, minor_version;
ssize_t size;
enum lttng_domain_type domain;
struct agent_app *app;
size = new_sock->ops->recvmsg(new_sock, &msg, sizeof(msg), 0);
if (size < sizeof(msg)) {
- ret = -errno;
+ ret = -EINVAL;
goto error_socket;
}
domain = be32toh(msg.domain);
pid = be32toh(msg.pid);
+ major_version = be32toh(msg.major_version);
+ minor_version = be32toh(msg.minor_version);
+
+ /* Test communication protocol version of the registring agent. */
+ if (major_version != AGENT_MAJOR_VERSION) {
+ ret = -EINVAL;
+ goto error_socket;
+ }
+ if (minor_version != AGENT_MINOR_VERSION) {
+ ret = -EINVAL;
+ goto error_socket;
+ }
DBG2("[agent-thread] New registration for pid %d domain %d on socket %d",
pid, domain, new_sock->fd);
}
while (1) {
- DBG3("[agent-thread] Manage agent polling on %d fds",
- LTTNG_POLL_GETNB(&events));
+ DBG3("[agent-thread] Manage agent polling");
/* Inifinite blocking call, waiting for transmission */
restart:
ret = lttng_poll_wait(&events, -1);
+ DBG3("[agent-thread] Manage agent return from poll on %d fds",
+ LTTNG_POLL_GETNB(&events));
if (ret < 0) {
/*
* Restart interrupted system call.
revents = LTTNG_POLL_GETEV(&events, i);
pollfd = LTTNG_POLL_GETFD(&events, i);
+ if (!revents) {
+ /* No activity for this FD (poll implementation). */
+ continue;
+ }
+
/* Thread quit pipe has been closed. Killing thread. */
ret = sessiond_check_thread_quit_pipe(pollfd, revents);
if (ret) {
goto error;
}
- destroy_agent_app(pollfd);
+ agent_destroy_app_by_sock(pollfd);
} else if (revents & (LPOLLIN)) {
int new_fd;
struct agent_app *app = NULL;
ret = lttng_poll_add(&events, new_fd,
LPOLLERR | LPOLLHUP | LPOLLRDHUP);
if (ret < 0) {
- destroy_agent_app(new_fd);
+ agent_destroy_app_by_sock(new_fd);
continue;
}
error_poll_create:
DBG("[agent-thread] is cleaning up and stopping.");
- if (agent_apps_ht_by_sock) {
- clean_agent_apps_ht();
- lttng_ht_destroy(agent_apps_ht_by_sock);
- }
-
rcu_thread_offline();
rcu_unregister_thread();
return NULL;