MINOR: connection: provide a generic data layer wakeup callback

Instead of calling conn_notify_si() from the connection handler, we
now call data->wake(), which will allow us to use a different callback
with health checks.

Note that we still rely on a flag in order to decide whether or not
to call this function. The reason is that with embryonic sessions,
the callback is already initialized to si_conn_cb without the flag,
and we can't call the SI notify function in the leave path before
the stream interface is initialized.

This issue should be addressed by involving a different data_cb for
embryonic sessions and for stream interfaces, that would be changed
during session_complete() for the final data_cb.

include/proto/stream_interface.h

@@ -36,11 +36,8 @@ int stream_int_check_timeouts(struct stream_interface *si);
 void stream_int_report_error(struct stream_interface *si);
 void stream_int_retnclose(struct stream_interface *si, const struct chunk *msg);
 int conn_si_send_proxy(struct connection *conn, unsigned int flag);
-void conn_notify_si(struct connection *conn);
 int stream_int_shutr(struct stream_interface *si);
 int stream_int_shutw(struct stream_interface *si);
-void si_conn_recv_cb(struct connection *conn);
-void si_conn_send_cb(struct connection *conn);
 void stream_sock_read0(struct stream_interface *si);
 
 extern struct si_ops si_embedded_ops;
@@ -135,7 +132,7 @@ static inline int si_connect(struct stream_interface *si)
 		si->conn.flags |= CO_FL_SI_SEND_PROXY;
 
 	/* we need to be notified about connection establishment */
-	si->conn.flags |= CO_FL_NOTIFY_SI;
+	si->conn.flags |= CO_FL_WAKE_DATA;
 
 	/* we're in the process of establishing a connection */
 	si->state = SI_ST_CON;

include/types/connection.h

@@ -76,7 +76,7 @@ enum {
 	CO_FL_WAIT_L4_CONN  = 0x00000004,  /* waiting for L4 to be connected */
 	CO_FL_WAIT_L6_CONN  = 0x00000008,  /* waiting for L6 to be connected (eg: SSL) */
 
-	CO_FL_NOTIFY_SI     = 0x00000010,  /* notify stream interface about changes */
+	CO_FL_WAKE_DATA     = 0x00000010,  /* wake-up data layer upon activity at the transport layer */
 
 	/* flags below are used for connection handshakes */
 	CO_FL_SI_SEND_PROXY = 0x00000020,  /* send a valid PROXY protocol header */
@@ -172,11 +172,14 @@ struct xprt_ops {
 /* data_cb describes the data layer's recv and send callbacks which are called
  * when I/O activity was detected after the transport layer is ready. These
  * callbacks are supposed to make use of the xprt_ops above to exchange data
- * from/to buffers and pipes.
+ * from/to buffers and pipes. The <wake> callback is used to report activity
+ * at the transport layer, which can be a connection opening/close, or any
+ * data movement.
  */
 struct data_cb {
 	void (*recv)(struct connection *conn);  /* data-layer recv callback */
 	void (*send)(struct connection *conn);  /* data-layer send callback */
+	void (*wake)(struct connection *conn);  /* data-layer callback to report activity */
 };
 
 /* a target describes what is on the remote side of the connection. */

src/connection.c

@@ -110,8 +110,8 @@ int conn_fd_handler(int fd)
 		return 0;
 	}
 
-	if (conn->flags & CO_FL_NOTIFY_SI)
+	if (conn->flags & CO_FL_WAKE_DATA)
-		conn_notify_si(conn);
+		conn->data->wake(conn);
 
 	/* Last check, verify if the connection just established */
 	if (unlikely(!(conn->flags & (CO_FL_WAIT_L4_CONN | CO_FL_WAIT_L6_CONN | CO_FL_CONNECTED))))

src/session.c

@@ -462,7 +462,7 @@ int session_complete(struct session *s)
 	}
 
 	/* we want the connection handler to notify the stream interface about updates. */
-	s->si[0].conn.flags |= CO_FL_NOTIFY_SI;
+	s->si[0].conn.flags |= CO_FL_WAKE_DATA;
 
 	/* it is important not to call the wakeup function directly but to
 	 * pass through task_wakeup(), because this one knows how to apply

src/stream_interface.c

@@ -44,6 +44,9 @@ static void stream_int_chk_snd(struct stream_interface *si);
 static void stream_int_update_conn(struct stream_interface *si);
 static void stream_int_chk_rcv_conn(struct stream_interface *si);
 static void stream_int_chk_snd_conn(struct stream_interface *si);
+static void si_conn_recv_cb(struct connection *conn);
+static void si_conn_send_cb(struct connection *conn);
+static void si_conn_wake_cb(struct connection *conn);
 
 /* stream-interface operations for embedded tasks */
 struct si_ops si_embedded_ops = {
@@ -69,6 +72,7 @@ struct si_ops si_conn_ops = {
 struct data_cb si_conn_cb = {
 	.recv    = si_conn_recv_cb,
 	.send    = si_conn_send_cb,
+	.wake    = si_conn_wake_cb,
 };
 
 /*
@@ -554,12 +558,12 @@ int conn_si_send_proxy(struct connection *conn, unsigned int flag)
 }
 
 /* Callback to be used by connection I/O handlers upon completion. It differs from
- * the function below in that it is designed to be called by lower layers after I/O
+ * the update function in that it is designed to be called by lower layers after I/O
  * events have been completed. It will also try to wake the associated task up if
  * an important event requires special handling. It relies on the connection handler
  * to commit any polling updates.
  */
-void conn_notify_si(struct connection *conn)
+static void si_conn_wake_cb(struct connection *conn)
 {
 	struct stream_interface *si = conn->owner;
 
@@ -938,7 +942,7 @@ static void stream_int_chk_snd_conn(struct stream_interface *si)
  * into the buffer from the connection. It iterates over the transport layer's
  * rcv_buf function.
  */
-void si_conn_recv_cb(struct connection *conn)
+static void si_conn_recv_cb(struct connection *conn)
 {
 	struct stream_interface *si = conn->owner;
 	struct channel *b = si->ib;
@@ -1152,7 +1156,7 @@ void si_conn_recv_cb(struct connection *conn)
  * from the buffer to the connection. It iterates over the transport layer's
  * snd_buf function.
  */
-void si_conn_send_cb(struct connection *conn)
+static void si_conn_send_cb(struct connection *conn)
 {
 	struct stream_interface *si = conn->owner;
 	struct channel *b = si->ob;