MAJOR: remove the stream interface and task management code from sock_*
The socket data layer code must only focus on moving data between a
socket and a buffer. We need a special stream interface handler to
update the stream interface and the file descriptor status.
At the moment the code works but suffers from a race condition caused
by its API : the read/write callbacks still make use of the fd instead
of using the connection. And when a double shutdown is performed, a call
to ->write() after ->read() processed an error results in dereferencing
a NULL fdtab[]->owner. This is only a temporary issue which doesn't need
to be fixed now since this will automatically go away when the functions
change to use the connection instead.
diff --git a/src/stream_interface.c b/src/stream_interface.c
index 59eca1b..9d6f961 100644
--- a/src/stream_interface.c
+++ b/src/stream_interface.c
@@ -33,6 +33,8 @@
#include <proto/stream_interface.h>
#include <proto/task.h>
+#include <types/pipe.h>
+
/* 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);
@@ -486,6 +488,95 @@
return FD_WAIT_WRITE;
}
+/* function to be called on stream sockets after all I/O handlers */
+void stream_sock_update_conn(struct connection *conn)
+{
+ int fd = conn->t.sock.fd;
+ struct stream_interface *si = container_of(conn, struct stream_interface, conn);
+
+ 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);
+
+ /* process consumer side, only once if possible */
+ if (fdtab[fd].ev & (FD_POLL_OUT | FD_POLL_ERR)) {
+ 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);
+ EV_FD_CLR(fd, DIR_WR);
+ si->ob->wex = TICK_ETERNITY;
+ }
+
+ if ((si->ob->flags & (BF_FULL|BF_SHUTW|BF_SHUTW_NOW|BF_HIJACK)) == 0)
+ si->flags |= SI_FL_WAIT_DATA;
+
+ if (si->ob->flags & BF_WRITE_ACTIVITY) {
+ /* update timeouts if we have written something */
+ if ((si->ob->flags & (BF_OUT_EMPTY|BF_SHUTW|BF_WRITE_PARTIAL)) == BF_WRITE_PARTIAL)
+ if (tick_isset(si->ob->wex))
+ si->ob->wex = tick_add_ifset(now_ms, si->ob->wto);
+
+ if (!(si->flags & SI_FL_INDEP_STR))
+ if (tick_isset(si->ib->rex))
+ si->ib->rex = tick_add_ifset(now_ms, si->ib->rto);
+
+ if (likely((si->ob->flags & (BF_SHUTW|BF_WRITE_PARTIAL|BF_FULL|BF_DONT_READ)) == BF_WRITE_PARTIAL &&
+ (si->ob->prod->flags & SI_FL_WAIT_ROOM)))
+ si_chk_rcv(si->ob->prod);
+ }
+ }
+
+ /* process producer side, only once if possible */
+ if (fdtab[fd].ev & (FD_POLL_IN | FD_POLL_HUP | FD_POLL_ERR)) {
+ /* We might have some data the consumer is waiting for.
+ * We can do fast-forwarding, but we avoid doing this for partial
+ * buffers, because it is very likely that it will be done again
+ * immediately afterwards once the following data is parsed (eg:
+ * HTTP chunking).
+ */
+ if (((si->ib->flags & (BF_READ_PARTIAL|BF_OUT_EMPTY)) == BF_READ_PARTIAL) &&
+ (si->ib->pipe /* always try to send spliced data */ ||
+ (si->ib->i == 0 && (si->ib->cons->flags & SI_FL_WAIT_DATA)))) {
+ int last_len = si->ib->pipe ? si->ib->pipe->data : 0;
+
+ si_chk_snd(si->ib->cons);
+
+ /* check if the consumer has freed some space */
+ if (!(si->ib->flags & BF_FULL) &&
+ (!last_len || !si->ib->pipe || si->ib->pipe->data < last_len))
+ si->flags &= ~SI_FL_WAIT_ROOM;
+ }
+
+ if (si->flags & SI_FL_WAIT_ROOM) {
+ EV_FD_CLR(fd, DIR_RD);
+ si->ib->rex = TICK_ETERNITY;
+ }
+ else if ((si->ib->flags & (BF_SHUTR|BF_READ_PARTIAL|BF_FULL|BF_DONT_READ|BF_READ_NOEXP)) == BF_READ_PARTIAL) {
+ if (tick_isset(si->ib->rex))
+ si->ib->rex = tick_add_ifset(now_ms, si->ib->rto);
+ }
+ }
+
+ /* wake the task up only when needed */
+ if (/* changes on the production side */
+ (si->ib->flags & (BF_READ_NULL|BF_READ_ERROR)) ||
+ si->state != SI_ST_EST ||
+ (si->flags & SI_FL_ERR) ||
+ ((si->ib->flags & BF_READ_PARTIAL) &&
+ (!si->ib->to_forward || si->ib->cons->state != SI_ST_EST)) ||
+
+ /* changes on the consumption side */
+ (si->ob->flags & (BF_WRITE_NULL|BF_WRITE_ERROR)) ||
+ ((si->ob->flags & BF_WRITE_ACTIVITY) &&
+ ((si->ob->flags & BF_SHUTW) ||
+ si->ob->prod->state != SI_ST_EST ||
+ ((si->ob->flags & BF_OUT_EMPTY) && !si->ob->to_forward)))) {
+ task_wakeup(si->owner, TASK_WOKEN_IO);
+ }
+ if (si->ib->flags & BF_READ_ACTIVITY)
+ si->ib->flags &= ~BF_READ_DONTWAIT;
+}
/*
* Local variables: