MAJOR: session: only allocate buffers when needed

A session doesn't need buffers all the time, especially when they're
empty. With this patch, we don't allocate buffers anymore when the
session is initialized, we only allocate them in two cases :

  - during process_session()
  - during I/O operations

During process_session(), we try hard to allocate both buffers at once
so that we know for sure that a started operation can complete. Indeed,
a previous version of this patch used to allocate one buffer at a time,
but it can result in a deadlock when all buffers are allocated for
requests for example, and there's no buffer left to emit error responses.
Here, if any of the buffers cannot be allocated, the whole operation is
cancelled and the session is added at the tail of the buffer wait queue.

At the end of process_session(), a call to session_release_buffers() is
done so that we can offer unused buffers to other sessions waiting for
them.

For I/O operations, we only need to allocate a buffer on the Rx path.
For this, we only allocate a single buffer but ensure that at least two
are available to avoid the deadlock situation. In case buffers are not
available, SI_FL_WAIT_ROOM is set on the stream interface and the session
is queued. Unused buffers resulting either from a successful send() or
from an unused read buffer are offered to pending sessions during the
->wake() callback.

diff --git a/src/peers.c b/src/peers.c
index 84356e632486a7f668bbece0b7c5fd0fe012bea4..173f7847dcd836bb4995c57d9a4218bd7a296ae9 100644 (file)
--- a/src/peers.c
+++ b/src/peers.c
@@ -1277,12 +1277,6 @@ static struct session *peer_session_create(struct peer *peer, struct peer_sessio
 
        s->rep->flags |= CF_READ_DONTWAIT;
 
-       if (unlikely(b_alloc(&s->req->buf) == NULL))
-               goto out_fail_req_buf; /* no memory */
-
-       if (unlikely(b_alloc(&s->rep->buf) == NULL))
-               goto out_fail_rep_buf; /* no memory */
-
        /* it is important not to call the wakeup function directly but to
         * pass through task_wakeup(), because this one knows how to apply
         * priorities to tasks.
@@ -1299,10 +1293,6 @@ static struct session *peer_session_create(struct peer *peer, struct peer_sessio
        return s;
 
        /* Error unrolling */
- out_fail_rep_buf:
-       b_free(&s->req->buf);
- out_fail_req_buf:
-       pool_free2(pool2_channel, s->rep);
  out_fail_rep:
        pool_free2(pool2_channel, s->req);
  out_fail_req:

diff --git a/src/session.c b/src/session.c
index d2368599d22caf191413048faacacccf81584c0d..fc5b886579a6728296f5c4b74804f768ac1caf31 100644 (file)
--- a/src/session.c
+++ b/src/session.c
@@ -516,12 +516,6 @@ int session_complete(struct session *s)
        s->rep->wex = TICK_ETERNITY;
        s->rep->analyse_exp = TICK_ETERNITY;
 
-       if (unlikely(b_alloc(&s->req->buf) == NULL))
-               goto out_free_rep; /* no memory */
-
-       if (unlikely(b_alloc(&s->rep->buf) == NULL))
-               goto out_free_req_buf; /* no memory */
-
        txn = &s->txn;
        /* Those variables will be checked and freed if non-NULL in
         * session.c:session_free(). It is important that they are
@@ -550,7 +544,7 @@ int session_complete(struct session *s)
                 * finished (=0, eg: monitoring), in both situations,
                 * we can release everything and close.
                 */
-               goto out_free_rep_buf;
+               goto out_free_rep;
        }
 
        /* if logs require transport layer information, note it on the connection */
@@ -568,10 +562,6 @@ int session_complete(struct session *s)
        return 1;
 
        /* Error unrolling */
- out_free_rep_buf:
-       b_free(&s->rep->buf);
- out_free_req_buf:
-       b_free(&s->req->buf);
  out_free_rep:
        pool_free2(pool2_channel, s->rep);
  out_free_req:
@@ -1804,6 +1794,16 @@ struct task *process_session(struct task *t)
                        goto update_exp_and_leave;
        }
 
+       /* below we may emit error messages so we have to ensure that we have
+        * our buffers properly allocated.
+        */
+       if (!session_alloc_buffers(s)) {
+               /* No buffer available, we've been subscribed to the list of
+                * buffer waiters, let's wait for our turn.
+                */
+               goto update_exp_and_leave;
+       }
+
        /* 1b: check for low-level errors reported at the stream interface.
         * First we check if it's a retryable error (in which case we don't
         * want to tell the buffer). Otherwise we report the error one level
@@ -2602,6 +2602,7 @@ struct task *process_session(struct task *t)
                if ((si_applet_call(s->req->cons) | si_applet_call(s->rep->cons)) != 0) {
                        if (task_in_rq(t)) {
                                t->expire = TICK_ETERNITY;
+                               session_release_buffers(s);
                                return t;
                        }
                }
@@ -2631,6 +2632,7 @@ struct task *process_session(struct task *t)
                if (!tick_isset(t->expire))
                        ABORT_NOW();
 #endif
+               session_release_buffers(s);
                return t; /* nothing more to do */
        }
 

diff --git a/src/stream_interface.c b/src/stream_interface.c
index 2c47953d8b6384b475c7145fa82dc9068637d21b..08c72e56c676c6cec0c9ab381ddb3a057c23120a 100644 (file)
--- a/src/stream_interface.c
+++ b/src/stream_interface.c
@@ -642,6 +642,8 @@ static int si_conn_wake_cb(struct connection *conn)
        }
        if (si->ib->flags & CF_READ_ACTIVITY)
                si->ib->flags &= ~CF_READ_DONTWAIT;
+
+       session_release_buffers(si_sess(si));
        return 0;
 }
 
@@ -1166,6 +1168,12 @@ static void si_conn_recv_cb(struct connection *conn)
                chn->pipe = NULL;
        }
 
+       /* now we'll need a buffer */
+       if (!session_alloc_recv_buffer(si_sess(si), &chn->buf)) {
+               si->flags |= SI_FL_WAIT_ROOM;
+               goto end_recv;
+       }
+
        /* Important note : if we're called with POLL_IN|POLL_HUP, it means the read polling
         * was enabled, which implies that the recv buffer was not full. So we have a guarantee
         * that if such an event is not handled above in splice, it will be handled here by
@@ -1230,9 +1238,6 @@ static void si_conn_recv_cb(struct connection *conn)
                }
        } /* while !flags */
 
-       if (conn->flags & CO_FL_ERROR)
-               return;
-
        if (cur_read) {
                if ((chn->flags & (CF_STREAMER | CF_STREAMER_FAST)) &&
                    (cur_read <= chn->buf->size / 2)) {
@@ -1272,6 +1277,10 @@ static void si_conn_recv_cb(struct connection *conn)
                chn->last_read = now_ms;
        }
 
+ end_recv:
+       if (conn->flags & CO_FL_ERROR)
+               return;
+
        if (conn_data_read0_pending(conn))
                /* connection closed */
                goto out_shutdown_r;