Skip to content

S3 refactor timer coros => change to store timer coros on flb_sched struct #28

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
PettitWesley wants to merge 2 commits into
base: s3-refactor-timer-coros-awspr
Choose a base branch
from
Open

S3 refactor timer coros => change to store timer coros on flb_sched struct #28

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
6d3ab25
Proposed change to associate sched with the output thread and then pu…
PettitWesley Oct 20, 2023
8576725
Switch async timers to flb_sched struct
PettitWesley Oct 20, 2023
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
81 changes: 39 additions & 42 deletions include/fluent-bit/flb_async_timer.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -34,27 +34,26 @@
#include <fluent-bit/flb_output_thread.h>


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 */
};
Expand All @@ -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 */
Expand All @@ -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;
Expand All @@ -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);
Expand All @@ -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;
Expand All @@ -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();
Expand Down
2 changes: 1 addition & 1 deletion include/fluent-bit/flb_config.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -243,7 +243,7 @@ struct flb_config {
*/
uint16_t in_table_id[512];

void *sched;
struct flb_sched *sched;

@matthewfala matthewfala Oct 21, 2023

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Wonder why this is a void * in the first place. Seems like all pointers are void * in this file. I'd keep it void * unless we understand this convention and have a reason to break it.

@PettitWesley PettitWesley Oct 31, 2023

Copy link
Copy Markdown
Owner Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

void * is useful if at compile time it could store multiple different structs. Here its always one struct, so there's no good reason AFAIK for it to be void.

unsigned int sched_cap;
unsigned int sched_base;

Expand Down
6 changes: 1 addition & 5 deletions include/fluent-bit/flb_output.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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;
};
Expand Down Expand Up @@ -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,
Expand Down
10 changes: 3 additions & 7 deletions include/fluent-bit/flb_output_thread.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Comment on lines +94 to +95

@matthewfala matthewfala Oct 21, 2023

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Calling out that this is the central change in this pr

};

int flb_output_thread_pool_create(struct flb_config *config,
Expand All @@ -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();
Expand Down
5 changes: 5 additions & 0 deletions include/fluent-bit/flb_scheduler.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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.
Expand Down
Loading