diff --git a/include/fluent-bit/flb_async_timer.h b/include/fluent-bit/flb_async_timer.h index 3d888fdc7c2..16ad39723f2 100644 --- a/include/fluent-bit/flb_async_timer.h +++ b/include/fluent-bit/flb_async_timer.h @@ -34,27 +34,26 @@ #include -void flb_async_timer_destroy(struct flb_out_async_timer *timer); -void flb_async_timer_cleanup(struct mk_list *list); +void flb_async_timer_destroy(struct flb_async_timer *timer); +void flb_async_timer_cleanup(struct flb_sched *sched); void flb_output_async_timer_cleanup(struct flb_config *config); int flb_sched_out_async_timer_cb_create(struct flb_sched *sched, int type, int ms, struct flb_output_instance *o_ins, char *job_name, void (*async_cb)(struct flb_config *, void *), void *data, struct flb_sched_timer **out_timer); -void flb_out_async_timers_print(struct flb_output_instance *ins); -void flb_async_timers_print(struct mk_list *async_timer_list); -int flb_async_timers_size(struct flb_output_instance *ins); +void flb_async_timers_print_all(struct flb_config *config); +void flb_async_timers_print(struct flb_sched *sched); +int flb_async_timers_size(struct flb_config *config); int flb_thread_pool_async_timers_size(struct flb_output_instance *ins); void flb_thread_pool_async_timers_print(struct flb_output_instance *ins); /* * stores timer coros on the async_timer_list, if the output uses them */ -struct flb_out_async_timer { +struct flb_async_timer { struct flb_config *config; /* FLB context */ - struct flb_output_instance *o_ins; /* output instance */ - struct flb_out_async_timer_cb_data *timer_data; /* callback info */ + struct flb_async_timer_cb_data *timer_data; /* callback info */ struct flb_coro *coro; /* parent coro addr */ struct mk_list _head; /* Link to async_timer_list */ }; @@ -63,48 +62,48 @@ struct flb_out_async_timer { * If the output uses timer coros, then this is used as the callback data * passed to flb_sched_timer_cb_create */ -struct flb_out_async_timer_cb_data { - struct flb_output_instance *ins; /* associate coro with this output instance */ - char *job_name; /* used on engine shutdown, print pending "custom" jobs */ - void (*async_cb) (struct flb_config *config, void *data); /* call this output callback in the coro */ - void *data; /* opaque data to pass to the above cb */ +struct flb_async_timer_cb_data { + int is_threaded; /* track on engine sched vs on worker sched */ + char *plugin_alias; /* used on engine shutdown, print pending "custom" jobs */ + char *job_name; /* used on engine shutdown, print pending "custom" jobs */ + void (*async_cb) (struct flb_config *config, void *data); /* call this output callback in the coro */ + void *data; /* opaque data to pass to the above cb */ }; -extern FLB_TLS_DEFINE(struct flb_out_async_timer, async_timer_coro_params); +extern FLB_TLS_DEFINE(struct flb_async_timer, async_timer_coro_params); /* The coro callee callback for async timer coros */ static FLB_INLINE void out_async_timer_cb(void) { struct flb_coro *coro; - struct flb_output_instance *o_ins; - struct flb_out_thread_instance *th_ins; - struct flb_out_async_timer *async_timer; + struct flb_async_timer *async_timer; + struct flb_sched *sched; - async_timer = (struct flb_out_async_timer *) FLB_TLS_GET(async_timer_coro_params); + async_timer = (struct flb_async_timer *) FLB_TLS_GET(async_timer_coro_params); if (!async_timer) { flb_error("[output] no async timer coro params defined, unexpected"); return; } coro = async_timer->coro; - o_ins = async_timer->o_ins; /* Run the callback provided by the output plugin */ async_timer->timer_data->async_cb(async_timer->config, async_timer->timer_data->data); /* move coro to destroy queue */ - if (flb_output_is_threaded(o_ins) == FLB_TRUE) { - th_ins = flb_output_thread_instance_get(); - pthread_mutex_lock(&th_ins->async_timer_mutex); + if (async_timer->timer_data->is_threaded == FLB_TRUE) { + sched = flb_sched_ctx_get(); + pthread_mutex_lock(&sched->async_timer_mutex); mk_list_del(&async_timer->_head); - mk_list_add(&async_timer->_head, &th_ins->async_timer_list_destroy); - pthread_mutex_unlock(&th_ins->async_timer_mutex); + mk_list_add(&async_timer->_head, &sched->async_timer_list_destroy); + pthread_mutex_unlock(&sched->async_timer_mutex); } else { mk_list_del(&async_timer->_head); - mk_list_add(&async_timer->_head, &o_ins->async_timer_list_destroy); + sched = async_timer->config->sched; + mk_list_add(&async_timer->_head, &sched->async_timer_list_destroy); } /* timer coro is complete; yield back to caller/control code */ @@ -120,14 +119,13 @@ void flb_out_async_sched_timer_cb(struct flb_config *config, void *data) { size_t stack_size; struct flb_coro *coro; - struct flb_out_async_timer *async_timer; - struct flb_out_async_timer *timer_thread_key; - struct flb_out_thread_instance *th_ins; - struct flb_out_async_timer_cb_data *ctx = (struct flb_out_async_timer_cb_data *) data; - struct flb_output_instance *o_ins; + struct flb_async_timer *async_timer; + struct flb_async_timer *timer_thread_key; + struct flb_async_timer_cb_data *ctx = (struct flb_async_timer_cb_data *) data; + struct flb_sched *sched; /* Custom output coroutine info */ - async_timer = (struct flb_out_async_timer*) flb_calloc(1, sizeof(struct flb_out_async_timer)); + async_timer = (struct flb_async_timer*) flb_calloc(1, sizeof(struct flb_async_timer)); if (!async_timer) { flb_errno(); return; @@ -140,10 +138,9 @@ void flb_out_async_sched_timer_cb(struct flb_config *config, void *data) return; } - o_ins = ctx->ins; - async_timer->o_ins = o_ins; async_timer->config = config; async_timer->coro = coro; + async_timer->timer_data = ctx; coro->callee = co_create(config->coro_stack_size, out_async_timer_cb, &stack_size); @@ -159,22 +156,22 @@ void flb_out_async_sched_timer_cb(struct flb_config *config, void *data) VALGRIND_STACK_REGISTER(coro->callee, ((char *) coro->callee) + stack_size); #endif - if (o_ins->is_threaded == FLB_TRUE) { - th_ins = flb_output_thread_instance_get(); - pthread_mutex_lock(&th_ins->async_timer_mutex); - mk_list_add(&async_timer->_head, &th_ins->async_timer_list); - pthread_mutex_unlock(&th_ins->async_timer_mutex); + if (ctx->is_threaded == FLB_TRUE) { + sched = flb_sched_ctx_get(); + pthread_mutex_lock(&sched->async_timer_mutex); + mk_list_add(&async_timer->_head, &sched->async_timer_list); + pthread_mutex_unlock(&sched->async_timer_mutex); } else { - mk_list_add(&async_timer->_head, &o_ins->async_timer_list); + mk_list_add(&async_timer->_head, &config->sched->async_timer_list); } /* Same struct used in async_timer_list and in the pthread key, * Unique memory needed since these are freed separately */ - timer_thread_key = (struct flb_out_async_timer *) FLB_TLS_GET(async_timer_coro_params); + timer_thread_key = (struct flb_async_timer *) FLB_TLS_GET(async_timer_coro_params); if (!timer_thread_key) { - timer_thread_key = (struct flb_out_async_timer *) flb_calloc(1, sizeof(struct flb_out_async_timer)); + timer_thread_key = (struct flb_async_timer *) flb_calloc(1, sizeof(struct flb_async_timer)); if (!timer_thread_key) { flb_errno(); return; @@ -183,8 +180,8 @@ void flb_out_async_sched_timer_cb(struct flb_config *config, void *data) /* copy to thread local storage */ timer_thread_key->coro = coro; - timer_thread_key->o_ins = o_ins; timer_thread_key->timer_data = async_timer->timer_data; + timer_thread_key->config = config; FLB_TLS_SET(async_timer_coro_params, timer_thread_key); coro->caller = co_active(); diff --git a/include/fluent-bit/flb_config.h b/include/fluent-bit/flb_config.h index cd6635dbc4e..8caa4cb88cc 100644 --- a/include/fluent-bit/flb_config.h +++ b/include/fluent-bit/flb_config.h @@ -243,7 +243,7 @@ struct flb_config { */ uint16_t in_table_id[512]; - void *sched; + struct flb_sched *sched; unsigned int sched_cap; unsigned int sched_base; diff --git a/include/fluent-bit/flb_output.h b/include/fluent-bit/flb_output.h index b77f5595c0b..21ab05b3637 100644 --- a/include/fluent-bit/flb_output.h +++ b/include/fluent-bit/flb_output.h @@ -409,10 +409,6 @@ struct flb_output_instance { struct mk_list flush_list; struct mk_list flush_list_destroy; - /* similar to flush coroutine list above, timer coroutine list */ - struct mk_list async_timer_list; - struct mk_list async_timer_list_destroy; - /* Keep a reference to the original context this instance belongs to */ struct flb_config *config; }; @@ -465,7 +461,7 @@ struct flb_out_flush_params { }; extern FLB_TLS_DEFINE(struct flb_out_flush_params, out_flush_params); -extern FLB_TLS_DEFINE(struct flb_out_async_timer, async_timer_coro_params); +extern FLB_TLS_DEFINE(struct flb_async_timer, async_timer_coro_params); static FLB_INLINE void output_params_set(struct flb_output_flush *out_flush, struct flb_coro *coro, diff --git a/include/fluent-bit/flb_output_thread.h b/include/fluent-bit/flb_output_thread.h index 407a97187f3..34bc54af9a0 100644 --- a/include/fluent-bit/flb_output_thread.h +++ b/include/fluent-bit/flb_output_thread.h @@ -87,14 +87,12 @@ struct flb_out_thread_instance { * must be protected: we use 'flush_mutex for that purpose. */ pthread_mutex_t flush_mutex; /* mutex for 'flush_list' */ - - /* Same as flush_mutex but for timer coros */ - struct mk_list async_timer_list; /* timer list */ - struct mk_list async_timer_list_destroy; /* timer destroy list */ - pthread_mutex_t async_timer_mutex; /* mutex for async timer lists */ /* List of mapped 'upstream' contexts */ struct mk_list upstreams; + + /* Each event loop has a scheduler instance */ + struct flb_sched *sched; }; int flb_output_thread_pool_create(struct flb_config *config, @@ -105,8 +103,6 @@ int flb_output_thread_pool_start(struct flb_output_instance *ins); int flb_output_thread_pool_flush(struct flb_task *task, struct flb_output_instance *out_ins, struct flb_config *config); -int flb_thread_pool_async_timers_size(struct flb_output_instance *ins); -void flb_thread_pool_async_timers_print(struct flb_output_instance *ins); void flb_output_thread_instance_init(); struct flb_out_thread_instance *flb_output_thread_instance_get(); diff --git a/include/fluent-bit/flb_scheduler.h b/include/fluent-bit/flb_scheduler.h index e21138f9d61..93678f8bce8 100644 --- a/include/fluent-bit/flb_scheduler.h +++ b/include/fluent-bit/flb_scheduler.h @@ -106,6 +106,11 @@ struct flb_sched { /* Timers: list of timers for different purposes */ struct mk_list timers; + /* async timers can create coros, this tracks the coros */ + struct mk_list async_timer_list; /* timer list */ + struct mk_list async_timer_list_destroy; /* timer destroy list */ + pthread_mutex_t async_timer_mutex; /* mutex because the engine needs to count the coros */ + /* * Timers_Drop: list of invalidated timers that needs to * be free()d once the event loop finish the cycle. diff --git a/src/flb_async_timer.c b/src/flb_async_timer.c index b8dc078ffd4..33ed3606d64 100644 --- a/src/flb_async_timer.c +++ b/src/flb_async_timer.c @@ -21,89 +21,87 @@ #include #include #include +#include -void flb_async_timer_destroy(struct flb_out_async_timer *timer) +void flb_async_timer_destroy(struct flb_async_timer *timer) { mk_list_del(&timer->_head); flb_coro_destroy(timer->coro); flb_free(timer); } -void flb_async_timer_cleanup(struct mk_list *destroy_list) +void flb_async_timer_cleanup(struct flb_sched *sched) { - struct flb_out_async_timer *async_timer; + struct flb_async_timer *async_timer; struct mk_list *tmp; struct mk_list *head; + struct mk_list *destroy_list = &sched->async_timer_list_destroy; + mk_list_foreach_safe(head, tmp, destroy_list) { - async_timer = mk_list_entry(head, struct flb_out_async_timer, _head); + async_timer = mk_list_entry(head, struct flb_async_timer, _head); + mk_list_del(&async_timer->_head); flb_async_timer_destroy(async_timer); } } -void flb_output_async_timer_cleanup(struct flb_config *config) -{ - struct flb_output_instance *o_ins; - struct mk_list *tmp; - struct mk_list *head; - mk_list_foreach_safe(head, tmp, &config->outputs) { - o_ins = mk_list_entry(head, struct flb_output_instance, _head); - flb_async_timer_cleanup(o_ins->async_timer_list_destroy); - } -} - int flb_sched_out_async_timer_cb_create(struct flb_sched *sched, int type, int ms, struct flb_output_instance *o_ins, char *job_name, void (*async_cb)(struct flb_config *, void *), void *data, struct flb_sched_timer **out_timer) { - struct flb_out_async_timer_cb_data *timer_data; + struct flb_async_timer_cb_data *timer_data; - timer_data = flb_calloc(1, sizeof(struct flb_out_async_timer_cb_data)); + timer_data = flb_calloc(1, sizeof(struct flb_async_timer_cb_data)); if (!timer_data) { - return; + return -1; } - timer_data->ins = o_ins; + timer_data->is_threaded = o_ins->is_threaded; + timer_data->plugin_alias = o_ins->alias; timer_data->job_name = job_name; - timer_data->cb = async_cb; + timer_data->async_cb = async_cb; timer_data->data = data; return flb_sched_timer_cb_create(sched, type, ms, flb_out_async_sched_timer_cb, timer_data, NULL); } /* Used in engine flb_running_count */ -int flb_async_timers_size(struct flb_output_instance *ins) +int flb_async_timers_size(struct flb_config *config) { - int size = 0; + int n = 0; + int timers = 0; + struct mk_list *tmp; + struct mk_list *head; + struct flb_output_instance *o_ins; - if (flb_output_is_threaded(ins) == FLB_TRUE) { - /* - * On threaded mode, we need to count the active co-routines of - * every running thread of the thread pool. - */ - size = flb_thread_pool_async_timers_size(ins); - } - else { - size = mk_list_size(&ins->async_timer_list); + timers = mk_list_size(&config->sched->async_timer_list); + + mk_list_foreach_safe(head, tmp, &config->outputs) { + o_ins = mk_list_entry(head, struct flb_output_instance, _head); + if (flb_output_is_threaded(o_ins) == FLB_TRUE) { + n = flb_thread_pool_async_timers_size(o_ins); + timers = timers + n; + } } - return size; + return timers; } -void flb_async_timers_print(struct mk_list *async_timer_list) +void flb_async_timers_print(struct flb_sched *sched) { - struct flb_out_async_timer *async_timer; + struct flb_async_timer *async_timer; struct mk_list *tmp; struct mk_list *head; + struct mk_list *async_timer_list = &sched->async_timer_list; int n = mk_list_size(async_timer_list); if (n != 0) { /* get one coro for the job_name */ mk_list_foreach_safe(head, tmp, async_timer_list) { - async_timer = mk_list_entry(head, struct flb_out_async_timer, _head); + async_timer = mk_list_entry(head, struct flb_async_timer, _head); if (async_timer != NULL) { - flb_info("[task] output=%s still running %d %s(s)", - async_timer->o_ins->alias, n, async_timer->timer_data->job_name); + flb_info("[task] %s still running %d %s(s)", + async_timer->timer_data->plugin_alias, n, async_timer->timer_data->job_name); break; } } @@ -111,13 +109,19 @@ void flb_async_timers_print(struct mk_list *async_timer_list) } /* Used in engine flb_running_print */ -void flb_out_async_timers_print(struct flb_output_instance *ins) +void flb_async_timers_print_all(struct flb_config *config) { - if (flb_output_is_threaded(ins) == FLB_TRUE) { - flb_thread_pool_async_timers_print(ins); - } - else { - flb_async_timers_print(&ins->async_timer_list); + struct mk_list *head; + struct mk_list *tmp; + struct flb_output_instance *o_ins; + + flb_async_timers_print(config->sched); + + mk_list_foreach_safe(head, tmp, &config->outputs) { + o_ins = mk_list_entry(head, struct flb_output_instance, _head); + if (flb_output_is_threaded(o_ins) == FLB_TRUE) { + flb_thread_pool_async_timers_print(o_ins); + } } } @@ -138,9 +142,9 @@ int flb_thread_pool_async_timers_size(struct flb_output_instance *ins) th_ins = th->params.data; - pthread_mutex_lock(&th_ins->flush_mutex); - n = mk_list_size(&th_ins->async_timer_list); - pthread_mutex_unlock(&th_ins->flush_mutex); + pthread_mutex_lock(&th_ins->sched->async_timer_mutex); + n = mk_list_size(&th_ins->sched->async_timer_list); + pthread_mutex_unlock(&th_ins->sched->async_timer_mutex); size += n; } @@ -162,8 +166,8 @@ void flb_thread_pool_async_timers_print(struct flb_output_instance *ins) } th_ins = th->params.data; - pthread_mutex_lock(&th_ins->async_timer_mutex); - flb_async_timers_print(&th_ins->async_timer_list); - pthread_mutex_unlock(&th_ins->async_timer_mutex); + pthread_mutex_lock(&th_ins->sched->async_timer_mutex); + flb_async_timers_print(&th_ins->sched->async_timer_list); + pthread_mutex_unlock(&th_ins->sched->async_timer_mutex); } } diff --git a/src/flb_engine.c b/src/flb_engine.c index 38db343c6f0..7e51df71c55 100644 --- a/src/flb_engine.c +++ b/src/flb_engine.c @@ -394,35 +394,22 @@ static inline int handle_output_event(flb_pipefd_t fd, uint64_t ts, return 0; } +/* Count of running coros */ static int flb_running_count(struct flb_config *config) { - int tasks = 0, timers = 0, n = 0; - struct mk_list *head; - struct mk_list *tmp; - struct flb_output_instance *o_ins; - - mk_list_foreach_safe(head, tmp, &config->outputs) { - o_ins = mk_list_entry(head, struct flb_output_instance, _head); - n = flb_async_timers_size(o_ins); - timers = timers + n; - } + int tasks = 0, timers = 0; + timers = flb_async_timers_size(config); tasks = flb_task_running_count(config); + return tasks + timers; } +/* Print running coros */ static void flb_running_print(struct flb_config *config) { - struct mk_list *head; - struct mk_list *tmp; - struct flb_output_instance *o_ins; - flb_task_running_print(config); - - mk_list_foreach_safe(head, tmp, &config->outputs) { - o_ins = mk_list_entry(head, struct flb_output_instance, _head); - flb_thread_pool_async_timers_print(o_ins); - } + flb_async_timers_print_all(config); } static inline int flb_engine_manager(flb_pipefd_t fd, struct flb_config *config) @@ -929,7 +916,7 @@ int flb_engine_start(struct flb_config *config) flb_net_dns_lookup_context_cleanup(&dns_ctx); flb_sched_timer_cleanup(config->sched); flb_upstream_conn_pending_destroy_list(&config->upstreams); - flb_output_async_timer_cleanup(config); + flb_async_timer_cleanup(config->sched->async_timer_list_destroy); /* * depend on main thread to clean up expired message diff --git a/src/flb_output.c b/src/flb_output.c index e5f5537ce5b..473626d85f1 100644 --- a/src/flb_output.c +++ b/src/flb_output.c @@ -40,7 +40,7 @@ #include FLB_TLS_DEFINE(struct flb_out_flush_params, out_flush_params); -FLB_TLS_DEFINE(struct flb_out_async_timer, async_timer_coro_params); +FLB_TLS_DEFINE(struct flb_async_timer, async_timer_coro_params); void flb_output_prepare() { @@ -717,8 +717,6 @@ struct flb_output_instance *flb_output_new(struct flb_config *config, mk_list_init(&instance->upstreams); mk_list_init(&instance->flush_list); mk_list_init(&instance->flush_list_destroy); - mk_list_init(&instance->async_timer_list); - mk_list_init(&instance->async_timer_list_destroy); mk_list_add(&instance->_head, &config->outputs); diff --git a/src/flb_output_thread.c b/src/flb_output_thread.c index 4aa55da009c..ade649a912c 100644 --- a/src/flb_output_thread.c +++ b/src/flb_output_thread.c @@ -182,7 +182,7 @@ static void output_thread(void *data) struct flb_output_flush *out_flush; struct flb_out_thread_instance *th_ins = data; struct flb_out_flush_params *flush_params = NULL; - struct flb_out_async_timer *timer_params = NULL; + struct flb_async_timer *timer_params = NULL; struct flb_net_dns dns_ctx; /* Register thread instance */ @@ -215,6 +215,7 @@ static void output_thread(void *data) return; } flb_sched_ctx_set(sched); + th_ins->sched = sched; /* * Sched a permanent callback triggered every 1.5 second to let other @@ -332,10 +333,10 @@ static void output_thread(void *data) /* Destroy upstream connections from the 'pending destroy list' */ flb_upstream_conn_pending_destroy_list(&th_ins->upstreams); flb_sched_timer_cleanup(sched); - flb_async_timer_cleanup(&th_ins->async_timer_list_destroy); + flb_async_timer_cleanup(&th_ins->sched); /* Check if we should stop the event loop */ - if (stopping == FLB_TRUE && mk_list_size(&th_ins->flush_list) == 0 && mk_list_size(&th_ins->async_timer_list) == 0) { + if (stopping == FLB_TRUE && mk_list_size(&th_ins->flush_list) == 0 && mk_list_size(&th_ins->sched->async_timer_list) == 0) { /* * If there are no busy network connections (and no coroutines) its * safe to stop it. @@ -359,7 +360,7 @@ static void output_thread(void *data) upstream_thread_destroy(th_ins); flb_upstream_conn_active_destroy_list(&th_ins->upstreams); flb_upstream_conn_pending_destroy_list(&th_ins->upstreams); - flb_async_timer_cleanup(&th_ins->async_timer_list_destroy); + flb_async_timer_cleanup(&th_ins->sched); flb_sched_destroy(sched); flush_params = FLB_TLS_GET(out_flush_params); @@ -444,10 +445,7 @@ int flb_output_thread_pool_create(struct flb_config *config, th_ins->flush_id = 0; mk_list_init(&th_ins->flush_list); mk_list_init(&th_ins->flush_list_destroy); - mk_list_init(&th_ins->async_timer_list); - mk_list_init(&th_ins->async_timer_list_destroy); pthread_mutex_init(&th_ins->flush_mutex, NULL); - pthread_mutex_init(&th_ins->async_timer_mutex, NULL); mk_list_init(&th_ins->upstreams); upstream_thread_create(th_ins, ins); diff --git a/src/flb_scheduler.c b/src/flb_scheduler.c index 1aacf554f11..d8f812e20fb 100644 --- a/src/flb_scheduler.c +++ b/src/flb_scheduler.c @@ -530,6 +530,9 @@ struct flb_sched *flb_sched_create(struct flb_config *config, mk_list_init(&sched->requests_wait); mk_list_init(&sched->timers); mk_list_init(&sched->timers_drop); + mk_list_init(&sched->async_timer_list); + mk_list_init(&sched->async_timer_list_destroy); + pthread_mutex_init(&sched->async_timer_mutex, NULL); /* Create the frame timer who enqueue 'requests' for future time */ timer = flb_sched_timer_create(sched);