The processing is really difficult to be smaller than processing_per_sec, and most msg will create a new applet, but there are already have a lot idle applets waiting for work. and the result is that the most applets is not reused.
This patch should be backported to 1.9. Best regards.
From cad58af8ee953b97ab5b9d9e7551823890e3da6c Mon Sep 17 00:00:00 2001 From: Kevin Zhu <[email protected]> Date: Sat, 13 Apr 2019 15:28:54 +0800 Subject: [PATCH] BUG/MAJOR: spoe: Rollback frequency counter to sending_rate The processing is really difficult to be smaller than processing_per_sec, and most msg will create a new applet, but there are already have a lot idle applets waiting for work. and the result is that the most applets is not reused. This patch should be backported to 1.9. --- include/types/spoe.h | 3 +-- src/flt_spoe.c | 30 +++++++++++++++--------------- 2 files changed, 16 insertions(+), 17 deletions(-) diff --git a/include/types/spoe.h b/include/types/spoe.h index a744cd7..7b6a9fb 100644 --- a/include/types/spoe.h +++ b/include/types/spoe.h @@ -265,8 +265,7 @@ struct spoe_agent { /* running info */ struct { unsigned int frame_size; /* current maximum frame size, only used to encode messages */ - unsigned int processing; - struct freq_ctr processing_per_sec; + unsigned int sending_rate; struct freq_ctr conn_per_sec; /* connections per second */ struct freq_ctr err_per_sec; /* connetion errors per second */ diff --git a/src/flt_spoe.c b/src/flt_spoe.c index b820710..514b466 100644 --- a/src/flt_spoe.c +++ b/src/flt_spoe.c @@ -1728,7 +1728,7 @@ spoe_handle_processing_appctx(struct appctx *appctx) } if (active_s || active_r) { - update_freq_ctr(&agent->rt[tid].processing_per_sec, active_s); + agent->rt[tid].sending_rate += active_s; SPOE_APPCTX(appctx)->task->expire = tick_add_ifset(now_ms, agent->timeout.idle); } @@ -2041,8 +2041,7 @@ spoe_queue_context(struct spoe_context *ctx) struct spoe_appctx *spoe_appctx; /* Check if we need to create a new SPOE applet or not. */ - if (!eb_is_empty(&agent->rt[tid].idle_applets) && - agent->rt[tid].processing < read_freq_ctr(&agent->rt[tid].processing_per_sec)) + if (!eb_is_empty(&agent->rt[tid].idle_applets) && agent->rt[tid].sending_rate) goto end; SPOE_PRINTF(stderr, "%d.%06d [SPOE/%-15s] %s: stream=%p" @@ -2096,18 +2095,21 @@ spoe_queue_context(struct spoe_context *ctx) return -1; } - /* Add the SPOE context in the sending queue */ + /* Add the SPOE context in the sending queue and update all running + * info */ LIST_ADDQ(&agent->rt[tid].sending_queue, &ctx->list); + if (agent->rt[tid].sending_rate) + agent->rt[tid].sending_rate--; _HA_ATOMIC_ADD(&agent->counters.nb_sending, 1); spoe_update_stat_time(&ctx->stats.tv_request, &ctx->stats.t_request); ctx->stats.tv_queue = now; SPOE_PRINTF(stderr, "%d.%06d [SPOE/%-15s] %s: stream=%p" " - Add stream in sending queue" - " - applets=%u - idles=%u - processing=%u\n", + " - applets=%u - idles=%u - sending_rate=%u\n", (int)now.tv_sec, (int)now.tv_usec, agent->id, __FUNCTION__, ctx->strm, agent->counters.applets, agent->counters.idles, - agent->rt[tid].processing); + agent->rt[tid].sending_rate); /* Finally try to wakeup an IDLE applet. */ if (!eb_is_empty(&agent->rt[tid].idle_applets)) { @@ -2549,14 +2551,13 @@ spoe_handle_processing_error(struct stream *s, struct spoe_agent *agent, } static inline int -spoe_start_processing(struct spoe_agent *agent, struct spoe_context *ctx, int dir) +spoe_start_processing(struct spoe_context *ctx, int dir) { /* If a process is already started for this SPOE context, retry * later. */ if (ctx->flags & SPOE_CTX_FL_PROCESS) return 0; - agent->rt[tid].processing++; ctx->stats.tv_start = now; ctx->stats.tv_request = now; ctx->stats.t_request = -1; @@ -2593,7 +2594,6 @@ spoe_stop_processing(struct spoe_agent *agent, struct spoe_context *ctx) } /* Reset the flag to allow next processing */ - agent->rt[tid].processing--; ctx->flags &= ~(SPOE_CTX_FL_PROCESS|SPOE_CTX_FL_FRAGMENTED); /* Reset processing timer */ @@ -2659,7 +2659,7 @@ spoe_process_messages(struct stream *s, struct spoe_context *ctx, s->task->expire = tick_first((tick_is_expired(s->task->expire, now_ms) ? 0 : s->task->expire), ctx->process_exp); } - ret = spoe_start_processing(agent, ctx, dir); + ret = spoe_start_processing(ctx, dir); if (!ret) goto out; @@ -2745,7 +2745,7 @@ spoe_process_group(struct stream *s, struct spoe_context *ctx, ret = spoe_process_messages(s, ctx, &group->messages, dir, SPOE_MSGS_BY_GROUP); if (ret && ctx->stats.t_process != -1) { SPOE_PRINTF(stderr, "%d.%06d [SPOE/%-15s] %s: stream=%p" - " - <GROUP:%s> sid=%u st=%u %ld/%ld/%ld/%ld/%ld %u/%u %u/%u %llu/%llu %u/%u\n", + " - <GROUP:%s> sid=%u st=%u %ld/%ld/%ld/%ld/%ld %u/%u %u/%u %llu/%llu %u\n", (int)now.tv_sec, (int)now.tv_usec, agent->id, __FUNCTION__, s, group->id, s->uniq_id, ctx->status_code, ctx->stats.t_request, ctx->stats.t_queue, ctx->stats.t_waiting, @@ -2753,7 +2753,7 @@ spoe_process_group(struct stream *s, struct spoe_context *ctx, agent->counters.idles, agent->counters.applets, agent->counters.nb_sending, agent->counters.nb_waiting, agent->counters.nb_errors, agent->counters.nb_processed, - agent->rt[tid].processing, read_freq_ctr(&agent->rt[tid].processing_per_sec)); + agent->rt[tid].sending_rate); if (ctx->status_code || !(conf->agent_fe.options2 & PR_O2_NOLOGNORM)) send_log(&conf->agent_fe, (!ctx->status_code ? LOG_NOTICE : LOG_WARNING), "SPOE: [%s] <GROUP:%s> sid=%u st=%u %ld/%ld/%ld/%ld/%ld %u/%u %u/%u %llu/%llu\n", @@ -2791,7 +2791,7 @@ spoe_process_event(struct stream *s, struct spoe_context *ctx, ret = spoe_process_messages(s, ctx, &(ctx->events[ev]), dir, SPOE_MSGS_BY_EVENT); if (ret && ctx->stats.t_process != -1) { SPOE_PRINTF(stderr, "%d.%06d [SPOE/%-15s] %s: stream=%p" - " - <EVENT:%s> sid=%u st=%u %ld/%ld/%ld/%ld/%ld %u/%u %u/%u %llu/%llu %u/%u\n", + " - <EVENT:%s> sid=%u st=%u %ld/%ld/%ld/%ld/%ld %u/%u %u/%u %llu/%llu %u\n", (int)now.tv_sec, (int)now.tv_usec, agent->id, __FUNCTION__, s, spoe_event_str[ev], s->uniq_id, ctx->status_code, ctx->stats.t_request, ctx->stats.t_queue, ctx->stats.t_waiting, @@ -2799,7 +2799,7 @@ spoe_process_event(struct stream *s, struct spoe_context *ctx, agent->counters.idles, agent->counters.applets, agent->counters.nb_sending, agent->counters.nb_waiting, agent->counters.nb_errors, agent->counters.nb_processed, - agent->rt[tid].processing, read_freq_ctr(&agent->rt[tid].processing_per_sec)); + agent->rt[tid].sending_rate); if (ctx->status_code || !(conf->agent_fe.options2 & PR_O2_NOLOGNORM)) send_log(&conf->agent_fe, (!ctx->status_code ? LOG_NOTICE : LOG_WARNING), "SPOE: [%s] <EVENT:%s> sid=%u st=%u %ld/%ld/%ld/%ld/%ld %u/%u %u/%u %llu/%llu\n", @@ -3089,7 +3089,7 @@ spoe_check(struct proxy *px, struct flt_conf *fconf) } for (i = 0; i < global.nbthread; ++i) { conf->agent->rt[i].frame_size = conf->agent->max_frame_size; - conf->agent->rt[i].processing = 0; + conf->agent->rt[i].sending_rate = 0; LIST_INIT(&conf->agent->rt[i].applets); LIST_INIT(&conf->agent->rt[i].sending_queue); LIST_INIT(&conf->agent->rt[i].waiting_queue); -- 2.7.4
