blob: 0a2b805c3dc71e07f82526a667ea0272de48ed71 [file] [log] [blame]
Willy Tarreau53a47662017-08-28 10:53:00 +02001/*
2 * Pass-through mux-demux for connections
3 *
4 * Copyright 2017 Willy Tarreau <w@1wt.eu>
5 *
6 * This program is free software; you can redistribute it and/or
7 * modify it under the terms of the GNU General Public License
8 * as published by the Free Software Foundation; either version
9 * 2 of the License, or (at your option) any later version.
10 *
11 */
12
13#include <common/config.h>
Willy Tarreau0108d902018-11-25 19:14:37 +010014#include <common/initcall.h>
Willy Tarreau53a47662017-08-28 10:53:00 +020015#include <proto/connection.h>
16#include <proto/stream.h>
Olivier Houchardb6c32ee2018-11-05 18:28:43 +010017#include <proto/task.h>
18
Olivier Houchardb6c32ee2018-11-05 18:28:43 +010019struct mux_pt_ctx {
20 struct conn_stream *cs;
21 struct connection *conn;
22 struct wait_event wait_event;
23};
24
Willy Tarreau8ceae722018-11-26 11:58:30 +010025DECLARE_STATIC_POOL(pool_head_pt_ctx, "mux_pt", sizeof(struct mux_pt_ctx));
26
Olivier Houchardb6c32ee2018-11-05 18:28:43 +010027static void mux_pt_destroy(struct mux_pt_ctx *ctx)
28{
29 struct connection *conn = ctx->conn;
30
31 LIST_DEL(&conn->list);
32 conn_stop_tracking(conn);
33 conn_full_close(conn);
Olivier Houchard7c6f8b12018-11-13 16:48:36 +010034 tasklet_free(ctx->wait_event.task);
35 conn->mux = NULL;
36 conn->mux_ctx = NULL;
Olivier Houchardb6c32ee2018-11-05 18:28:43 +010037 if (conn->destroy_cb)
38 conn->destroy_cb(conn);
39 /* We don't bother unsubscribing here, as we're about to destroy
40 * both the connection and the mux_pt_ctx
41 */
42 conn_free(conn);
43 pool_free(pool_head_pt_ctx, ctx);
44}
45
46/* Callback, used when we get I/Os while in idle mode */
47static struct task *mux_pt_io_cb(struct task *t, void *tctx, unsigned short status)
48{
49 struct mux_pt_ctx *ctx = tctx;
50
51 conn_sock_drain(ctx->conn);
Olivier Houchard7c6f8b12018-11-13 16:48:36 +010052 if (ctx->conn->flags & (CO_FL_ERROR | CO_FL_SOCK_RD_SH | CO_FL_SOCK_WR_SH))
Olivier Houchardb6c32ee2018-11-05 18:28:43 +010053 mux_pt_destroy(ctx);
54 else
55 ctx->conn->xprt->subscribe(ctx->conn, SUB_CAN_RECV,
56 &ctx->wait_event);
57
58 return NULL;
59}
Willy Tarreau53a47662017-08-28 10:53:00 +020060
Olivier Houchard9aaf7782017-09-13 18:30:23 +020061/* Initialize the mux once it's attached. It is expected that conn->mux_ctx
62 * points to the existing conn_stream (for outgoing connections) or NULL (for
63 * incoming ones, in which case one will be allocated and a new stream will be
64 * instanciated). Returns < 0 on error.
Willy Tarreau53a47662017-08-28 10:53:00 +020065 */
Olivier Houchardf502aca2018-12-14 19:42:40 +010066static int mux_pt_init(struct connection *conn, struct proxy *prx, struct session *sess)
Willy Tarreau53a47662017-08-28 10:53:00 +020067{
Olivier Houchard9aaf7782017-09-13 18:30:23 +020068 struct conn_stream *cs = conn->mux_ctx;
Olivier Houchardb6c32ee2018-11-05 18:28:43 +010069 struct mux_pt_ctx *ctx = pool_alloc(pool_head_pt_ctx);
70
71 if (!ctx)
72 goto fail;
73
74 ctx->wait_event.task = tasklet_new();
75 if (!ctx->wait_event.task)
76 goto fail_free_ctx;
77 ctx->wait_event.task->context = ctx;
78 ctx->wait_event.task->process = mux_pt_io_cb;
79 ctx->wait_event.wait_reason = 0;
80 ctx->conn = conn;
Olivier Houchard9aaf7782017-09-13 18:30:23 +020081
82 if (!cs) {
83 cs = cs_new(conn);
84 if (!cs)
Olivier Houchardb6c32ee2018-11-05 18:28:43 +010085 goto fail_free_ctx;
Olivier Houchard9aaf7782017-09-13 18:30:23 +020086
87 if (stream_create_from_cs(cs) < 0)
88 goto fail_free;
89
Olivier Houchard9aaf7782017-09-13 18:30:23 +020090 }
Olivier Houchardb6c32ee2018-11-05 18:28:43 +010091 conn->mux_ctx = ctx;
92 ctx->cs = cs;
Olivier Houchard8706c812018-12-04 19:17:25 +010093 cs->flags |= CS_FL_RCV_MORE;
Willy Tarreau53a47662017-08-28 10:53:00 +020094 return 0;
Olivier Houchard9aaf7782017-09-13 18:30:23 +020095
96 fail_free:
97 cs_free(cs);
Olivier Houchardb6c32ee2018-11-05 18:28:43 +010098fail_free_ctx:
99 if (ctx->wait_event.task)
100 tasklet_free(ctx->wait_event.task);
101 pool_free(pool_head_pt_ctx, ctx);
Olivier Houchard9aaf7782017-09-13 18:30:23 +0200102 fail:
103 return -1;
Willy Tarreau53a47662017-08-28 10:53:00 +0200104}
105
106/* callback to be used by default for the pass-through mux. It calls the data
107 * layer wake() callback if it is set otherwise returns 0.
108 */
109static int mux_pt_wake(struct connection *conn)
110{
Olivier Houchardb6c32ee2018-11-05 18:28:43 +0100111 struct mux_pt_ctx *ctx = conn->mux_ctx;
112 struct conn_stream *cs = ctx->cs;
113 int ret = 0;
Olivier Houchard9aaf7782017-09-13 18:30:23 +0200114
Olivier Houchardb6c32ee2018-11-05 18:28:43 +0100115 if (cs) {
116 ret = cs->data_cb->wake ? cs->data_cb->wake(cs) : 0;
Olivier Houchard9aaf7782017-09-13 18:30:23 +0200117
Olivier Houchardb6c32ee2018-11-05 18:28:43 +0100118 if (ret < 0)
119 return ret;
120 } else {
121 conn_sock_drain(conn);
122 if (conn->flags & (CO_FL_ERROR | CO_FL_SOCK_RD_SH)) {
123 mux_pt_destroy(ctx);
124 return -1;
125 }
126 }
Willy Tarreauad7f0ad2018-08-24 15:48:59 +0200127
Olivier Houchard7fc96d52017-11-23 18:25:47 +0100128 /* If we had early data, and we're done with the handshake
129 * then whe know the data are safe, and we can remove the flag.
130 */
131 if ((conn->flags & (CO_FL_EARLY_DATA | CO_FL_EARLY_SSL_HS | CO_FL_HANDSHAKE)) ==
132 CO_FL_EARLY_DATA)
133 conn->flags &= ~CO_FL_EARLY_DATA;
Willy Tarreaued339a32017-11-03 15:55:24 +0100134 return ret;
Willy Tarreau53a47662017-08-28 10:53:00 +0200135}
136
Olivier Houchard7a3f0df2017-09-13 18:30:23 +0200137/*
138 * Attach a new stream to a connection
139 * (Used for outgoing connections)
140 */
Olivier Houchardf502aca2018-12-14 19:42:40 +0100141static struct conn_stream *mux_pt_attach(struct connection *conn, struct session *sess)
Olivier Houchard7a3f0df2017-09-13 18:30:23 +0200142{
Olivier Houchard7c6f8b12018-11-13 16:48:36 +0100143 struct conn_stream *cs;
144 struct mux_pt_ctx *ctx = conn->mux_ctx;
145
Olivier Houchard1ced4852018-11-29 13:49:59 +0100146 conn->xprt->unsubscribe(conn, SUB_CAN_RECV, &ctx->wait_event);
Olivier Houchard7c6f8b12018-11-13 16:48:36 +0100147 cs = cs_new(conn);
148 if (!cs)
149 goto fail;
150
151 ctx->cs = cs;
Olivier Houchard8706c812018-12-04 19:17:25 +0100152 cs->flags |= CS_FL_RCV_MORE;
Olivier Houchard7c6f8b12018-11-13 16:48:36 +0100153 return (cs);
154fail:
Olivier Houchard7a3f0df2017-09-13 18:30:23 +0200155 return NULL;
156}
157
Willy Tarreaufafd3982018-11-18 21:29:20 +0100158/* Retrieves a valid conn_stream from this connection, or returns NULL. For
159 * this mux, it's easy as we can only store a single conn_stream.
160 */
161static const struct conn_stream *mux_pt_get_first_cs(const struct connection *conn)
162{
Olivier Houchardb6c32ee2018-11-05 18:28:43 +0100163 struct mux_pt_ctx *ctx = conn->mux_ctx;
164 struct conn_stream *cs = ctx->cs;
Willy Tarreaufafd3982018-11-18 21:29:20 +0100165
166 return cs;
167}
168
Olivier Houchard7c6f8b12018-11-13 16:48:36 +0100169/* Destroy the mux and the associated connection, if no longer used */
Olivier Houchard060ed432018-11-06 16:32:42 +0100170static void mux_pt_destroy_meth(struct connection *conn)
171{
Olivier Houchard7c6f8b12018-11-13 16:48:36 +0100172 struct mux_pt_ctx *ctx = conn->mux_ctx;
173
174 if (!(ctx->cs))
175 mux_pt_destroy(ctx);
Olivier Houchard060ed432018-11-06 16:32:42 +0100176}
177
Olivier Houchard7a3f0df2017-09-13 18:30:23 +0200178/*
Willy Tarreau2c52a2b2017-10-08 11:00:17 +0200179 * Detach the stream from the connection and possibly release the connection.
Olivier Houchard7a3f0df2017-09-13 18:30:23 +0200180 */
181static void mux_pt_detach(struct conn_stream *cs)
182{
Willy Tarreau2c52a2b2017-10-08 11:00:17 +0200183 struct connection *conn = cs->conn;
Olivier Houchardb6c32ee2018-11-05 18:28:43 +0100184 struct mux_pt_ctx *ctx = cs->conn->mux_ctx;
Willy Tarreau2c52a2b2017-10-08 11:00:17 +0200185
Olivier Houchardb6c32ee2018-11-05 18:28:43 +0100186 /* Subscribe, to know if we got disconnected */
Olivier Houchard7c6f8b12018-11-13 16:48:36 +0100187 if (conn->owner != NULL &&
188 !(conn->flags & (CO_FL_ERROR | CO_FL_SOCK_RD_SH | CO_FL_SOCK_WR_SH))) {
189 ctx->cs = NULL;
190 conn->xprt->subscribe(conn, SUB_CAN_RECV, &ctx->wait_event);
191 } else
192 /* There's no session attached to that connection, destroy it */
193 mux_pt_destroy(ctx);
Olivier Houchard7a3f0df2017-09-13 18:30:23 +0200194}
195
Olivier Houchardd540b362018-11-05 18:37:53 +0100196static int mux_pt_avail_streams(struct connection *conn)
197{
198 struct mux_pt_ctx *ctx = conn->mux_ctx;
199
200 return (ctx->cs == NULL ? 1 : 0);
201}
202
Olivier Houchard8defe4b2018-12-02 01:31:17 +0100203static int mux_pt_max_streams(struct connection *conn)
204{
205 return 1;
206}
207
Willy Tarreauecdb3fe2017-10-05 15:25:48 +0200208static void mux_pt_shutr(struct conn_stream *cs, enum cs_shr_mode mode)
Olivier Houchard7a3f0df2017-09-13 18:30:23 +0200209{
Willy Tarreau4b795242017-10-05 18:47:38 +0200210 if (cs->flags & CS_FL_SHR)
211 return;
Olivier Houchard8706c812018-12-04 19:17:25 +0100212 cs->flags &= ~CS_FL_RCV_MORE;
Olivier Houchard7a3f0df2017-09-13 18:30:23 +0200213 if (conn_xprt_ready(cs->conn) && cs->conn->xprt->shutr)
Willy Tarreauecdb3fe2017-10-05 15:25:48 +0200214 cs->conn->xprt->shutr(cs->conn, (mode == CS_SHR_DRAIN));
Willy Tarreau4b795242017-10-05 18:47:38 +0200215 if (cs->flags & CS_FL_SHW)
216 conn_full_close(cs->conn);
Olivier Houchard7c6f8b12018-11-13 16:48:36 +0100217 /* Maybe we've been put in the list of available idle connections,
218 * get ouf of here
219 */
220 LIST_DEL(&cs->conn->list);
221 LIST_INIT(&cs->conn->list);
Olivier Houchard7a3f0df2017-09-13 18:30:23 +0200222}
223
Willy Tarreauecdb3fe2017-10-05 15:25:48 +0200224static void mux_pt_shutw(struct conn_stream *cs, enum cs_shw_mode mode)
Olivier Houchard7a3f0df2017-09-13 18:30:23 +0200225{
Willy Tarreau4b795242017-10-05 18:47:38 +0200226 if (cs->flags & CS_FL_SHW)
227 return;
Olivier Houchard7a3f0df2017-09-13 18:30:23 +0200228 if (conn_xprt_ready(cs->conn) && cs->conn->xprt->shutw)
Willy Tarreauecdb3fe2017-10-05 15:25:48 +0200229 cs->conn->xprt->shutw(cs->conn, (mode == CS_SHW_NORMAL));
Willy Tarreau4b795242017-10-05 18:47:38 +0200230 if (!(cs->flags & CS_FL_SHR))
Willy Tarreaua48c1412017-12-22 18:46:33 +0100231 conn_sock_shutw(cs->conn, (mode == CS_SHW_NORMAL));
Willy Tarreau4b795242017-10-05 18:47:38 +0200232 else
233 conn_full_close(cs->conn);
Olivier Houchard7c6f8b12018-11-13 16:48:36 +0100234 /* Maybe we've been put in the list of available idle connections,
235 * get ouf of here
236 */
237 LIST_DEL(&cs->conn->list);
238 LIST_INIT(&cs->conn->list);
Olivier Houchard7a3f0df2017-09-13 18:30:23 +0200239}
240
241/*
242 * Called from the upper layer, to get more data
243 */
Willy Tarreau7f3225f2018-06-19 06:15:17 +0200244static size_t mux_pt_rcv_buf(struct conn_stream *cs, struct buffer *buf, size_t count, int flags)
Olivier Houchard7a3f0df2017-09-13 18:30:23 +0200245{
Willy Tarreaud9cf5402018-07-18 11:29:06 +0200246 size_t ret;
Olivier Houchard7a3f0df2017-09-13 18:30:23 +0200247
Christopher Faulet4eb7d742018-10-11 15:29:21 +0200248 if (!count) {
249 cs->flags |= CS_FL_RCV_MORE;
250 return 0;
251 }
Willy Tarreaue0f24ee2018-12-14 10:51:23 +0100252 b_realign_if_empty(buf);
Willy Tarreau7f3225f2018-06-19 06:15:17 +0200253 ret = cs->conn->xprt->rcv_buf(cs->conn, buf, count, flags);
Olivier Houchard8706c812018-12-04 19:17:25 +0100254 if (conn_xprt_read0_pending(cs->conn)) {
255 if (ret == 0)
256 cs->flags &= ~CS_FL_RCV_MORE;
Olivier Houchard7a3f0df2017-09-13 18:30:23 +0200257 cs->flags |= CS_FL_EOS;
Olivier Houchard8706c812018-12-04 19:17:25 +0100258 }
259 if (cs->conn->flags & CO_FL_ERROR) {
260 if (ret == 0)
261 cs->flags &= ~CS_FL_RCV_MORE;
Willy Tarreau4ff3b892017-10-16 15:17:17 +0200262 cs->flags |= CS_FL_ERROR;
Olivier Houchard8706c812018-12-04 19:17:25 +0100263 }
Willy Tarreaud9cf5402018-07-18 11:29:06 +0200264 return ret;
Olivier Houchard7a3f0df2017-09-13 18:30:23 +0200265}
266
267/* Called from the upper layer, to send data */
Christopher Fauletd44a9b32018-07-27 11:59:41 +0200268static size_t mux_pt_snd_buf(struct conn_stream *cs, struct buffer *buf, size_t count, int flags)
Olivier Houchard7a3f0df2017-09-13 18:30:23 +0200269{
Olivier Houchardb72d98a2018-11-30 13:17:48 +0100270 size_t ret;
271
272 if (cs->conn->flags & CO_FL_HANDSHAKE)
273 return 0;
274 ret = cs->conn->xprt->snd_buf(cs->conn, buf, count, flags);
Christopher Fauletd44a9b32018-07-27 11:59:41 +0200275
276 if (ret > 0)
277 b_del(buf, ret);
278 return ret;
Olivier Houchard7a3f0df2017-09-13 18:30:23 +0200279}
280
Olivier Houchard6ff20392018-07-17 18:46:31 +0200281/* Called from the upper layer, to subscribe to events */
282static int mux_pt_subscribe(struct conn_stream *cs, int event_type, void *param)
283{
284 return (cs->conn->xprt->subscribe(cs->conn, event_type, param));
285}
286
Olivier Houchard83a0cd82018-09-28 17:57:58 +0200287static int mux_pt_unsubscribe(struct conn_stream *cs, int event_type, void *param)
288{
289 return (cs->conn->xprt->unsubscribe(cs->conn, event_type, param));
290}
291
Olivier Houchard7da120b2017-11-01 13:55:10 +0100292#if defined(CONFIG_HAP_LINUX_SPLICE)
Olivier Houchard7a3f0df2017-09-13 18:30:23 +0200293/* Send and get, using splicing */
294static int mux_pt_rcv_pipe(struct conn_stream *cs, struct pipe *pipe, unsigned int count)
295{
296 int ret;
297
298 ret = cs->conn->xprt->rcv_pipe(cs->conn, pipe, count);
299 if (conn_xprt_read0_pending(cs->conn))
300 cs->flags |= CS_FL_EOS;
Willy Tarreau4ff3b892017-10-16 15:17:17 +0200301 if (cs->conn->flags & CO_FL_ERROR)
302 cs->flags |= CS_FL_ERROR;
Olivier Houchard7a3f0df2017-09-13 18:30:23 +0200303 return (ret);
304}
305
306static int mux_pt_snd_pipe(struct conn_stream *cs, struct pipe *pipe)
307{
308 return (cs->conn->xprt->snd_pipe(cs->conn, pipe));
309}
Olivier Houchard7da120b2017-11-01 13:55:10 +0100310#endif
Olivier Houchard7a3f0df2017-09-13 18:30:23 +0200311
Willy Tarreau53a47662017-08-28 10:53:00 +0200312/* The mux operations */
313const struct mux_ops mux_pt_ops = {
314 .init = mux_pt_init,
Willy Tarreau53a47662017-08-28 10:53:00 +0200315 .wake = mux_pt_wake,
Olivier Houchard7a3f0df2017-09-13 18:30:23 +0200316 .rcv_buf = mux_pt_rcv_buf,
317 .snd_buf = mux_pt_snd_buf,
Olivier Houchard6ff20392018-07-17 18:46:31 +0200318 .subscribe = mux_pt_subscribe,
Olivier Houchard83a0cd82018-09-28 17:57:58 +0200319 .unsubscribe = mux_pt_unsubscribe,
Olivier Houchard7a3f0df2017-09-13 18:30:23 +0200320#if defined(CONFIG_HAP_LINUX_SPLICE)
321 .rcv_pipe = mux_pt_rcv_pipe,
322 .snd_pipe = mux_pt_snd_pipe,
323#endif
324 .attach = mux_pt_attach,
Willy Tarreaufafd3982018-11-18 21:29:20 +0100325 .get_first_cs = mux_pt_get_first_cs,
Olivier Houchard7a3f0df2017-09-13 18:30:23 +0200326 .detach = mux_pt_detach,
Olivier Houchardd540b362018-11-05 18:37:53 +0100327 .avail_streams = mux_pt_avail_streams,
Olivier Houchard8defe4b2018-12-02 01:31:17 +0100328 .max_streams = mux_pt_max_streams,
Olivier Houchard060ed432018-11-06 16:32:42 +0100329 .destroy = mux_pt_destroy_meth,
Olivier Houchard7a3f0df2017-09-13 18:30:23 +0200330 .shutr = mux_pt_shutr,
331 .shutw = mux_pt_shutw,
Willy Tarreau28f1cb92017-12-20 16:14:44 +0100332 .flags = MX_FL_NONE,
Willy Tarreau53a47662017-08-28 10:53:00 +0200333 .name = "PASS",
334};
Willy Tarreauf6490822017-09-21 19:43:21 +0200335
Christopher Faulet32f61c02018-04-10 14:33:41 +0200336/* PROT selection : default mux has empty name */
337static struct mux_proto_list mux_proto_pt =
338 { .token = IST(""), .mode = PROTO_MODE_ANY, .side = PROTO_SIDE_BOTH, .mux = &mux_pt_ops };
Willy Tarreauf6490822017-09-21 19:43:21 +0200339
Willy Tarreau0108d902018-11-25 19:14:37 +0100340INITCALL1(STG_REGISTER, register_mux_proto, &mux_proto_pt);