CLEANUP: connection: split sock_ops into data_ops, app_cp and si_ops

Some parts of the sock_ops structure were only used by the stream
interface and have been moved into si_ops. Some of them were callbacks
to the stream interface from the connection and have been moved into
app_cp as they're the application seen from the connection (later,
health-checks will need to use them). The rest has moved to data_ops.

Normally at this point the connection could live without knowing about
stream interfaces at all.

include/proto/raw_sock.h

@@ -24,7 +24,7 @@
 
 #include <types/stream_interface.h>
 
-extern struct sock_ops raw_sock;
+extern struct data_ops raw_sock;
 
 #endif /* _PROTO_RAW_SOCK_H */
 

include/proto/stream_interface.h

@@ -36,17 +36,16 @@ 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);
-void stream_int_update_conn(struct stream_interface *si);
 int stream_int_shutr(struct stream_interface *si);
 int stream_int_shutw(struct stream_interface *si);
-void stream_int_chk_rcv_conn(struct stream_interface *si);
-void stream_int_chk_snd_conn(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 sock_ops stream_int_embedded;
-extern struct sock_ops stream_int_task;
+extern struct si_ops si_embedded_ops;
+extern struct si_ops si_task_ops;
+extern struct si_ops si_conn_ops;
+extern struct app_cb si_conn_cb;
 
 struct task *stream_int_register_handler(struct stream_interface *si,
 					 struct si_applet *app);
@@ -59,11 +58,6 @@ static inline const struct protocol *si_ctrl(struct stream_interface *si)
 	return si->conn.ctrl;
 }
 
-static inline const struct sock_ops *si_data(struct stream_interface *si)
-{
-	return si->conn.data;
-}
-
 static inline int si_fd(struct stream_interface *si)
 {
 	return si->conn.t.sock.fd;
@@ -130,13 +124,35 @@ static inline struct listener *target_client(struct target *t)
 	return t->ptr.l;
 }
 
-static inline void stream_interface_prepare(struct stream_interface *si, const struct sock_ops *ops)
+static inline void si_prepare_conn(struct stream_interface *si, const struct protocol *ctrl, const struct data_ops *ops)
 {
+	si->ops = &si_conn_ops;
+	si->conn.app_cb = &si_conn_cb;
+	si->conn.ctrl = ctrl;
 	si->conn.data = ops;
 	si->conn.data_st = 0;
 	si->conn.data_ctx = NULL;
 }
 
+static inline void si_prepare_embedded(struct stream_interface *si)
+{
+	si->ops = &si_embedded_ops;
+	si->conn.app_cb = NULL;
+	si->conn.ctrl = NULL;
+	si->conn.data = NULL;
+	si->conn.data_st = 0;
+	si->conn.data_ctx = NULL;
+}
+
+static inline void si_prepare_task(struct stream_interface *si)
+{
+	si->ops = &si_task_ops;
+	si->conn.app_cb = NULL;
+	si->conn.ctrl = NULL;
+	si->conn.data = NULL;
+	si->conn.data_st = 0;
+	si->conn.data_ctx = NULL;
+}
 
 /* Retrieves the source address for the stream interface. */
 static inline void si_get_from_addr(struct stream_interface *si)
@@ -187,19 +203,19 @@ static inline void si_shutw(struct stream_interface *si)
 /* Calls the data state update on the stream interfaace */
 static inline void si_update(struct stream_interface *si)
 {
-	si_data(si)->update(si);
+	si->ops->update(si);
 }
 
 /* Calls chk_rcv on the connection using the data layer */
 static inline void si_chk_rcv(struct stream_interface *si)
 {
-	si_data(si)->chk_rcv(si);
+	si->ops->chk_rcv(si);
 }
 
 /* Calls chk_snd on the connection using the data layer */
 static inline void si_chk_snd(struct stream_interface *si)
 {
-	si_data(si)->chk_snd(si);
+	si->ops->chk_snd(si);
 }
 
 /* Calls chk_snd on the connection using the ctrl layer */

include/types/connection.h

@@ -28,8 +28,10 @@
 #include <common/config.h>
 
 /* referenced below */
-struct sock_ops;
 struct protocol;
+struct connection;
+struct buffer;
+struct pipe;
 
 /* Polling flags that are manipulated by I/O callbacks and handshake callbacks
  * indicate what they expect from a file descriptor at each layer. For each
@@ -108,6 +110,29 @@ enum {
 	CO_FL_CURR_WR_POL   = CO_FL_WR_POL << 28,  /* sending needs to poll first */
 };
 
+
+/* data_ops describes data-layer operations for a connection. They generally
+ * run over a socket-based control layer, but not always.
+ */
+struct data_ops {
+	int  (*rcv_buf)(struct connection *conn, struct buffer *buf, int count); /* recv callback */
+	int  (*snd_buf)(struct connection *conn, struct buffer *buf, int flags); /* send callback */
+	int  (*rcv_pipe)(struct connection *conn, struct pipe *pipe, unsigned int count); /* recv-to-pipe callback */
+	int  (*snd_pipe)(struct connection *conn, struct pipe *pipe); /* send-to-pipe callback */
+	void (*shutr)(struct connection *, int);    /* shutr function */
+	void (*shutw)(struct connection *, int);    /* shutw function */
+	void (*close)(struct connection *);         /* close the data channel on the connection */
+};
+
+/* app_cb describes read and write callbacks which are called upon detected I/O
+ * activity at the data layer. These callbacks are supposed to make use of the
+ * data_ops above to exchange data from/to buffers and pipes.
+ */
+struct app_cb {
+	void (*recv)(struct connection *conn);  /* application-layer recv callback */
+	void (*send)(struct connection *conn);  /* application-layer send callback */
+};
+
 /* This structure describes a connection with its methods and data.
  * A connection may be performed to proxy or server via a local or remote
  * socket, and can also be made to an internal applet. It can support
@@ -116,8 +141,9 @@ enum {
  * connections, but other methods for applets.
  */
 struct connection {
-	const struct sock_ops *data;  /* operations at the data layer */
-	const struct protocol *ctrl;  /* operations at the control layer, generally a protocol */
+	const struct data_ops *data;  /* operations at the data layer */
+	const struct protocol *ctrl;  /* operations at the socket layer */
+	const struct app_cb *app_cb;  /* application layer callbacks */
 	union {                       /* definitions which depend on connection type */
 		struct {              /*** information used by socket-based connections ***/
 			int fd;       /* file descriptor for a stream driver when known */

include/types/peers.h

@@ -74,7 +74,7 @@ struct peer {
 	time_t last_change;
 	struct sockaddr_storage addr;  /* peer address */
 	struct protocol *proto;	       /* peer address protocol */
-	struct sock_ops *sock;         /* peer socket operations */
+	struct data_ops *data;         /* peer socket operations at data layer */
 	void *sock_init_arg;           /* socket operations's opaque init argument if needed */
 	struct peer *next;	  /* next peer in the list */
 };

include/types/protocols.h

@@ -102,7 +102,7 @@ struct listener {
 	int options;			/* socket options : LI_O_* */
 	struct licounters *counters;	/* statistics counters */
 	struct protocol *proto;		/* protocol this listener belongs to */
-	struct sock_ops *sock;          /* listener socket operations */
+	struct data_ops *data;          /* data-layer operations operations for this socket */
 	int nbconn;			/* current number of connections on this listener */
 	int maxconn;			/* maximum connections allowed on this listener */
 	unsigned int backlog;		/* if set, listen backlog */

include/types/server.h

@@ -151,8 +151,7 @@ struct server {
 	int bind_hdr_occ;			/* occurrence number of header above: >0 = from first, <0 = from end, 0=disabled */
 #endif
 	struct protocol *proto;	                /* server address protocol */
-	struct sock_ops *sock;                  /* server socket operations */
-	void *sock_init_arg;                    /* socket operations's opaque init argument if needed */
+	struct data_ops *data;                  /* data-layer operations */
 	unsigned down_time;			/* total time the server was down */
 	time_t last_change;			/* last time, when the state was changed */
 	struct timeval check_start;		/* last health check start time */

include/types/stream_interface.h

@@ -91,12 +91,10 @@ enum {
 
 #define SI_FL_CAP_SPLICE (SI_FL_CAP_SPLTCP)
 
-struct buffer;
 struct server;
 struct proxy;
 struct si_applet;
 struct stream_interface;
-struct pipe;
 
 struct target {
 	int type;
@@ -110,19 +108,11 @@ struct target {
 	} ptr;
 };
 
-struct sock_ops {
+/* operations available on a stream-interface */
+struct si_ops {
 	void (*update)(struct stream_interface *);  /* I/O update function */
-	void (*shutr)(struct connection *, int);    /* shutr function */
-	void (*shutw)(struct connection *, int);    /* shutw function */
 	void (*chk_rcv)(struct stream_interface *); /* chk_rcv function */
 	void (*chk_snd)(struct stream_interface *); /* chk_snd function */
-	void (*read)(struct connection *conn);      /* read callback after poll() */
-	void (*write)(struct connection *conn);     /* write callback after poll() */
-	void (*close)(struct connection *);         /* close the data channel on the connection */
-	int  (*rcv_buf)(struct connection *conn, struct buffer *buf, int count); /* recv callback */
-	int  (*snd_buf)(struct connection *conn, struct buffer *buf, int flags); /* send callback */
-	int  (*rcv_pipe)(struct connection *conn, struct pipe *pipe, unsigned int count); /* recv-to-pipe callback */
-	int  (*snd_pipe)(struct connection *conn, struct pipe *pipe); /* send-to-pipe callback */
 };
 
 /* A stream interface has 3 parts :
@@ -148,6 +138,7 @@ struct stream_interface {
 	void *err_loc;          /* commonly the server, NULL when SI_ET_NONE */
 
 	struct connection conn; /* descriptor for a connection */
+	struct si_ops *ops;     /* general operations at the stream interface layer */
 
 	void (*release)(struct stream_interface *); /* handler to call after the last close() */
 

src/backend.c

@@ -984,13 +984,11 @@ int connect_server(struct session *s)
 
 	/* set the correct protocol on the output stream interface */
 	if (s->target.type == TARG_TYPE_SERVER) {
-		s->req->cons->conn.ctrl = target_srv(&s->target)->proto;
-		stream_interface_prepare(s->req->cons, target_srv(&s->target)->sock);
+		si_prepare_conn(s->req->cons, target_srv(&s->target)->proto, target_srv(&s->target)->data);
 	}
 	else if (s->target.type == TARG_TYPE_PROXY) {
 		/* proxies exclusively run on raw_sock right now */
-		s->req->cons->conn.ctrl = protocol_by_family(s->req->cons->addr.to.ss_family);
-		stream_interface_prepare(s->req->cons, &raw_sock);
+		si_prepare_conn(s->req->cons, protocol_by_family(s->req->cons->addr.to.ss_family), &raw_sock);
 		if (!si_ctrl(s->req->cons))
 			return SN_ERR_INTERNAL;
 	}

src/cfgparse.c

@@ -266,7 +266,7 @@ static int str2listener(char *str, struct proxy *curproxy, const char *file, int
 
 			l->fd = -1;
 			l->addr = ss;
-			l->sock = &raw_sock;
+			l->data = &raw_sock;
 			l->state = LI_INIT;
 
 			if (ss.ss_family == AF_INET) {
@@ -1275,7 +1275,7 @@ int cfg_parse_peers(const char *file, int linenum, char **args, int kwm)
 		}
 		newpeer->addr = *sk;
 		newpeer->proto = protocol_by_family(newpeer->addr.ss_family);
-		newpeer->sock  = &raw_sock;
+		newpeer->data  = &raw_sock;
 		newpeer->sock_init_arg = NULL;
 
 		if (!sk) {
@@ -4093,8 +4093,7 @@ stats_error_parsing:
 			}
 			newsrv->addr = *sk;
 			newsrv->proto = protocol_by_family(newsrv->addr.ss_family);
-			newsrv->sock  = &raw_sock;
-			newsrv->sock_init_arg = NULL;
+			newsrv->data  = &raw_sock;
 
 			if (!sk) {
 				Alert("parsing [%s:%d] : Unknown protocol family %d '%s'\n",

src/connection.c

@@ -13,8 +13,7 @@
 #include <common/compat.h>
 #include <common/config.h>
 
-#include <types/connection.h>
-
+#include <proto/connection.h>
 #include <proto/proto_tcp.h>
 #include <proto/stream_interface.h>
 
@@ -49,7 +48,7 @@ int conn_fd_handler(int fd)
 		__conn_sock_stop_both(conn);
 
 	if (fdtab[fd].ev & (FD_POLL_IN | FD_POLL_HUP | FD_POLL_ERR))
-		conn->data->read(conn);
+		conn->app_cb->recv(conn);
 
 	if (unlikely(conn->flags & CO_FL_ERROR))
 		goto leave;
@@ -61,7 +60,7 @@ int conn_fd_handler(int fd)
 		goto process_handshake;
 
 	if (fdtab[fd].ev & (FD_POLL_OUT | FD_POLL_ERR))
-		conn->data->write(conn);
+		conn->app_cb->send(conn);
 
 	if (unlikely(conn->flags & CO_FL_ERROR))
 		goto leave;

src/dumpstats.c

@@ -277,7 +277,7 @@ static int stats_parse_global(char **args, int section_type, struct proxy *curpx
 			}
 		}
 
-		global.stats_sock.sock = &raw_sock;
+		global.stats_sock.data = &raw_sock;
 		uxst_add_listener(&global.stats_sock);
 		global.maxsock++;
 	}

src/peers.c

@@ -1156,7 +1156,6 @@ static struct session *peer_session_create(struct peer *peer, struct peer_sessio
 	s->si[0].state = s->si[0].prev_state = SI_ST_EST;
 	s->si[0].err_type = SI_ET_NONE;
 	s->si[0].err_loc = NULL;
-	s->si[0].conn.ctrl = NULL;
 	s->si[0].release = NULL;
 	s->si[0].send_proxy_ofs = 0;
 	set_target_client(&s->si[0].target, l);
@@ -1178,11 +1177,10 @@ static struct session *peer_session_create(struct peer *peer, struct peer_sessio
 	s->si[1].conn_retries = p->conn_retries;
 	s->si[1].err_type = SI_ET_NONE;
 	s->si[1].err_loc = NULL;
-	s->si[1].conn.ctrl = peer->proto;
 	s->si[1].release = NULL;
 	s->si[1].send_proxy_ofs = 0;
 	set_target_proxy(&s->si[1].target, s->be);
-	stream_interface_prepare(&s->si[1], peer->sock);
+	si_prepare_conn(&s->si[1], peer->proto, peer->data);
 	s->si[1].exp = TICK_ETERNITY;
 	s->si[1].flags = SI_FL_NONE;
 	if (s->be->options2 & PR_O2_INDEPSTR)

src/raw_sock.c

@@ -318,22 +318,17 @@ static int raw_sock_from_buf(struct connection *conn, struct buffer *buf, int fl
 }
 
 
-/* stream sock operations */
-struct sock_ops raw_sock = {
-	.update  = stream_int_update_conn,
-	.shutr   = NULL,
-	.shutw   = NULL,
-	.chk_rcv = stream_int_chk_rcv_conn,
-	.chk_snd = stream_int_chk_snd_conn,
-	.read    = si_conn_recv_cb,
-	.write   = si_conn_send_cb,
-	.snd_buf = raw_sock_from_buf,
-	.rcv_buf = raw_sock_to_buf,
+/* data-layer operations for RAW sockets */
+struct data_ops raw_sock = {
+	.snd_buf  = raw_sock_from_buf,
+	.rcv_buf  = raw_sock_to_buf,
 #if defined(CONFIG_HAP_LINUX_SPLICE)
 	.rcv_pipe = raw_sock_to_pipe,
 	.snd_pipe = raw_sock_from_pipe,
 #endif
-	.close   = NULL,
+	.shutr    = NULL,
+	.shutw    = NULL,
+	.close    = NULL,
 };
 
 /*

src/session.c

@@ -180,7 +180,7 @@ int session_accept(struct listener *l, int cfd, struct sockaddr_storage *addr)
 		s->si[0].flags |= SI_FL_INDEP_STR;
 
 	/* add the various callbacks */
-	stream_interface_prepare(&s->si[0], l->sock);
+	si_prepare_conn(&s->si[0], l->proto, l->data);
 
 	if ((s->si[0].conn.data->rcv_pipe && s->si[0].conn.data->snd_pipe) &&
 	    (addr->ss_family == AF_INET || addr->ss_family == AF_INET6))
@@ -196,11 +196,10 @@ int session_accept(struct listener *l, int cfd, struct sockaddr_storage *addr)
 	s->si[1].err_type  = SI_ET_NONE;
 	s->si[1].conn_retries = 0;  /* used for logging too */
 	s->si[1].err_loc   = NULL;
-	s->si[1].conn.ctrl = NULL;
 	s->si[1].release   = NULL;
 	s->si[1].send_proxy_ofs = 0;
 	clear_target(&s->si[1].target);
-	stream_interface_prepare(&s->si[1], &stream_int_embedded);
+	si_prepare_embedded(&s->si[1]);
 	s->si[1].exp       = TICK_ETERNITY;
 	s->si[1].flags     = SI_FL_NONE;
 

src/stream_interface.c

@@ -41,29 +41,34 @@ static void stream_int_update(struct stream_interface *si);
 static void stream_int_update_embedded(struct stream_interface *si);
 static void stream_int_chk_rcv(struct stream_interface *si);
 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);
 
-/* socket operations for embedded tasks */
-struct sock_ops stream_int_embedded = {
+/* stream-interface operations for embedded tasks */
+struct si_ops si_embedded_ops = {
 	.update  = stream_int_update_embedded,
-	.shutr   = NULL,
-	.shutw   = NULL,
 	.chk_rcv = stream_int_chk_rcv,
 	.chk_snd = stream_int_chk_snd,
-	.read    = NULL,
-	.write   = NULL,
-	.close   = NULL,
 };
 
-/* socket operations for external tasks */
-struct sock_ops stream_int_task = {
+/* stream-interface operations for external tasks */
+struct si_ops si_task_ops = {
 	.update  = stream_int_update,
-	.shutr   = NULL,
-	.shutw   = NULL,
 	.chk_rcv = stream_int_chk_rcv,
 	.chk_snd = stream_int_chk_snd,
-	.read    = NULL,
-	.write   = NULL,
-	.close   = NULL,
+};
+
+/* stream-interface operations for connections */
+struct si_ops si_conn_ops = {
+	.update  = stream_int_update_conn,
+	.chk_rcv = stream_int_chk_rcv_conn,
+	.chk_snd = stream_int_chk_snd_conn,
+};
+
+struct app_cb si_conn_cb = {
+	.recv    = si_conn_recv_cb,
+	.send    = si_conn_send_cb,
 };
 
 /*
@@ -415,8 +420,7 @@ struct task *stream_int_register_handler(struct stream_interface *si, struct si_
 {
 	DPRINTF(stderr, "registering handler %p for si %p (was %p)\n", app, si, si->owner);
 
-	stream_interface_prepare(si, &stream_int_embedded);
-	si->conn.ctrl = NULL;
+	si_prepare_embedded(si);
 	set_target_applet(&si->target, app);
 	si->release   = app->release;
 	si->flags |= SI_FL_WAIT_DATA;
@@ -437,8 +441,7 @@ struct task *stream_int_register_handler_task(struct stream_interface *si,
 
 	DPRINTF(stderr, "registering handler %p for si %p (was %p)\n", fct, si, si->owner);
 
-	stream_interface_prepare(si, &stream_int_task);
-	si->conn.ctrl = NULL;
+	si_prepare_task(si);
 	clear_target(&si->target);
 	si->release   = NULL;
 	si->flags |= SI_FL_WAIT_DATA;
@@ -835,7 +838,7 @@ void stream_int_update_conn(struct stream_interface *si)
  * timeouts, so that we can still check them later at wake-up. This function is
  * dedicated to connection-based stream interfaces.
  */
-void stream_int_chk_rcv_conn(struct stream_interface *si)
+static void stream_int_chk_rcv_conn(struct stream_interface *si)
 {
 	struct channel *ib = si->ib;
 
@@ -866,7 +869,7 @@ void stream_int_chk_rcv_conn(struct stream_interface *si)
  * for data in the buffer. Note that it intentionally does not update timeouts,
  * so that we can still check them later at wake-up.
  */
-void stream_int_chk_snd_conn(struct stream_interface *si)
+static void stream_int_chk_snd_conn(struct stream_interface *si)
 {
 	struct channel *ob = si->ob;
 
@@ -1275,7 +1278,6 @@ void stream_sock_read0(struct stream_interface *si)
 	return;
 }
 
-
 /*
  * Local variables:
  *  c-indent-level: 8