MAJOR: stream_int: use a common stream_int_shut*() functions regardless of the data layer
Up to now, we had to use a shutr/shutw interface per data layer, which
basically means 3 distinct functions when we include SSL :
- generic stream_interface
- sock_raw
- sock_ssl
With this change, the code located in the stream_interface manages all the
stream_interface and buffer updates, and calls the data layer hooks when
needed.
At the moment, the socket layer hook had been implicitly considered as
being a regular socket, so the si_shut*() functions call the normal
shutdown() and EV_FD_CLR() functions on the fd if a socket layer is
defined. This may change in the future. The stream_int_shut*()
functions don't call EV_FD_CLR() so that they can later be embedded
in lower layers.
Thus, the si->data->shutr() is not called anymore and si->data->shutw()
is called to close the data layer only (eg: only for SSL).
Proceeding like this is very important because it's the only way to be
able not to rely on these functions when called from the connection
handlers, and call the data layers' instead.
diff --git a/src/stream_interface.c b/src/stream_interface.c
index 030c7c1..3ee0c4e 100644
--- a/src/stream_interface.c
+++ b/src/stream_interface.c
@@ -39,16 +39,14 @@
/* socket functions used when running a stream interface as a task */
static void stream_int_update(struct stream_interface *si);
static void stream_int_update_embedded(struct stream_interface *si);
-static void stream_int_shutr(struct stream_interface *si);
-static void stream_int_shutw(struct stream_interface *si);
static void stream_int_chk_rcv(struct stream_interface *si);
static void stream_int_chk_snd(struct stream_interface *si);
/* socket operations for embedded tasks */
struct sock_ops stream_int_embedded = {
.update = stream_int_update_embedded,
- .shutr = stream_int_shutr,
- .shutw = stream_int_shutw,
+ .shutr = NULL,
+ .shutw = NULL,
.chk_rcv = stream_int_chk_rcv,
.chk_snd = stream_int_chk_snd,
.read = NULL,
@@ -59,8 +57,8 @@
/* socket operations for external tasks */
struct sock_ops stream_int_task = {
.update = stream_int_update,
- .shutr = stream_int_shutr,
- .shutw = stream_int_shutw,
+ .shutr = NULL,
+ .shutw = NULL,
.chk_rcv = stream_int_chk_rcv,
.chk_snd = stream_int_chk_snd,
.read = NULL,
@@ -210,65 +208,134 @@
si->ib->flags &= ~BF_READ_DONTWAIT;
}
-/* default shutr function for scheduled tasks */
-static void stream_int_shutr(struct stream_interface *si)
+/*
+ * This function performs a shutdown-read on a stream interface in a connected
+ * or init state (it does nothing for other states). It either shuts the read
+ * side or marks itself as closed. The buffer flags are updated to reflect the
+ * new state. If the stream interface has SI_FL_NOHALF, we also forward the
+ * close to the write side. If a control layer is defined, then it is supposed
+ * to be a socket layer and file descriptors are then shutdown or closed
+ * accordingly. If no control layer is defined, then the SI is supposed to be
+ * an embedded one and the owner task is woken up if it exists. The function
+ * does not disable polling on the FD by itself, it returns non-zero instead
+ * if the caller needs to do so (except when the FD is deleted where this is
+ * implicit).
+ */
+int stream_int_shutr(struct stream_interface *si)
{
- DPRINTF(stderr, "%s: si=%p, si->state=%d ib->flags=%08x ob->flags=%08x\n",
- __FUNCTION__,
- si, si->state, si->ib->flags, si->ob->flags);
+ struct connection *conn = &si->conn;
si->ib->flags &= ~BF_SHUTR_NOW;
if (si->ib->flags & BF_SHUTR)
- return;
+ return 0;
si->ib->flags |= BF_SHUTR;
si->ib->rex = TICK_ETERNITY;
si->flags &= ~SI_FL_WAIT_ROOM;
if (si->state != SI_ST_EST && si->state != SI_ST_CON)
- return;
+ return 0;
if (si->ob->flags & BF_SHUTW) {
+ conn_data_close(&si->conn);
+ if (conn->ctrl)
+ fd_delete(si_fd(si));
si->state = SI_ST_DIS;
si->exp = TICK_ETERNITY;
- conn_data_close(&si->conn);
if (si->release)
si->release(si);
}
+ else if (si->flags & SI_FL_NOHALF) {
+ /* we want to immediately forward this close to the write side */
+ return stream_int_shutw(si);
+ }
+ else if (conn->ctrl) {
+ /* we want the caller to disable polling on this FD */
+ return 1;
+ }
- /* note that if the task exist, it must unregister itself once it runs */
- if (!(si->flags & SI_FL_DONT_WAKE) && si->owner)
+ /* note that if the task exists, it must unregister itself once it runs */
+ if (!conn->ctrl && !(si->flags & SI_FL_DONT_WAKE) && si->owner)
task_wakeup(si->owner, TASK_WOKEN_IO);
+ return 0;
}
-/* default shutw function for scheduled tasks */
-static void stream_int_shutw(struct stream_interface *si)
+/*
+ * This function performs a shutdown-write on a stream interface in a connected or
+ * init state (it does nothing for other states). It either shuts the write side
+ * or marks itself as closed. The buffer flags are updated to reflect the new state.
+ * It does also close everything if the SI was marked as being in error state. If
+ * there is a data-layer shutdown, it is called. If a control layer is defined, then
+ * it is supposed to be a socket layer and file descriptors are then shutdown or
+ * closed accordingly. If no control layer is defined, then the SI is supposed to
+ * be an embedded one and the owner task is woken up if it exists. The function
+ * does not disable polling on the FD by itself, it returns non-zero instead if
+ * the caller needs to do so (except when the FD is deleted where this is implicit).
+ */
+int stream_int_shutw(struct stream_interface *si)
{
- DPRINTF(stderr, "%s: si=%p, si->state=%d ib->flags=%08x ob->flags=%08x\n",
- __FUNCTION__,
- si, si->state, si->ib->flags, si->ob->flags);
+ struct connection *conn = &si->conn;
si->ob->flags &= ~BF_SHUTW_NOW;
if (si->ob->flags & BF_SHUTW)
- return;
+ return 0;
si->ob->flags |= BF_SHUTW;
si->ob->wex = TICK_ETERNITY;
si->flags &= ~SI_FL_WAIT_DATA;
switch (si->state) {
case SI_ST_EST:
- if (!(si->ib->flags & (BF_SHUTR|BF_DONT_READ)))
- break;
+ /* we have to shut before closing, otherwise some short messages
+ * may never leave the system, especially when there are remaining
+ * unread data in the socket input buffer, or when nolinger is set.
+ * However, if SI_FL_NOLINGER is explicitly set, we know there is
+ * no risk so we close both sides immediately.
+ */
+ if (si->flags & SI_FL_ERR) {
+ /* quick close, the socket is already shut. Remove pending flags. */
+ si->flags &= ~SI_FL_NOLINGER;
+ } else if (si->flags & SI_FL_NOLINGER) {
+ si->flags &= ~SI_FL_NOLINGER;
+ if (conn->ctrl) {
+ setsockopt(si_fd(si), SOL_SOCKET, SO_LINGER,
+ (struct linger *) &nolinger, sizeof(struct linger));
+ }
+ /* unclean data-layer shutdown */
+ if (conn->data && conn->data->shutw)
+ conn->data->shutw(conn, 0);
+ } else {
+ /* clean data-layer shutdown */
+ if (conn->data && conn->data->shutw)
+ conn->data->shutw(conn, 1);
+
+ if (!(si->flags & SI_FL_NOHALF)) {
+ /* We shutdown transport layer */
+ if (conn->ctrl)
+ shutdown(si_fd(si), SHUT_WR);
+
+ if (!(si->ib->flags & (BF_SHUTR|BF_DONT_READ))) {
+ /* OK just a shutw, but we want the caller
+ * to disable polling on this FD if exists.
+ */
+ return !!conn->ctrl;
+ }
+ }
+ }
/* fall through */
case SI_ST_CON:
+ /* we may have to close a pending connection, and mark the
+ * response buffer as shutr
+ */
+ conn_data_close(&si->conn);
+ if (conn->ctrl)
+ fd_delete(si_fd(si));
+ /* fall through */
case SI_ST_CER:
case SI_ST_QUE:
case SI_ST_TAR:
si->state = SI_ST_DIS;
- /* fall through */
- conn_data_close(&si->conn);
if (si->release)
si->release(si);
default:
@@ -278,9 +345,10 @@
si->exp = TICK_ETERNITY;
}
- /* note that if the task exist, it must unregister itself once it runs */
- if (!(si->flags & SI_FL_DONT_WAKE) && si->owner)
+ /* note that if the task exists, it must unregister itself once it runs */
+ if (!conn->ctrl && !(si->flags & SI_FL_DONT_WAKE) && si->owner)
task_wakeup(si->owner, TASK_WOKEN_IO);
+ return 0;
}
/* default chk_rcv function for scheduled tasks */
@@ -513,7 +581,7 @@
if (si->ob->flags & BF_OUT_EMPTY) {
if (((si->ob->flags & (BF_SHUTW|BF_HIJACK|BF_SHUTW_NOW)) == BF_SHUTW_NOW) &&
(si->state == SI_ST_EST))
- si_shutw(si);
+ stream_int_shutw(si);
EV_FD_CLR(fd, DIR_WR);
si->ob->wex = TICK_ETERNITY;
}