BUG/MEDIUM: stream-int: Don't loss write's notifs when a stream is woken up

When a write activity is reported on a channel, it is important to keep this
information for the stream because it take part on the analyzers' triggering.
When some data are written, the flag CF_WRITE_PARTIAL is set. It participates to
the task's timeout updates and to the stream's waking. It is also used in
CF_MASK_ANALYSER mask to trigger channels anaylzers. In the past, it was cleared
by process_stream. Because of a bug (fixed in commit 95fad5ba4 ["BUG/MAJOR:
stream-int: don't re-arm recv if send fails"]), It is now cleared before each
send and in stream_int_notify. So it is possible to loss this information when
process_stream is called, preventing analyzers to be called, and possibly
leading to a stalled stream.

Today, this happens in HTTP2 when you call the stat page or when you use the
cache filter. In fact, this happens when the response is sent by an applet. In
HTTP1, everything seems to work as expected.

To fix the problem, we need to make the difference between the write activity
reported to lower layers and the one reported to the stream. So the flag
CF_WRITE_EVENT has been added to notify the stream of the write activity on a
channel. It is set when a send succedded and reset by process_stream. It is also
used in CF_MASK_ANALYSER. finally, it is checked in stream_int_notify to wake up
a stream and in channel_check_timeouts.

This bug is probably present in 1.7 but it seems to have no effect. So for now,
no needs to backport it.

diff --git a/include/proto/channel.h b/include/proto/channel.h
index d6f355e71041886cd34caf3ff106678800c70277..274495f28c6c51c5f7ea5257264cc7bcf66addc6 100644 (file)
--- a/include/proto/channel.h
+++ b/include/proto/channel.h
@@ -231,7 +231,7 @@ static inline void channel_check_timeouts(struct channel *chn)
            unlikely(tick_is_expired(chn->rex, now_ms)))
                chn->flags |= CF_READ_TIMEOUT;
 
-       if (likely(!(chn->flags & (CF_SHUTW|CF_WRITE_TIMEOUT|CF_WRITE_ACTIVITY))) &&
+       if (likely(!(chn->flags & (CF_SHUTW|CF_WRITE_TIMEOUT|CF_WRITE_ACTIVITY|CF_WRITE_EVENT))) &&
            unlikely(tick_is_expired(chn->wex, now_ms)))
                chn->flags |= CF_WRITE_TIMEOUT;
 
@@ -491,7 +491,7 @@ static inline void co_skip(struct channel *chn, int len)
                chn->buf->p = chn->buf->data;
 
        /* notify that some data was written to the SI from the buffer */
-       chn->flags |= CF_WRITE_PARTIAL;
+       chn->flags |= CF_WRITE_PARTIAL | CF_WRITE_EVENT;
 }
 
 /* Tries to copy chunk <chunk> into the channel's buffer after length controls.

diff --git a/include/types/channel.h b/include/types/channel.h
index 03bb4e278889f597b18128a09322481aedcc721c..c483399fa67231b7d7f1ac69582bbe2db224ed75 100644 (file)
--- a/include/types/channel.h
+++ b/include/types/channel.h
@@ -117,11 +117,11 @@
 
 #define CF_WAKE_ONCE      0x10000000  /* pretend there is activity on this channel (one-shoot) */
 #define CF_FLT_ANALYZE    0x20000000  /* at least one filter is still analyzing this channel */
-/* unused: 0x40000000 */
+#define CF_WRITE_EVENT    0x40000000  /* write activity not processed yet by the stream */
 #define CF_ISRESP         0x80000000  /* 0 = request channel, 1 = response channel */
 
 /* Masks which define input events for stream analysers */
-#define CF_MASK_ANALYSER  (CF_READ_ATTACHED|CF_READ_ACTIVITY|CF_READ_TIMEOUT|CF_ANA_TIMEOUT|CF_WRITE_ACTIVITY|CF_WAKE_ONCE)
+#define CF_MASK_ANALYSER  (CF_READ_ATTACHED|CF_READ_ACTIVITY|CF_READ_TIMEOUT|CF_ANA_TIMEOUT|CF_WRITE_ACTIVITY|CF_WAKE_ONCE|CF_WRITE_EVENT)
 
 /* Mask for static flags which cause analysers to be woken up when they change */
 #define CF_MASK_STATIC    (CF_SHUTR|CF_SHUTW|CF_SHUTR_NOW|CF_SHUTW_NOW)

diff --git a/src/proto_http.c b/src/proto_http.c
index 05731793926882db42afa1adae01ec9753e7a0c1..9909eedc36f6e3116007f8da4d27015514561d58 100644 (file)
--- a/src/proto_http.c
+++ b/src/proto_http.c
@@ -4326,7 +4326,7 @@ void http_end_txn_clean_session(struct stream *s)
        s->si[1].exp       = TICK_ETERNITY;
        s->si[1].flags    &= SI_FL_ISBACK | SI_FL_DONT_WAKE; /* we're in the context of process_stream */
        s->req.flags &= ~(CF_SHUTW|CF_SHUTW_NOW|CF_AUTO_CONNECT|CF_WRITE_ERROR|CF_STREAMER|CF_STREAMER_FAST|CF_NEVER_WAIT|CF_WAKE_CONNECT|CF_WROTE_DATA);
-       s->res.flags &= ~(CF_SHUTR|CF_SHUTR_NOW|CF_READ_ATTACHED|CF_READ_ERROR|CF_READ_NOEXP|CF_STREAMER|CF_STREAMER_FAST|CF_WRITE_PARTIAL|CF_NEVER_WAIT|CF_WROTE_DATA);
+       s->res.flags &= ~(CF_SHUTR|CF_SHUTR_NOW|CF_READ_ATTACHED|CF_READ_ERROR|CF_READ_NOEXP|CF_STREAMER|CF_STREAMER_FAST|CF_WRITE_PARTIAL|CF_NEVER_WAIT|CF_WROTE_DATA|CF_WRITE_EVENT);
        s->flags &= ~(SF_DIRECT|SF_ASSIGNED|SF_ADDR_SET|SF_BE_ASSIGNED|SF_FORCE_PRST|SF_IGNORE_PRST);
        s->flags &= ~(SF_CURR_SESS|SF_REDIRECTABLE|SF_SRV_REUSED);
        s->flags &= ~(SF_ERR_MASK|SF_FINST_MASK|SF_REDISP);

diff --git a/src/stream.c b/src/stream.c
index 2d4f78a72eaffa054335693061bbbc38e3b3badd..8f81622dd46a197efd053f054c1195c237e1eaf4 100644 (file)
--- a/src/stream.c
+++ b/src/stream.c
@@ -613,7 +613,7 @@ static int sess_update_st_con_tcp(struct stream *s)
        if (!(req->flags & CF_WROTE_DATA) &&
            unlikely((rep->flags & CF_SHUTW) ||
                     ((req->flags & CF_SHUTW_NOW) && /* FIXME: this should not prevent a connection from establishing */
-                     ((!(req->flags & CF_WRITE_ACTIVITY) && channel_is_empty(req)) ||
+                     ((!(req->flags & (CF_WRITE_ACTIVITY|CF_WRITE_EVENT)) && channel_is_empty(req)) ||
                       s->be->options & PR_O_ABRT_CLOSE)))) {
                /* give up */
                si_shutw(si);
@@ -624,7 +624,7 @@ static int sess_update_st_con_tcp(struct stream *s)
        }
 
        /* we need to wait a bit more if there was no activity either */
-       if (!(req->flags & CF_WRITE_ACTIVITY))
+       if (!(req->flags & (CF_WRITE_ACTIVITY|CF_WRITE_EVENT)))
                return 1;
 
        /* OK, this means that a connection succeeded. The caller will be
@@ -1693,7 +1693,7 @@ struct task *process_stream(struct task *t)
                 */
                if (!((req->flags | res->flags) &
                      (CF_SHUTR|CF_READ_ACTIVITY|CF_READ_TIMEOUT|CF_SHUTW|
-                      CF_WRITE_ACTIVITY|CF_WRITE_TIMEOUT|CF_ANA_TIMEOUT)) &&
+                      CF_WRITE_ACTIVITY|CF_WRITE_EVENT|CF_WRITE_TIMEOUT|CF_ANA_TIMEOUT)) &&
                    !((si_f->flags | si_b->flags) & (SI_FL_EXP|SI_FL_ERR)) &&
                    ((s->pending_events & TASK_WOKEN_ANY) == TASK_WOKEN_TIMER)) {
                        si_f->flags &= ~SI_FL_DONT_WAKE;
@@ -2389,8 +2389,8 @@ struct task *process_stream(struct task *t)
                if (si_b->state == SI_ST_EST)
                        si_update(si_b);
 
-               req->flags &= ~(CF_READ_NULL|CF_READ_PARTIAL|CF_WRITE_NULL|CF_WRITE_PARTIAL|CF_READ_ATTACHED);
-               res->flags &= ~(CF_READ_NULL|CF_READ_PARTIAL|CF_WRITE_NULL|CF_WRITE_PARTIAL|CF_READ_ATTACHED);
+               req->flags &= ~(CF_READ_NULL|CF_READ_PARTIAL|CF_WRITE_NULL|CF_WRITE_PARTIAL|CF_READ_ATTACHED|CF_WRITE_EVENT);
+               res->flags &= ~(CF_READ_NULL|CF_READ_PARTIAL|CF_WRITE_NULL|CF_WRITE_PARTIAL|CF_READ_ATTACHED|CF_WRITE_EVENT);
                si_f->prev_state = si_f->state;
                si_b->prev_state = si_b->state;
                si_f->flags &= ~(SI_FL_ERR|SI_FL_EXP);

diff --git a/src/stream_interface.c b/src/stream_interface.c
index 9eef3a2f08cf475c818d20cac86ec51d1b7768c8..4ac2320bfc7647b36417c3497a7bfb22f9740f82 100644 (file)
--- a/src/stream_interface.c
+++ b/src/stream_interface.c
@@ -546,7 +546,7 @@ void stream_int_notify(struct stream_interface *si)
 
            /* changes on the consumption side */
            (oc->flags & (CF_WRITE_NULL|CF_WRITE_ERROR)) ||
-           ((oc->flags & CF_WRITE_ACTIVITY) &&
+           ((oc->flags & (CF_WRITE_ACTIVITY|CF_WRITE_EVENT)) &&
             ((oc->flags & CF_SHUTW) ||
              ((oc->flags & CF_WAKE_WRITE) &&
               (si_opposite(si)->state != SI_ST_EST ||
@@ -636,7 +636,7 @@ static void si_cs_send(struct conn_stream *cs)
        if (oc->pipe && conn->xprt->snd_pipe && conn->mux->snd_pipe) {
                ret = conn->mux->snd_pipe(cs, oc->pipe);
                if (ret > 0)
-                       oc->flags |= CF_WRITE_PARTIAL | CF_WROTE_DATA;
+                       oc->flags |= CF_WRITE_PARTIAL | CF_WROTE_DATA | CF_WRITE_EVENT;
 
                if (!oc->pipe->data) {
                        put_pipe(oc->pipe);
@@ -682,7 +682,7 @@ static void si_cs_send(struct conn_stream *cs)
 
                ret = conn->mux->snd_buf(cs, oc->buf, send_flag);
                if (ret > 0) {
-                       oc->flags |= CF_WRITE_PARTIAL | CF_WROTE_DATA;
+                       oc->flags |= CF_WRITE_PARTIAL | CF_WROTE_DATA | CF_WRITE_EVENT;
 
                        if (!oc->buf->o) {
                                /* Always clear both flags once everything has been sent, they're one-shot */