MAJOR: applet: now call si_applet_done() instead of si_update() in I/O handlers
The applet I/O handlers now rely on si_applet_done() which itself decides
to wake up or sleep the appctx. Now it becomes critical that applte handlers
properly call this on every exit path so that the appctx is removed from the
active list after I/O have been handled. One such call was added to the Lua
socket handler. It used to work without it probably because the main task is
woken up by the parent task but now it's needed.
diff --git a/src/dumpstats.c b/src/dumpstats.c
index 0a2bd67..809025f 100644
--- a/src/dumpstats.c
+++ b/src/dumpstats.c
@@ -2428,14 +2428,14 @@
res->flags |= CF_READ_NULL;
}
+ out:
/* update all other flags and resync with the other side */
- si_update(si);
+ si_applet_done(si);
/* we don't want to expire timeouts while we're processing requests */
si_ic(si)->rex = TICK_ETERNITY;
si_oc(si)->wex = TICK_ETERNITY;
- out:
DPRINTF(stderr, "%s@%d: st=%d, rqf=%x, rpf=%x, rqh=%d, rqs=%d, rh=%d, rs=%d\n",
__FUNCTION__, __LINE__,
si->state, req->flags, res->flags, req->buf->i, req->buf->o, res->buf->i, res->buf->o);
@@ -4897,7 +4897,7 @@
if (bi_putchk(si_ic(si), &trash) == -1) {
si->flags |= SI_FL_WAIT_ROOM;
si_ic(si)->to_forward = last_fwd;
- goto fail;
+ goto out;
}
}
@@ -4948,7 +4948,7 @@
chunk_printf(&trash, "\r\n0\r\n\r\n");
if (bi_putchk(si_ic(si), &trash) == -1) {
si->flags |= SI_FL_WAIT_ROOM;
- goto fail;
+ goto out;
}
}
/* eat the whole request */
@@ -4967,15 +4967,14 @@
}
}
- fail:
+ out:
/* update all other flags and resync with the other side */
- si_update(si);
+ si_applet_done(si);
/* we don't want to expire timeouts while we're processing requests */
si_ic(si)->rex = TICK_ETERNITY;
si_oc(si)->wex = TICK_ETERNITY;
- out:
if (unlikely(si->state == SI_ST_DIS || si->state == SI_ST_CLO)) {
/* check that we have released everything then unregister */
stream_int_unregister_handler(si);
diff --git a/src/hlua.c b/src/hlua.c
index 68f0a39..fb006e6 100644
--- a/src/hlua.c
+++ b/src/hlua.c
@@ -1446,11 +1446,11 @@
si_ic(si)->flags |= CF_READ_NULL;
hlua_com_wake(&appctx->ctx.hlua.wake_on_read);
hlua_com_wake(&appctx->ctx.hlua.wake_on_write);
- return;
+ goto leave;
}
if (!(c->flags & CO_FL_CONNECTED))
- return;
+ goto leave;
/* This function is called after the connect. */
appctx->ctx.hlua.connected = 1;
@@ -1462,6 +1462,9 @@
/* Wake the tasks which wants to read if the buffer contains data. */
if (channel_is_empty(si_ic(si)))
hlua_com_wake(&appctx->ctx.hlua.wake_on_read);
+
+ leave:
+ si_applet_done(si);
}
/* This function is called when the "struct stream" is destroyed.
@@ -1631,7 +1634,7 @@
bo_skip(oc, len + skip_at_end);
/* Don't wait anything. */
- si_update(&socket->s->si[0]);
+ si_applet_done(&socket->s->si[0]);
/* If the pattern reclaim to read all the data
* in the connection, got out.
@@ -1808,7 +1811,7 @@
}
/* update buffers. */
- si_update(&socket->s->si[0]);
+ si_applet_done(&socket->s->si[0]);
socket->s->req.rex = TICK_ETERNITY;
socket->s->res.wex = TICK_ETERNITY;
diff --git a/src/peers.c b/src/peers.c
index 2a927e0..97a8728 100644
--- a/src/peers.c
+++ b/src/peers.c
@@ -1038,17 +1038,16 @@
si_shutw(si);
si_shutr(si);
si_ic(si)->flags |= CF_READ_NULL;
- goto quit;
+ goto out;
}
}
}
out:
- si_update(si);
+ si_applet_done(si);
si_oc(si)->flags |= CF_READ_DONTWAIT;
/* we don't want to expire timeouts while we're processing requests */
si_ic(si)->rex = TICK_ETERNITY;
si_oc(si)->wex = TICK_ETERNITY;
-quit:
return;
full:
si->flags |= SI_FL_WAIT_ROOM;
diff --git a/src/stream_interface.c b/src/stream_interface.c
index 37dad7e..2fdd94f 100644
--- a/src/stream_interface.c
+++ b/src/stream_interface.c
@@ -394,6 +394,7 @@
return NULL;
si->flags |= SI_FL_WAIT_DATA;
+ appctx_wakeup(appctx);
return si_appctx(si);
}
@@ -1569,6 +1570,8 @@
ic->rex = TICK_ETERNITY;
si->flags &= ~SI_FL_WAIT_ROOM;
+ /* Note: on shutr, we don't call the applet */
+
if (si->state != SI_ST_EST && si->state != SI_ST_CON)
return;
@@ -1581,10 +1584,6 @@
/* we want to immediately forward this close to the write side */
return stream_int_shutw_applet(si);
}
-
- /* note that if the task exists, it must unregister itself once it runs */
- if (!(si->flags & SI_FL_DONT_WAKE))
- task_wakeup(si_task(si), TASK_WOKEN_IO);
}
/*
@@ -1606,6 +1605,9 @@
oc->wex = TICK_ETERNITY;
si->flags &= ~SI_FL_WAIT_DATA;
+ /* on shutw we always wake the applet up */
+ appctx_wakeup(si_appctx(si));
+
switch (si->state) {
case SI_ST_EST:
/* we have to shut before closing, otherwise some short messages
@@ -1633,10 +1635,6 @@
ic->rex = TICK_ETERNITY;
si->exp = TICK_ETERNITY;
}
-
- /* note that if the task exists, it must unregister itself once it runs */
- if (!(si->flags & SI_FL_DONT_WAKE))
- task_wakeup(si_task(si), TASK_WOKEN_IO);
}
/* chk_rcv function for applets */
@@ -1650,17 +1648,14 @@
if (unlikely(si->state != SI_ST_EST || (ic->flags & (CF_SHUTR|CF_DONT_READ))))
return;
+ /* here we only wake the applet up if it was waiting for some room */
+ if (!(si->flags & SI_FL_WAIT_ROOM))
+ return;
- if (!channel_may_recv(ic) || ic->pipe) {
- /* stop reading */
- si->flags |= SI_FL_WAIT_ROOM;
- }
- else {
+ if (channel_may_recv(ic) && !ic->pipe) {
/* (re)start reading */
- si->flags &= ~SI_FL_WAIT_ROOM;
- if (!(si->flags & SI_FL_DONT_WAKE))
- task_wakeup(si_task(si), TASK_WOKEN_IO);
- }
+ appctx_wakeup(si_appctx(si));
+ }
}
/* chk_snd function for applets */
@@ -1675,19 +1670,18 @@
if (unlikely(si->state != SI_ST_EST || (oc->flags & CF_SHUTW)))
return;
- if (!(si->flags & SI_FL_WAIT_DATA) || /* not waiting for data */
- channel_is_empty(oc)) /* called with nothing to send ! */
+ /* we only wake the applet up if it was waiting for some data */
+
+ if (!(si->flags & SI_FL_WAIT_DATA))
return;
- /* Otherwise there are remaining data to be sent in the buffer,
- * so we tell the handler.
- */
- si->flags &= ~SI_FL_WAIT_DATA;
if (!tick_isset(oc->wex))
oc->wex = tick_add_ifset(now_ms, oc->wto);
- if (!(si->flags & SI_FL_DONT_WAKE))
- task_wakeup(si_task(si), TASK_WOKEN_IO);
+ if (!channel_is_empty(oc)) {
+ /* (re)start sending */
+ appctx_wakeup(si_appctx(si));
+ }
}
/*