blob: ea0056dd41fc17726e233eebaf5e84d2c21d02b5 [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
Willy Tarreau4c7e4b72020-05-27 12:58:42 +020013#include <haproxy/api.h>
Willy Tarreau16f958c2020-06-03 08:44:35 +020014#include <haproxy/buf.h>
Willy Tarreau7ea393d2020-06-04 18:02:10 +020015#include <haproxy/connection.h>
Christopher Fauletc0ae0972021-04-08 16:45:11 +020016#include <haproxy/pipe-t.h>
Willy Tarreaudfd3de82020-06-04 23:46:14 +020017#include <haproxy/stream.h>
Willy Tarreaucea0e1b2020-06-04 17:25:40 +020018#include <haproxy/task.h>
Christopher Fauletc0ae0972021-04-08 16:45:11 +020019#include <haproxy/trace.h>
Olivier Houchardb6c32ee2018-11-05 18:28:43 +010020
Olivier Houchardb6c32ee2018-11-05 18:28:43 +010021struct mux_pt_ctx {
22 struct conn_stream *cs;
23 struct connection *conn;
24 struct wait_event wait_event;
25};
26
Willy Tarreau8ceae722018-11-26 11:58:30 +010027DECLARE_STATIC_POOL(pool_head_pt_ctx, "mux_pt", sizeof(struct mux_pt_ctx));
28
Christopher Fauletc0ae0972021-04-08 16:45:11 +020029/* trace source and events */
30static void pt_trace(enum trace_level level, uint64_t mask,
31 const struct trace_source *src,
32 const struct ist where, const struct ist func,
33 const void *a1, const void *a2, const void *a3, const void *a4);
34
35/* The event representation is split like this :
36 * pt_ctx - internal PT context
37 * strm - application layer
38 */
39static const struct trace_event pt_trace_events[] = {
40#define PT_EV_CONN_NEW (1ULL << 0)
41 { .mask = PT_EV_CONN_NEW, .name = "pt_conn_new", .desc = "new PT connection" },
42#define PT_EV_CONN_WAKE (1ULL << 1)
43 { .mask = PT_EV_CONN_WAKE, .name = "pt_conn_wake", .desc = "PT connection woken up" },
44#define PT_EV_CONN_END (1ULL << 2)
45 { .mask = PT_EV_CONN_END, .name = "pt_conn_end", .desc = "PT connection terminated" },
46#define PT_EV_CONN_ERR (1ULL << 3)
47 { .mask = PT_EV_CONN_ERR, .name = "pt_conn_err", .desc = "error on PT connection" },
48#define PT_EV_STRM_NEW (1ULL << 4)
49 { .mask = PT_EV_STRM_NEW, .name = "strm_new", .desc = "app-layer stream creation" },
50#define PT_EV_STRM_SHUT (1ULL << 5)
51 { .mask = PT_EV_STRM_SHUT, .name = "strm_shut", .desc = "stream shutdown" },
52#define PT_EV_STRM_END (1ULL << 6)
53 { .mask = PT_EV_STRM_END, .name = "strm_end", .desc = "detaching app-layer stream" },
54#define PT_EV_STRM_ERR (1ULL << 7)
55 { .mask = PT_EV_STRM_ERR, .name = "strm_err", .desc = "stream error" },
56#define PT_EV_RX_DATA (1ULL << 8)
57 { .mask = PT_EV_RX_DATA, .name = "pt_rx_data", .desc = "Rx on PT connection" },
58#define PT_EV_TX_DATA (1ULL << 9)
59 { .mask = PT_EV_TX_DATA, .name = "pt_tx_data", .desc = "Tx on PT connection" },
60
61 {}
62};
63
64
65static const struct name_desc pt_trace_decoding[] = {
66#define PT_VERB_CLEAN 1
67 { .name="clean", .desc="only user-friendly stuff, generally suitable for level \"user\"" },
68#define PT_VERB_MINIMAL 2
69 { .name="minimal", .desc="report only h1c/h1s state and flags, no real decoding" },
70#define PT_VERB_SIMPLE 3
71 { .name="simple", .desc="add request/response status line or htx info when available" },
72#define PT_VERB_ADVANCED 4
73 { .name="advanced", .desc="add header fields or frame decoding when available" },
74#define PT_VERB_COMPLETE 5
75 { .name="complete", .desc="add full data dump when available" },
76 { /* end */ }
77};
78
Willy Tarreau6eb3d372021-04-10 19:29:26 +020079static struct trace_source trace_pt __read_mostly = {
Christopher Fauletc0ae0972021-04-08 16:45:11 +020080 .name = IST("pt"),
81 .desc = "Passthrough multiplexer",
82 .arg_def = TRC_ARG1_CONN, // TRACE()'s first argument is always a connection
83 .default_cb = pt_trace,
84 .known_events = pt_trace_events,
85 .lockon_args = NULL,
86 .decoding = pt_trace_decoding,
87 .report_events = ~0, // report everything by default
88};
89
90#define TRACE_SOURCE &trace_pt
91INITCALL1(STG_REGISTER, trace_register_source, TRACE_SOURCE);
92
93static inline void pt_trace_buf(const struct buffer *buf, size_t ofs, size_t len)
94{
95 size_t block1, block2;
96 int line, ptr, newptr;
97
98 block1 = b_contig_data(buf, ofs);
99 block2 = 0;
100 if (block1 > len)
101 block1 = len;
102 block2 = len - block1;
103
104 ofs = b_peek_ofs(buf, ofs);
105
106 line = 0;
107 ptr = ofs;
108 while (ptr < ofs + block1) {
109 newptr = dump_text_line(&trace_buf, b_orig(buf), b_size(buf), ofs + block1, &line, ptr);
110 if (newptr == ptr)
111 break;
112 ptr = newptr;
113 }
114
115 line = ptr = 0;
116 while (ptr < block2) {
117 newptr = dump_text_line(&trace_buf, b_orig(buf), b_size(buf), block2, &line, ptr);
118 if (newptr == ptr)
119 break;
120 ptr = newptr;
121 }
122}
123
124/* the PT traces always expect that arg1, if non-null, is of type connection
125 * (from which we can derive the pt context), that arg2, if non-null, is a
126 * conn-stream, and that arg3, if non-null, is a buffer.
127 */
128static void pt_trace(enum trace_level level, uint64_t mask, const struct trace_source *src,
129 const struct ist where, const struct ist func,
130 const void *a1, const void *a2, const void *a3, const void *a4)
131{
132 const struct connection *conn = a1;
133 const struct mux_pt_ctx *ctx = conn ? conn->ctx : NULL;
134 const struct conn_stream *cs = a2;
135 const struct buffer *buf = a3;
136 const size_t *val = a4;
137
138 if (!ctx|| src->verbosity < PT_VERB_CLEAN)
139 return;
140
141 /* Display frontend/backend info by default */
142 chunk_appendf(&trace_buf, " : [%c]", (conn_is_back(conn) ? 'B' : 'F'));
143
144 if (src->verbosity == PT_VERB_CLEAN)
145 return;
146
147 /* Display the value to the 4th argument (level > STATE) */
148 if (src->level > TRACE_LEVEL_STATE && val)
149 chunk_appendf(&trace_buf, " - VAL=%lu", (long)*val);
150
151 /* Display conn and cs info, if defined (pointer + flags) */
152 chunk_appendf(&trace_buf, " - conn=%p(0x%08x)", conn, conn->flags);
153 if (cs)
154 chunk_appendf(&trace_buf, " cs=%p(0x%08x)", cs, cs->flags);
155
156 if (src->verbosity == PT_VERB_MINIMAL)
157 return;
158
159 /* Display buffer info, if defined (level > USER & verbosity > SIMPLE) */
160 if (src->level > TRACE_LEVEL_USER && buf) {
161 int full = 0, max = 3000, chunk = 1024;
162
163 /* Full info (level > STATE && verbosity > SIMPLE) */
164 if (src->level > TRACE_LEVEL_STATE) {
165 if (src->verbosity == PT_VERB_COMPLETE)
166 full = 1;
167 else if (src->verbosity == PT_VERB_ADVANCED) {
168 full = 1;
169 max = 256;
170 chunk = 64;
171 }
172 }
173
174 chunk_appendf(&trace_buf, " buf=%u@%p+%u/%u",
175 (unsigned int)b_data(buf), b_orig(buf),
176 (unsigned int)b_head_ofs(buf), (unsigned int)b_size(buf));
177
178 if (b_data(buf) && full) {
179 chunk_memcat(&trace_buf, "\n", 1);
180 if (b_data(buf) < max)
181 pt_trace_buf(buf, 0, b_data(buf));
182 else {
183 pt_trace_buf(buf, 0, chunk);
184 chunk_memcat(&trace_buf, " ...\n", 6);
185 pt_trace_buf(buf, b_data(buf) - chunk, chunk);
186 }
187 }
188 }
189}
190
Olivier Houchardb6c32ee2018-11-05 18:28:43 +0100191static void mux_pt_destroy(struct mux_pt_ctx *ctx)
192{
Christopher Faulet5a7ca292020-11-03 09:11:43 +0100193 struct connection *conn = NULL;
Christopher Faulet39a96ee2019-04-08 10:52:21 +0200194
Christopher Fauletc0ae0972021-04-08 16:45:11 +0200195 TRACE_POINT(PT_EV_CONN_END);
196
Christopher Faulet5a7ca292020-11-03 09:11:43 +0100197 if (ctx) {
198 /* The connection must be attached to this mux to be released */
199 if (ctx->conn && ctx->conn->ctx == ctx)
200 conn = ctx->conn;
201
Christopher Fauletc0ae0972021-04-08 16:45:11 +0200202 TRACE_DEVEL("freeing pt context", PT_EV_CONN_END, conn);
203
Willy Tarreau3c39a7d2019-06-14 14:42:29 +0200204 tasklet_free(ctx->wait_event.tasklet);
Christopher Faulet5a7ca292020-11-03 09:11:43 +0100205
206 if (conn && ctx->wait_event.events != 0)
207 conn->xprt->unsubscribe(conn, conn->xprt_ctx, ctx->wait_event.events,
208 &ctx->wait_event);
209 pool_free(pool_head_pt_ctx, ctx);
210 }
211
212 if (conn) {
Christopher Faulet39a96ee2019-04-08 10:52:21 +0200213 conn->mux = NULL;
214 conn->ctx = NULL;
Christopher Fauletc0ae0972021-04-08 16:45:11 +0200215 TRACE_DEVEL("freeing conn", PT_EV_CONN_END, conn);
Christopher Faulet5a7ca292020-11-03 09:11:43 +0100216
217 conn_stop_tracking(conn);
218 conn_full_close(conn);
Christopher Faulet39a96ee2019-04-08 10:52:21 +0200219 if (conn->destroy_cb)
220 conn->destroy_cb(conn);
Christopher Faulet39a96ee2019-04-08 10:52:21 +0200221 conn_free(conn);
222 }
Olivier Houchardb6c32ee2018-11-05 18:28:43 +0100223}
224
Willy Tarreau691d5032021-01-20 14:55:01 +0100225/* Callback, used when we get I/Os while in idle mode. This one is exported so
226 * that "show fd" can resolve it.
227 */
Willy Tarreau144f84a2021-03-02 16:09:26 +0100228struct task *mux_pt_io_cb(struct task *t, void *tctx, unsigned int status)
Olivier Houchardb6c32ee2018-11-05 18:28:43 +0100229{
230 struct mux_pt_ctx *ctx = tctx;
231
Christopher Fauletc0ae0972021-04-08 16:45:11 +0200232 TRACE_ENTER(PT_EV_CONN_WAKE, ctx->conn, ctx->cs);
Olivier Houchardea510fc2019-10-18 13:56:40 +0200233 if (ctx->cs) {
234 /* There's a small race condition.
235 * mux_pt_io_cb() is only supposed to be called if we have no
236 * stream attached. However, maybe the tasklet got woken up,
237 * and this connection was then attached to a new stream.
Olivier Houchard2ed389d2019-10-18 14:18:29 +0200238 * If this happened, just wake the tasklet up if anybody
239 * subscribed to receive events, and otherwise call the wake
240 * method, to make sure the event is noticed.
Olivier Houchardea510fc2019-10-18 13:56:40 +0200241 */
Willy Tarreau7872d1f2020-01-10 07:06:05 +0100242 if (ctx->conn->subs) {
243 ctx->conn->subs->events = 0;
244 tasklet_wakeup(ctx->conn->subs->tasklet);
245 ctx->conn->subs = NULL;
Olivier Houchard2ed389d2019-10-18 14:18:29 +0200246 } else if (ctx->cs->data_cb->wake)
Olivier Houchardea510fc2019-10-18 13:56:40 +0200247 ctx->cs->data_cb->wake(ctx->cs);
Christopher Fauletc0ae0972021-04-08 16:45:11 +0200248 TRACE_DEVEL("leaving waking up CS", PT_EV_CONN_WAKE, ctx->conn, ctx->cs);
Willy Tarreau74163142021-03-13 11:30:19 +0100249 return t;
Olivier Houchardea510fc2019-10-18 13:56:40 +0200250 }
Willy Tarreau2ded48d2020-12-11 16:20:34 +0100251 conn_ctrl_drain(ctx->conn);
Willy Tarreau74163142021-03-13 11:30:19 +0100252 if (ctx->conn->flags & (CO_FL_ERROR | CO_FL_SOCK_RD_SH | CO_FL_SOCK_WR_SH)) {
Christopher Faulete2c65ba2021-04-10 09:02:32 +0200253 TRACE_DEVEL("leaving destroying pt context", PT_EV_CONN_WAKE, ctx->conn);
Olivier Houchard9dce2c52019-10-18 10:59:30 +0200254 mux_pt_destroy(ctx);
Willy Tarreau74163142021-03-13 11:30:19 +0100255 t = NULL;
256 }
Christopher Fauletc0ae0972021-04-08 16:45:11 +0200257 else {
Olivier Houcharde179d0e2019-03-21 18:27:17 +0100258 ctx->conn->xprt->subscribe(ctx->conn, ctx->conn->xprt_ctx, SUB_RETRY_RECV,
Christopher Fauletc0ae0972021-04-08 16:45:11 +0200259 &ctx->wait_event);
Christopher Faulete2c65ba2021-04-10 09:02:32 +0200260 TRACE_DEVEL("leaving subscribing for reads", PT_EV_CONN_WAKE, ctx->conn);
Christopher Fauletc0ae0972021-04-08 16:45:11 +0200261 }
Olivier Houchardb6c32ee2018-11-05 18:28:43 +0100262
Willy Tarreau74163142021-03-13 11:30:19 +0100263 return t;
Olivier Houchardb6c32ee2018-11-05 18:28:43 +0100264}
Willy Tarreau53a47662017-08-28 10:53:00 +0200265
Willy Tarreau3d2ee552018-12-19 14:12:10 +0100266/* Initialize the mux once it's attached. It is expected that conn->ctx
Olivier Houchard9aaf7782017-09-13 18:30:23 +0200267 * points to the existing conn_stream (for outgoing connections) or NULL (for
268 * incoming ones, in which case one will be allocated and a new stream will be
Ilya Shipitsin46a030c2020-07-05 16:36:08 +0500269 * instantiated). Returns < 0 on error.
Willy Tarreau53a47662017-08-28 10:53:00 +0200270 */
Christopher Faulet51f73eb2019-04-08 11:22:47 +0200271static int mux_pt_init(struct connection *conn, struct proxy *prx, struct session *sess,
272 struct buffer *input)
Willy Tarreau53a47662017-08-28 10:53:00 +0200273{
Willy Tarreau3d2ee552018-12-19 14:12:10 +0100274 struct conn_stream *cs = conn->ctx;
Olivier Houchardb6c32ee2018-11-05 18:28:43 +0100275 struct mux_pt_ctx *ctx = pool_alloc(pool_head_pt_ctx);
276
Christopher Fauletc0ae0972021-04-08 16:45:11 +0200277 TRACE_ENTER(PT_EV_CONN_NEW);
278
279 if (!ctx) {
280 TRACE_ERROR("PT context allocation failure", PT_EV_CONN_NEW|PT_EV_CONN_END|PT_EV_CONN_ERR);
Olivier Houchardb6c32ee2018-11-05 18:28:43 +0100281 goto fail;
Christopher Fauletc0ae0972021-04-08 16:45:11 +0200282 }
Olivier Houchardb6c32ee2018-11-05 18:28:43 +0100283
Willy Tarreau3c39a7d2019-06-14 14:42:29 +0200284 ctx->wait_event.tasklet = tasklet_new();
285 if (!ctx->wait_event.tasklet)
Olivier Houchardb6c32ee2018-11-05 18:28:43 +0100286 goto fail_free_ctx;
Willy Tarreau3c39a7d2019-06-14 14:42:29 +0200287 ctx->wait_event.tasklet->context = ctx;
288 ctx->wait_event.tasklet->process = mux_pt_io_cb;
Willy Tarreau4f6516d2018-12-19 13:59:17 +0100289 ctx->wait_event.events = 0;
Olivier Houchardb6c32ee2018-11-05 18:28:43 +0100290 ctx->conn = conn;
Olivier Houchard9aaf7782017-09-13 18:30:23 +0200291
292 if (!cs) {
Christopher Faulet236c93b2020-07-02 09:19:54 +0200293 cs = cs_new(conn, conn->target);
Christopher Fauletc0ae0972021-04-08 16:45:11 +0200294 if (!cs) {
295 TRACE_ERROR("CS allocation failure", PT_EV_STRM_NEW|PT_EV_STRM_END|PT_EV_STRM_ERR, conn);
Olivier Houchardb6c32ee2018-11-05 18:28:43 +0100296 goto fail_free_ctx;
Christopher Fauletc0ae0972021-04-08 16:45:11 +0200297 }
Olivier Houchard9aaf7782017-09-13 18:30:23 +0200298
Christopher Fauletc0ae0972021-04-08 16:45:11 +0200299 if (stream_create_from_cs(cs, &BUF_NULL) < 0) {
300 TRACE_ERROR("stream creation failure", PT_EV_STRM_NEW|PT_EV_STRM_END|PT_EV_STRM_ERR, conn, cs);
Olivier Houchard9aaf7782017-09-13 18:30:23 +0200301 goto fail_free;
Christopher Fauletc0ae0972021-04-08 16:45:11 +0200302 }
303 TRACE_POINT(PT_EV_STRM_NEW, conn, cs);
Olivier Houchard9aaf7782017-09-13 18:30:23 +0200304 }
Willy Tarreau3d2ee552018-12-19 14:12:10 +0100305 conn->ctx = ctx;
Olivier Houchardb6c32ee2018-11-05 18:28:43 +0100306 ctx->cs = cs;
Olivier Houchard8706c812018-12-04 19:17:25 +0100307 cs->flags |= CS_FL_RCV_MORE;
Willy Tarreau17ccd1a2020-01-17 16:19:34 +0100308 if (global.tune.options & GTUNE_USE_SPLICE)
309 cs->flags |= CS_FL_MAY_SPLICE;
Christopher Fauletc0ae0972021-04-08 16:45:11 +0200310
311 TRACE_LEAVE(PT_EV_CONN_NEW, conn, cs);
Willy Tarreau53a47662017-08-28 10:53:00 +0200312 return 0;
Olivier Houchard9aaf7782017-09-13 18:30:23 +0200313
314 fail_free:
315 cs_free(cs);
Olivier Houchardb6c32ee2018-11-05 18:28:43 +0100316fail_free_ctx:
Willy Tarreau3c39a7d2019-06-14 14:42:29 +0200317 if (ctx->wait_event.tasklet)
318 tasklet_free(ctx->wait_event.tasklet);
Olivier Houchardb6c32ee2018-11-05 18:28:43 +0100319 pool_free(pool_head_pt_ctx, ctx);
Olivier Houchard9aaf7782017-09-13 18:30:23 +0200320 fail:
Christopher Fauletc0ae0972021-04-08 16:45:11 +0200321 TRACE_DEVEL("leaving in error", PT_EV_CONN_NEW|PT_EV_CONN_END|PT_EV_CONN_ERR);
Olivier Houchard9aaf7782017-09-13 18:30:23 +0200322 return -1;
Willy Tarreau53a47662017-08-28 10:53:00 +0200323}
324
325/* callback to be used by default for the pass-through mux. It calls the data
326 * layer wake() callback if it is set otherwise returns 0.
327 */
328static int mux_pt_wake(struct connection *conn)
329{
Willy Tarreau3d2ee552018-12-19 14:12:10 +0100330 struct mux_pt_ctx *ctx = conn->ctx;
Olivier Houchardb6c32ee2018-11-05 18:28:43 +0100331 struct conn_stream *cs = ctx->cs;
332 int ret = 0;
Olivier Houchard9aaf7782017-09-13 18:30:23 +0200333
Christopher Fauletc0ae0972021-04-08 16:45:11 +0200334 TRACE_ENTER(PT_EV_CONN_WAKE, ctx->conn, cs);
Olivier Houchardb6c32ee2018-11-05 18:28:43 +0100335 if (cs) {
336 ret = cs->data_cb->wake ? cs->data_cb->wake(cs) : 0;
Olivier Houchard9aaf7782017-09-13 18:30:23 +0200337
Christopher Fauletc0ae0972021-04-08 16:45:11 +0200338 if (ret < 0) {
339 TRACE_DEVEL("leaving waking up CS", PT_EV_CONN_WAKE, ctx->conn, cs);
Olivier Houchardb6c32ee2018-11-05 18:28:43 +0100340 return ret;
Christopher Fauletc0ae0972021-04-08 16:45:11 +0200341 }
Olivier Houchardb6c32ee2018-11-05 18:28:43 +0100342 } else {
Willy Tarreau2ded48d2020-12-11 16:20:34 +0100343 conn_ctrl_drain(conn);
Olivier Houchardb6c32ee2018-11-05 18:28:43 +0100344 if (conn->flags & (CO_FL_ERROR | CO_FL_SOCK_RD_SH)) {
Christopher Fauletc0ae0972021-04-08 16:45:11 +0200345 TRACE_DEVEL("leaving destroying PT context", PT_EV_CONN_WAKE, ctx->conn);
Olivier Houchardb6c32ee2018-11-05 18:28:43 +0100346 mux_pt_destroy(ctx);
347 return -1;
348 }
349 }
Willy Tarreauad7f0ad2018-08-24 15:48:59 +0200350
Olivier Houchard7fc96d52017-11-23 18:25:47 +0100351 /* If we had early data, and we're done with the handshake
Ilya Shipitsin46a030c2020-07-05 16:36:08 +0500352 * then we know the data are safe, and we can remove the flag.
Olivier Houchard7fc96d52017-11-23 18:25:47 +0100353 */
Willy Tarreau911db9b2020-01-23 16:27:54 +0100354 if ((conn->flags & (CO_FL_EARLY_DATA | CO_FL_EARLY_SSL_HS | CO_FL_WAIT_XPRT)) ==
Olivier Houchard7fc96d52017-11-23 18:25:47 +0100355 CO_FL_EARLY_DATA)
356 conn->flags &= ~CO_FL_EARLY_DATA;
Christopher Fauletc0ae0972021-04-08 16:45:11 +0200357
358 TRACE_LEAVE(PT_EV_CONN_WAKE, ctx->conn);
Willy Tarreaued339a32017-11-03 15:55:24 +0100359 return ret;
Willy Tarreau53a47662017-08-28 10:53:00 +0200360}
361
Olivier Houchard7a3f0df2017-09-13 18:30:23 +0200362/*
363 * Attach a new stream to a connection
364 * (Used for outgoing connections)
365 */
Olivier Houchardf502aca2018-12-14 19:42:40 +0100366static struct conn_stream *mux_pt_attach(struct connection *conn, struct session *sess)
Olivier Houchard7a3f0df2017-09-13 18:30:23 +0200367{
Olivier Houchard7c6f8b12018-11-13 16:48:36 +0100368 struct conn_stream *cs;
Willy Tarreau3d2ee552018-12-19 14:12:10 +0100369 struct mux_pt_ctx *ctx = conn->ctx;
Olivier Houchard7c6f8b12018-11-13 16:48:36 +0100370
Christopher Fauletc0ae0972021-04-08 16:45:11 +0200371 TRACE_ENTER(PT_EV_STRM_NEW, conn);
Olivier Houchardea32b0f2019-08-10 23:56:16 +0200372 if (ctx->wait_event.events)
373 conn->xprt->unsubscribe(ctx->conn, conn->xprt_ctx, SUB_RETRY_RECV, &ctx->wait_event);
Christopher Faulet236c93b2020-07-02 09:19:54 +0200374 cs = cs_new(conn, conn->target);
Christopher Fauletc0ae0972021-04-08 16:45:11 +0200375 if (!cs) {
376 TRACE_ERROR("CS allocation failure", PT_EV_STRM_NEW|PT_EV_STRM_END|PT_EV_STRM_ERR, conn);
Olivier Houchard7c6f8b12018-11-13 16:48:36 +0100377 goto fail;
Christopher Fauletc0ae0972021-04-08 16:45:11 +0200378 }
Olivier Houchard7c6f8b12018-11-13 16:48:36 +0100379
380 ctx->cs = cs;
Olivier Houchard8706c812018-12-04 19:17:25 +0100381 cs->flags |= CS_FL_RCV_MORE;
Christopher Fauletc0ae0972021-04-08 16:45:11 +0200382
383 TRACE_LEAVE(PT_EV_STRM_NEW, conn, cs);
Olivier Houchard7c6f8b12018-11-13 16:48:36 +0100384 return (cs);
385fail:
Christopher Fauletc0ae0972021-04-08 16:45:11 +0200386 TRACE_DEVEL("leaving on error", PT_EV_STRM_NEW|PT_EV_STRM_END|PT_EV_STRM_ERR, conn);
Olivier Houchard7a3f0df2017-09-13 18:30:23 +0200387 return NULL;
388}
389
Willy Tarreaufafd3982018-11-18 21:29:20 +0100390/* Retrieves a valid conn_stream from this connection, or returns NULL. For
391 * this mux, it's easy as we can only store a single conn_stream.
392 */
393static const struct conn_stream *mux_pt_get_first_cs(const struct connection *conn)
394{
Willy Tarreau3d2ee552018-12-19 14:12:10 +0100395 struct mux_pt_ctx *ctx = conn->ctx;
Olivier Houchardb6c32ee2018-11-05 18:28:43 +0100396 struct conn_stream *cs = ctx->cs;
Willy Tarreaufafd3982018-11-18 21:29:20 +0100397
398 return cs;
399}
400
Christopher Faulet73c12072019-04-08 11:23:22 +0200401/* Destroy the mux and the associated connection if still attached to this mux
402 * and no longer used */
403static void mux_pt_destroy_meth(void *ctx)
Olivier Houchard060ed432018-11-06 16:32:42 +0100404{
Christopher Faulet73c12072019-04-08 11:23:22 +0200405 struct mux_pt_ctx *pt = ctx;
Olivier Houchard7c6f8b12018-11-13 16:48:36 +0100406
Christopher Fauletc0ae0972021-04-08 16:45:11 +0200407 TRACE_POINT(PT_EV_CONN_END, pt->conn, pt->cs);
Christopher Faulet39a96ee2019-04-08 10:52:21 +0200408 if (!(pt->cs) || !(pt->conn) || pt->conn->ctx != pt)
Christopher Faulet73c12072019-04-08 11:23:22 +0200409 mux_pt_destroy(pt);
Olivier Houchard060ed432018-11-06 16:32:42 +0100410}
411
Olivier Houchard7a3f0df2017-09-13 18:30:23 +0200412/*
Willy Tarreau2c52a2b2017-10-08 11:00:17 +0200413 * Detach the stream from the connection and possibly release the connection.
Olivier Houchard7a3f0df2017-09-13 18:30:23 +0200414 */
415static void mux_pt_detach(struct conn_stream *cs)
416{
Willy Tarreau2c52a2b2017-10-08 11:00:17 +0200417 struct connection *conn = cs->conn;
Willy Tarreau3d2ee552018-12-19 14:12:10 +0100418 struct mux_pt_ctx *ctx = cs->conn->ctx;
Willy Tarreau2c52a2b2017-10-08 11:00:17 +0200419
Christopher Fauletc0ae0972021-04-08 16:45:11 +0200420 TRACE_ENTER(PT_EV_STRM_END, conn, cs);
421
Olivier Houchardb6c32ee2018-11-05 18:28:43 +0100422 /* Subscribe, to know if we got disconnected */
Christopher Faulet126137f2022-03-09 15:55:58 +0100423 if (!conn_is_back(conn) && conn->owner != NULL &&
Olivier Houchard7c6f8b12018-11-13 16:48:36 +0100424 !(conn->flags & (CO_FL_ERROR | CO_FL_SOCK_RD_SH | CO_FL_SOCK_WR_SH))) {
425 ctx->cs = NULL;
Olivier Houcharde179d0e2019-03-21 18:27:17 +0100426 conn->xprt->subscribe(conn, conn->xprt_ctx, SUB_RETRY_RECV, &ctx->wait_event);
Christopher Fauletc0ae0972021-04-08 16:45:11 +0200427 } else {
Olivier Houchard7c6f8b12018-11-13 16:48:36 +0100428 /* There's no session attached to that connection, destroy it */
Christopher Fauletc0ae0972021-04-08 16:45:11 +0200429 TRACE_DEVEL("killing dead connection", PT_EV_STRM_END, conn, cs);
Olivier Houchard7c6f8b12018-11-13 16:48:36 +0100430 mux_pt_destroy(ctx);
Christopher Fauletc0ae0972021-04-08 16:45:11 +0200431 }
432
433 TRACE_LEAVE(PT_EV_STRM_END);
Olivier Houchard7a3f0df2017-09-13 18:30:23 +0200434}
435
Willy Tarreau00f18a32019-01-26 12:19:01 +0100436/* returns the number of streams in use on a connection */
437static int mux_pt_used_streams(struct connection *conn)
Olivier Houchardd540b362018-11-05 18:37:53 +0100438{
Willy Tarreau3d2ee552018-12-19 14:12:10 +0100439 struct mux_pt_ctx *ctx = conn->ctx;
Olivier Houchardd540b362018-11-05 18:37:53 +0100440
Willy Tarreau00f18a32019-01-26 12:19:01 +0100441 return ctx->cs ? 1 : 0;
Olivier Houchardd540b362018-11-05 18:37:53 +0100442}
443
Willy Tarreau00f18a32019-01-26 12:19:01 +0100444/* returns the number of streams still available on a connection */
445static int mux_pt_avail_streams(struct connection *conn)
Olivier Houchard8defe4b2018-12-02 01:31:17 +0100446{
Willy Tarreau00f18a32019-01-26 12:19:01 +0100447 return 1 - mux_pt_used_streams(conn);
Olivier Houchard8defe4b2018-12-02 01:31:17 +0100448}
449
Willy Tarreauecdb3fe2017-10-05 15:25:48 +0200450static void mux_pt_shutr(struct conn_stream *cs, enum cs_shr_mode mode)
Olivier Houchard7a3f0df2017-09-13 18:30:23 +0200451{
Christopher Fauletc0ae0972021-04-08 16:45:11 +0200452 TRACE_ENTER(PT_EV_STRM_SHUT, cs->conn, cs);
453
Willy Tarreau4b795242017-10-05 18:47:38 +0200454 if (cs->flags & CS_FL_SHR)
455 return;
Christopher Fauletd94f8772018-12-17 13:21:02 +0100456 cs->flags &= ~(CS_FL_RCV_MORE | CS_FL_WANT_ROOM);
Olivier Houchard7a3f0df2017-09-13 18:30:23 +0200457 if (conn_xprt_ready(cs->conn) && cs->conn->xprt->shutr)
Olivier Houcharde179d0e2019-03-21 18:27:17 +0100458 cs->conn->xprt->shutr(cs->conn, cs->conn->xprt_ctx,
459 (mode == CS_SHR_DRAIN));
Willy Tarreau7d7b11c2020-12-11 11:07:19 +0100460 else if (mode == CS_SHR_DRAIN)
Willy Tarreau2ded48d2020-12-11 16:20:34 +0100461 conn_ctrl_drain(cs->conn);
Willy Tarreau4b795242017-10-05 18:47:38 +0200462 if (cs->flags & CS_FL_SHW)
463 conn_full_close(cs->conn);
Christopher Fauletc0ae0972021-04-08 16:45:11 +0200464
465 TRACE_LEAVE(PT_EV_STRM_SHUT, cs->conn, cs);
Olivier Houchard7a3f0df2017-09-13 18:30:23 +0200466}
467
Willy Tarreauecdb3fe2017-10-05 15:25:48 +0200468static void mux_pt_shutw(struct conn_stream *cs, enum cs_shw_mode mode)
Olivier Houchard7a3f0df2017-09-13 18:30:23 +0200469{
Christopher Fauletc0ae0972021-04-08 16:45:11 +0200470 TRACE_ENTER(PT_EV_STRM_SHUT, cs->conn, cs);
471
Willy Tarreau4b795242017-10-05 18:47:38 +0200472 if (cs->flags & CS_FL_SHW)
473 return;
Olivier Houchard7a3f0df2017-09-13 18:30:23 +0200474 if (conn_xprt_ready(cs->conn) && cs->conn->xprt->shutw)
Olivier Houcharde179d0e2019-03-21 18:27:17 +0100475 cs->conn->xprt->shutw(cs->conn, cs->conn->xprt_ctx,
476 (mode == CS_SHW_NORMAL));
Willy Tarreau4b795242017-10-05 18:47:38 +0200477 if (!(cs->flags & CS_FL_SHR))
Willy Tarreaua48c1412017-12-22 18:46:33 +0100478 conn_sock_shutw(cs->conn, (mode == CS_SHW_NORMAL));
Willy Tarreau4b795242017-10-05 18:47:38 +0200479 else
480 conn_full_close(cs->conn);
Christopher Fauletc0ae0972021-04-08 16:45:11 +0200481
482 TRACE_LEAVE(PT_EV_STRM_SHUT, cs->conn, cs);
Olivier Houchard7a3f0df2017-09-13 18:30:23 +0200483}
484
485/*
486 * Called from the upper layer, to get more data
Christopher Faulet93a466b2021-09-21 15:50:55 +0200487 *
488 * The caller is responsible for defragmenting <buf> if necessary. But <flags>
489 * must be tested to know the calling context. If CO_RFL_BUF_FLUSH is set, it
490 * means the caller wants to flush input data (from the mux buffer and the
491 * channel buffer) to be able to use kernel splicing or any kind of mux-to-mux
492 * xfer. If CO_RFL_KEEP_RECV is set, the mux must always subscribe for read
493 * events before giving back. CO_RFL_BUF_WET is set if <buf> is congested with
494 * data scheduled for leaving soon. CO_RFL_BUF_NOT_STUCK is set to instruct the
495 * mux it may optimize the data copy to <buf> if necessary. Otherwise, it should
496 * copy as much data as possible.
Olivier Houchard7a3f0df2017-09-13 18:30:23 +0200497 */
Willy Tarreau7f3225f2018-06-19 06:15:17 +0200498static 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 +0200499{
Christopher Fauletc0ae0972021-04-08 16:45:11 +0200500 size_t ret = 0;
501
502 TRACE_ENTER(PT_EV_RX_DATA, cs->conn, cs, buf, (size_t[]){count});
Olivier Houchard7a3f0df2017-09-13 18:30:23 +0200503
Christopher Faulet4eb7d742018-10-11 15:29:21 +0200504 if (!count) {
Christopher Fauletd94f8772018-12-17 13:21:02 +0100505 cs->flags |= (CS_FL_RCV_MORE | CS_FL_WANT_ROOM);
Christopher Fauletc0ae0972021-04-08 16:45:11 +0200506 goto end;
Christopher Faulet4eb7d742018-10-11 15:29:21 +0200507 }
Willy Tarreaue0f24ee2018-12-14 10:51:23 +0100508 b_realign_if_empty(buf);
Olivier Houcharde179d0e2019-03-21 18:27:17 +0100509 ret = cs->conn->xprt->rcv_buf(cs->conn, cs->conn->xprt_ctx, buf, count, flags);
Olivier Houchard8706c812018-12-04 19:17:25 +0100510 if (conn_xprt_read0_pending(cs->conn)) {
Willy Tarreau9cca8df2019-07-15 06:47:54 +0200511 cs->flags &= ~(CS_FL_RCV_MORE | CS_FL_WANT_ROOM);
Christopher Faulet87a8f352019-03-22 14:51:36 +0100512 cs->flags |= CS_FL_EOS;
Christopher Fauletc0ae0972021-04-08 16:45:11 +0200513 TRACE_DEVEL("read0 on connection", PT_EV_RX_DATA, cs->conn, cs);
Olivier Houchard8706c812018-12-04 19:17:25 +0100514 }
515 if (cs->conn->flags & CO_FL_ERROR) {
Willy Tarreau9cca8df2019-07-15 06:47:54 +0200516 cs->flags &= ~(CS_FL_RCV_MORE | CS_FL_WANT_ROOM);
Willy Tarreau4ff3b892017-10-16 15:17:17 +0200517 cs->flags |= CS_FL_ERROR;
Christopher Fauletc0ae0972021-04-08 16:45:11 +0200518 TRACE_DEVEL("error on connection", PT_EV_RX_DATA|PT_EV_CONN_ERR, cs->conn, cs);
Olivier Houchard8706c812018-12-04 19:17:25 +0100519 }
Christopher Fauletc0ae0972021-04-08 16:45:11 +0200520 end:
521 TRACE_LEAVE(PT_EV_RX_DATA, cs->conn, cs, buf, (size_t[]){ret});
Willy Tarreaud9cf5402018-07-18 11:29:06 +0200522 return ret;
Olivier Houchard7a3f0df2017-09-13 18:30:23 +0200523}
524
525/* Called from the upper layer, to send data */
Christopher Fauletd44a9b32018-07-27 11:59:41 +0200526static 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 +0200527{
Olivier Houchardb72d98a2018-11-30 13:17:48 +0100528 size_t ret;
529
Christopher Fauletc0ae0972021-04-08 16:45:11 +0200530 TRACE_ENTER(PT_EV_TX_DATA, cs->conn, cs, buf, (size_t[]){count});
531
Olivier Houcharde179d0e2019-03-21 18:27:17 +0100532 ret = cs->conn->xprt->snd_buf(cs->conn, cs->conn->xprt_ctx, buf, count, flags);
Christopher Fauletd44a9b32018-07-27 11:59:41 +0200533
534 if (ret > 0)
535 b_del(buf, ret);
Christopher Fauletc0ae0972021-04-08 16:45:11 +0200536
537 TRACE_LEAVE(PT_EV_TX_DATA, cs->conn, cs, buf, (size_t[]){ret});
Christopher Fauletd44a9b32018-07-27 11:59:41 +0200538 return ret;
Olivier Houchard7a3f0df2017-09-13 18:30:23 +0200539}
540
Willy Tarreauee1a6fc2020-01-17 07:52:13 +0100541/* Called from the upper layer, to subscribe <es> to events <event_type>. The
542 * event subscriber <es> is not allowed to change from a previous call as long
543 * as at least one event is still subscribed. The <event_type> must only be a
544 * combination of SUB_RETRY_RECV and SUB_RETRY_SEND. It always returns 0.
545 */
546static int mux_pt_subscribe(struct conn_stream *cs, int event_type, struct wait_event *es)
Olivier Houchard6ff20392018-07-17 18:46:31 +0200547{
Christopher Fauletc0ae0972021-04-08 16:45:11 +0200548 TRACE_POINT(PT_EV_RX_DATA|PT_EV_TX_DATA, cs->conn, cs, 0, (size_t[]){event_type});
Willy Tarreauee1a6fc2020-01-17 07:52:13 +0100549 return cs->conn->xprt->subscribe(cs->conn, cs->conn->xprt_ctx, event_type, es);
Olivier Houchard6ff20392018-07-17 18:46:31 +0200550}
551
Willy Tarreauee1a6fc2020-01-17 07:52:13 +0100552/* Called from the upper layer, to unsubscribe <es> from events <event_type>.
553 * The <es> pointer is not allowed to differ from the one passed to the
554 * subscribe() call. It always returns zero.
555 */
556static int mux_pt_unsubscribe(struct conn_stream *cs, int event_type, struct wait_event *es)
Olivier Houchard83a0cd82018-09-28 17:57:58 +0200557{
Christopher Fauletc0ae0972021-04-08 16:45:11 +0200558 TRACE_POINT(PT_EV_RX_DATA|PT_EV_TX_DATA, cs->conn, cs, 0, (size_t[]){event_type});
Willy Tarreauee1a6fc2020-01-17 07:52:13 +0100559 return cs->conn->xprt->unsubscribe(cs->conn, cs->conn->xprt_ctx, event_type, es);
Olivier Houchard83a0cd82018-09-28 17:57:58 +0200560}
561
Willy Tarreaue5733232019-05-22 19:24:06 +0200562#if defined(USE_LINUX_SPLICE)
Olivier Houchard7a3f0df2017-09-13 18:30:23 +0200563/* Send and get, using splicing */
564static int mux_pt_rcv_pipe(struct conn_stream *cs, struct pipe *pipe, unsigned int count)
565{
566 int ret;
567
Christopher Fauletc0ae0972021-04-08 16:45:11 +0200568 TRACE_ENTER(PT_EV_RX_DATA, cs->conn, cs, 0, (size_t[]){count});
569
Olivier Houcharde179d0e2019-03-21 18:27:17 +0100570 ret = cs->conn->xprt->rcv_pipe(cs->conn, cs->conn->xprt_ctx, pipe, count);
Christopher Fauletc0ae0972021-04-08 16:45:11 +0200571 if (conn_xprt_read0_pending(cs->conn)) {
Christopher Faulet87a8f352019-03-22 14:51:36 +0100572 cs->flags |= CS_FL_EOS;
Christopher Fauletc0ae0972021-04-08 16:45:11 +0200573 TRACE_DEVEL("read0 on connection", PT_EV_RX_DATA, cs->conn, cs);
574 }
575 if (cs->conn->flags & CO_FL_ERROR) {
Willy Tarreau4ff3b892017-10-16 15:17:17 +0200576 cs->flags |= CS_FL_ERROR;
Christopher Fauletc0ae0972021-04-08 16:45:11 +0200577 TRACE_DEVEL("error on connection", PT_EV_RX_DATA|PT_EV_CONN_ERR, cs->conn, cs);
578 }
579
580 TRACE_LEAVE(PT_EV_RX_DATA, cs->conn, cs, 0, (size_t[]){ret});
Olivier Houchard7a3f0df2017-09-13 18:30:23 +0200581 return (ret);
582}
583
584static int mux_pt_snd_pipe(struct conn_stream *cs, struct pipe *pipe)
585{
Christopher Fauletc0ae0972021-04-08 16:45:11 +0200586 int ret;
587
588 TRACE_ENTER(PT_EV_TX_DATA, cs->conn, cs, 0, (size_t[]){pipe->data});
589
590 ret = cs->conn->xprt->snd_pipe(cs->conn, cs->conn->xprt_ctx, pipe);
591
592 TRACE_LEAVE(PT_EV_TX_DATA, cs->conn, cs, 0, (size_t[]){ret});
593 return ret;
Olivier Houchard7a3f0df2017-09-13 18:30:23 +0200594}
Olivier Houchard7da120b2017-11-01 13:55:10 +0100595#endif
Olivier Houchard7a3f0df2017-09-13 18:30:23 +0200596
Olivier Houchard9b8e11e2019-10-25 16:19:26 +0200597static int mux_pt_ctl(struct connection *conn, enum mux_ctl_type mux_ctl, void *output)
598{
599 int ret = 0;
600 switch (mux_ctl) {
601 case MUX_STATUS:
Willy Tarreau911db9b2020-01-23 16:27:54 +0100602 if (!(conn->flags & CO_FL_WAIT_XPRT))
Olivier Houchard9b8e11e2019-10-25 16:19:26 +0200603 ret |= MUX_STATUS_READY;
604 return ret;
Christopher Faulet4c8ad842020-10-06 14:59:17 +0200605 case MUX_EXIT_STATUS:
606 return MUX_ES_UNKNOWN;
Olivier Houchard9b8e11e2019-10-25 16:19:26 +0200607 default:
608 return -1;
609 }
610}
611
Willy Tarreau53a47662017-08-28 10:53:00 +0200612/* The mux operations */
Christopher Faulet28da3f52021-02-05 16:44:46 +0100613const struct mux_ops mux_tcp_ops = {
Willy Tarreau53a47662017-08-28 10:53:00 +0200614 .init = mux_pt_init,
Willy Tarreau53a47662017-08-28 10:53:00 +0200615 .wake = mux_pt_wake,
Olivier Houchard7a3f0df2017-09-13 18:30:23 +0200616 .rcv_buf = mux_pt_rcv_buf,
617 .snd_buf = mux_pt_snd_buf,
Olivier Houchard6ff20392018-07-17 18:46:31 +0200618 .subscribe = mux_pt_subscribe,
Olivier Houchard83a0cd82018-09-28 17:57:58 +0200619 .unsubscribe = mux_pt_unsubscribe,
Willy Tarreaue5733232019-05-22 19:24:06 +0200620#if defined(USE_LINUX_SPLICE)
Olivier Houchard7a3f0df2017-09-13 18:30:23 +0200621 .rcv_pipe = mux_pt_rcv_pipe,
622 .snd_pipe = mux_pt_snd_pipe,
623#endif
624 .attach = mux_pt_attach,
Willy Tarreaufafd3982018-11-18 21:29:20 +0100625 .get_first_cs = mux_pt_get_first_cs,
Olivier Houchard7a3f0df2017-09-13 18:30:23 +0200626 .detach = mux_pt_detach,
Olivier Houchardd540b362018-11-05 18:37:53 +0100627 .avail_streams = mux_pt_avail_streams,
Willy Tarreau00f18a32019-01-26 12:19:01 +0100628 .used_streams = mux_pt_used_streams,
Olivier Houchard060ed432018-11-06 16:32:42 +0100629 .destroy = mux_pt_destroy_meth,
Olivier Houchard9b8e11e2019-10-25 16:19:26 +0200630 .ctl = mux_pt_ctl,
Olivier Houchard7a3f0df2017-09-13 18:30:23 +0200631 .shutr = mux_pt_shutr,
632 .shutw = mux_pt_shutw,
Willy Tarreau28f1cb92017-12-20 16:14:44 +0100633 .flags = MX_FL_NONE,
Willy Tarreau53a47662017-08-28 10:53:00 +0200634 .name = "PASS",
635};
Willy Tarreauf6490822017-09-21 19:43:21 +0200636
Christopher Faulet28da3f52021-02-05 16:44:46 +0100637
638const struct mux_ops mux_pt_ops = {
639 .init = mux_pt_init,
640 .wake = mux_pt_wake,
641 .rcv_buf = mux_pt_rcv_buf,
642 .snd_buf = mux_pt_snd_buf,
643 .subscribe = mux_pt_subscribe,
644 .unsubscribe = mux_pt_unsubscribe,
645#if defined(USE_LINUX_SPLICE)
646 .rcv_pipe = mux_pt_rcv_pipe,
647 .snd_pipe = mux_pt_snd_pipe,
648#endif
649 .attach = mux_pt_attach,
650 .get_first_cs = mux_pt_get_first_cs,
651 .detach = mux_pt_detach,
652 .avail_streams = mux_pt_avail_streams,
653 .used_streams = mux_pt_used_streams,
654 .destroy = mux_pt_destroy_meth,
655 .ctl = mux_pt_ctl,
656 .shutr = mux_pt_shutr,
657 .shutw = mux_pt_shutw,
658 .flags = MX_FL_NONE|MX_FL_NO_UPG,
659 .name = "PASS",
660};
661
Christopher Faulet32f61c02018-04-10 14:33:41 +0200662/* PROT selection : default mux has empty name */
Christopher Faulet28da3f52021-02-05 16:44:46 +0100663static struct mux_proto_list mux_proto_none =
664 { .token = IST("none"), .mode = PROTO_MODE_TCP, .side = PROTO_SIDE_BOTH, .mux = &mux_pt_ops };
665static struct mux_proto_list mux_proto_tcp =
666 { .token = IST(""), .mode = PROTO_MODE_TCP, .side = PROTO_SIDE_BOTH, .mux = &mux_tcp_ops };
Willy Tarreauf6490822017-09-21 19:43:21 +0200667
Christopher Faulet28da3f52021-02-05 16:44:46 +0100668INITCALL1(STG_REGISTER, register_mux_proto, &mux_proto_none);
669INITCALL1(STG_REGISTER, register_mux_proto, &mux_proto_tcp);