MINOR: spoe: Replace sending_rate by a frequency counter

sending_rate was a counter used to evaluate the SPOE capacity to process
frames. Because it was not really accurrate, it has been replaced by a frequency
counter representing the number of frames handled by the SPOE per second. We
just check this counter is higher than the number of streams waiting for a
reply. If not, a new applet is created.

diff --git a/include/types/spoe.h b/include/types/spoe.h
index 30510565d25b30c67549cee6c6bde438857aea66..2354f6e1aabcb3fe6399c285ab278ae0cf563a38 100644 (file)
--- a/include/types/spoe.h
+++ b/include/types/spoe.h
@@ -262,7 +262,9 @@ struct spoe_agent {
                unsigned int    frame_size;     /* current maximum frame size, only used to encode messages */
                unsigned int    applets_act;    /* # of applets alive at a time */
                unsigned int    applets_idle;   /* # of applets in the state SPOE_APPCTX_ST_IDLE */
-               unsigned int    sending_rate;   /* the global sending rate */
+
+               unsigned int    processing;
+               struct freq_ctr processing_per_sec;
 
                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 1d8e4e9ffb1a9e8610b5d06501ba5e9044918a87..2baa327cc4337962930f328810380c01366a108f 100644 (file)
--- a/src/flt_spoe.c
+++ b/src/flt_spoe.c
@@ -1638,7 +1638,7 @@ spoe_handle_processing_appctx(struct appctx *appctx)
                        goto next;
 
                case 0: /* ignore */
-                       agent->rt[tid].sending_rate++;
+                       update_freq_ctr(&agent->rt[tid].processing_per_sec, 1);
                        fpa++;
                        break;
 
@@ -1646,7 +1646,7 @@ spoe_handle_processing_appctx(struct appctx *appctx)
                        break;
 
                default:
-                       agent->rt[tid].sending_rate++;
+                       update_freq_ctr(&agent->rt[tid].processing_per_sec, 1);
                        fpa++;
                        break;
        }
@@ -1991,7 +1991,7 @@ spoe_queue_context(struct spoe_context *ctx)
 
        /* Check if we need to create a new SPOE applet or not. */
        if (agent->rt[tid].applets_idle &&
-           agent->rt[tid].sending_rate)
+           agent->rt[tid].processing < read_freq_ctr(&agent->rt[tid].processing_per_sec))
                goto end;
 
        SPOE_PRINTF(stderr, "%d.%06d [SPOE/%-15s] %s: stream=%p"
@@ -2045,18 +2045,15 @@ spoe_queue_context(struct spoe_context *ctx)
                return -1;
        }
 
-       /* Add the SPOE context in the sending queue and update all running
-        * info */
+       /* Add the SPOE context in the sending queue */
        LIST_ADDQ(&agent->rt[tid].sending_queue, &ctx->list);
-       if (agent->rt[tid].sending_rate)
-               agent->rt[tid].sending_rate--;
 
        SPOE_PRINTF(stderr, "%d.%06d [SPOE/%-15s] %s: stream=%p"
                    " - Add stream in sending queue"
-                   " - applets_act=%u - applets_idle=%u - sending_rate=%u\n",
+                   " - applets_act=%u - applets_idle=%u - processing=%u\n",
                    (int)now.tv_sec, (int)now.tv_usec, agent->id, __FUNCTION__,
                    ctx->strm, agent->rt[tid].applets_act, agent->rt[tid].applets_idle,
-                   agent->rt[tid].sending_rate);
+                   agent->rt[tid].processing);
 
        /* Finally try to wakeup the first IDLE applet found and move it at the
         * end of the list. */
@@ -2436,13 +2433,15 @@ spoe_process_actions(struct stream *s, struct spoe_context *ctx, int dir)
  * Functions that process SPOE events
  **************************************************************************/
 static inline int
-spoe_start_processing(struct spoe_context *ctx, int dir)
+spoe_start_processing(struct spoe_agent *agent, 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++;
+
        /* Set the right flag to prevent request and response processing
         * in same time. */
        ctx->flags |= ((dir == SMP_OPT_DIR_REQ)
@@ -2452,16 +2451,20 @@ spoe_start_processing(struct spoe_context *ctx, int dir)
 }
 
 static inline void
-spoe_stop_processing(struct spoe_context *ctx)
+spoe_stop_processing(struct spoe_agent *agent, struct spoe_context *ctx)
 {
        struct spoe_appctx *sa = ctx->spoe_appctx;
 
+       if (!(ctx->flags & SPOE_CTX_FL_PROCESS))
+               return;
+
        if (sa && sa->frag_ctx.ctx == ctx) {
                sa->frag_ctx.ctx = NULL;
                spoe_wakeup_appctx(sa->owner);
        }
 
        /* Reset the flag to allow next processing */
+       agent->rt[tid].processing--;
        ctx->flags &= ~(SPOE_CTX_FL_PROCESS|SPOE_CTX_FL_FRAGMENTED);
 
        ctx->status_code = 0;
@@ -2555,7 +2558,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(ctx, dir);
+               ret = spoe_start_processing(agent, ctx, dir);
                if (!ret)
                        goto out;
 
@@ -2609,7 +2612,7 @@ spoe_process_messages(struct stream *s, struct spoe_context *ctx,
        ret = 1;
 
   end:
-       spoe_stop_processing(ctx);
+       spoe_stop_processing(agent, ctx);
        return ret;
 }
 
@@ -2710,7 +2713,7 @@ spoe_wakeup_context(struct spoe_context *ctx)
 }
 
 static struct spoe_context *
-spoe_create_context(struct filter *filter)
+spoe_create_context(struct stream *s, struct filter *filter)
 {
        struct spoe_config  *conf = FLT_CONF(filter);
        struct spoe_context *ctx;
@@ -2736,17 +2739,25 @@ spoe_create_context(struct filter *filter)
        ctx->frame_id    = 1;
        ctx->process_exp = TICK_ETERNITY;
 
+       ctx->strm   = s;
+       ctx->state  = SPOE_CTX_ST_READY;
+       filter->ctx = ctx;
+
        return ctx;
 }
 
 static void
-spoe_destroy_context(struct spoe_context *ctx)
+spoe_destroy_context(struct filter *filter)
 {
+       struct spoe_config  *conf = FLT_CONF(filter);
+       struct spoe_context *ctx  = filter->ctx;
+
        if (!ctx)
                return;
 
-       spoe_stop_processing(ctx);
+       spoe_stop_processing(conf->agent, ctx);
        pool_free(pool_head_spoe_ctx, ctx);
+       filter->ctx = NULL;
 }
 
 static void
@@ -2907,8 +2918,7 @@ spoe_start(struct stream *s, struct filter *filter)
                    (int)now.tv_sec, (int)now.tv_usec, agent->id,
                    __FUNCTION__, s);
 
-       ctx = spoe_create_context(filter);
-       if (ctx == NULL) {
+       if ((ctx = spoe_create_context(s, filter)) == NULL) {
                SPOE_PRINTF(stderr, "%d.%06d [SPOE/%-15s] %s: stream=%p"
                            " - failed to create SPOE context\n",
                            (int)now.tv_sec, (int)now.tv_usec, agent->id,
@@ -2919,10 +2929,6 @@ spoe_start(struct stream *s, struct filter *filter)
                return 0;
        }
 
-       ctx->strm   = s;
-       ctx->state  = SPOE_CTX_ST_READY;
-       filter->ctx = ctx;
-
        if (!LIST_ISEMPTY(&ctx->events[SPOE_EV_ON_TCP_REQ_FE]))
                filter->pre_analyzers |= AN_REQ_INSPECT_FE;
 
@@ -2953,7 +2959,7 @@ spoe_stop(struct stream *s, struct filter *filter)
                    (int)now.tv_sec, (int)now.tv_usec,
                    ((struct spoe_config *)FLT_CONF(filter))->agent->id,
                    __FUNCTION__, s);
-       spoe_destroy_context(filter->ctx);
+       spoe_destroy_context(filter);
 }
 
 
@@ -3186,7 +3192,7 @@ cfg_parse_spoe_agent(const char *file, int linenum, char **args, int kwm)
                        curagent->rt[i].frame_size   = curagent->max_frame_size;
                        curagent->rt[i].applets_act  = 0;
                        curagent->rt[i].applets_idle = 0;
-                       curagent->rt[i].sending_rate = 0;
+                       curagent->rt[i].processing   = 0;
                        LIST_INIT(&curagent->rt[i].applets);
                        LIST_INIT(&curagent->rt[i].sending_queue);
                        LIST_INIT(&curagent->rt[i].waiting_queue);
@@ -4168,7 +4174,7 @@ spoe_send_group(struct act_rule *rule, struct proxy *px,
                                    agent->id, __FUNCTION__, s, group->id);
                        ctx->status_code = SPOE_CTX_ERR_INTERRUPT;
                        spoe_handle_processing_error(s, agent, ctx, dir);
-                       spoe_stop_processing(ctx);
+                       spoe_stop_processing(agent, ctx);
                        return ACT_RET_CONT;
                }
                return ACT_RET_YIELD;