gamzinav/haproxy
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

MEDIUM: stconn: merge the app_ops and the data_cb fields

Browse Source 
For historical reasons (stream-interface and connections), we used to
require two independent fields for the application level callbacks and
the transport-level functions. Over time the distinction faded away so
much that the low-level functions became specific to the application
and conversely. For example, applets may only work with streams on top
since they rely on the channels, and the stream-level functions differ
between applets and connections. Right now the application level only
contains a wake() callback and the low-level ones contain the functions
that act at the lower level to perform the shutr/shutw and at the upper
level to notify about readability and writability. Let's just merge them
together into a single set and get rid of this confusing distinction.
Note that the check ops do not define any app-level function since these
are only called by streams.
This commit is contained in:
Willy Tarreau 2022-05-18 10:17:16 +02:00
parent c086960a03
commit 2f2318df87
12 changed files with 63 additions and 79 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

23
include/haproxy/conn_stream-t.h
View File

@ -136,19 +136,6 @@ enum cs_state_bit {
struct stconn;
/* cs_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. The <wake> callback is used to report activity
* at the transport layer, which can be a connection opening/close, or any
* data movement. It may abort a connection by returning < 0.
*/
struct data_cb {
int (*wake)(struct stconn *sc); /* data-layer callback to report activity */
char name[8]; /* data layer name, zero-terminated */
};
/* A Stream Endpoint Descriptor (sedesc) is the link between the stream
* connector (ex. conn_stream) and the Stream Endpoint (mux or appctx).
* It always exists for either of them, and binds them together. It also
@ -173,12 +160,17 @@ struct sedesc {
unsigned int flags;
};
/* operations available on a stream connector */
/* sc_app_ops describes the application layer's operations and notification
* callbacks when I/O activity is reported and to use to perform shutr/shutw.
* There are very few combinations in practice (strm/chk <-> none/mux/applet).
*/
struct sc_app_ops {
void (*chk_rcv)(struct stconn *); /* chk_rcv function, may not be null */
void (*chk_snd)(struct stconn *); /* chk_snd function, may not be null */
void (*shutr)(struct stconn *); /* shut read function, may not be null */
void (*shutw)(struct stconn *); /* shut write function, may not be null */
int (*wake)(struct stconn *); /* data-layer callback to report activity */
char name[8]; /* data layer name, zero-terminated */
};
/*
@ -194,8 +186,7 @@ struct stconn {
struct wait_event wait_event; /* We're in a wait list */
struct sedesc *sedesc; /* points to the stream endpoint descriptor */
enum obj_type *app; /* points to the applicative point (stream or check) */
const struct data_cb *data_cb; /* data layer callbacks. Must be set before xprt->init() */
struct sc_app_ops *ops; /* general operations used at the app layer */
const struct sc_app_ops *app_ops; /* general operations used at the app layer */
struct sockaddr_storage *src; /* source address (pool), when known, otherwise NULL */
struct sockaddr_storage *dst; /* destination address (pool), when known, otherwise NULL */
};

4
include/haproxy/conn_stream.h
View File

@ -217,9 +217,9 @@ static inline struct check *cs_check(const struct stconn *cs)
}
static inline const char *cs_get_data_name(const struct stconn *cs)
{
if (!cs->data_cb)
if (!cs->app_ops)
return "NONE";
return cs->data_cb->name;
return cs->app_ops->name;
}
/* shut read */

16
include/haproxy/cs_utils.h
View File

@ -275,15 +275,15 @@ static inline void cs_must_kill_conn(struct stconn *cs)
/* Sends a shutr to the endpoint using the data layer */
static inline void cs_shutr(struct stconn *cs)
{
if (likely(cs->ops->shutr))
cs->ops->shutr(cs);
if (likely(cs->app_ops->shutr))
cs->app_ops->shutr(cs);
}
/* Sends a shutw to the endpoint using the data layer */
static inline void cs_shutw(struct stconn *cs)
{
if (likely(cs->ops->shutw))
cs->ops->shutw(cs);
if (likely(cs->app_ops->shutw))
cs->app_ops->shutw(cs);
}
/* This is to be used after making some room available in a channel. It will
@ -304,15 +304,15 @@ static inline void cs_chk_rcv(struct stconn *cs)
return;
sc_ep_set(cs, SE_FL_RX_WAIT_EP);
if (likely(cs->ops->chk_rcv))
cs->ops->chk_rcv(cs);
if (likely(cs->app_ops->chk_rcv))
cs->app_ops->chk_rcv(cs);
}
/* Calls chk_snd on the endpoint using the data layer */
static inline void cs_chk_snd(struct stconn *cs)
{
if (likely(cs->ops->chk_snd))
cs->ops->chk_snd(cs);
if (likely(cs->app_ops->chk_snd))
cs->app_ops->chk_snd(cs);
}
/* Combines both cs_update_rx() and cs_update_tx() at once */

2
src/applet.c
View File

@ -252,7 +252,7 @@ struct task *task_run_applet(struct task *t, void *context, unsigned int state)
stream_dump_and_crash(&app->obj_type, read_freq_ctr(&app->call_rate));
}
cs->data_cb->wake(cs);
cs->app_ops->wake(cs);
channel_release_buffer(cs_ic(cs), &app->buffer_wait);
return t;
}

6
src/check.c
View File

@ -140,12 +140,6 @@ struct trace_source trace_check = {
INITCALL1(STG_REGISTER, trace_register_source, TRACE_SOURCE);
struct data_cb check_conn_cb = {
.wake = wake_srv_chk,
.name = "CHCK",
};
/* Dummy frontend used to create all checks sessions. */
struct proxy checks_fe;

67
src/conn_stream.c
View File

@ -41,12 +41,19 @@ static void sc_app_shutw_applet(struct stconn *cs);
static void sc_app_chk_rcv_applet(struct stconn *cs);
static void sc_app_chk_snd_applet(struct stconn *cs);
static int cs_conn_process(struct stconn *cs);
static int cs_conn_recv(struct stconn *cs);
static int cs_conn_send(struct stconn *cs);
static int cs_applet_process(struct stconn *cs);
/* stream connector operations for connections */
struct sc_app_ops sc_app_conn_ops = {
.chk_rcv = sc_app_chk_rcv_conn,
.chk_snd = sc_app_chk_snd_conn,
.shutr = sc_app_shutr_conn,
.shutw = sc_app_shutw_conn,
.wake = cs_conn_process,
.name = "STRM",
};
/* stream connector operations for embedded tasks */
@ -55,31 +62,29 @@ struct sc_app_ops sc_app_embedded_ops = {
.chk_snd = sc_app_chk_snd,
.shutr = sc_app_shutr,
.shutw = sc_app_shutw,
.wake = NULL, /* may never be used */
.name = "NONE", /* may never be used */
};
/* stream connector operations for connections */
/* stream connector operations for applets */
struct sc_app_ops sc_app_applet_ops = {
.chk_rcv = sc_app_chk_rcv_applet,
.chk_snd = sc_app_chk_snd_applet,
.shutr = sc_app_shutr_applet,
.shutw = sc_app_shutw_applet,
};
static int cs_conn_process(struct stconn *cs);
static int cs_conn_recv(struct stconn *cs);
static int cs_conn_send(struct stconn *cs);
static int cs_applet_process(struct stconn *cs);
struct data_cb cs_data_conn_cb = {
.wake = cs_conn_process,
.name = "STRM",
};
struct data_cb cs_data_applet_cb = {
.wake = cs_applet_process,
.name = "STRM",
};
/* stream connector for health checks on connections */
struct sc_app_ops sc_app_check_ops = {
.chk_rcv = NULL,
.chk_snd = NULL,
.shutr = NULL,
.shutw = NULL,
.wake = wake_srv_chk,
.name = "CHCK",
};
/* Initializes an endpoint */
void sedesc_init(struct sedesc *sedesc)
@ -130,7 +135,7 @@ static struct stconn *cs_new(struct sedesc *sedesc)
cs->state = SC_ST_INI;
cs->hcto = TICK_ETERNITY;
cs->app = NULL;
cs->data_cb = NULL;
cs->app_ops = NULL;
cs->src = NULL;
cs->dst = NULL;
cs->wait_event.tasklet = NULL;
@ -185,8 +190,7 @@ struct stconn *cs_new_from_strm(struct stream *strm, unsigned int flags)
cs->flags |= flags;
sc_ep_set(cs, SE_FL_DETACHED);
cs->app = &strm->obj_type;
cs->ops = &sc_app_embedded_ops;
cs->data_cb = NULL;
cs->app_ops = &sc_app_embedded_ops;
return cs;
}
@ -204,7 +208,7 @@ struct stconn *cs_new_from_check(struct check *check, unsigned int flags)
cs->flags |= flags;
sc_ep_set(cs, SE_FL_DETACHED);
cs->app = &check->obj_type;
cs->data_cb = &check_conn_cb;
cs->app_ops = &sc_app_check_ops;
return cs;
}
@ -264,8 +268,7 @@ int cs_attach_mux(struct stconn *cs, void *endp, void *ctx)
cs->wait_event.events = 0;
}
cs->ops = &sc_app_conn_ops;
cs->data_cb = &cs_data_conn_cb;
cs->app_ops = &sc_app_conn_ops;
}
else if (cs_check(cs)) {
if (!cs->wait_event.tasklet) {
@ -277,7 +280,7 @@ int cs_attach_mux(struct stconn *cs, void *endp, void *ctx)
cs->wait_event.events = 0;
}
cs->data_cb = &check_conn_cb;
cs->app_ops = &sc_app_check_ops;
}
return 0;
}
@ -292,10 +295,8 @@ static void cs_attach_applet(struct stconn *cs, void *endp)
cs->sedesc->se = endp;
sc_ep_set(cs, SE_FL_T_APPLET);
sc_ep_clr(cs, SE_FL_DETACHED);
if (cs_strm(cs)) {
cs->ops = &sc_app_applet_ops;
cs->data_cb = &cs_data_applet_cb;
}
if (cs_strm(cs))
cs->app_ops = &sc_app_applet_ops;
}
/* Attaches a stconn to a app layer and sets the relevant
@ -315,16 +316,13 @@ int cs_attach_strm(struct stconn *cs, struct stream *strm)
cs->wait_event.tasklet->context = cs;
cs->wait_event.events = 0;
cs->ops = &sc_app_conn_ops;
cs->data_cb = &cs_data_conn_cb;
cs->app_ops = &sc_app_conn_ops;
}
else if (sc_ep_test(cs, SE_FL_T_APPLET)) {
cs->ops = &sc_app_applet_ops;
cs->data_cb = &cs_data_applet_cb;
cs->app_ops = &sc_app_applet_ops;
}
else {
cs->ops = &sc_app_embedded_ops;
cs->data_cb = NULL;
cs->app_ops = &sc_app_embedded_ops;
}
return 0;
}
@ -393,8 +391,9 @@ static void cs_detach_endp(struct stconn **csp)
*/
cs->flags &= SC_FL_ISBACK;
if (cs_strm(cs))
cs->ops = &sc_app_embedded_ops;
cs->data_cb = NULL;
cs->app_ops = &sc_app_embedded_ops;
else
cs->app_ops = NULL;
cs_free_cond(csp);
}
@ -409,7 +408,7 @@ static void cs_detach_app(struct stconn **csp)
return;
cs->app = NULL;
cs->data_cb = NULL;
cs->app_ops = NULL;
sockaddr_free(&cs->src);
sockaddr_free(&cs->dst);

2
src/connection.c
View File

@ -91,7 +91,7 @@ int conn_create_mux(struct connection *conn)
return 0;
fail:
/* let the upper layer know the connection failed */
cs->data_cb->wake(cs);
cs->app_ops->wake(cs);
return -1;
} else
return conn_complete_session(conn);

4
src/mux_fcgi.c
View File

@ -977,9 +977,9 @@ static void fcgi_strm_alert(struct fcgi_strm *fstrm)
fcgi_strm_notify_recv(fstrm);
fcgi_strm_notify_send(fstrm);
}
else if (fcgi_strm_sc(fstrm) && fcgi_strm_sc(fstrm)->data_cb->wake != NULL) {
else if (fcgi_strm_sc(fstrm) && fcgi_strm_sc(fstrm)->app_ops->wake != NULL) {
TRACE_POINT(FCGI_EV_STRM_WAKE, fstrm->fconn->conn, fstrm);
fcgi_strm_sc(fstrm)->data_cb->wake(fcgi_strm_sc(fstrm));
fcgi_strm_sc(fstrm)->app_ops->wake(fcgi_strm_sc(fstrm));
}
}

4
src/mux_h1.c
View File

@ -2625,9 +2625,9 @@ static void h1_alert(struct h1s *h1s)
h1_wake_stream_for_recv(h1s);
h1_wake_stream_for_send(h1s);
}
else if (h1s_sc(h1s) && h1s_sc(h1s)->data_cb->wake != NULL) {
else if (h1s_sc(h1s) && h1s_sc(h1s)->app_ops->wake != NULL) {
TRACE_POINT(H1_EV_STRM_WAKE, h1s->h1c->conn, h1s);
h1s_sc(h1s)->data_cb->wake(h1s_sc(h1s));
h1s_sc(h1s)->app_ops->wake(h1s_sc(h1s));
}
}

4
src/mux_h2.c
View File

@ -1334,9 +1334,9 @@ static void __maybe_unused h2s_alert(struct h2s *h2s)
h2s_notify_recv(h2s);
h2s_notify_send(h2s);
}
else if (h2s_sc(h2s) && h2s_sc(h2s)->data_cb->wake != NULL) {
else if (h2s_sc(h2s) && h2s_sc(h2s)->app_ops->wake != NULL) {
TRACE_POINT(H2_EV_STRM_WAKE, h2s->h2c->conn, h2s);
h2s_sc(h2s)->data_cb->wake(h2s_sc(h2s));
h2s_sc(h2s)->app_ops->wake(h2s_sc(h2s));
}
TRACE_LEAVE(H2_EV_H2S_WAKE, h2s->h2c->conn, h2s);

6
src/mux_pt.c
View File

@ -252,8 +252,8 @@ struct task *mux_pt_io_cb(struct task *t, void *tctx, unsigned int status)
ctx->conn->subs->events = 0;
tasklet_wakeup(ctx->conn->subs->tasklet);
ctx->conn->subs = NULL;
} else if (pt_sc(ctx)->data_cb->wake)
pt_sc(ctx)->data_cb->wake(pt_sc(ctx));
} else if (pt_sc(ctx)->app_ops->wake)
pt_sc(ctx)->app_ops->wake(pt_sc(ctx));
TRACE_DEVEL("leaving waking up CS", PT_EV_CONN_WAKE, ctx->conn);
return t;
}
@ -349,7 +349,7 @@ static int mux_pt_wake(struct connection *conn)
TRACE_ENTER(PT_EV_CONN_WAKE, ctx->conn);
if (!se_fl_test(ctx->endp, SE_FL_ORPHAN)) {
ret = pt_sc(ctx)->data_cb->wake ? pt_sc(ctx)->data_cb->wake(pt_sc(ctx)) : 0;
ret = pt_sc(ctx)->app_ops->wake ? pt_sc(ctx)->app_ops->wake(pt_sc(ctx)) : 0;
if (ret < 0) {
TRACE_DEVEL("leaving waking up CS", PT_EV_CONN_WAKE, ctx->conn);

4
src/mux_quic.c
View File

@ -1594,8 +1594,8 @@ static int qc_wake_some_streams(struct qcc *qcc)
qcs_notify_recv(qcs);
qcs_notify_send(qcs);
}
else if (qcs->endp->sc->data_cb->wake) {
qcs->endp->sc->data_cb->wake(qcs->endp->sc);
else if (qcs->endp->sc->app_ops->wake) {
qcs->endp->sc->app_ops->wake(qcs->endp->sc);
}
}
}