struct pool_head *pool_head_pendconn;
-static void __pendconn_free(struct pendconn *p);
-
/* perform minimal intializations, report 0 in case of error, 1 if OK. */
int init_pendconn()
{
return max;
}
-
-/* Returns the first pending connection for server <s>, which may be NULL if
- * nothing is pending.
- */
-static inline struct pendconn *pendconn_from_srv(const struct server *s) {
- if (!s->nbpend)
- return NULL;
- return LIST_ELEM(s->pendconns.n, struct pendconn *, list);
-}
-
-/* Returns the first pending connection for proxy <px>, which may be NULL if
- * nothing is pending.
+/* Remove the pendconn from the server/proxy queue. At this stage, the
+ * connection is not really dequeued. It will be done during the
+ * process_stream. This function must be called by function owning the locks on
+ * the pendconn _AND_ the server/proxy. It also decreases the pending count.
+ *
+ * The caller must own the lock on the pendconn _AND_ the queue containing the
+ * pendconn. The pendconn must still be queued.
*/
-static inline struct pendconn *pendconn_from_px(const struct proxy *px) {
- if (!px->nbpend)
- return NULL;
-
- return LIST_ELEM(px->pendconns.n, struct pendconn *, list);
+static void pendconn_unlink(struct pendconn *p)
+{
+ if (p->srv)
+ p->srv->nbpend--;
+ else
+ p->px->nbpend--;
+ HA_ATOMIC_SUB(&p->px->totpend, 1);
+ LIST_DEL(&p->list);
+ LIST_INIT(&p->list);
}
-
-/* Detaches the next pending connection from either a server or a proxy, and
- * returns its associated stream. If no pending connection is found, NULL is
- * returned. Note that neither <srv> nor <px> may be NULL.
- * Priority is given to the oldest request in the queue if both <srv> and <px>
- * have pending requests. This ensures that no request will be left unserved.
- * The <px> queue is not considered if the server (or a tracked server) is not
- * RUNNING, is disabled, or has a null weight (server going down). The <srv>
- * queue is still considered in this case, because if some connections remain
- * there, it means that some requests have been forced there after it was seen
- * down (eg: due to option persist).
- * The stream is immediately marked as "assigned", and both its <srv> and
- * <srv_conn> are set to <srv>,
+/* Process the next pending connection from either a server or a proxy, and
+ * returns 0 on success. If no pending connection is found, 1 is returned.
+ * Note that neither <srv> nor <px> may be NULL. Priority is given to the
+ * oldest request in the queue if both <srv> and <px> have pending
+ * requests. This ensures that no request will be left unserved. The <px> queue
+ * is not considered if the server (or a tracked server) is not RUNNING, is
+ * disabled, or has a null weight (server going down). The <srv> queue is still
+ * considered in this case, because if some connections remain there, it means
+ * that some requests have been forced there after it was seen down (eg: due to
+ * option persist). The stream is immediately marked as "assigned", and both
+ * its <srv> and <srv_conn> are set to <srv>.
+ *
+ * This function must only be called if the server queue _AND_ the proxy queue
+ * are locked. Today it is only called by process_srv_queue.
*/
-static struct stream *pendconn_get_next_strm(struct server *srv, struct proxy *px)
+static int pendconn_process_next_strm(struct server *srv, struct proxy *px)
{
- struct pendconn *ps, *pp;
- struct stream *strm;
- struct server *rsrv;
+ struct pendconn *p = NULL;
+ struct server *rsrv;
rsrv = srv->track;
if (!rsrv)
rsrv = srv;
- ps = pendconn_from_srv(srv);
- pp = pendconn_from_px(px);
- /* we want to get the definitive pendconn in <ps> */
- if (!pp || !srv_currently_usable(rsrv)) {
- if (!ps)
- return NULL;
- } else {
- /* pendconn exists in the proxy queue */
- if (!ps || tv_islt(&pp->strm->logs.tv_request, &ps->strm->logs.tv_request))
- ps = pp;
+ if (srv->nbpend) {
+ list_for_each_entry(p, &srv->pendconns, list) {
+ if (!HA_SPIN_TRYLOCK(PENDCONN_LOCK, &p->lock))
+ goto ps_found;
+ }
+ p = NULL;
}
- strm = ps->strm;
- __pendconn_free(ps);
- /* we want to note that the stream has now been assigned a server */
- strm->flags |= SF_ASSIGNED;
- strm->target = &srv->obj_type;
- __stream_add_srv_conn(strm, srv);
+ ps_found:
+ if (srv_currently_usable(rsrv) && px->nbpend) {
+ struct pendconn *pp;
+
+ list_for_each_entry(pp, &px->pendconns, list) {
+ /* If the server pendconn is older than the proxy one,
+ * we process the server one. */
+ if (p && !tv_islt(&pp->strm->logs.tv_request, &p->strm->logs.tv_request))
+ goto pendconn_found;
+
+ if (!HA_SPIN_TRYLOCK(PENDCONN_LOCK, &pp->lock)) {
+ /* Let's switch from the server pendconn to the
+ * proxy pendconn. Don't forget to unlock the
+ * server pendconn, if any. */
+ if (p)
+ HA_SPIN_UNLOCK(PENDCONN_LOCK, &p->lock);
+ p = pp;
+ goto pendconn_found;
+ }
+ }
+ }
+
+ if (!p)
+ return 1;
+
+ pendconn_found:
+ pendconn_unlink(p);
+ p->strm_flags |= SF_ASSIGNED;
+ p->srv = srv;
+
HA_ATOMIC_ADD(&srv->served, 1);
HA_ATOMIC_ADD(&srv->proxy->served, 1);
if (px->lbprm.server_take_conn)
px->lbprm.server_take_conn(srv);
+ __stream_add_srv_conn(p->strm, srv);
- return strm;
+ task_wakeup(p->strm->task, TASK_WOKEN_RES);
+ HA_SPIN_UNLOCK(PENDCONN_LOCK, &p->lock);
+ return 0;
}
-/*
- * Manages a server's connection queue. This function will try to dequeue as
+/* Manages a server's connection queue. This function will try to dequeue as
* many pending streams as possible, and wake them up.
*/
void process_srv_queue(struct server *s)
HA_SPIN_LOCK(PROXY_LOCK, &p->lock);
HA_SPIN_LOCK(SERVER_LOCK, &s->lock);
-
- /* First, check if we can handle some connections queued at the proxy. We
- * will take as many as we can handle.
- */
maxconn = srv_dynamic_maxconn(s);
while (s->served < maxconn) {
- struct stream *strm = pendconn_get_next_strm(s, p);
-
- if (strm == NULL)
+ if (pendconn_process_next_strm(s, p))
break;
- task_wakeup(strm->task, TASK_WOKEN_RES);
}
HA_SPIN_UNLOCK(SERVER_LOCK, &s->lock);
HA_SPIN_UNLOCK(PROXY_LOCK, &p->lock);
* are updated accordingly. Returns NULL if no memory is available, otherwise the
* pendconn itself. If the stream was already marked as served, its flag is
* cleared. It is illegal to call this function with a non-NULL strm->srv_conn.
+ *
+ * This function must be called by the stream itself, so in the context of
+ * process_stream.
*/
struct pendconn *pendconn_add(struct stream *strm)
{
struct pendconn *p;
- struct server *srv;
- int count;
+ struct proxy *px;
+ struct server *srv;
p = pool_alloc(pool_head_pendconn);
if (!p)
return NULL;
- strm->pend_pos = p;
- p->strm = strm;
srv = objt_server(strm->target);
+ px = strm->be;
+
+ p->srv = NULL;
+ p->px = px;
+ p->strm = strm;
+ p->strm_flags = strm->flags;
+ HA_SPIN_INIT(&p->lock);
if ((strm->flags & SF_ASSIGNED) && srv) {
p->srv = srv;
HA_SPIN_LOCK(SERVER_LOCK, &srv->lock);
+ srv->nbpend++;
+ strm->logs.srv_queue_size += srv->nbpend;
+ if (srv->nbpend > srv->counters.nbpend_max)
+ srv->counters.nbpend_max = srv->nbpend;
LIST_ADDQ(&srv->pendconns, &p->list);
HA_SPIN_UNLOCK(SERVER_LOCK, &srv->lock);
- count = HA_ATOMIC_ADD(&srv->nbpend, 1);
- strm->logs.srv_queue_size += count;
- HA_ATOMIC_UPDATE_MAX(&srv->counters.nbpend_max, count);
- } else {
- p->srv = NULL;
- HA_SPIN_LOCK(PROXY_LOCK, &strm->be->lock);
- LIST_ADDQ(&strm->be->pendconns, &p->list);
- HA_SPIN_UNLOCK(PROXY_LOCK, &strm->be->lock);
- count = HA_ATOMIC_ADD(&strm->be->nbpend, 1);
- strm->logs.prx_queue_size += count;
- HA_ATOMIC_UPDATE_MAX(&strm->be->be_counters.nbpend_max, count);
}
- HA_ATOMIC_ADD(&strm->be->totpend, 1);
+ else {
+ HA_SPIN_LOCK(PROXY_LOCK, &px->lock);
+ px->nbpend++;
+ strm->logs.prx_queue_size += px->nbpend;
+ if (px->nbpend > px->be_counters.nbpend_max)
+ px->be_counters.nbpend_max = px->nbpend;
+ LIST_ADDQ(&px->pendconns, &p->list);
+ HA_SPIN_UNLOCK(PROXY_LOCK, &px->lock);
+ }
+ HA_ATOMIC_ADD(&px->totpend, 1);
+ strm->pend_pos = p;
return p;
}
*/
int pendconn_redistribute(struct server *s)
{
- struct pendconn *pc, *pc_bck;
+ struct pendconn *p, *pback;
int xferred = 0;
+ /* The REDISP option was specified. We will ignore cookie and force to
+ * balance or use the dispatcher. */
+ if ((s->proxy->options & (PR_O_REDISP|PR_O_PERSIST)) != PR_O_REDISP)
+ return 0;
+
HA_SPIN_LOCK(SERVER_LOCK, &s->lock);
- list_for_each_entry_safe(pc, pc_bck, &s->pendconns, list) {
- struct stream *strm = pc->strm;
+ list_for_each_entry_safe(p, pback, &s->pendconns, list) {
+ if (p->strm_flags & SF_FORCE_PRST)
+ continue;
- if ((strm->be->options & (PR_O_REDISP|PR_O_PERSIST)) == PR_O_REDISP &&
- !(strm->flags & SF_FORCE_PRST)) {
- /* The REDISP option was specified. We will ignore
- * cookie and force to balance or use the dispatcher.
- */
+ if (HA_SPIN_TRYLOCK(PENDCONN_LOCK, &p->lock))
+ continue;
- /* it's left to the dispatcher to choose a server */
- strm->flags &= ~(SF_DIRECT | SF_ASSIGNED | SF_ADDR_SET);
+ /* it's left to the dispatcher to choose a server */
+ pendconn_unlink(p);
+ p->strm_flags &= ~(SF_DIRECT | SF_ASSIGNED | SF_ADDR_SET);
- __pendconn_free(pc);
- task_wakeup(strm->task, TASK_WOKEN_RES);
- xferred++;
- }
+ task_wakeup(p->strm->task, TASK_WOKEN_RES);
+ HA_SPIN_UNLOCK(PENDCONN_LOCK, &p->lock);
}
HA_SPIN_UNLOCK(SERVER_LOCK, &s->lock);
return xferred;
*/
int pendconn_grab_from_px(struct server *s)
{
- int xferred;
+ struct pendconn *p, *pback;
+ int maxconn, xferred = 0;
if (!srv_currently_usable(s))
return 0;
HA_SPIN_LOCK(PROXY_LOCK, &s->proxy->lock);
- for (xferred = 0; !s->maxconn || xferred < srv_dynamic_maxconn(s); xferred++) {
- struct stream *strm;
- struct pendconn *p;
-
- p = pendconn_from_px(s->proxy);
- if (!p)
+ maxconn = srv_dynamic_maxconn(s);
+ list_for_each_entry_safe(p, pback, &s->proxy->pendconns, list) {
+ if (s->maxconn && s->served + xferred >= maxconn)
break;
- p->strm->target = &s->obj_type;
- strm = p->strm;
- __pendconn_free(p);
- task_wakeup(strm->task, TASK_WOKEN_RES);
+
+ if (HA_SPIN_TRYLOCK(PENDCONN_LOCK, &p->lock))
+ continue;
+
+ pendconn_unlink(p);
+ p->srv = s;
+
+ task_wakeup(p->strm->task, TASK_WOKEN_RES);
+ HA_SPIN_UNLOCK(PENDCONN_LOCK, &p->lock);
+ xferred++;
}
HA_SPIN_UNLOCK(PROXY_LOCK, &s->proxy->lock);
return xferred;
}
-/*
- * Detaches pending connection <p>, decreases the pending count, and frees
- * the pending connection. The connection might have been queued to a specific
- * server as well as to the proxy. The stream also gets marked unqueued.
+/* Try to dequeue pending connection attached to the stream <strm>. It must
+ * always exists here. If the pendconn is still linked to the server or the
+ * proxy queue, nothing is done and the function returns 1. Otherwise,
+ * <strm>->flags and <strm>->target are updated, the pendconn is released and 0
+ * is returned.
+ *
+ * This function must be called by the stream itself, so in the context of
+ * process_stream.
*/
-void pendconn_free(struct pendconn *p)
+int pendconn_dequeue(struct stream *strm)
{
- if (p->srv) {
- HA_SPIN_LOCK(SERVER_LOCK, &p->srv->lock);
- LIST_DEL(&p->list);
- HA_SPIN_UNLOCK(SERVER_LOCK, &p->srv->lock);
- HA_ATOMIC_SUB(&p->srv->nbpend, 1);
+ struct pendconn *p;
+
+ if (unlikely(!strm->pend_pos)) {
+ /* unexpected case because it is called by the stream itself and
+ * only the stream can release a pendconn. So it is only
+ * possible if a pendconn is released by someone else or if the
+ * stream is supposed to be queued but without its associated
+ * pendconn. In both cases it is a bug! */
+ abort();
}
- else {
- HA_SPIN_LOCK(SERVER_LOCK, &p->strm->be->lock);
- LIST_DEL(&p->list);
- HA_SPIN_UNLOCK(SERVER_LOCK, &p->strm->be->lock);
- HA_ATOMIC_SUB(&p->strm->be->nbpend, 1);
+ p = strm->pend_pos;
+ HA_SPIN_LOCK(PENDCONN_LOCK, &p->lock);
+
+ /* the pendconn is still linked to the server/proxy queue, so unlock it
+ * and go away. */
+ if (!LIST_ISEMPTY(&p->list)) {
+ HA_SPIN_UNLOCK(PENDCONN_LOCK, &p->lock);
+ return 1;
}
- p->strm->pend_pos = NULL;
- HA_ATOMIC_SUB(&p->strm->be->totpend, 1);
+
+ /* the pendconn must be dequeued now */
+ if (p->srv)
+ strm->target = &p->srv->obj_type;
+
+ strm->flags &= ~(SF_DIRECT | SF_ASSIGNED | SF_ADDR_SET);
+ strm->flags |= p->strm_flags & (SF_DIRECT | SF_ASSIGNED | SF_ADDR_SET);
+ strm->pend_pos = NULL;
+ HA_SPIN_UNLOCK(PENDCONN_LOCK, &p->lock);
pool_free(pool_head_pendconn, p);
+ return 0;
}
-/* Lock-free version of pendconn_free. */
-static void __pendconn_free(struct pendconn *p)
+/* Release the pending connection <p>, and decreases the pending count if
+ * needed. The connection might have been queued to a specific server as well as
+ * to the proxy. The stream also gets marked unqueued. <p> must always be
+ * defined here. So it is the caller responsibility to check its existance.
+ *
+ * This function must be called by the stream itself, so in the context of
+ * process_stream.
+ */
+void pendconn_free(struct pendconn *p)
{
+ struct stream *strm = p->strm;
+
+ HA_SPIN_LOCK(PENDCONN_LOCK, &p->lock);
+
+ /* The pendconn was already unlinked, just release it. */
+ if (LIST_ISEMPTY(&p->list))
+ goto release;
+
if (p->srv) {
+ HA_SPIN_LOCK(SERVER_LOCK, &p->srv->lock);
+ p->srv->nbpend--;
LIST_DEL(&p->list);
- HA_ATOMIC_SUB(&p->srv->nbpend, 1);
+ HA_SPIN_UNLOCK(SERVER_LOCK, &p->srv->lock);
}
else {
+ HA_SPIN_LOCK(PROXY_LOCK, &p->px->lock);
+ p->px->nbpend--;
LIST_DEL(&p->list);
- HA_ATOMIC_SUB(&p->strm->be->nbpend, 1);
+ HA_SPIN_UNLOCK(PROXY_LOCK, &p->px->lock);
}
- p->strm->pend_pos = NULL;
- HA_ATOMIC_SUB(&p->strm->be->totpend, 1);
+ HA_ATOMIC_SUB(&p->px->totpend, 1);
+
+ release:
+ strm->pend_pos = NULL;
+ HA_SPIN_UNLOCK(PENDCONN_LOCK, &p->lock);
pool_free(pool_head_pendconn, p);
}
projects / thirdparty / haproxy.git / commitdiff
