trigger/action executor: move trigger registration state change to client thread
[lttng-tools.git] / src / bin / lttng-sessiond / action-executor.c
CommitLineData
f2b3ef9f
JG
1/*
2 * Copyright (C) 2020 Jérémie Galarneau <jeremie.galarneau@efficios.com>
3 *
4 * SPDX-License-Identifier: GPL-2.0-only
5 *
6 */
7
8#include "action-executor.h"
9#include "cmd.h"
10#include "health-sessiond.h"
11#include "lttng-sessiond.h"
12#include "notification-thread-internal.h"
13#include "session.h"
14#include "thread.h"
15#include <common/macros.h>
16#include <common/optional.h>
17#include <lttng/action/action-internal.h>
2d57482c 18#include <lttng/action/group-internal.h>
f2b3ef9f 19#include <lttng/action/group.h>
2af3b9c9 20#include <lttng/action/notify-internal.h>
f2b3ef9f
JG
21#include <lttng/action/notify.h>
22#include <lttng/action/rotate-session.h>
23#include <lttng/action/snapshot-session.h>
24#include <lttng/action/start-session.h>
25#include <lttng/action/stop-session.h>
26#include <lttng/condition/evaluation.h>
e393070a 27#include <lttng/condition/on-event-internal.h>
f2b3ef9f
JG
28#include <lttng/lttng-error.h>
29#include <lttng/trigger/trigger-internal.h>
30#include <pthread.h>
31#include <stdbool.h>
32#include <stddef.h>
33#include <urcu/list.h>
34
35#define THREAD_NAME "Action Executor"
36#define MAX_QUEUED_WORK_COUNT 8192
37
38struct action_work_item {
39 uint64_t id;
40 struct lttng_trigger *trigger;
41 struct lttng_evaluation *evaluation;
42 struct notification_client_list *client_list;
43 LTTNG_OPTIONAL(struct lttng_credentials) object_creds;
44 struct cds_list_head list_node;
45};
46
47struct action_executor {
48 struct lttng_thread *thread;
49 struct notification_thread_handle *notification_thread_handle;
50 struct {
51 uint64_t pending_count;
52 struct cds_list_head list;
53 pthread_cond_t cond;
54 pthread_mutex_t lock;
55 } work;
56 bool should_quit;
57 uint64_t next_work_item_id;
58};
59
60/*
61 * Only return non-zero on a fatal error that should shut down the action
62 * executor.
63 */
64typedef int (*action_executor_handler)(struct action_executor *executor,
65 const struct action_work_item *,
2d57482c 66 struct lttng_action *action);
f2b3ef9f
JG
67
68static int action_executor_notify_handler(struct action_executor *executor,
69 const struct action_work_item *,
2d57482c
JR
70 struct lttng_action *);
71static int action_executor_start_session_handler(
72 struct action_executor *executor,
f2b3ef9f 73 const struct action_work_item *,
2d57482c
JR
74 struct lttng_action *);
75static int action_executor_stop_session_handler(
76 struct action_executor *executor,
f2b3ef9f 77 const struct action_work_item *,
2d57482c
JR
78 struct lttng_action *);
79static int action_executor_rotate_session_handler(
80 struct action_executor *executor,
f2b3ef9f 81 const struct action_work_item *,
2d57482c
JR
82 struct lttng_action *);
83static int action_executor_snapshot_session_handler(
84 struct action_executor *executor,
f2b3ef9f 85 const struct action_work_item *,
2d57482c 86 struct lttng_action *);
f2b3ef9f
JG
87static int action_executor_group_handler(struct action_executor *executor,
88 const struct action_work_item *,
2d57482c 89 struct lttng_action *);
f2b3ef9f
JG
90static int action_executor_generic_handler(struct action_executor *executor,
91 const struct action_work_item *,
2d57482c 92 struct lttng_action *);
f2b3ef9f
JG
93
94static const action_executor_handler action_executors[] = {
95 [LTTNG_ACTION_TYPE_NOTIFY] = action_executor_notify_handler,
96 [LTTNG_ACTION_TYPE_START_SESSION] = action_executor_start_session_handler,
97 [LTTNG_ACTION_TYPE_STOP_SESSION] = action_executor_stop_session_handler,
98 [LTTNG_ACTION_TYPE_ROTATE_SESSION] = action_executor_rotate_session_handler,
99 [LTTNG_ACTION_TYPE_SNAPSHOT_SESSION] = action_executor_snapshot_session_handler,
100 [LTTNG_ACTION_TYPE_GROUP] = action_executor_group_handler,
101};
102
f2b3ef9f
JG
103static const char *get_action_name(const struct lttng_action *action)
104{
0e43bcbf
JG
105 const enum lttng_action_type action_type = lttng_action_get_type(action);
106
107 assert(action_type != LTTNG_ACTION_TYPE_UNKNOWN);
108
c0e2990d 109 return lttng_action_type_string(action_type);
f2b3ef9f
JG
110}
111
112/* Check if this trigger allowed to interect with a given session. */
113static bool is_trigger_allowed_for_session(const struct lttng_trigger *trigger,
114 struct ltt_session *session)
115{
116 bool is_allowed = false;
117 const struct lttng_credentials session_creds = {
ff588497
JR
118 .uid = LTTNG_OPTIONAL_INIT_VALUE(session->uid),
119 .gid = LTTNG_OPTIONAL_INIT_VALUE(session->gid),
f2b3ef9f
JG
120 };
121 /* Can never be NULL. */
122 const struct lttng_credentials *trigger_creds =
123 lttng_trigger_get_credentials(trigger);
124
ff588497
JR
125 is_allowed = (lttng_credentials_is_equal_uid(trigger_creds, &session_creds)) ||
126 (lttng_credentials_get_uid(trigger_creds) == 0);
f2b3ef9f 127 if (!is_allowed) {
ff588497 128 WARN("Trigger is not allowed to interact with session `%s`: session uid = %ld, session gid = %ld, trigger uid = %ld",
f2b3ef9f
JG
129 session->name,
130 (long int) session->uid,
131 (long int) session->gid,
ff588497 132 (long int) lttng_credentials_get_uid(trigger_creds));
f2b3ef9f
JG
133 }
134
135 return is_allowed;
136}
137
34f87583
JR
138static const char *get_trigger_name(const struct lttng_trigger *trigger)
139{
140 const char *trigger_name;
141 enum lttng_trigger_status trigger_status;
142
143 trigger_status = lttng_trigger_get_name(trigger, &trigger_name);
144 assert(trigger_status == LTTNG_TRIGGER_STATUS_OK);
145
146 return trigger_name;
147}
148
f2b3ef9f
JG
149static int client_handle_transmission_status(
150 struct notification_client *client,
151 enum client_transmission_status status,
152 void *user_data)
153{
154 int ret = 0;
155 struct action_executor *executor = user_data;
156 bool update_communication = true;
157
f2b3ef9f
JG
158 switch (status) {
159 case CLIENT_TRANSMISSION_STATUS_COMPLETE:
160 DBG("Successfully sent full notification to client, client_id = %" PRIu64,
161 client->id);
162 update_communication = false;
163 break;
164 case CLIENT_TRANSMISSION_STATUS_QUEUED:
165 DBG("Queued notification in client outgoing buffer, client_id = %" PRIu64,
166 client->id);
167 break;
168 case CLIENT_TRANSMISSION_STATUS_FAIL:
169 DBG("Communication error occurred while sending notification to client, client_id = %" PRIu64,
170 client->id);
f2b3ef9f
JG
171 break;
172 default:
173 ERR("Fatal error encoutered while sending notification to client, client_id = %" PRIu64,
174 client->id);
f2b3ef9f
JG
175 ret = -1;
176 goto end;
177 }
178
179 if (!update_communication) {
180 goto end;
181 }
182
6c24d3fd 183 /* Safe to read client's id without locking as it is immutable. */
f2b3ef9f
JG
184 ret = notification_thread_client_communication_update(
185 executor->notification_thread_handle, client->id,
186 status);
187end:
188 return ret;
189}
190
191static int action_executor_notify_handler(struct action_executor *executor,
192 const struct action_work_item *work_item,
2d57482c 193 struct lttng_action *action)
f2b3ef9f
JG
194{
195 return notification_client_list_send_evaluation(work_item->client_list,
52d55cf9 196 work_item->trigger,
f2b3ef9f 197 work_item->evaluation,
c203f058
JR
198 work_item->object_creds.is_set ?
199 &(work_item->object_creds.value) :
200 NULL,
64eafdf6 201 client_handle_transmission_status, executor);
f2b3ef9f
JG
202}
203
2d57482c
JR
204static int action_executor_start_session_handler(
205 struct action_executor *executor,
f2b3ef9f 206 const struct action_work_item *work_item,
2d57482c 207 struct lttng_action *action)
f2b3ef9f
JG
208{
209 int ret = 0;
210 const char *session_name;
211 enum lttng_action_status action_status;
212 struct ltt_session *session;
213 enum lttng_error_code cmd_ret;
214
215 action_status = lttng_action_start_session_get_session_name(
216 action, &session_name);
217 if (action_status != LTTNG_ACTION_STATUS_OK) {
218 ERR("Failed to get session name from `%s` action",
219 get_action_name(action));
220 ret = -1;
221 goto end;
222 }
223
224 session_lock_list();
225 session = session_find_by_name(session_name);
226 if (!session) {
34f87583 227 DBG("Failed to find session `%s` by name while executing `%s` action of trigger `%s`",
f2b3ef9f 228 session_name, get_action_name(action),
34f87583 229 get_trigger_name(work_item->trigger));
f2b3ef9f
JG
230 goto error_unlock_list;
231 }
232
233 session_lock(session);
234 if (!is_trigger_allowed_for_session(work_item->trigger, session)) {
235 goto error_dispose_session;
236 }
237
238 cmd_ret = cmd_start_trace(session);
239 switch (cmd_ret) {
240 case LTTNG_OK:
34f87583
JR
241 DBG("Successfully started session `%s` on behalf of trigger `%s`",
242 session_name, get_trigger_name(work_item->trigger));
f2b3ef9f
JG
243 break;
244 case LTTNG_ERR_TRACE_ALREADY_STARTED:
34f87583
JR
245 DBG("Attempted to start session `%s` on behalf of trigger `%s` but it was already started",
246 session_name, get_trigger_name(work_item->trigger));
f2b3ef9f
JG
247 break;
248 default:
34f87583
JR
249 WARN("Failed to start session `%s` on behalf of trigger `%s`: %s",
250 session_name, get_trigger_name(work_item->trigger),
f2b3ef9f 251 lttng_strerror(-cmd_ret));
2d57482c 252 lttng_action_increase_execution_failure_count(action);
f2b3ef9f
JG
253 break;
254 }
255
256error_dispose_session:
257 session_unlock(session);
258 session_put(session);
259error_unlock_list:
260 session_unlock_list();
261end:
262 return ret;
263}
264
2d57482c
JR
265static int action_executor_stop_session_handler(
266 struct action_executor *executor,
f2b3ef9f 267 const struct action_work_item *work_item,
2d57482c 268 struct lttng_action *action)
f2b3ef9f
JG
269{
270 int ret = 0;
271 const char *session_name;
272 enum lttng_action_status action_status;
273 struct ltt_session *session;
274 enum lttng_error_code cmd_ret;
275
276 action_status = lttng_action_stop_session_get_session_name(
277 action, &session_name);
278 if (action_status != LTTNG_ACTION_STATUS_OK) {
279 ERR("Failed to get session name from `%s` action",
280 get_action_name(action));
281 ret = -1;
282 goto end;
283 }
284
285 session_lock_list();
286 session = session_find_by_name(session_name);
287 if (!session) {
34f87583 288 DBG("Failed to find session `%s` by name while executing `%s` action of trigger `%s`",
f2b3ef9f 289 session_name, get_action_name(action),
34f87583 290 get_trigger_name(work_item->trigger));
2d57482c 291 lttng_action_increase_execution_failure_count(action);
f2b3ef9f
JG
292 goto error_unlock_list;
293 }
294
295 session_lock(session);
296 if (!is_trigger_allowed_for_session(work_item->trigger, session)) {
297 goto error_dispose_session;
298 }
299
300 cmd_ret = cmd_stop_trace(session);
301 switch (cmd_ret) {
302 case LTTNG_OK:
34f87583
JR
303 DBG("Successfully stopped session `%s` on behalf of trigger `%s`",
304 session_name, get_trigger_name(work_item->trigger));
f2b3ef9f
JG
305 break;
306 case LTTNG_ERR_TRACE_ALREADY_STOPPED:
34f87583
JR
307 DBG("Attempted to stop session `%s` on behalf of trigger `%s` but it was already stopped",
308 session_name, get_trigger_name(work_item->trigger));
f2b3ef9f
JG
309 break;
310 default:
34f87583
JR
311 WARN("Failed to stop session `%s` on behalf of trigger `%s`: %s",
312 session_name, get_trigger_name(work_item->trigger),
f2b3ef9f 313 lttng_strerror(-cmd_ret));
2d57482c 314 lttng_action_increase_execution_failure_count(action);
f2b3ef9f
JG
315 break;
316 }
317
318error_dispose_session:
319 session_unlock(session);
320 session_put(session);
321error_unlock_list:
322 session_unlock_list();
323end:
324 return ret;
325}
326
2d57482c
JR
327static int action_executor_rotate_session_handler(
328 struct action_executor *executor,
f2b3ef9f 329 const struct action_work_item *work_item,
2d57482c 330 struct lttng_action *action)
f2b3ef9f
JG
331{
332 int ret = 0;
333 const char *session_name;
334 enum lttng_action_status action_status;
335 struct ltt_session *session;
336 enum lttng_error_code cmd_ret;
337
338 action_status = lttng_action_rotate_session_get_session_name(
339 action, &session_name);
340 if (action_status != LTTNG_ACTION_STATUS_OK) {
341 ERR("Failed to get session name from `%s` action",
342 get_action_name(action));
343 ret = -1;
344 goto end;
345 }
346
347 session_lock_list();
348 session = session_find_by_name(session_name);
349 if (!session) {
34f87583 350 DBG("Failed to find session `%s` by name while executing `%s` action of trigger `%s`",
f2b3ef9f 351 session_name, get_action_name(action),
34f87583 352 get_trigger_name(work_item->trigger));
2d57482c 353 lttng_action_increase_execution_failure_count(action);
f2b3ef9f
JG
354 goto error_unlock_list;
355 }
356
357 session_lock(session);
358 if (!is_trigger_allowed_for_session(work_item->trigger, session)) {
359 goto error_dispose_session;
360 }
361
362 cmd_ret = cmd_rotate_session(session, NULL, false,
363 LTTNG_TRACE_CHUNK_COMMAND_TYPE_MOVE_TO_COMPLETED);
364 switch (cmd_ret) {
365 case LTTNG_OK:
34f87583
JR
366 DBG("Successfully started rotation of session `%s` on behalf of trigger `%s`",
367 session_name, get_trigger_name(work_item->trigger));
f2b3ef9f
JG
368 break;
369 case LTTNG_ERR_ROTATION_PENDING:
34f87583
JR
370 DBG("Attempted to start a rotation of session `%s` on behalf of trigger `%s` but a rotation is already ongoing",
371 session_name, get_trigger_name(work_item->trigger));
2d57482c 372 lttng_action_increase_execution_failure_count(action);
f2b3ef9f
JG
373 break;
374 case LTTNG_ERR_ROTATION_MULTIPLE_AFTER_STOP:
375 case LTTNG_ERR_ROTATION_AFTER_STOP_CLEAR:
34f87583
JR
376 DBG("Attempted to start a rotation of session `%s` on behalf of trigger `%s` but a rotation has already been completed since the last stop or clear",
377 session_name, get_trigger_name(work_item->trigger));
f2b3ef9f
JG
378 break;
379 default:
34f87583
JR
380 WARN("Failed to start a rotation of session `%s` on behalf of trigger `%s`: %s",
381 session_name, get_trigger_name(work_item->trigger),
f2b3ef9f 382 lttng_strerror(-cmd_ret));
2d57482c 383 lttng_action_increase_execution_failure_count(action);
f2b3ef9f
JG
384 break;
385 }
386
387error_dispose_session:
388 session_unlock(session);
389 session_put(session);
390error_unlock_list:
391 session_unlock_list();
392end:
393 return ret;
394}
395
2d57482c
JR
396static int action_executor_snapshot_session_handler(
397 struct action_executor *executor,
f2b3ef9f 398 const struct action_work_item *work_item,
2d57482c 399 struct lttng_action *action)
f2b3ef9f
JG
400{
401 int ret = 0;
402 const char *session_name;
403 enum lttng_action_status action_status;
404 struct ltt_session *session;
405 const struct lttng_snapshot_output default_snapshot_output = {
406 .max_size = UINT64_MAX,
407 };
408 const struct lttng_snapshot_output *snapshot_output =
409 &default_snapshot_output;
410 enum lttng_error_code cmd_ret;
411
412 action_status = lttng_action_snapshot_session_get_session_name(
413 action, &session_name);
414 if (action_status != LTTNG_ACTION_STATUS_OK) {
415 ERR("Failed to get session name from `%s` action",
416 get_action_name(action));
417 ret = -1;
418 goto end;
419 }
420
421 action_status = lttng_action_snapshot_session_get_output(
422 action, &snapshot_output);
423 if (action_status != LTTNG_ACTION_STATUS_OK &&
424 action_status != LTTNG_ACTION_STATUS_UNSET) {
425 ERR("Failed to get output from `%s` action",
426 get_action_name(action));
427 ret = -1;
428 goto end;
429 }
430
431 session_lock_list();
432 session = session_find_by_name(session_name);
433 if (!session) {
ca46af4e 434 DBG("Failed to find session `%s` by name while executing `%s` action of trigger `%s`",
f2b3ef9f 435 session_name, get_action_name(action),
ca46af4e 436 get_trigger_name(work_item->trigger));
2d57482c 437 lttng_action_increase_execution_failure_count(action);
f2b3ef9f
JG
438 goto error_unlock_list;
439 }
440
441
442 session_lock(session);
443 if (!is_trigger_allowed_for_session(work_item->trigger, session)) {
444 goto error_dispose_session;
445 }
446
447 cmd_ret = cmd_snapshot_record(session, snapshot_output, 0);
448 switch (cmd_ret) {
449 case LTTNG_OK:
34f87583
JR
450 DBG("Successfully recorded snapshot of session `%s` on behalf of trigger `%s`",
451 session_name, get_trigger_name(work_item->trigger));
f2b3ef9f
JG
452 break;
453 default:
34f87583
JR
454 WARN("Failed to record snapshot of session `%s` on behalf of trigger `%s`: %s",
455 session_name, get_trigger_name(work_item->trigger),
f2b3ef9f 456 lttng_strerror(-cmd_ret));
2d57482c 457 lttng_action_increase_execution_failure_count(action);
f2b3ef9f
JG
458 break;
459 }
460
461error_dispose_session:
462 session_unlock(session);
463 session_put(session);
464error_unlock_list:
465 session_unlock_list();
466end:
467 return ret;
468}
469
470static int action_executor_group_handler(struct action_executor *executor,
471 const struct action_work_item *work_item,
2d57482c 472 struct lttng_action *action_group)
f2b3ef9f
JG
473{
474 int ret = 0;
475 unsigned int i, count;
476 enum lttng_action_status action_status;
477
478 action_status = lttng_action_group_get_count(action_group, &count);
479 if (action_status != LTTNG_ACTION_STATUS_OK) {
480 /* Fatal error. */
481 ERR("Failed to get count of action in action group");
482 ret = -1;
483 goto end;
484 }
485
486 DBG("Action group has %u action%s", count, count != 1 ? "s" : "");
487 for (i = 0; i < count; i++) {
2d57482c
JR
488 struct lttng_action *action =
489 lttng_action_group_borrow_mutable_at_index(
f2b3ef9f
JG
490 action_group, i);
491
492 ret = action_executor_generic_handler(
493 executor, work_item, action);
494 if (ret) {
34f87583
JR
495 ERR("Stopping the execution of the action group of trigger `%s` following a fatal error",
496 get_trigger_name(work_item->trigger));
f2b3ef9f
JG
497 goto end;
498 }
499 }
500end:
501 return ret;
502}
503
504static int action_executor_generic_handler(struct action_executor *executor,
505 const struct action_work_item *work_item,
2d57482c 506 struct lttng_action *action)
f2b3ef9f 507{
2d57482c 508 int ret;
0e43bcbf
JG
509 const enum lttng_action_type action_type = lttng_action_get_type(action);
510
511 assert(action_type != LTTNG_ACTION_TYPE_UNKNOWN);
512
2d57482c
JR
513 lttng_action_increase_execution_request_count(action);
514 if (!lttng_action_should_execute(action)) {
515 DBG("Policy prevented execution of action `%s` of trigger `%s` action work item %" PRIu64,
516 get_action_name(action),
517 get_trigger_name(work_item->trigger),
518 work_item->id);
519 ret = 0;
520 goto end;
521 }
522
523 lttng_action_increase_execution_count(action);
2516f2d8 524 DBG("Executing action `%s` of trigger `%s` action work item %" PRIu64,
f2b3ef9f 525 get_action_name(action),
34f87583 526 get_trigger_name(work_item->trigger),
f2b3ef9f 527 work_item->id);
2d57482c
JR
528 ret = action_executors[action_type](executor, work_item, action);
529end:
530 return ret;
f2b3ef9f
JG
531}
532
533static int action_work_item_execute(struct action_executor *executor,
534 struct action_work_item *work_item)
535{
536 int ret;
2d57482c
JR
537 struct lttng_action *action =
538 lttng_trigger_get_action(work_item->trigger);
f2b3ef9f 539
34f87583
JR
540 DBG("Starting execution of action work item %" PRIu64 " of trigger `%s`",
541 work_item->id, get_trigger_name(work_item->trigger));
f2b3ef9f 542 ret = action_executor_generic_handler(executor, work_item, action);
34f87583
JR
543 DBG("Completed execution of action work item %" PRIu64 " of trigger `%s`",
544 work_item->id, get_trigger_name(work_item->trigger));
f2b3ef9f
JG
545 return ret;
546}
547
548static void action_work_item_destroy(struct action_work_item *work_item)
549{
550 lttng_trigger_put(work_item->trigger);
551 lttng_evaluation_destroy(work_item->evaluation);
552 notification_client_list_put(work_item->client_list);
553 free(work_item);
554}
555
556static void *action_executor_thread(void *_data)
557{
558 struct action_executor *executor = _data;
559
560 assert(executor);
561
412d7227
SM
562 health_register(the_health_sessiond,
563 HEALTH_SESSIOND_TYPE_ACTION_EXECUTOR);
f2b3ef9f
JG
564
565 rcu_register_thread();
566 rcu_thread_online();
567
568 DBG("Entering work execution loop");
569 pthread_mutex_lock(&executor->work.lock);
570 while (!executor->should_quit) {
571 int ret;
572 struct action_work_item *work_item;
573
574 health_code_update();
575 if (executor->work.pending_count == 0) {
576 health_poll_entry();
577 DBG("No work items enqueued, entering wait");
578 pthread_cond_wait(&executor->work.cond,
579 &executor->work.lock);
580 DBG("Woke-up from wait");
581 health_poll_exit();
582 continue;
583 }
584
0db0f8e0 585 /* Pop item from front of the list with work lock held. */
f2b3ef9f
JG
586 work_item = cds_list_first_entry(&executor->work.list,
587 struct action_work_item, list_node);
588 cds_list_del(&work_item->list_node);
589 executor->work.pending_count--;
590
591 /*
592 * Work can be performed without holding the work lock,
593 * allowing new items to be queued.
594 */
595 pthread_mutex_unlock(&executor->work.lock);
d2a28b27
JR
596
597 /* Execute item only if a trigger is registered. */
598 lttng_trigger_lock(work_item->trigger);
599 if (!lttng_trigger_is_registered(work_item->trigger)) {
600 const char *trigger_name = NULL;
601 uid_t trigger_owner_uid;
602 enum lttng_trigger_status trigger_status;
603
604 trigger_status = lttng_trigger_get_name(
605 work_item->trigger, &trigger_name);
606 switch (trigger_status) {
607 case LTTNG_TRIGGER_STATUS_OK:
608 break;
609 case LTTNG_TRIGGER_STATUS_UNSET:
610 trigger_name = "(unset)";
611 break;
612 default:
613 abort();
614 }
615
616 trigger_status = lttng_trigger_get_owner_uid(
617 work_item->trigger, &trigger_owner_uid);
618 assert(trigger_status == LTTNG_TRIGGER_STATUS_OK);
619
620 DBG("Work item skipped since the associated trigger is no longer registered: work item id = %" PRIu64 ", trigger name = '%s', trigger owner uid = %d",
621 work_item->id, trigger_name,
622 (int) trigger_owner_uid);
623 ret = 0;
624 goto skip_execute;
625 }
626
f2b3ef9f 627 ret = action_work_item_execute(executor, work_item);
d2a28b27
JR
628
629 skip_execute:
630 lttng_trigger_unlock(work_item->trigger);
f2b3ef9f
JG
631 action_work_item_destroy(work_item);
632 if (ret) {
633 /* Fatal error. */
634 break;
635 }
636
637 health_code_update();
638 pthread_mutex_lock(&executor->work.lock);
639 }
640
f5f5c54d
JG
641 if (executor->should_quit) {
642 pthread_mutex_unlock(&executor->work.lock);
643 }
f2b3ef9f
JG
644 DBG("Left work execution loop");
645
646 health_code_update();
647
648 rcu_thread_offline();
649 rcu_unregister_thread();
412d7227 650 health_unregister(the_health_sessiond);
f2b3ef9f
JG
651
652 return NULL;
653}
654
655static bool shutdown_action_executor_thread(void *_data)
656{
657 struct action_executor *executor = _data;
658
8db3acaf 659 pthread_mutex_lock(&executor->work.lock);
f2b3ef9f
JG
660 executor->should_quit = true;
661 pthread_cond_signal(&executor->work.cond);
8db3acaf 662 pthread_mutex_unlock(&executor->work.lock);
f2b3ef9f
JG
663 return true;
664}
665
666static void clean_up_action_executor_thread(void *_data)
667{
668 struct action_executor *executor = _data;
669
670 assert(cds_list_empty(&executor->work.list));
671
672 pthread_mutex_destroy(&executor->work.lock);
673 pthread_cond_destroy(&executor->work.cond);
674 free(executor);
675}
676
677struct action_executor *action_executor_create(
678 struct notification_thread_handle *handle)
679{
680 struct action_executor *executor = zmalloc(sizeof(*executor));
681
682 if (!executor) {
683 goto end;
684 }
685
686 CDS_INIT_LIST_HEAD(&executor->work.list);
687 pthread_cond_init(&executor->work.cond, NULL);
688 pthread_mutex_init(&executor->work.lock, NULL);
689 executor->notification_thread_handle = handle;
690
691 executor->thread = lttng_thread_create(THREAD_NAME,
692 action_executor_thread, shutdown_action_executor_thread,
693 clean_up_action_executor_thread, executor);
694end:
695 return executor;
696}
697
698void action_executor_destroy(struct action_executor *executor)
699{
700 struct action_work_item *work_item, *tmp;
701
702 /* TODO Wait for work list to drain? */
703 lttng_thread_shutdown(executor->thread);
704 pthread_mutex_lock(&executor->work.lock);
705 if (executor->work.pending_count != 0) {
706 WARN("%" PRIu64
707 " trigger action%s still queued for execution and will be discarded",
708 executor->work.pending_count,
709 executor->work.pending_count == 1 ? " is" :
710 "s are");
711 }
712
713 cds_list_for_each_entry_safe (
714 work_item, tmp, &executor->work.list, list_node) {
715 WARN("Discarding action work item %" PRIu64
34f87583
JR
716 " associated to trigger `%s`",
717 work_item->id, get_trigger_name(work_item->trigger));
f2b3ef9f
JG
718 cds_list_del(&work_item->list_node);
719 action_work_item_destroy(work_item);
720 }
721 pthread_mutex_unlock(&executor->work.lock);
722 lttng_thread_put(executor->thread);
723}
724
725/* RCU read-lock must be held by the caller. */
726enum action_executor_status action_executor_enqueue(
727 struct action_executor *executor,
728 struct lttng_trigger *trigger,
729 struct lttng_evaluation *evaluation,
730 const struct lttng_credentials *object_creds,
731 struct notification_client_list *client_list)
732{
733 enum action_executor_status executor_status = ACTION_EXECUTOR_STATUS_OK;
734 const uint64_t work_item_id = executor->next_work_item_id++;
735 struct action_work_item *work_item;
736 bool signal = false;
737
738 pthread_mutex_lock(&executor->work.lock);
739 /* Check for queue overflow. */
740 if (executor->work.pending_count >= MAX_QUEUED_WORK_COUNT) {
741 /* Most likely spammy, remove if it is the case. */
34f87583
JR
742 DBG("Refusing to enqueue action for trigger `%s` as work item %" PRIu64
743 " (overflow)", get_trigger_name(trigger), work_item_id);
f2b3ef9f
JG
744 executor_status = ACTION_EXECUTOR_STATUS_OVERFLOW;
745 goto error_unlock;
746 }
747
748 work_item = zmalloc(sizeof(*work_item));
749 if (!work_item) {
2516f2d8 750 PERROR("Failed to allocate action executor work item on behalf of trigger `%s`",
34f87583 751 get_trigger_name(trigger));
f2b3ef9f
JG
752 executor_status = ACTION_EXECUTOR_STATUS_ERROR;
753 goto error_unlock;
754 }
755
756 lttng_trigger_get(trigger);
757 if (client_list) {
758 const bool reference_acquired =
759 notification_client_list_get(client_list);
760
761 assert(reference_acquired);
762 }
763
764 *work_item = (typeof(*work_item)){
765 .id = work_item_id,
766 .trigger = trigger,
767 /* Ownership transferred to the work item. */
768 .evaluation = evaluation,
769 .object_creds = {
770 .is_set = !!object_creds,
771 .value = object_creds ? *object_creds :
772 (typeof(work_item->object_creds.value)) {},
773 },
774 .client_list = client_list,
775 .list_node = CDS_LIST_HEAD_INIT(work_item->list_node),
776 };
777
778 evaluation = NULL;
779 cds_list_add_tail(&work_item->list_node, &executor->work.list);
780 executor->work.pending_count++;
2516f2d8 781 DBG("Enqueued action for trigger `%s` as work item #%" PRIu64,
34f87583 782 get_trigger_name(trigger), work_item_id);
f2b3ef9f
JG
783 signal = true;
784
785error_unlock:
f2b3ef9f
JG
786 if (signal) {
787 pthread_cond_signal(&executor->work.cond);
788 }
8db3acaf 789 pthread_mutex_unlock(&executor->work.lock);
f2b3ef9f
JG
790
791 lttng_evaluation_destroy(evaluation);
792 return executor_status;
793}
This page took 0.094723 seconds and 4 git commands to generate.