1 /*
2 * Stream management functions.
3 *
4 * Copyright 2000-2012 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 <stdlib.h>
14 #include <unistd.h>
15 #include <fcntl.h>
16
17 #include <import/ebistree.h>
18
19 #include <haproxy/acl.h>
20 #include <haproxy/action.h>
21 #include <haproxy/activity.h>
22 #include <haproxy/api.h>
23 #include <haproxy/applet.h>
24 #include <haproxy/arg.h>
25 #include <haproxy/backend.h>
26 #include <haproxy/capture.h>
27 #include <haproxy/cfgparse.h>
28 #include <haproxy/channel.h>
29 #include <haproxy/check.h>
30 #include <haproxy/cli.h>
31 #include <haproxy/connection.h>
32 #include <haproxy/dict.h>
33 #include <haproxy/dns.h>
34 #include <haproxy/dynbuf.h>
35 #include <haproxy/fd.h>
36 #include <haproxy/filters.h>
37 #include <haproxy/freq_ctr.h>
38 #include <haproxy/frontend.h>
39 #include <haproxy/global.h>
40 #include <haproxy/hlua.h>
41 #include <haproxy/http_ana.h>
42 #include <haproxy/http_rules.h>
43 #include <haproxy/htx.h>
44 #include <haproxy/istbuf.h>
45 #include <haproxy/log.h>
46 #include <haproxy/pipe.h>
47 #include <haproxy/pool.h>
48 #include <haproxy/proxy.h>
49 #include <haproxy/queue.h>
50 #include <haproxy/server.h>
51 #include <haproxy/session.h>
52 #include <haproxy/stats-t.h>
53 #include <haproxy/stick_table.h>
54 #include <haproxy/stream.h>
55 #include <haproxy/stream_interface.h>
56 #include <haproxy/task.h>
57 #include <haproxy/tcp_rules.h>
58 #include <haproxy/thread.h>
59 #include <haproxy/trace.h>
60 #include <haproxy/vars.h>
61
62
63 DECLARE_POOL(pool_head_stream, "stream", sizeof(struct stream));
64 DECLARE_POOL(pool_head_uniqueid, "uniqueid", UNIQUEID_LEN);
65
66 struct list streams = LIST_HEAD_INIT(streams);
67 __decl_spinlock(streams_lock);
68
69 /* List of all use-service keywords. */
70 static struct list service_keywords = LIST_HEAD_INIT(service_keywords);
71
72
73 /* trace source and events */
74 static void strm_trace(enum trace_level level, uint64_t mask,
75 const struct trace_source *src,
76 const struct ist where, const struct ist func,
77 const void *a1, const void *a2, const void *a3, const void *a4);
78
79 /* The event representation is split like this :
80 * strm - stream
81 * si - stream interface
82 * http - http analyzis
83 * tcp - tcp analyzis
84 *
85 * STRM_EV_* macros are defined in <proto/stream.h>
86 */
87 static const struct trace_event strm_trace_events[] = {
88 { .mask = STRM_EV_STRM_NEW, .name = "strm_new", .desc = "new stream" },
89 { .mask = STRM_EV_STRM_FREE, .name = "strm_free", .desc = "release stream" },
90 { .mask = STRM_EV_STRM_ERR, .name = "strm_err", .desc = "error during stream processing" },
91 { .mask = STRM_EV_STRM_ANA, .name = "strm_ana", .desc = "stream analyzers" },
92 { .mask = STRM_EV_STRM_PROC, .name = "strm_proc", .desc = "stream processing" },
93
94 { .mask = STRM_EV_SI_ST, .name = "si_state", .desc = "processing stream-interface states" },
95
96 { .mask = STRM_EV_HTTP_ANA, .name = "http_ana", .desc = "HTTP analyzers" },
97 { .mask = STRM_EV_HTTP_ERR, .name = "http_err", .desc = "error during HTTP analyzis" },
98
99 { .mask = STRM_EV_TCP_ANA, .name = "tcp_ana", .desc = "TCP analyzers" },
100 { .mask = STRM_EV_TCP_ERR, .name = "tcp_err", .desc = "error during TCP analyzis" },
101 {}
102 };
103
104 static const struct name_desc strm_trace_lockon_args[4] = {
105 /* arg1 */ { /* already used by the stream */ },
106 /* arg2 */ { },
107 /* arg3 */ { },
108 /* arg4 */ { }
109 };
110
111 static const struct name_desc strm_trace_decoding[] = {
112 #define STRM_VERB_CLEAN 1
113 { .name="clean", .desc="only user-friendly stuff, generally suitable for level \"user\"" },
114 #define STRM_VERB_MINIMAL 2
115 { .name="minimal", .desc="report info on stream and stream-interfaces" },
116 #define STRM_VERB_SIMPLE 3
117 { .name="simple", .desc="add info on request and response channels" },
118 #define STRM_VERB_ADVANCED 4
119 { .name="advanced", .desc="add info on channel's buffer for data and developer levels only" },
120 #define STRM_VERB_COMPLETE 5
121 { .name="complete", .desc="add info on channel's buffer" },
122 { /* end */ }
123 };
124
125 struct trace_source trace_strm = {
126 .name = IST("stream"),
127 .desc = "Applicative stream",
128 .arg_def = TRC_ARG1_STRM, // TRACE()'s first argument is always a stream
129 .default_cb = strm_trace,
130 .known_events = strm_trace_events,
131 .lockon_args = strm_trace_lockon_args,
132 .decoding = strm_trace_decoding,
133 .report_events = ~0, // report everything by default
134 };
135
136 #define TRACE_SOURCE &trace_strm
137 INITCALL1(STG_REGISTER, trace_register_source, TRACE_SOURCE);
138
139 /* the stream traces always expect that arg1, if non-null, is of a stream (from
140 * which we can derive everything), that arg2, if non-null, is an http
141 * transaction, that arg3, if non-null, is an http message.
142 */
strm_trace(enum trace_level level,uint64_t mask,const struct trace_source * src,const struct ist where,const struct ist func,const void * a1,const void * a2,const void * a3,const void * a4)143 static void strm_trace(enum trace_level level, uint64_t mask, const struct trace_source *src,
144 const struct ist where, const struct ist func,
145 const void *a1, const void *a2, const void *a3, const void *a4)
146 {
147 const struct stream *s = a1;
148 const struct http_txn *txn = a2;
149 const struct http_msg *msg = a3;
150 struct task *task;
151 const struct stream_interface *si_f, *si_b;
152 const struct channel *req, *res;
153 struct htx *htx;
154
155 if (!s || src->verbosity < STRM_VERB_CLEAN)
156 return;
157
158 task = s->task;
159 si_f = &s->si[0];
160 si_b = &s->si[1];
161 req = &s->req;
162 res = &s->res;
163 htx = (msg ? htxbuf(&msg->chn->buf) : NULL);
164
165 /* General info about the stream (htx/tcp, id...) */
166 chunk_appendf(&trace_buf, " : [%u,%s]",
167 s->uniq_id, ((s->flags & SF_HTX) ? "HTX" : "TCP"));
168 if (isttest(s->unique_id)) {
169 chunk_appendf(&trace_buf, " id=");
170 b_putist(&trace_buf, s->unique_id);
171 }
172
173 /* Front and back stream-int state */
174 chunk_appendf(&trace_buf, " SI=(%s,%s)",
175 si_state_str(si_f->state), si_state_str(si_b->state));
176
177 /* If txn is defined, HTTP req/rep states */
178 if (txn)
179 chunk_appendf(&trace_buf, " HTTP=(%s,%s)",
180 h1_msg_state_str(txn->req.msg_state), h1_msg_state_str(txn->rsp.msg_state));
181 if (msg)
182 chunk_appendf(&trace_buf, " %s", ((msg->chn->flags & CF_ISRESP) ? "RESPONSE" : "REQUEST"));
183
184 if (src->verbosity == STRM_VERB_CLEAN)
185 return;
186
187 /* If msg defined, display status-line if possible (verbosity > MINIMAL) */
188 if (src->verbosity > STRM_VERB_MINIMAL && htx && htx_nbblks(htx)) {
189 const struct htx_blk *blk = htx_get_head_blk(htx);
190 const struct htx_sl *sl = htx_get_blk_ptr(htx, blk);
191 enum htx_blk_type type = htx_get_blk_type(blk);
192
193 if (type == HTX_BLK_REQ_SL || type == HTX_BLK_RES_SL)
194 chunk_appendf(&trace_buf, " - \"%.*s %.*s %.*s\"",
195 HTX_SL_P1_LEN(sl), HTX_SL_P1_PTR(sl),
196 HTX_SL_P2_LEN(sl), HTX_SL_P2_PTR(sl),
197 HTX_SL_P3_LEN(sl), HTX_SL_P3_PTR(sl));
198 }
199
200
201 /* If txn defined info about HTTP msgs, otherwise info about SI. */
202 if (txn) {
203 chunk_appendf(&trace_buf, " - t=%p s=(%p,0x%08x) txn.flags=0x%08x, http.flags=(0x%08x,0x%08x) status=%d",
204 task, s, s->flags, txn->flags, txn->req.flags, txn->rsp.flags, txn->status);
205 }
206 else {
207 chunk_appendf(&trace_buf, " - t=%p s=(%p,0x%08x) si_f=(%p,0x%08x,0x%x) si_b=(%p,0x%08x,0x%x) retries=%d",
208 task, s, s->flags, si_f, si_f->flags, si_f->err_type,
209 si_b, si_b->flags, si_b->err_type, si_b->conn_retries);
210 }
211
212 if (src->verbosity == STRM_VERB_MINIMAL)
213 return;
214
215
216 /* If txn defined, don't display all channel info */
217 if (src->verbosity == STRM_VERB_SIMPLE || txn) {
218 chunk_appendf(&trace_buf, " req=(%p .fl=0x%08x .exp(r,w,a)=(%u,%u,%u))",
219 req, req->flags, req->rex, req->wex, req->analyse_exp);
220 chunk_appendf(&trace_buf, " res=(%p .fl=0x%08x .exp(r,w,a)=(%u,%u,%u))",
221 res, res->flags, res->rex, res->wex, res->analyse_exp);
222 }
223 else {
224 chunk_appendf(&trace_buf, " req=(%p .fl=0x%08x .ana=0x%08x .exp(r,w,a)=(%u,%u,%u) .o=%lu .tot=%llu .to_fwd=%u)",
225 req, req->flags, req->analysers, req->rex, req->wex, req->analyse_exp,
226 (long)req->output, req->total, req->to_forward);
227 chunk_appendf(&trace_buf, " res=(%p .fl=0x%08x .ana=0x%08x .exp(r,w,a)=(%u,%u,%u) .o=%lu .tot=%llu .to_fwd=%u)",
228 res, res->flags, res->analysers, res->rex, res->wex, res->analyse_exp,
229 (long)res->output, res->total, res->to_forward);
230 }
231
232 if (src->verbosity == STRM_VERB_SIMPLE ||
233 (src->verbosity == STRM_VERB_ADVANCED && src->level < TRACE_LEVEL_DATA))
234 return;
235
236 /* channels' buffer info */
237 if (s->flags & SF_HTX) {
238 struct htx *rqhtx = htxbuf(&req->buf);
239 struct htx *rphtx = htxbuf(&res->buf);
240
241 chunk_appendf(&trace_buf, " htx=(%u/%u#%u, %u/%u#%u)",
242 rqhtx->data, rqhtx->size, htx_nbblks(rqhtx),
243 rphtx->data, rphtx->size, htx_nbblks(rphtx));
244 }
245 else {
246 chunk_appendf(&trace_buf, " buf=(%u@%p+%u/%u, %u@%p+%u/%u)",
247 (unsigned int)b_data(&req->buf), b_orig(&req->buf),
248 (unsigned int)b_head_ofs(&req->buf), (unsigned int)b_size(&req->buf),
249 (unsigned int)b_data(&req->buf), b_orig(&req->buf),
250 (unsigned int)b_head_ofs(&req->buf), (unsigned int)b_size(&req->buf));
251 }
252
253 /* If msg defined, display htx info if defined (level > USER) */
254 if (src->level > TRACE_LEVEL_USER && htx && htx_nbblks(htx)) {
255 int full = 0;
256
257 /* Full htx info (level > STATE && verbosity > SIMPLE) */
258 if (src->level > TRACE_LEVEL_STATE) {
259 if (src->verbosity == STRM_VERB_COMPLETE)
260 full = 1;
261 }
262
263 chunk_memcat(&trace_buf, "\n\t", 2);
264 htx_dump(&trace_buf, htx, full);
265 }
266 }
267
268 /* Create a new stream for connection <conn>. Return < 0 on error. This is only
269 * valid right after the handshake, before the connection's data layer is
270 * initialized, because it relies on the session to be in conn->owner.
271 */
stream_create_from_cs(struct conn_stream * cs)272 int stream_create_from_cs(struct conn_stream *cs)
273 {
274 struct stream *strm;
275
276 strm = stream_new(cs->conn->owner, &cs->obj_type);
277 if (strm == NULL)
278 return -1;
279
280 task_wakeup(strm->task, TASK_WOKEN_INIT);
281 return 0;
282 }
283
284 /* Callback used to wake up a stream when an input buffer is available. The
285 * stream <s>'s stream interfaces are checked for a failed buffer allocation
286 * as indicated by the presence of the SI_FL_RXBLK_ROOM flag and the lack of a
287 * buffer, and and input buffer is assigned there (at most one). The function
288 * returns 1 and wakes the stream up if a buffer was taken, otherwise zero.
289 * It's designed to be called from __offer_buffer().
290 */
stream_buf_available(void * arg)291 int stream_buf_available(void *arg)
292 {
293 struct stream *s = arg;
294
295 if (!s->req.buf.size && !s->req.pipe && (s->si[0].flags & SI_FL_RXBLK_BUFF) &&
296 b_alloc_margin(&s->req.buf, global.tune.reserved_bufs))
297 si_rx_buff_rdy(&s->si[0]);
298 else if (!s->res.buf.size && !s->res.pipe && (s->si[1].flags & SI_FL_RXBLK_BUFF) &&
299 b_alloc_margin(&s->res.buf, 0))
300 si_rx_buff_rdy(&s->si[1]);
301 else
302 return 0;
303
304 task_wakeup(s->task, TASK_WOKEN_RES);
305 return 1;
306
307 }
308
309 /* This function is called from the session handler which detects the end of
310 * handshake, in order to complete initialization of a valid stream. It must be
311 * called with a completely initialized session. It returns the pointer to
312 * the newly created stream, or NULL in case of fatal error. The client-facing
313 * end point is assigned to <origin>, which must be valid. The stream's task
314 * is configured with a nice value inherited from the listener's nice if any.
315 * The task's context is set to the new stream, and its function is set to
316 * process_stream(). Target and analysers are null.
317 */
stream_new(struct session * sess,enum obj_type * origin)318 struct stream *stream_new(struct session *sess, enum obj_type *origin)
319 {
320 struct stream *s;
321 struct task *t;
322 struct conn_stream *cs = objt_cs(origin);
323 struct appctx *appctx = objt_appctx(origin);
324 const struct cs_info *csinfo;
325
326 DBG_TRACE_ENTER(STRM_EV_STRM_NEW);
327 if (unlikely((s = pool_alloc(pool_head_stream)) == NULL))
328 goto out_fail_alloc;
329
330 /* minimum stream initialization required for an embryonic stream is
331 * fairly low. We need very little to execute L4 ACLs, then we need a
332 * task to make the client-side connection live on its own.
333 * - flags
334 * - stick-entry tracking
335 */
336 s->flags = 0;
337 s->logs.logwait = sess->fe->to_log;
338 s->logs.level = 0;
339 tv_zero(&s->logs.tv_request);
340 s->logs.t_queue = -1;
341 s->logs.t_connect = -1;
342 s->logs.t_data = -1;
343 s->logs.t_close = 0;
344 s->logs.bytes_in = s->logs.bytes_out = 0;
345 s->logs.prx_queue_pos = 0; /* we get the number of pending conns before us */
346 s->logs.srv_queue_pos = 0; /* we will get this number soon */
347 s->obj_type = OBJ_TYPE_STREAM;
348
349 csinfo = si_get_cs_info(cs);
350 if (csinfo) {
351 s->logs.accept_date = csinfo->create_date;
352 s->logs.tv_accept = csinfo->tv_create;
353 s->logs.t_handshake = csinfo->t_handshake;
354 s->logs.t_idle = csinfo->t_idle;
355 }
356 else {
357 s->logs.accept_date = sess->accept_date;
358 s->logs.tv_accept = sess->tv_accept;
359 s->logs.t_handshake = sess->t_handshake;
360 s->logs.t_idle = -1;
361 }
362
363 /* default logging function */
364 s->do_log = strm_log;
365
366 /* default error reporting function, may be changed by analysers */
367 s->srv_error = default_srv_error;
368
369 /* Initialise the current rule list pointer to NULL. We are sure that
370 * any rulelist match the NULL pointer.
371 */
372 s->current_rule_list = NULL;
373 s->current_rule = NULL;
374
375 /* Copy SC counters for the stream. We don't touch refcounts because
376 * any reference we have is inherited from the session. Since the stream
377 * doesn't exist without the session, the session's existence guarantees
378 * we don't lose the entry. During the store operation, the stream won't
379 * touch these ones.
380 */
381 memcpy(s->stkctr, sess->stkctr, sizeof(s->stkctr));
382
383 s->sess = sess;
384 s->si[0].flags = SI_FL_NONE;
385 s->si[1].flags = SI_FL_ISBACK;
386
387 s->uniq_id = _HA_ATOMIC_XADD(&global.req_count, 1);
388
389 /* OK, we're keeping the stream, so let's properly initialize the stream */
390 LIST_INIT(&s->back_refs);
391
392 MT_LIST_INIT(&s->buffer_wait.list);
393 s->buffer_wait.target = s;
394 s->buffer_wait.wakeup_cb = stream_buf_available;
395
396 s->call_rate.curr_sec = s->call_rate.curr_ctr = s->call_rate.prev_ctr = 0;
397 s->pcli_next_pid = 0;
398 s->pcli_flags = 0;
399 s->unique_id = IST_NULL;
400
401 if ((t = task_new(tid_bit)) == NULL)
402 goto out_fail_alloc;
403
404 s->task = t;
405 s->pending_events = 0;
406 t->process = process_stream;
407 t->context = s;
408 t->expire = TICK_ETERNITY;
409 if (sess->listener)
410 t->nice = sess->listener->nice;
411
412 /* Note: initially, the stream's backend points to the frontend.
413 * This changes later when switching rules are executed or
414 * when the default backend is assigned.
415 */
416 s->be = sess->fe;
417 s->req.buf = BUF_NULL;
418 s->res.buf = BUF_NULL;
419 s->req_cap = NULL;
420 s->res_cap = NULL;
421
422 /* Initialise all the variables contexts even if not used.
423 * This permits to prune these contexts without errors.
424 */
425 vars_init(&s->vars_txn, SCOPE_TXN);
426 vars_init(&s->vars_reqres, SCOPE_REQ);
427
428 /* this part should be common with other protocols */
429 if (si_reset(&s->si[0]) < 0)
430 goto out_fail_alloc;
431 si_set_state(&s->si[0], SI_ST_EST);
432 s->si[0].hcto = sess->fe->timeout.clientfin;
433
434 if (cs && cs->conn->mux) {
435 if (cs->conn->mux->flags & MX_FL_CLEAN_ABRT)
436 s->si[0].flags |= SI_FL_CLEAN_ABRT;
437 if (cs->conn->mux->flags & MX_FL_HTX)
438 s->flags |= SF_HTX;
439 }
440 /* Set SF_HTX flag for HTTP frontends. */
441 if (sess->fe->mode == PR_MODE_HTTP)
442 s->flags |= SF_HTX;
443
444 /* attach the incoming connection to the stream interface now. */
445 if (cs)
446 si_attach_cs(&s->si[0], cs);
447 else if (appctx)
448 si_attach_appctx(&s->si[0], appctx);
449
450 if (likely(sess->fe->options2 & PR_O2_INDEPSTR))
451 s->si[0].flags |= SI_FL_INDEP_STR;
452
453 /* pre-initialize the other side's stream interface to an INIT state. The
454 * callbacks will be initialized before attempting to connect.
455 */
456 if (si_reset(&s->si[1]) < 0)
457 goto out_fail_alloc_si1;
458 s->si[1].hcto = TICK_ETERNITY;
459
460 if (likely(sess->fe->options2 & PR_O2_INDEPSTR))
461 s->si[1].flags |= SI_FL_INDEP_STR;
462
463 stream_init_srv_conn(s);
464 s->target = sess->listener ? sess->listener->default_target : NULL;
465 s->target_addr = NULL;
466
467 s->pend_pos = NULL;
468 s->priority_class = 0;
469 s->priority_offset = 0;
470
471 /* init store persistence */
472 s->store_count = 0;
473
474 channel_init(&s->req);
475 s->req.flags |= CF_READ_ATTACHED; /* the producer is already connected */
476 s->req.analysers = sess->listener ? sess->listener->analysers : 0;
477
478 if (!sess->fe->fe_req_ana) {
479 channel_auto_connect(&s->req); /* don't wait to establish connection */
480 channel_auto_close(&s->req); /* let the producer forward close requests */
481 }
482
483 s->req.rto = sess->fe->timeout.client;
484 s->req.wto = TICK_ETERNITY;
485 s->req.rex = TICK_ETERNITY;
486 s->req.wex = TICK_ETERNITY;
487 s->req.analyse_exp = TICK_ETERNITY;
488
489 channel_init(&s->res);
490 s->res.flags |= CF_ISRESP;
491 s->res.analysers = 0;
492
493 if (sess->fe->options2 & PR_O2_NODELAY) {
494 s->req.flags |= CF_NEVER_WAIT;
495 s->res.flags |= CF_NEVER_WAIT;
496 }
497
498 s->res.wto = sess->fe->timeout.client;
499 s->res.rto = TICK_ETERNITY;
500 s->res.rex = TICK_ETERNITY;
501 s->res.wex = TICK_ETERNITY;
502 s->res.analyse_exp = TICK_ETERNITY;
503
504 s->txn = NULL;
505 s->hlua = NULL;
506
507 s->dns_ctx.dns_requester = NULL;
508 s->dns_ctx.hostname_dn = NULL;
509 s->dns_ctx.hostname_dn_len = 0;
510 s->dns_ctx.parent = NULL;
511
512 HA_SPIN_LOCK(STRMS_LOCK, &streams_lock);
513 LIST_ADDQ(&streams, &s->list);
514 HA_SPIN_UNLOCK(STRMS_LOCK, &streams_lock);
515
516 if (flt_stream_init(s) < 0 || flt_stream_start(s) < 0)
517 goto out_fail_accept;
518
519 s->si[1].l7_buffer = BUF_NULL;
520 /* finish initialization of the accepted file descriptor */
521 if (appctx)
522 si_want_get(&s->si[0]);
523
524 if (sess->fe->accept && sess->fe->accept(s) < 0)
525 goto out_fail_accept;
526
527 /* it is important not to call the wakeup function directly but to
528 * pass through task_wakeup(), because this one knows how to apply
529 * priorities to tasks. Using multi thread we must be sure that
530 * stream is fully initialized before calling task_wakeup. So
531 * the caller must handle the task_wakeup
532 */
533 DBG_TRACE_LEAVE(STRM_EV_STRM_NEW, s);
534 return s;
535
536 /* Error unrolling */
537 out_fail_accept:
538 flt_stream_release(s, 0);
539 task_destroy(t);
540 tasklet_free(s->si[1].wait_event.tasklet);
541 LIST_DEL(&s->list);
542 out_fail_alloc_si1:
543 tasklet_free(s->si[0].wait_event.tasklet);
544 out_fail_alloc:
545 pool_free(pool_head_stream, s);
546 DBG_TRACE_DEVEL("leaving on error", STRM_EV_STRM_NEW|STRM_EV_STRM_ERR);
547 return NULL;
548 }
549
550 /*
551 * frees the context associated to a stream. It must have been removed first.
552 */
stream_free(struct stream * s)553 static void stream_free(struct stream *s)
554 {
555 struct session *sess = strm_sess(s);
556 struct proxy *fe = sess->fe;
557 struct bref *bref, *back;
558 struct conn_stream *cli_cs = objt_cs(s->si[0].end);
559 int must_free_sess;
560 int i;
561
562 DBG_TRACE_POINT(STRM_EV_STRM_FREE, s);
563
564 /* detach the stream from its own task before even releasing it so
565 * that walking over a task list never exhibits a dying stream.
566 */
567 s->task->context = NULL;
568 __ha_barrier_store();
569
570 pendconn_free(s);
571
572 if (objt_server(s->target)) { /* there may be requests left pending in queue */
573 if (s->flags & SF_CURR_SESS) {
574 s->flags &= ~SF_CURR_SESS;
575 _HA_ATOMIC_SUB(&__objt_server(s->target)->cur_sess, 1);
576 }
577 if (may_dequeue_tasks(objt_server(s->target), s->be))
578 process_srv_queue(objt_server(s->target), 0);
579 }
580
581 if (unlikely(s->srv_conn)) {
582 /* the stream still has a reserved slot on a server, but
583 * it should normally be only the same as the one above,
584 * so this should not happen in fact.
585 */
586 sess_change_server(s, NULL);
587 }
588
589 if (s->req.pipe)
590 put_pipe(s->req.pipe);
591
592 if (s->res.pipe)
593 put_pipe(s->res.pipe);
594
595 /* We may still be present in the buffer wait queue */
596 if (MT_LIST_ADDED(&s->buffer_wait.list))
597 MT_LIST_DEL(&s->buffer_wait.list);
598
599 if (s->req.buf.size || s->res.buf.size) {
600 b_free(&s->req.buf);
601 b_free(&s->res.buf);
602 offer_buffers(NULL, tasks_run_queue);
603 }
604
605 pool_free(pool_head_uniqueid, s->unique_id.ptr);
606 s->unique_id = IST_NULL;
607
608 hlua_ctx_destroy(s->hlua);
609 s->hlua = NULL;
610 if (s->txn)
611 http_end_txn(s);
612
613 /* ensure the client-side transport layer is destroyed */
614 if (cli_cs)
615 cs_close(cli_cs);
616
617 for (i = 0; i < s->store_count; i++) {
618 if (!s->store[i].ts)
619 continue;
620 stksess_free(s->store[i].table, s->store[i].ts);
621 s->store[i].ts = NULL;
622 }
623
624 if (s->txn) {
625 pool_free(pool_head_http_txn, s->txn);
626 s->txn = NULL;
627 }
628
629 if (s->dns_ctx.dns_requester) {
630 __decl_thread(struct dns_resolvers *resolvers = s->dns_ctx.parent->arg.dns.resolvers);
631
632 HA_SPIN_LOCK(DNS_LOCK, &resolvers->lock);
633 free(s->dns_ctx.hostname_dn); s->dns_ctx.hostname_dn = NULL;
634 s->dns_ctx.hostname_dn_len = 0;
635 dns_unlink_resolution(s->dns_ctx.dns_requester);
636 HA_SPIN_UNLOCK(DNS_LOCK, &resolvers->lock);
637
638 pool_free(dns_requester_pool, s->dns_ctx.dns_requester);
639 s->dns_ctx.dns_requester = NULL;
640 }
641
642 flt_stream_stop(s);
643 flt_stream_release(s, 0);
644
645 if (fe) {
646 if (s->req_cap) {
647 struct cap_hdr *h;
648 for (h = fe->req_cap; h; h = h->next)
649 pool_free(h->pool, s->req_cap[h->index]);
650 }
651
652 if (s->res_cap) {
653 struct cap_hdr *h;
654 for (h = fe->rsp_cap; h; h = h->next)
655 pool_free(h->pool, s->res_cap[h->index]);
656 }
657
658 pool_free(fe->rsp_cap_pool, s->res_cap);
659 pool_free(fe->req_cap_pool, s->req_cap);
660 }
661
662 /* Cleanup all variable contexts. */
663 if (!LIST_ISEMPTY(&s->vars_txn.head))
664 vars_prune(&s->vars_txn, s->sess, s);
665 if (!LIST_ISEMPTY(&s->vars_reqres.head))
666 vars_prune(&s->vars_reqres, s->sess, s);
667
668 stream_store_counters(s);
669
670 HA_SPIN_LOCK(STRMS_LOCK, &streams_lock);
671 list_for_each_entry_safe(bref, back, &s->back_refs, users) {
672 /* we have to unlink all watchers. We must not relink them if
673 * this stream was the last one in the list.
674 */
675 LIST_DEL(&bref->users);
676 LIST_INIT(&bref->users);
677 if (s->list.n != &streams)
678 LIST_ADDQ(&LIST_ELEM(s->list.n, struct stream *, list)->back_refs, &bref->users);
679 bref->ref = s->list.n;
680 }
681 LIST_DEL(&s->list);
682 HA_SPIN_UNLOCK(STRMS_LOCK, &streams_lock);
683
684 /* applets do not release session yet */
685 must_free_sess = objt_appctx(sess->origin) && sess->origin == s->si[0].end;
686
687
688 si_release_endpoint(&s->si[1]);
689 si_release_endpoint(&s->si[0]);
690
691 tasklet_free(s->si[0].wait_event.tasklet);
692 tasklet_free(s->si[1].wait_event.tasklet);
693
694 b_free(&s->si[1].l7_buffer);
695 if (must_free_sess) {
696 sess->origin = NULL;
697 session_free(sess);
698 }
699
700 sockaddr_free(&s->target_addr);
701 pool_free(pool_head_stream, s);
702
703 /* We may want to free the maximum amount of pools if the proxy is stopping */
704 if (fe && unlikely(fe->state == PR_STSTOPPED)) {
705 pool_flush(pool_head_buffer);
706 pool_flush(pool_head_http_txn);
707 pool_flush(pool_head_requri);
708 pool_flush(pool_head_capture);
709 pool_flush(pool_head_stream);
710 pool_flush(pool_head_session);
711 pool_flush(pool_head_connection);
712 pool_flush(pool_head_pendconn);
713 pool_flush(fe->req_cap_pool);
714 pool_flush(fe->rsp_cap_pool);
715 }
716 }
717
718
719 /* Allocates a work buffer for stream <s>. It is meant to be called inside
720 * process_stream(). It will only allocate the side needed for the function
721 * to work fine, which is the response buffer so that an error message may be
722 * built and returned. Response buffers may be allocated from the reserve, this
723 * is critical to ensure that a response may always flow and will never block a
724 * server from releasing a connection. Returns 0 in case of failure, non-zero
725 * otherwise.
726 */
stream_alloc_work_buffer(struct stream * s)727 static int stream_alloc_work_buffer(struct stream *s)
728 {
729 if (MT_LIST_ADDED(&s->buffer_wait.list))
730 MT_LIST_DEL(&s->buffer_wait.list);
731
732 if (b_alloc_margin(&s->res.buf, 0))
733 return 1;
734
735 MT_LIST_ADDQ(&buffer_wq, &s->buffer_wait.list);
736 return 0;
737 }
738
739 /* releases unused buffers after processing. Typically used at the end of the
740 * update() functions. It will try to wake up as many tasks/applets as the
741 * number of buffers that it releases. In practice, most often streams are
742 * blocked on a single buffer, so it makes sense to try to wake two up when two
743 * buffers are released at once.
744 */
stream_release_buffers(struct stream * s)745 void stream_release_buffers(struct stream *s)
746 {
747 int offer = 0;
748
749 if (c_size(&s->req) && c_empty(&s->req)) {
750 offer = 1;
751 b_free(&s->req.buf);
752 }
753 if (c_size(&s->res) && c_empty(&s->res)) {
754 offer = 1;
755 b_free(&s->res.buf);
756 }
757
758 /* if we're certain to have at least 1 buffer available, and there is
759 * someone waiting, we can wake up a waiter and offer them.
760 */
761 if (offer)
762 offer_buffers(s, tasks_run_queue);
763 }
764
stream_process_counters(struct stream * s)765 void stream_process_counters(struct stream *s)
766 {
767 struct session *sess = s->sess;
768 unsigned long long bytes;
769 void *ptr1,*ptr2;
770 struct stksess *ts;
771 int i;
772
773 bytes = s->req.total - s->logs.bytes_in;
774 s->logs.bytes_in = s->req.total;
775 if (bytes) {
776 _HA_ATOMIC_ADD(&sess->fe->fe_counters.bytes_in, bytes);
777 _HA_ATOMIC_ADD(&s->be->be_counters.bytes_in, bytes);
778
779 if (objt_server(s->target))
780 _HA_ATOMIC_ADD(&objt_server(s->target)->counters.bytes_in, bytes);
781
782 if (sess->listener && sess->listener->counters)
783 _HA_ATOMIC_ADD(&sess->listener->counters->bytes_in, bytes);
784
785 for (i = 0; i < MAX_SESS_STKCTR; i++) {
786 struct stkctr *stkctr = &s->stkctr[i];
787
788 ts = stkctr_entry(stkctr);
789 if (!ts) {
790 stkctr = &sess->stkctr[i];
791 ts = stkctr_entry(stkctr);
792 if (!ts)
793 continue;
794 }
795
796 HA_RWLOCK_WRLOCK(STK_SESS_LOCK, &ts->lock);
797 ptr1 = stktable_data_ptr(stkctr->table, ts, STKTABLE_DT_BYTES_IN_CNT);
798 if (ptr1)
799 stktable_data_cast(ptr1, bytes_in_cnt) += bytes;
800
801 ptr2 = stktable_data_ptr(stkctr->table, ts, STKTABLE_DT_BYTES_IN_RATE);
802 if (ptr2)
803 update_freq_ctr_period(&stktable_data_cast(ptr2, bytes_in_rate),
804 stkctr->table->data_arg[STKTABLE_DT_BYTES_IN_RATE].u, bytes);
805 HA_RWLOCK_WRUNLOCK(STK_SESS_LOCK, &ts->lock);
806
807 /* If data was modified, we need to touch to re-schedule sync */
808 if (ptr1 || ptr2)
809 stktable_touch_local(stkctr->table, ts, 0);
810 }
811 }
812
813 bytes = s->res.total - s->logs.bytes_out;
814 s->logs.bytes_out = s->res.total;
815 if (bytes) {
816 _HA_ATOMIC_ADD(&sess->fe->fe_counters.bytes_out, bytes);
817 _HA_ATOMIC_ADD(&s->be->be_counters.bytes_out, bytes);
818
819 if (objt_server(s->target))
820 _HA_ATOMIC_ADD(&objt_server(s->target)->counters.bytes_out, bytes);
821
822 if (sess->listener && sess->listener->counters)
823 _HA_ATOMIC_ADD(&sess->listener->counters->bytes_out, bytes);
824
825 for (i = 0; i < MAX_SESS_STKCTR; i++) {
826 struct stkctr *stkctr = &s->stkctr[i];
827
828 ts = stkctr_entry(stkctr);
829 if (!ts) {
830 stkctr = &sess->stkctr[i];
831 ts = stkctr_entry(stkctr);
832 if (!ts)
833 continue;
834 }
835
836 HA_RWLOCK_WRLOCK(STK_SESS_LOCK, &ts->lock);
837 ptr1 = stktable_data_ptr(stkctr->table, ts, STKTABLE_DT_BYTES_OUT_CNT);
838 if (ptr1)
839 stktable_data_cast(ptr1, bytes_out_cnt) += bytes;
840
841 ptr2 = stktable_data_ptr(stkctr->table, ts, STKTABLE_DT_BYTES_OUT_RATE);
842 if (ptr2)
843 update_freq_ctr_period(&stktable_data_cast(ptr2, bytes_out_rate),
844 stkctr->table->data_arg[STKTABLE_DT_BYTES_OUT_RATE].u, bytes);
845 HA_RWLOCK_WRUNLOCK(STK_SESS_LOCK, &ts->lock);
846
847 /* If data was modified, we need to touch to re-schedule sync */
848 if (ptr1 || ptr2)
849 stktable_touch_local(stkctr->table, stkctr_entry(stkctr), 0);
850 }
851 }
852 }
853
854 /*
855 * This function handles the transition between the SI_ST_CON state and the
856 * SI_ST_EST state. It must only be called after switching from SI_ST_CON (or
857 * SI_ST_INI or SI_ST_RDY) to SI_ST_EST, but only when a ->proto is defined.
858 * Note that it will switch the interface to SI_ST_DIS if we already have
859 * the CF_SHUTR flag, it means we were able to forward the request, and
860 * receive the response, before process_stream() had the opportunity to
861 * make the switch from SI_ST_CON to SI_ST_EST. When that happens, we want
862 * to go through back_establish() anyway, to make sure the analysers run.
863 * Timeouts are cleared. Error are reported on the channel so that analysers
864 * can handle them.
865 */
back_establish(struct stream * s)866 static void back_establish(struct stream *s)
867 {
868 struct stream_interface *si = &s->si[1];
869 struct conn_stream *srv_cs = objt_cs(si->end);
870 struct connection *conn = srv_cs ? srv_cs->conn : objt_conn(si->end);
871 struct channel *req = &s->req;
872 struct channel *rep = &s->res;
873
874 DBG_TRACE_ENTER(STRM_EV_STRM_PROC|STRM_EV_SI_ST, s);
875 /* First, centralize the timers information, and clear any irrelevant
876 * timeout.
877 */
878 s->logs.t_connect = tv_ms_elapsed(&s->logs.tv_accept, &now);
879 si->exp = TICK_ETERNITY;
880 si->flags &= ~SI_FL_EXP;
881
882 /* errors faced after sending data need to be reported */
883 if (si->flags & SI_FL_ERR && req->flags & CF_WROTE_DATA) {
884 /* Don't add CF_WRITE_ERROR if we're here because
885 * early data were rejected by the server, or
886 * http_wait_for_response() will never be called
887 * to send a 425.
888 */
889 if (conn && conn->err_code != CO_ER_SSL_EARLY_FAILED)
890 req->flags |= CF_WRITE_ERROR;
891 rep->flags |= CF_READ_ERROR;
892 si->err_type = SI_ET_DATA_ERR;
893 DBG_TRACE_STATE("read/write error", STRM_EV_STRM_PROC|STRM_EV_SI_ST|STRM_EV_STRM_ERR, s);
894 }
895
896 if (objt_server(s->target))
897 health_adjust(objt_server(s->target), HANA_STATUS_L4_OK);
898
899 if (s->be->mode == PR_MODE_TCP) { /* let's allow immediate data connection in this case */
900 /* if the user wants to log as soon as possible, without counting
901 * bytes from the server, then this is the right moment. */
902 if (!LIST_ISEMPTY(&strm_fe(s)->logformat) && !(s->logs.logwait & LW_BYTES)) {
903 /* note: no pend_pos here, session is established */
904 s->logs.t_close = s->logs.t_connect; /* to get a valid end date */
905 s->do_log(s);
906 }
907 }
908 else {
909 rep->flags |= CF_READ_DONTWAIT; /* a single read is enough to get response headers */
910 }
911
912 rep->analysers |= strm_fe(s)->fe_rsp_ana | s->be->be_rsp_ana;
913
914 /* Be sure to filter response headers if the backend is an HTTP proxy
915 * and if there are filters attached to the stream. */
916 if (s->be->mode == PR_MODE_HTTP && HAS_FILTERS(s))
917 rep->analysers |= AN_RES_FLT_HTTP_HDRS;
918
919 si_rx_endp_more(si);
920 rep->flags |= CF_READ_ATTACHED; /* producer is now attached */
921 if (objt_cs(si->end)) {
922 /* real connections have timeouts */
923 req->wto = s->be->timeout.server;
924 rep->rto = s->be->timeout.server;
925 /* The connection is now established, try to read data from the
926 * underlying layer, and subscribe to recv events. We use a
927 * delayed recv here to give a chance to the data to flow back
928 * by the time we process other tasks.
929 */
930 si_chk_rcv(si);
931 }
932 req->wex = TICK_ETERNITY;
933 /* If we managed to get the whole response, and we don't have anything
934 * left to send, or can't, switch to SI_ST_DIS now. */
935 if (rep->flags & (CF_SHUTR | CF_SHUTW)) {
936 si->state = SI_ST_DIS;
937 DBG_TRACE_STATE("response channel shutdwn for read/write", STRM_EV_STRM_PROC|STRM_EV_SI_ST|STRM_EV_STRM_ERR, s);
938 }
939
940 DBG_TRACE_LEAVE(STRM_EV_STRM_PROC|STRM_EV_SI_ST, s);
941 }
942
943 /* Set correct stream termination flags in case no analyser has done it. It
944 * also counts a failed request if the server state has not reached the request
945 * stage.
946 */
sess_set_term_flags(struct stream * s)947 static void sess_set_term_flags(struct stream *s)
948 {
949 if (!(s->flags & SF_FINST_MASK)) {
950 if (s->si[1].state == SI_ST_INI) {
951 /* anything before REQ in fact */
952 _HA_ATOMIC_ADD(&strm_fe(s)->fe_counters.failed_req, 1);
953 if (strm_li(s) && strm_li(s)->counters)
954 _HA_ATOMIC_ADD(&strm_li(s)->counters->failed_req, 1);
955
956 s->flags |= SF_FINST_R;
957 }
958 else if (s->si[1].state == SI_ST_QUE)
959 s->flags |= SF_FINST_Q;
960 else if (si_state_in(s->si[1].state, SI_SB_REQ|SI_SB_TAR|SI_SB_ASS|SI_SB_CON|SI_SB_CER|SI_SB_RDY))
961 s->flags |= SF_FINST_C;
962 else if (s->si[1].state == SI_ST_EST || s->si[1].prev_state == SI_ST_EST)
963 s->flags |= SF_FINST_D;
964 else
965 s->flags |= SF_FINST_L;
966 }
967 }
968
969 /* This function parses the use-service action ruleset. It executes
970 * the associated ACL and set an applet as a stream or txn final node.
971 * it returns ACT_RET_ERR if an error occurs, the proxy left in
972 * consistent state. It returns ACT_RET_STOP in success case because
973 * use-service must be a terminal action. Returns ACT_RET_YIELD
974 * if the initialisation function require more data.
975 */
process_use_service(struct act_rule * rule,struct proxy * px,struct session * sess,struct stream * s,int flags)976 enum act_return process_use_service(struct act_rule *rule, struct proxy *px,
977 struct session *sess, struct stream *s, int flags)
978
979 {
980 struct appctx *appctx;
981
982 /* Initialises the applet if it is required. */
983 if (flags & ACT_OPT_FIRST) {
984 /* Register applet. this function schedules the applet. */
985 s->target = &rule->applet.obj_type;
986 if (unlikely(!si_register_handler(&s->si[1], objt_applet(s->target))))
987 return ACT_RET_ERR;
988
989 /* Initialise the context. */
990 appctx = si_appctx(&s->si[1]);
991 memset(&appctx->ctx, 0, sizeof(appctx->ctx));
992 appctx->rule = rule;
993 }
994 else
995 appctx = si_appctx(&s->si[1]);
996
997 /* Stops the applet scheduling, in case of the init function miss
998 * some data.
999 */
1000 si_stop_get(&s->si[1]);
1001
1002 /* Call initialisation. */
1003 if (rule->applet.init)
1004 switch (rule->applet.init(appctx, px, s)) {
1005 case 0: return ACT_RET_ERR;
1006 case 1: break;
1007 default: return ACT_RET_YIELD;
1008 }
1009
1010 if (rule->from != ACT_F_HTTP_REQ) {
1011 if (sess->fe == s->be) /* report it if the request was intercepted by the frontend */
1012 _HA_ATOMIC_ADD(&sess->fe->fe_counters.intercepted_req, 1);
1013
1014 /* The flag SF_ASSIGNED prevent from server assignment. */
1015 s->flags |= SF_ASSIGNED;
1016 }
1017
1018 /* Now we can schedule the applet. */
1019 si_cant_get(&s->si[1]);
1020 appctx_wakeup(appctx);
1021 return ACT_RET_STOP;
1022 }
1023
1024 /* This stream analyser checks the switching rules and changes the backend
1025 * if appropriate. The default_backend rule is also considered, then the
1026 * target backend's forced persistence rules are also evaluated last if any.
1027 * It returns 1 if the processing can continue on next analysers, or zero if it
1028 * either needs more data or wants to immediately abort the request.
1029 */
process_switching_rules(struct stream * s,struct channel * req,int an_bit)1030 static int process_switching_rules(struct stream *s, struct channel *req, int an_bit)
1031 {
1032 struct persist_rule *prst_rule;
1033 struct session *sess = s->sess;
1034 struct proxy *fe = sess->fe;
1035
1036 req->analysers &= ~an_bit;
1037 req->analyse_exp = TICK_ETERNITY;
1038
1039 DBG_TRACE_ENTER(STRM_EV_STRM_ANA, s);
1040
1041 /* now check whether we have some switching rules for this request */
1042 if (!(s->flags & SF_BE_ASSIGNED)) {
1043 struct switching_rule *rule;
1044
1045 list_for_each_entry(rule, &fe->switching_rules, list) {
1046 int ret = 1;
1047
1048 if (rule->cond) {
1049 ret = acl_exec_cond(rule->cond, fe, sess, s, SMP_OPT_DIR_REQ|SMP_OPT_FINAL);
1050 ret = acl_pass(ret);
1051 if (rule->cond->pol == ACL_COND_UNLESS)
1052 ret = !ret;
1053 }
1054
1055 if (ret) {
1056 /* If the backend name is dynamic, try to resolve the name.
1057 * If we can't resolve the name, or if any error occurs, break
1058 * the loop and fallback to the default backend.
1059 */
1060 struct proxy *backend = NULL;
1061
1062 if (rule->dynamic) {
1063 struct buffer *tmp;
1064
1065 tmp = alloc_trash_chunk();
1066 if (!tmp)
1067 goto sw_failed;
1068
1069 if (build_logline(s, tmp->area, tmp->size, &rule->be.expr))
1070 backend = proxy_be_by_name(tmp->area);
1071
1072 free_trash_chunk(tmp);
1073 tmp = NULL;
1074
1075 if (!backend)
1076 break;
1077 }
1078 else
1079 backend = rule->be.backend;
1080
1081 if (!stream_set_backend(s, backend))
1082 goto sw_failed;
1083 break;
1084 }
1085 }
1086
1087 /* To ensure correct connection accounting on the backend, we
1088 * have to assign one if it was not set (eg: a listen). This
1089 * measure also takes care of correctly setting the default
1090 * backend if any.
1091 */
1092 if (!(s->flags & SF_BE_ASSIGNED))
1093 if (!stream_set_backend(s, fe->defbe.be ? fe->defbe.be : s->be))
1094 goto sw_failed;
1095 }
1096
1097 /* we don't want to run the TCP or HTTP filters again if the backend has not changed */
1098 if (fe == s->be) {
1099 s->req.analysers &= ~AN_REQ_INSPECT_BE;
1100 s->req.analysers &= ~AN_REQ_HTTP_PROCESS_BE;
1101 s->req.analysers &= ~AN_REQ_FLT_START_BE;
1102 }
1103
1104 /* as soon as we know the backend, we must check if we have a matching forced or ignored
1105 * persistence rule, and report that in the stream.
1106 */
1107 list_for_each_entry(prst_rule, &s->be->persist_rules, list) {
1108 int ret = 1;
1109
1110 if (prst_rule->cond) {
1111 ret = acl_exec_cond(prst_rule->cond, s->be, sess, s, SMP_OPT_DIR_REQ|SMP_OPT_FINAL);
1112 ret = acl_pass(ret);
1113 if (prst_rule->cond->pol == ACL_COND_UNLESS)
1114 ret = !ret;
1115 }
1116
1117 if (ret) {
1118 /* no rule, or the rule matches */
1119 if (prst_rule->type == PERSIST_TYPE_FORCE) {
1120 s->flags |= SF_FORCE_PRST;
1121 } else {
1122 s->flags |= SF_IGNORE_PRST;
1123 }
1124 break;
1125 }
1126 }
1127
1128 DBG_TRACE_LEAVE(STRM_EV_STRM_ANA, s);
1129 return 1;
1130
1131 sw_failed:
1132 /* immediately abort this request in case of allocation failure */
1133 channel_abort(&s->req);
1134 channel_abort(&s->res);
1135
1136 if (!(s->flags & SF_ERR_MASK))
1137 s->flags |= SF_ERR_RESOURCE;
1138 if (!(s->flags & SF_FINST_MASK))
1139 s->flags |= SF_FINST_R;
1140
1141 if (s->txn)
1142 s->txn->status = 500;
1143 s->req.analysers &= AN_REQ_FLT_END;
1144 s->req.analyse_exp = TICK_ETERNITY;
1145 DBG_TRACE_DEVEL("leaving on error", STRM_EV_STRM_ANA|STRM_EV_STRM_ERR, s);
1146 return 0;
1147 }
1148
1149 /* This stream analyser works on a request. It applies all use-server rules on
1150 * it then returns 1. The data must already be present in the buffer otherwise
1151 * they won't match. It always returns 1.
1152 */
process_server_rules(struct stream * s,struct channel * req,int an_bit)1153 static int process_server_rules(struct stream *s, struct channel *req, int an_bit)
1154 {
1155 struct proxy *px = s->be;
1156 struct session *sess = s->sess;
1157 struct server_rule *rule;
1158
1159 DBG_TRACE_ENTER(STRM_EV_STRM_ANA, s);
1160
1161 if (!(s->flags & SF_ASSIGNED)) {
1162 list_for_each_entry(rule, &px->server_rules, list) {
1163 int ret;
1164
1165 ret = acl_exec_cond(rule->cond, s->be, sess, s, SMP_OPT_DIR_REQ|SMP_OPT_FINAL);
1166 ret = acl_pass(ret);
1167 if (rule->cond->pol == ACL_COND_UNLESS)
1168 ret = !ret;
1169
1170 if (ret) {
1171 struct server *srv;
1172
1173 if (rule->dynamic) {
1174 struct buffer *tmp = get_trash_chunk();
1175
1176 if (!build_logline(s, tmp->area, tmp->size, &rule->expr))
1177 break;
1178
1179 srv = findserver(s->be, tmp->area);
1180 if (!srv)
1181 break;
1182 }
1183 else
1184 srv = rule->srv.ptr;
1185
1186 if ((srv->cur_state != SRV_ST_STOPPED) ||
1187 (px->options & PR_O_PERSIST) ||
1188 (s->flags & SF_FORCE_PRST)) {
1189 s->flags |= SF_DIRECT | SF_ASSIGNED;
1190 s->target = &srv->obj_type;
1191 break;
1192 }
1193 /* if the server is not UP, let's go on with next rules
1194 * just in case another one is suited.
1195 */
1196 }
1197 }
1198 }
1199
1200 req->analysers &= ~an_bit;
1201 req->analyse_exp = TICK_ETERNITY;
1202 DBG_TRACE_LEAVE(STRM_EV_STRM_ANA, s);
1203 return 1;
1204 }
1205
sticking_rule_find_target(struct stream * s,struct stktable * t,struct stksess * ts)1206 static inline void sticking_rule_find_target(struct stream *s,
1207 struct stktable *t, struct stksess *ts)
1208 {
1209 struct proxy *px = s->be;
1210 struct eb32_node *node;
1211 struct dict_entry *de;
1212 void *ptr;
1213 struct server *srv;
1214
1215 /* Look for the server name previously stored in <t> stick-table */
1216 HA_RWLOCK_RDLOCK(STK_SESS_LOCK, &ts->lock);
1217 ptr = __stktable_data_ptr(t, ts, STKTABLE_DT_SERVER_NAME);
1218 de = stktable_data_cast(ptr, server_name);
1219 HA_RWLOCK_RDUNLOCK(STK_SESS_LOCK, &ts->lock);
1220
1221 if (de) {
1222 struct ebpt_node *name;
1223
1224 name = ebis_lookup(&px->conf.used_server_name, de->value.key);
1225 if (name) {
1226 srv = container_of(name, struct server, conf.name);
1227 goto found;
1228 }
1229 }
1230
1231 /* Look for the server ID */
1232 HA_RWLOCK_RDLOCK(STK_SESS_LOCK, &ts->lock);
1233 ptr = __stktable_data_ptr(t, ts, STKTABLE_DT_SERVER_ID);
1234 node = eb32_lookup(&px->conf.used_server_id, stktable_data_cast(ptr, server_id));
1235 HA_RWLOCK_RDUNLOCK(STK_SESS_LOCK, &ts->lock);
1236
1237 if (!node)
1238 return;
1239
1240 srv = container_of(node, struct server, conf.id);
1241 found:
1242 if ((srv->cur_state != SRV_ST_STOPPED) ||
1243 (px->options & PR_O_PERSIST) || (s->flags & SF_FORCE_PRST)) {
1244 s->flags |= SF_DIRECT | SF_ASSIGNED;
1245 s->target = &srv->obj_type;
1246 }
1247 }
1248
1249 /* This stream analyser works on a request. It applies all sticking rules on
1250 * it then returns 1. The data must already be present in the buffer otherwise
1251 * they won't match. It always returns 1.
1252 */
process_sticking_rules(struct stream * s,struct channel * req,int an_bit)1253 static int process_sticking_rules(struct stream *s, struct channel *req, int an_bit)
1254 {
1255 struct proxy *px = s->be;
1256 struct session *sess = s->sess;
1257 struct sticking_rule *rule;
1258
1259 DBG_TRACE_ENTER(STRM_EV_STRM_ANA, s);
1260
1261 list_for_each_entry(rule, &px->sticking_rules, list) {
1262 int ret = 1 ;
1263 int i;
1264
1265 /* Only the first stick store-request of each table is applied
1266 * and other ones are ignored. The purpose is to allow complex
1267 * configurations which look for multiple entries by decreasing
1268 * order of precision and to stop at the first which matches.
1269 * An example could be a store of the IP address from an HTTP
1270 * header first, then from the source if not found.
1271 */
1272 if (rule->flags & STK_IS_STORE) {
1273 for (i = 0; i < s->store_count; i++) {
1274 if (rule->table.t == s->store[i].table)
1275 break;
1276 }
1277
1278 if (i != s->store_count)
1279 continue;
1280 }
1281
1282 if (rule->cond) {
1283 ret = acl_exec_cond(rule->cond, px, sess, s, SMP_OPT_DIR_REQ|SMP_OPT_FINAL);
1284 ret = acl_pass(ret);
1285 if (rule->cond->pol == ACL_COND_UNLESS)
1286 ret = !ret;
1287 }
1288
1289 if (ret) {
1290 struct stktable_key *key;
1291
1292 key = stktable_fetch_key(rule->table.t, px, sess, s, SMP_OPT_DIR_REQ|SMP_OPT_FINAL, rule->expr, NULL);
1293 if (!key)
1294 continue;
1295
1296 if (rule->flags & STK_IS_MATCH) {
1297 struct stksess *ts;
1298
1299 if ((ts = stktable_lookup_key(rule->table.t, key)) != NULL) {
1300 if (!(s->flags & SF_ASSIGNED))
1301 sticking_rule_find_target(s, rule->table.t, ts);
1302 stktable_touch_local(rule->table.t, ts, 1);
1303 }
1304 }
1305 if (rule->flags & STK_IS_STORE) {
1306 if (s->store_count < (sizeof(s->store) / sizeof(s->store[0]))) {
1307 struct stksess *ts;
1308
1309 ts = stksess_new(rule->table.t, key);
1310 if (ts) {
1311 s->store[s->store_count].table = rule->table.t;
1312 s->store[s->store_count++].ts = ts;
1313 }
1314 }
1315 }
1316 }
1317 }
1318
1319 req->analysers &= ~an_bit;
1320 req->analyse_exp = TICK_ETERNITY;
1321 DBG_TRACE_LEAVE(STRM_EV_STRM_ANA, s);
1322 return 1;
1323 }
1324
1325 /* This stream analyser works on a response. It applies all store rules on it
1326 * then returns 1. The data must already be present in the buffer otherwise
1327 * they won't match. It always returns 1.
1328 */
process_store_rules(struct stream * s,struct channel * rep,int an_bit)1329 static int process_store_rules(struct stream *s, struct channel *rep, int an_bit)
1330 {
1331 struct proxy *px = s->be;
1332 struct session *sess = s->sess;
1333 struct sticking_rule *rule;
1334 int i;
1335 int nbreq = s->store_count;
1336
1337 DBG_TRACE_ENTER(STRM_EV_STRM_ANA, s);
1338
1339 list_for_each_entry(rule, &px->storersp_rules, list) {
1340 int ret = 1 ;
1341
1342 /* Only the first stick store-response of each table is applied
1343 * and other ones are ignored. The purpose is to allow complex
1344 * configurations which look for multiple entries by decreasing
1345 * order of precision and to stop at the first which matches.
1346 * An example could be a store of a set-cookie value, with a
1347 * fallback to a parameter found in a 302 redirect.
1348 *
1349 * The store-response rules are not allowed to override the
1350 * store-request rules for the same table, but they may coexist.
1351 * Thus we can have up to one store-request entry and one store-
1352 * response entry for the same table at any time.
1353 */
1354 for (i = nbreq; i < s->store_count; i++) {
1355 if (rule->table.t == s->store[i].table)
1356 break;
1357 }
1358
1359 /* skip existing entries for this table */
1360 if (i < s->store_count)
1361 continue;
1362
1363 if (rule->cond) {
1364 ret = acl_exec_cond(rule->cond, px, sess, s, SMP_OPT_DIR_RES|SMP_OPT_FINAL);
1365 ret = acl_pass(ret);
1366 if (rule->cond->pol == ACL_COND_UNLESS)
1367 ret = !ret;
1368 }
1369
1370 if (ret) {
1371 struct stktable_key *key;
1372
1373 key = stktable_fetch_key(rule->table.t, px, sess, s, SMP_OPT_DIR_RES|SMP_OPT_FINAL, rule->expr, NULL);
1374 if (!key)
1375 continue;
1376
1377 if (s->store_count < (sizeof(s->store) / sizeof(s->store[0]))) {
1378 struct stksess *ts;
1379
1380 ts = stksess_new(rule->table.t, key);
1381 if (ts) {
1382 s->store[s->store_count].table = rule->table.t;
1383 s->store[s->store_count++].ts = ts;
1384 }
1385 }
1386 }
1387 }
1388
1389 /* process store request and store response */
1390 for (i = 0; i < s->store_count; i++) {
1391 struct stksess *ts;
1392 void *ptr;
1393 struct dict_entry *de;
1394
1395 if (objt_server(s->target) && objt_server(s->target)->flags & SRV_F_NON_STICK) {
1396 stksess_free(s->store[i].table, s->store[i].ts);
1397 s->store[i].ts = NULL;
1398 continue;
1399 }
1400
1401 ts = stktable_set_entry(s->store[i].table, s->store[i].ts);
1402 if (ts != s->store[i].ts) {
1403 /* the entry already existed, we can free ours */
1404 stksess_free(s->store[i].table, s->store[i].ts);
1405 }
1406 s->store[i].ts = NULL;
1407
1408 HA_RWLOCK_WRLOCK(STK_SESS_LOCK, &ts->lock);
1409 ptr = __stktable_data_ptr(s->store[i].table, ts, STKTABLE_DT_SERVER_ID);
1410 stktable_data_cast(ptr, server_id) = __objt_server(s->target)->puid;
1411 HA_RWLOCK_WRUNLOCK(STK_SESS_LOCK, &ts->lock);
1412
1413 HA_RWLOCK_WRLOCK(STK_SESS_LOCK, &ts->lock);
1414 de = dict_insert(&server_name_dict, __objt_server(s->target)->id);
1415 if (de) {
1416 ptr = __stktable_data_ptr(s->store[i].table, ts, STKTABLE_DT_SERVER_NAME);
1417 stktable_data_cast(ptr, server_name) = de;
1418 }
1419 HA_RWLOCK_WRUNLOCK(STK_SESS_LOCK, &ts->lock);
1420
1421 stktable_touch_local(s->store[i].table, ts, 1);
1422 }
1423 s->store_count = 0; /* everything is stored */
1424
1425 rep->analysers &= ~an_bit;
1426 rep->analyse_exp = TICK_ETERNITY;
1427
1428 DBG_TRACE_LEAVE(STRM_EV_STRM_ANA, s);
1429 return 1;
1430 }
1431
1432 /* This macro is very specific to the function below. See the comments in
1433 * process_stream() below to understand the logic and the tests.
1434 */
1435 #define UPDATE_ANALYSERS(real, list, back, flag) { \
1436 list = (((list) & ~(flag)) | ~(back)) & (real); \
1437 back = real; \
1438 if (!(list)) \
1439 break; \
1440 if (((list) ^ ((list) & ((list) - 1))) < (flag)) \
1441 continue; \
1442 }
1443
1444 /* These 2 following macros call an analayzer for the specified channel if the
1445 * right flag is set. The first one is used for "filterable" analyzers. If a
1446 * stream has some registered filters, pre and post analyaze callbacks are
1447 * called. The second are used for other analyzers (AN_REQ/RES_FLT_* and
1448 * AN_REQ/RES_HTTP_XFER_BODY) */
1449 #define FLT_ANALYZE(strm, chn, fun, list, back, flag, ...) \
1450 { \
1451 if ((list) & (flag)) { \
1452 if (HAS_FILTERS(strm)) { \
1453 if (!flt_pre_analyze((strm), (chn), (flag))) \
1454 break; \
1455 if (!fun((strm), (chn), (flag), ##__VA_ARGS__)) \
1456 break; \
1457 if (!flt_post_analyze((strm), (chn), (flag))) \
1458 break; \
1459 } \
1460 else { \
1461 if (!fun((strm), (chn), (flag), ##__VA_ARGS__)) \
1462 break; \
1463 } \
1464 UPDATE_ANALYSERS((chn)->analysers, (list), \
1465 (back), (flag)); \
1466 } \
1467 }
1468
1469 #define ANALYZE(strm, chn, fun, list, back, flag, ...) \
1470 { \
1471 if ((list) & (flag)) { \
1472 if (!fun((strm), (chn), (flag), ##__VA_ARGS__)) \
1473 break; \
1474 UPDATE_ANALYSERS((chn)->analysers, (list), \
1475 (back), (flag)); \
1476 } \
1477 }
1478
1479 /* Processes the client, server, request and response jobs of a stream task,
1480 * then puts it back to the wait queue in a clean state, or cleans up its
1481 * resources if it must be deleted. Returns in <next> the date the task wants
1482 * to be woken up, or TICK_ETERNITY. In order not to call all functions for
1483 * nothing too many times, the request and response buffers flags are monitored
1484 * and each function is called only if at least another function has changed at
1485 * least one flag it is interested in.
1486 */
process_stream(struct task * t,void * context,unsigned short state)1487 struct task *process_stream(struct task *t, void *context, unsigned short state)
1488 {
1489 struct server *srv;
1490 struct stream *s = context;
1491 struct session *sess = s->sess;
1492 unsigned int rqf_last, rpf_last;
1493 unsigned int rq_prod_last, rq_cons_last;
1494 unsigned int rp_cons_last, rp_prod_last;
1495 unsigned int req_ana_back;
1496 struct channel *req, *res;
1497 struct stream_interface *si_f, *si_b;
1498 unsigned int rate;
1499
1500 DBG_TRACE_ENTER(STRM_EV_STRM_PROC, s);
1501
1502 activity[tid].stream_calls++;
1503
1504 req = &s->req;
1505 res = &s->res;
1506
1507 si_f = &s->si[0];
1508 si_b = &s->si[1];
1509
1510 /* First, attempt to receive pending data from I/O layers */
1511 si_sync_recv(si_f);
1512 si_sync_recv(si_b);
1513
1514 rate = update_freq_ctr(&s->call_rate, 1);
1515 if (rate >= 100000 && s->call_rate.prev_ctr) { // make sure to wait at least a full second
1516 stream_dump_and_crash(&s->obj_type, read_freq_ctr(&s->call_rate));
1517 }
1518
1519 /* this data may be no longer valid, clear it */
1520 if (s->txn)
1521 memset(&s->txn->auth, 0, sizeof(s->txn->auth));
1522
1523 /* This flag must explicitly be set every time */
1524 req->flags &= ~(CF_READ_NOEXP|CF_WAKE_WRITE);
1525 res->flags &= ~(CF_READ_NOEXP|CF_WAKE_WRITE);
1526
1527 /* Keep a copy of req/rep flags so that we can detect shutdowns */
1528 rqf_last = req->flags & ~CF_MASK_ANALYSER;
1529 rpf_last = res->flags & ~CF_MASK_ANALYSER;
1530
1531 /* we don't want the stream interface functions to recursively wake us up */
1532 si_f->flags |= SI_FL_DONT_WAKE;
1533 si_b->flags |= SI_FL_DONT_WAKE;
1534
1535 /* update pending events */
1536 s->pending_events |= (state & TASK_WOKEN_ANY);
1537
1538 /* 1a: Check for low level timeouts if needed. We just set a flag on
1539 * stream interfaces when their timeouts have expired.
1540 */
1541 if (unlikely(s->pending_events & TASK_WOKEN_TIMER)) {
1542 si_check_timeouts(si_f);
1543 si_check_timeouts(si_b);
1544
1545 /* check channel timeouts, and close the corresponding stream interfaces
1546 * for future reads or writes. Note: this will also concern upper layers
1547 * but we do not touch any other flag. We must be careful and correctly
1548 * detect state changes when calling them.
1549 */
1550
1551 channel_check_timeouts(req);
1552
1553 if (unlikely((req->flags & (CF_SHUTW|CF_WRITE_TIMEOUT)) == CF_WRITE_TIMEOUT)) {
1554 si_b->flags |= SI_FL_NOLINGER;
1555 si_shutw(si_b);
1556 }
1557
1558 if (unlikely((req->flags & (CF_SHUTR|CF_READ_TIMEOUT)) == CF_READ_TIMEOUT)) {
1559 if (si_f->flags & SI_FL_NOHALF)
1560 si_f->flags |= SI_FL_NOLINGER;
1561 si_shutr(si_f);
1562 }
1563
1564 channel_check_timeouts(res);
1565
1566 if (unlikely((res->flags & (CF_SHUTW|CF_WRITE_TIMEOUT)) == CF_WRITE_TIMEOUT)) {
1567 si_f->flags |= SI_FL_NOLINGER;
1568 si_shutw(si_f);
1569 }
1570
1571 if (unlikely((res->flags & (CF_SHUTR|CF_READ_TIMEOUT)) == CF_READ_TIMEOUT)) {
1572 if (si_b->flags & SI_FL_NOHALF)
1573 si_b->flags |= SI_FL_NOLINGER;
1574 si_shutr(si_b);
1575 }
1576
1577 if (HAS_FILTERS(s))
1578 flt_stream_check_timeouts(s);
1579
1580 /* Once in a while we're woken up because the task expires. But
1581 * this does not necessarily mean that a timeout has been reached.
1582 * So let's not run a whole stream processing if only an expiration
1583 * timeout needs to be refreshed.
1584 */
1585 if (!((req->flags | res->flags) &
1586 (CF_SHUTR|CF_READ_ACTIVITY|CF_READ_TIMEOUT|CF_SHUTW|
1587 CF_WRITE_ACTIVITY|CF_WRITE_TIMEOUT|CF_ANA_TIMEOUT)) &&
1588 !((si_f->flags | si_b->flags) & (SI_FL_EXP|SI_FL_ERR)) &&
1589 ((s->pending_events & TASK_WOKEN_ANY) == TASK_WOKEN_TIMER)) {
1590 si_f->flags &= ~SI_FL_DONT_WAKE;
1591 si_b->flags &= ~SI_FL_DONT_WAKE;
1592 goto update_exp_and_leave;
1593 }
1594 }
1595
1596 resync_stream_interface:
1597 /* below we may emit error messages so we have to ensure that we have
1598 * our buffers properly allocated.
1599 */
1600 if (!stream_alloc_work_buffer(s)) {
1601 /* No buffer available, we've been subscribed to the list of
1602 * buffer waiters, let's wait for our turn.
1603 */
1604 si_f->flags &= ~SI_FL_DONT_WAKE;
1605 si_b->flags &= ~SI_FL_DONT_WAKE;
1606 goto update_exp_and_leave;
1607 }
1608
1609 /* 1b: check for low-level errors reported at the stream interface.
1610 * First we check if it's a retryable error (in which case we don't
1611 * want to tell the buffer). Otherwise we report the error one level
1612 * upper by setting flags into the buffers. Note that the side towards
1613 * the client cannot have connect (hence retryable) errors. Also, the
1614 * connection setup code must be able to deal with any type of abort.
1615 */
1616 srv = objt_server(s->target);
1617 if (unlikely(si_f->flags & SI_FL_ERR)) {
1618 if (si_state_in(si_f->state, SI_SB_EST|SI_SB_DIS)) {
1619 si_shutr(si_f);
1620 si_shutw(si_f);
1621 si_report_error(si_f);
1622 if (!(req->analysers) && !(res->analysers)) {
1623 _HA_ATOMIC_ADD(&s->be->be_counters.cli_aborts, 1);
1624 _HA_ATOMIC_ADD(&sess->fe->fe_counters.cli_aborts, 1);
1625 if (sess->listener && sess->listener->counters)
1626 _HA_ATOMIC_ADD(&sess->listener->counters->cli_aborts, 1);
1627 if (srv)
1628 _HA_ATOMIC_ADD(&srv->counters.cli_aborts, 1);
1629 if (!(s->flags & SF_ERR_MASK))
1630 s->flags |= SF_ERR_CLICL;
1631 if (!(s->flags & SF_FINST_MASK))
1632 s->flags |= SF_FINST_D;
1633 }
1634 }
1635 }
1636
1637 if (unlikely(si_b->flags & SI_FL_ERR)) {
1638 if (si_state_in(si_b->state, SI_SB_EST|SI_SB_DIS)) {
1639 si_shutr(si_b);
1640 si_shutw(si_b);
1641 si_report_error(si_b);
1642 _HA_ATOMIC_ADD(&s->be->be_counters.failed_resp, 1);
1643 if (srv)
1644 _HA_ATOMIC_ADD(&srv->counters.failed_resp, 1);
1645 if (!(req->analysers) && !(res->analysers)) {
1646 _HA_ATOMIC_ADD(&s->be->be_counters.srv_aborts, 1);
1647 _HA_ATOMIC_ADD(&sess->fe->fe_counters.srv_aborts, 1);
1648 if (sess->listener && sess->listener->counters)
1649 _HA_ATOMIC_ADD(&sess->listener->counters->srv_aborts, 1);
1650 if (srv)
1651 _HA_ATOMIC_ADD(&srv->counters.srv_aborts, 1);
1652 if (!(s->flags & SF_ERR_MASK))
1653 s->flags |= SF_ERR_SRVCL;
1654 if (!(s->flags & SF_FINST_MASK))
1655 s->flags |= SF_FINST_D;
1656 }
1657 }
1658 /* note: maybe we should process connection errors here ? */
1659 }
1660
1661 if (si_state_in(si_b->state, SI_SB_CON|SI_SB_RDY)) {
1662 /* we were trying to establish a connection on the server side,
1663 * maybe it succeeded, maybe it failed, maybe we timed out, ...
1664 */
1665 if (si_b->state == SI_ST_RDY)
1666 back_handle_st_rdy(s);
1667 else if (si_b->state == SI_ST_CON)
1668 back_handle_st_con(s);
1669
1670 if (si_b->state == SI_ST_CER)
1671 back_handle_st_cer(s);
1672 else if (si_b->state == SI_ST_EST)
1673 back_establish(s);
1674
1675 /* state is now one of SI_ST_CON (still in progress), SI_ST_EST
1676 * (established), SI_ST_DIS (abort), SI_ST_CLO (last error),
1677 * SI_ST_ASS/SI_ST_TAR/SI_ST_REQ for retryable errors.
1678 */
1679 }
1680
1681 rq_prod_last = si_f->state;
1682 rq_cons_last = si_b->state;
1683 rp_cons_last = si_f->state;
1684 rp_prod_last = si_b->state;
1685
1686 /* Check for connection closure */
1687 DBG_TRACE_POINT(STRM_EV_STRM_PROC, s);
1688
1689 /* nothing special to be done on client side */
1690 if (unlikely(si_f->state == SI_ST_DIS))
1691 si_f->state = SI_ST_CLO;
1692
1693 /* When a server-side connection is released, we have to count it and
1694 * check for pending connections on this server.
1695 */
1696 if (unlikely(si_b->state == SI_ST_DIS)) {
1697 si_b->state = SI_ST_CLO;
1698 srv = objt_server(s->target);
1699 if (srv) {
1700 if (s->flags & SF_CURR_SESS) {
1701 s->flags &= ~SF_CURR_SESS;
1702 _HA_ATOMIC_SUB(&srv->cur_sess, 1);
1703 }
1704 sess_change_server(s, NULL);
1705 if (may_dequeue_tasks(srv, s->be))
1706 process_srv_queue(srv, 0);
1707 }
1708 }
1709
1710 /*
1711 * Note: of the transient states (REQ, CER, DIS), only REQ may remain
1712 * at this point.
1713 */
1714
1715 resync_request:
1716 /* Analyse request */
1717 if (((req->flags & ~rqf_last) & CF_MASK_ANALYSER) ||
1718 ((req->flags ^ rqf_last) & CF_MASK_STATIC) ||
1719 (req->analysers && (req->flags & CF_SHUTW)) ||
1720 si_f->state != rq_prod_last ||
1721 si_b->state != rq_cons_last ||
1722 s->pending_events & TASK_WOKEN_MSG) {
1723 unsigned int flags = req->flags;
1724
1725 if (si_state_in(si_f->state, SI_SB_EST|SI_SB_DIS|SI_SB_CLO)) {
1726 int max_loops = global.tune.maxpollevents;
1727 unsigned int ana_list;
1728 unsigned int ana_back;
1729
1730 /* it's up to the analysers to stop new connections,
1731 * disable reading or closing. Note: if an analyser
1732 * disables any of these bits, it is responsible for
1733 * enabling them again when it disables itself, so
1734 * that other analysers are called in similar conditions.
1735 */
1736 channel_auto_read(req);
1737 channel_auto_connect(req);
1738 channel_auto_close(req);
1739
1740 /* We will call all analysers for which a bit is set in
1741 * req->analysers, following the bit order from LSB
1742 * to MSB. The analysers must remove themselves from
1743 * the list when not needed. Any analyser may return 0
1744 * to break out of the loop, either because of missing
1745 * data to take a decision, or because it decides to
1746 * kill the stream. We loop at least once through each
1747 * analyser, and we may loop again if other analysers
1748 * are added in the middle.
1749 *
1750 * We build a list of analysers to run. We evaluate all
1751 * of these analysers in the order of the lower bit to
1752 * the higher bit. This ordering is very important.
1753 * An analyser will often add/remove other analysers,
1754 * including itself. Any changes to itself have no effect
1755 * on the loop. If it removes any other analysers, we
1756 * want those analysers not to be called anymore during
1757 * this loop. If it adds an analyser that is located
1758 * after itself, we want it to be scheduled for being
1759 * processed during the loop. If it adds an analyser
1760 * which is located before it, we want it to switch to
1761 * it immediately, even if it has already been called
1762 * once but removed since.
1763 *
1764 * In order to achieve this, we compare the analyser
1765 * list after the call with a copy of it before the
1766 * call. The work list is fed with analyser bits that
1767 * appeared during the call. Then we compare previous
1768 * work list with the new one, and check the bits that
1769 * appeared. If the lowest of these bits is lower than
1770 * the current bit, it means we have enabled a previous
1771 * analyser and must immediately loop again.
1772 */
1773
1774 ana_list = ana_back = req->analysers;
1775 while (ana_list && max_loops--) {
1776 /* Warning! ensure that analysers are always placed in ascending order! */
1777 ANALYZE (s, req, flt_start_analyze, ana_list, ana_back, AN_REQ_FLT_START_FE);
1778 FLT_ANALYZE(s, req, tcp_inspect_request, ana_list, ana_back, AN_REQ_INSPECT_FE);
1779 FLT_ANALYZE(s, req, http_wait_for_request, ana_list, ana_back, AN_REQ_WAIT_HTTP);
1780 FLT_ANALYZE(s, req, http_wait_for_request_body, ana_list, ana_back, AN_REQ_HTTP_BODY);
1781 FLT_ANALYZE(s, req, http_process_req_common, ana_list, ana_back, AN_REQ_HTTP_PROCESS_FE, sess->fe);
1782 FLT_ANALYZE(s, req, process_switching_rules, ana_list, ana_back, AN_REQ_SWITCHING_RULES);
1783 ANALYZE (s, req, flt_start_analyze, ana_list, ana_back, AN_REQ_FLT_START_BE);
1784 FLT_ANALYZE(s, req, tcp_inspect_request, ana_list, ana_back, AN_REQ_INSPECT_BE);
1785 FLT_ANALYZE(s, req, http_process_req_common, ana_list, ana_back, AN_REQ_HTTP_PROCESS_BE, s->be);
1786 FLT_ANALYZE(s, req, http_process_tarpit, ana_list, ana_back, AN_REQ_HTTP_TARPIT);
1787 FLT_ANALYZE(s, req, process_server_rules, ana_list, ana_back, AN_REQ_SRV_RULES);
1788 FLT_ANALYZE(s, req, http_process_request, ana_list, ana_back, AN_REQ_HTTP_INNER);
1789 FLT_ANALYZE(s, req, tcp_persist_rdp_cookie, ana_list, ana_back, AN_REQ_PRST_RDP_COOKIE);
1790 FLT_ANALYZE(s, req, process_sticking_rules, ana_list, ana_back, AN_REQ_STICKING_RULES);
1791 ANALYZE (s, req, flt_analyze_http_headers, ana_list, ana_back, AN_REQ_FLT_HTTP_HDRS);
1792 ANALYZE (s, req, http_request_forward_body, ana_list, ana_back, AN_REQ_HTTP_XFER_BODY);
1793 ANALYZE (s, req, pcli_wait_for_request, ana_list, ana_back, AN_REQ_WAIT_CLI);
1794 ANALYZE (s, req, flt_xfer_data, ana_list, ana_back, AN_REQ_FLT_XFER_DATA);
1795 ANALYZE (s, req, flt_end_analyze, ana_list, ana_back, AN_REQ_FLT_END);
1796 break;
1797 }
1798 }
1799
1800 rq_prod_last = si_f->state;
1801 rq_cons_last = si_b->state;
1802 req->flags &= ~CF_WAKE_ONCE;
1803 rqf_last = req->flags;
1804
1805 if ((req->flags ^ flags) & (CF_SHUTR|CF_SHUTW))
1806 goto resync_request;
1807 }
1808
1809 /* we'll monitor the request analysers while parsing the response,
1810 * because some response analysers may indirectly enable new request
1811 * analysers (eg: HTTP keep-alive).
1812 */
1813 req_ana_back = req->analysers;
1814
1815 resync_response:
1816 /* Analyse response */
1817
1818 if (((res->flags & ~rpf_last) & CF_MASK_ANALYSER) ||
1819 (res->flags ^ rpf_last) & CF_MASK_STATIC ||
1820 (res->analysers && (res->flags & CF_SHUTW)) ||
1821 si_f->state != rp_cons_last ||
1822 si_b->state != rp_prod_last ||
1823 s->pending_events & TASK_WOKEN_MSG) {
1824 unsigned int flags = res->flags;
1825
1826 if (si_state_in(si_b->state, SI_SB_EST|SI_SB_DIS|SI_SB_CLO)) {
1827 int max_loops = global.tune.maxpollevents;
1828 unsigned int ana_list;
1829 unsigned int ana_back;
1830
1831 /* it's up to the analysers to stop disable reading or
1832 * closing. Note: if an analyser disables any of these
1833 * bits, it is responsible for enabling them again when
1834 * it disables itself, so that other analysers are called
1835 * in similar conditions.
1836 */
1837 channel_auto_read(res);
1838 channel_auto_close(res);
1839
1840 /* We will call all analysers for which a bit is set in
1841 * res->analysers, following the bit order from LSB
1842 * to MSB. The analysers must remove themselves from
1843 * the list when not needed. Any analyser may return 0
1844 * to break out of the loop, either because of missing
1845 * data to take a decision, or because it decides to
1846 * kill the stream. We loop at least once through each
1847 * analyser, and we may loop again if other analysers
1848 * are added in the middle.
1849 */
1850
1851 ana_list = ana_back = res->analysers;
1852 while (ana_list && max_loops--) {
1853 /* Warning! ensure that analysers are always placed in ascending order! */
1854 ANALYZE (s, res, flt_start_analyze, ana_list, ana_back, AN_RES_FLT_START_FE);
1855 ANALYZE (s, res, flt_start_analyze, ana_list, ana_back, AN_RES_FLT_START_BE);
1856 FLT_ANALYZE(s, res, tcp_inspect_response, ana_list, ana_back, AN_RES_INSPECT);
1857 FLT_ANALYZE(s, res, http_wait_for_response, ana_list, ana_back, AN_RES_WAIT_HTTP);
1858 FLT_ANALYZE(s, res, process_store_rules, ana_list, ana_back, AN_RES_STORE_RULES);
1859 FLT_ANALYZE(s, res, http_process_res_common, ana_list, ana_back, AN_RES_HTTP_PROCESS_BE, s->be);
1860 ANALYZE (s, res, flt_analyze_http_headers, ana_list, ana_back, AN_RES_FLT_HTTP_HDRS);
1861 ANALYZE (s, res, http_response_forward_body, ana_list, ana_back, AN_RES_HTTP_XFER_BODY);
1862 ANALYZE (s, res, pcli_wait_for_response, ana_list, ana_back, AN_RES_WAIT_CLI);
1863 ANALYZE (s, res, flt_xfer_data, ana_list, ana_back, AN_RES_FLT_XFER_DATA);
1864 ANALYZE (s, res, flt_end_analyze, ana_list, ana_back, AN_RES_FLT_END);
1865 break;
1866 }
1867 }
1868
1869 rp_cons_last = si_f->state;
1870 rp_prod_last = si_b->state;
1871 res->flags &= ~CF_WAKE_ONCE;
1872 rpf_last = res->flags;
1873
1874 if ((res->flags ^ flags) & (CF_SHUTR|CF_SHUTW))
1875 goto resync_response;
1876 }
1877
1878 /* maybe someone has added some request analysers, so we must check and loop */
1879 if (req->analysers & ~req_ana_back)
1880 goto resync_request;
1881
1882 if ((req->flags & ~rqf_last) & CF_MASK_ANALYSER)
1883 goto resync_request;
1884
1885 /* FIXME: here we should call protocol handlers which rely on
1886 * both buffers.
1887 */
1888
1889
1890 /*
1891 * Now we propagate unhandled errors to the stream. Normally
1892 * we're just in a data phase here since it means we have not
1893 * seen any analyser who could set an error status.
1894 */
1895 srv = objt_server(s->target);
1896 if (unlikely(!(s->flags & SF_ERR_MASK))) {
1897 if (req->flags & (CF_READ_ERROR|CF_READ_TIMEOUT|CF_WRITE_ERROR|CF_WRITE_TIMEOUT)) {
1898 /* Report it if the client got an error or a read timeout expired */
1899 req->analysers &= AN_REQ_FLT_END;
1900 if (req->flags & CF_READ_ERROR) {
1901 _HA_ATOMIC_ADD(&s->be->be_counters.cli_aborts, 1);
1902 _HA_ATOMIC_ADD(&sess->fe->fe_counters.cli_aborts, 1);
1903 if (sess->listener && sess->listener->counters)
1904 _HA_ATOMIC_ADD(&sess->listener->counters->cli_aborts, 1);
1905 if (srv)
1906 _HA_ATOMIC_ADD(&srv->counters.cli_aborts, 1);
1907 s->flags |= SF_ERR_CLICL;
1908 }
1909 else if (req->flags & CF_READ_TIMEOUT) {
1910 _HA_ATOMIC_ADD(&s->be->be_counters.cli_aborts, 1);
1911 _HA_ATOMIC_ADD(&sess->fe->fe_counters.cli_aborts, 1);
1912 if (sess->listener && sess->listener->counters)
1913 _HA_ATOMIC_ADD(&sess->listener->counters->cli_aborts, 1);
1914 if (srv)
1915 _HA_ATOMIC_ADD(&srv->counters.cli_aborts, 1);
1916 s->flags |= SF_ERR_CLITO;
1917 }
1918 else if (req->flags & CF_WRITE_ERROR) {
1919 _HA_ATOMIC_ADD(&s->be->be_counters.srv_aborts, 1);
1920 _HA_ATOMIC_ADD(&sess->fe->fe_counters.srv_aborts, 1);
1921 if (sess->listener && sess->listener->counters)
1922 _HA_ATOMIC_ADD(&sess->listener->counters->srv_aborts, 1);
1923 if (srv)
1924 _HA_ATOMIC_ADD(&srv->counters.srv_aborts, 1);
1925 s->flags |= SF_ERR_SRVCL;
1926 }
1927 else {
1928 _HA_ATOMIC_ADD(&s->be->be_counters.srv_aborts, 1);
1929 _HA_ATOMIC_ADD(&sess->fe->fe_counters.srv_aborts, 1);
1930 if (sess->listener && sess->listener->counters)
1931 _HA_ATOMIC_ADD(&sess->listener->counters->srv_aborts, 1);
1932 if (srv)
1933 _HA_ATOMIC_ADD(&srv->counters.srv_aborts, 1);
1934 s->flags |= SF_ERR_SRVTO;
1935 }
1936 sess_set_term_flags(s);
1937
1938 /* Abort the request if a client error occurred while
1939 * the backend stream-interface is in the SI_ST_INI
1940 * state. It is switched into the SI_ST_CLO state and
1941 * the request channel is erased. */
1942 if (si_b->state == SI_ST_INI) {
1943 si_b->state = SI_ST_CLO;
1944 channel_abort(req);
1945 if (IS_HTX_STRM(s))
1946 channel_htx_erase(req, htxbuf(&req->buf));
1947 else
1948 channel_erase(req);
1949 }
1950 }
1951 else if (res->flags & (CF_READ_ERROR|CF_READ_TIMEOUT|CF_WRITE_ERROR|CF_WRITE_TIMEOUT)) {
1952 /* Report it if the server got an error or a read timeout expired */
1953 res->analysers &= AN_RES_FLT_END;
1954 if (res->flags & CF_READ_ERROR) {
1955 _HA_ATOMIC_ADD(&s->be->be_counters.srv_aborts, 1);
1956 _HA_ATOMIC_ADD(&sess->fe->fe_counters.srv_aborts, 1);
1957 if (sess->listener && sess->listener->counters)
1958 _HA_ATOMIC_ADD(&sess->listener->counters->srv_aborts, 1);
1959 if (srv)
1960 _HA_ATOMIC_ADD(&srv->counters.srv_aborts, 1);
1961 s->flags |= SF_ERR_SRVCL;
1962 }
1963 else if (res->flags & CF_READ_TIMEOUT) {
1964 _HA_ATOMIC_ADD(&s->be->be_counters.srv_aborts, 1);
1965 _HA_ATOMIC_ADD(&sess->fe->fe_counters.srv_aborts, 1);
1966 if (sess->listener && sess->listener->counters)
1967 _HA_ATOMIC_ADD(&sess->listener->counters->srv_aborts, 1);
1968 if (srv)
1969 _HA_ATOMIC_ADD(&srv->counters.srv_aborts, 1);
1970 s->flags |= SF_ERR_SRVTO;
1971 }
1972 else if (res->flags & CF_WRITE_ERROR) {
1973 _HA_ATOMIC_ADD(&s->be->be_counters.cli_aborts, 1);
1974 _HA_ATOMIC_ADD(&sess->fe->fe_counters.cli_aborts, 1);
1975 if (sess->listener && sess->listener->counters)
1976 _HA_ATOMIC_ADD(&sess->listener->counters->cli_aborts, 1);
1977 if (srv)
1978 _HA_ATOMIC_ADD(&srv->counters.cli_aborts, 1);
1979 s->flags |= SF_ERR_CLICL;
1980 }
1981 else {
1982 _HA_ATOMIC_ADD(&s->be->be_counters.cli_aborts, 1);
1983 _HA_ATOMIC_ADD(&sess->fe->fe_counters.cli_aborts, 1);
1984 if (sess->listener && sess->listener->counters)
1985 _HA_ATOMIC_ADD(&sess->listener->counters->cli_aborts, 1);
1986 if (srv)
1987 _HA_ATOMIC_ADD(&srv->counters.cli_aborts, 1);
1988 s->flags |= SF_ERR_CLITO;
1989 }
1990 sess_set_term_flags(s);
1991 }
1992 }
1993
1994 /*
1995 * Here we take care of forwarding unhandled data. This also includes
1996 * connection establishments and shutdown requests.
1997 */
1998
1999
2000 /* If noone is interested in analysing data, it's time to forward
2001 * everything. We configure the buffer to forward indefinitely.
2002 * Note that we're checking CF_SHUTR_NOW as an indication of a possible
2003 * recent call to channel_abort().
2004 */
2005 if (unlikely((!req->analysers || (req->analysers == AN_REQ_FLT_END && !(req->flags & CF_FLT_ANALYZE))) &&
2006 !(req->flags & (CF_SHUTW|CF_SHUTR_NOW)) &&
2007 (si_state_in(si_f->state, SI_SB_EST|SI_SB_DIS|SI_SB_CLO)) &&
2008 (req->to_forward != CHN_INFINITE_FORWARD))) {
2009 /* This buffer is freewheeling, there's no analyser
2010 * attached to it. If any data are left in, we'll permit them to
2011 * move.
2012 */
2013 channel_auto_read(req);
2014 channel_auto_connect(req);
2015 channel_auto_close(req);
2016
2017 if (IS_HTX_STRM(s)) {
2018 struct htx *htx = htxbuf(&req->buf);
2019
2020 /* We'll let data flow between the producer (if still connected)
2021 * to the consumer.
2022 */
2023 co_set_data(req, htx->data);
2024 if (!(req->flags & (CF_SHUTR|CF_SHUTW_NOW)))
2025 channel_htx_forward_forever(req, htx);
2026 }
2027 else {
2028 /* We'll let data flow between the producer (if still connected)
2029 * to the consumer (which might possibly not be connected yet).
2030 */
2031 c_adv(req, ci_data(req));
2032 if (!(req->flags & (CF_SHUTR|CF_SHUTW_NOW)))
2033 channel_forward_forever(req);
2034 }
2035 }
2036
2037 /* check if it is wise to enable kernel splicing to forward request data */
2038 if (!(req->flags & (CF_KERN_SPLICING|CF_SHUTR)) &&
2039 req->to_forward &&
2040 (global.tune.options & GTUNE_USE_SPLICE) &&
2041 (objt_cs(si_f->end) && __objt_cs(si_f->end)->conn->xprt && __objt_cs(si_f->end)->conn->xprt->rcv_pipe &&
2042 __objt_cs(si_f->end)->conn->mux && __objt_cs(si_f->end)->conn->mux->rcv_pipe) &&
2043 (objt_cs(si_b->end) && __objt_cs(si_b->end)->conn->xprt && __objt_cs(si_b->end)->conn->xprt->snd_pipe &&
2044 __objt_cs(si_b->end)->conn->mux && __objt_cs(si_b->end)->conn->mux->snd_pipe) &&
2045 (pipes_used < global.maxpipes) &&
2046 (((sess->fe->options2|s->be->options2) & PR_O2_SPLIC_REQ) ||
2047 (((sess->fe->options2|s->be->options2) & PR_O2_SPLIC_AUT) &&
2048 (req->flags & CF_STREAMER_FAST)))) {
2049 req->flags |= CF_KERN_SPLICING;
2050 }
2051
2052 /* reflect what the L7 analysers have seen last */
2053 rqf_last = req->flags;
2054
2055 /* it's possible that an upper layer has requested a connection setup or abort.
2056 * There are 2 situations where we decide to establish a new connection :
2057 * - there are data scheduled for emission in the buffer
2058 * - the CF_AUTO_CONNECT flag is set (active connection)
2059 */
2060 if (si_b->state == SI_ST_INI) {
2061 if (!(req->flags & CF_SHUTW)) {
2062 if ((req->flags & CF_AUTO_CONNECT) || !channel_is_empty(req)) {
2063 /* If we have an appctx, there is no connect method, so we
2064 * immediately switch to the connected state, otherwise we
2065 * perform a connection request.
2066 */
2067 si_b->state = SI_ST_REQ; /* new connection requested */
2068 si_b->conn_retries = s->be->conn_retries;
2069 if ((s->be->retry_type &~ PR_RE_CONN_FAILED) &&
2070 (s->be->mode == PR_MODE_HTTP) &&
2071 !(si_b->flags & SI_FL_D_L7_RETRY))
2072 si_b->flags |= SI_FL_L7_RETRY;
2073 }
2074 }
2075 else {
2076 si_release_endpoint(si_b);
2077 si_b->state = SI_ST_CLO; /* shutw+ini = abort */
2078 channel_shutw_now(req); /* fix buffer flags upon abort */
2079 channel_shutr_now(res);
2080 }
2081 }
2082
2083
2084 /* we may have a pending connection request, or a connection waiting
2085 * for completion.
2086 */
2087 if (si_state_in(si_b->state, SI_SB_REQ|SI_SB_QUE|SI_SB_TAR|SI_SB_ASS)) {
2088 /* prune the request variables and swap to the response variables. */
2089 if (s->vars_reqres.scope != SCOPE_RES) {
2090 if (!LIST_ISEMPTY(&s->vars_reqres.head))
2091 vars_prune(&s->vars_reqres, s->sess, s);
2092 vars_init(&s->vars_reqres, SCOPE_RES);
2093 }
2094
2095 do {
2096 /* nb: step 1 might switch from QUE to ASS, but we first want
2097 * to give a chance to step 2 to perform a redirect if needed.
2098 */
2099 if (si_b->state != SI_ST_REQ)
2100 back_try_conn_req(s);
2101 if (si_b->state == SI_ST_REQ)
2102 back_handle_st_req(s);
2103
2104 /* get a chance to complete an immediate connection setup */
2105 if (si_b->state == SI_ST_RDY)
2106 goto resync_stream_interface;
2107
2108 /* applets directly go to the ESTABLISHED state. Similarly,
2109 * servers experience the same fate when their connection
2110 * is reused.
2111 */
2112 if (unlikely(si_b->state == SI_ST_EST))
2113 back_establish(s);
2114
2115 srv = objt_server(s->target);
2116 if (si_b->state == SI_ST_ASS && srv && srv->rdr_len && (s->flags & SF_REDIRECTABLE))
2117 http_perform_server_redirect(s, si_b);
2118 } while (si_b->state == SI_ST_ASS);
2119 }
2120
2121 /* Let's see if we can send the pending request now */
2122 si_sync_send(si_b);
2123
2124 /*
2125 * Now forward all shutdown requests between both sides of the request buffer
2126 */
2127
2128 /* first, let's check if the request buffer needs to shutdown(write), which may
2129 * happen either because the input is closed or because we want to force a close
2130 * once the server has begun to respond. If a half-closed timeout is set, we adjust
2131 * the other side's timeout as well.
2132 */
2133 if (unlikely((req->flags & (CF_SHUTW|CF_SHUTW_NOW|CF_AUTO_CLOSE|CF_SHUTR)) ==
2134 (CF_AUTO_CLOSE|CF_SHUTR))) {
2135 channel_shutw_now(req);
2136 }
2137
2138 /* shutdown(write) pending */
2139 if (unlikely((req->flags & (CF_SHUTW|CF_SHUTW_NOW)) == CF_SHUTW_NOW &&
2140 channel_is_empty(req))) {
2141 if (req->flags & CF_READ_ERROR)
2142 si_b->flags |= SI_FL_NOLINGER;
2143 si_shutw(si_b);
2144 }
2145
2146 /* shutdown(write) done on server side, we must stop the client too */
2147 if (unlikely((req->flags & (CF_SHUTW|CF_SHUTR|CF_SHUTR_NOW)) == CF_SHUTW &&
2148 !req->analysers))
2149 channel_shutr_now(req);
2150
2151 /* shutdown(read) pending */
2152 if (unlikely((req->flags & (CF_SHUTR|CF_SHUTR_NOW)) == CF_SHUTR_NOW)) {
2153 if (si_f->flags & SI_FL_NOHALF)
2154 si_f->flags |= SI_FL_NOLINGER;
2155 si_shutr(si_f);
2156 }
2157
2158 /* Benchmarks have shown that it's optimal to do a full resync now */
2159 if (si_f->state == SI_ST_DIS ||
2160 si_state_in(si_b->state, SI_SB_RDY|SI_SB_DIS) ||
2161 (si_f->flags & SI_FL_ERR && si_f->state != SI_ST_CLO) ||
2162 (si_b->flags & SI_FL_ERR && si_b->state != SI_ST_CLO))
2163 goto resync_stream_interface;
2164
2165 /* otherwise we want to check if we need to resync the req buffer or not */
2166 if ((req->flags ^ rqf_last) & (CF_SHUTR|CF_SHUTW))
2167 goto resync_request;
2168
2169 /* perform output updates to the response buffer */
2170
2171 /* If noone is interested in analysing data, it's time to forward
2172 * everything. We configure the buffer to forward indefinitely.
2173 * Note that we're checking CF_SHUTR_NOW as an indication of a possible
2174 * recent call to channel_abort().
2175 */
2176 if (unlikely((!res->analysers || (res->analysers == AN_RES_FLT_END && !(res->flags & CF_FLT_ANALYZE))) &&
2177 !(res->flags & (CF_SHUTW|CF_SHUTR_NOW)) &&
2178 si_state_in(si_b->state, SI_SB_EST|SI_SB_DIS|SI_SB_CLO) &&
2179 (res->to_forward != CHN_INFINITE_FORWARD))) {
2180 /* This buffer is freewheeling, there's no analyser
2181 * attached to it. If any data are left in, we'll permit them to
2182 * move.
2183 */
2184 channel_auto_read(res);
2185 channel_auto_close(res);
2186
2187 if (IS_HTX_STRM(s)) {
2188 struct htx *htx = htxbuf(&res->buf);
2189
2190 /* We'll let data flow between the producer (if still connected)
2191 * to the consumer.
2192 */
2193 co_set_data(res, htx->data);
2194 if (!(res->flags & (CF_SHUTR|CF_SHUTW_NOW)))
2195 channel_htx_forward_forever(res, htx);
2196 }
2197 else {
2198 /* We'll let data flow between the producer (if still connected)
2199 * to the consumer.
2200 */
2201 c_adv(res, ci_data(res));
2202 if (!(res->flags & (CF_SHUTR|CF_SHUTW_NOW)))
2203 channel_forward_forever(res);
2204 }
2205
2206 /* if we have no analyser anymore in any direction and have a
2207 * tunnel timeout set, use it now. Note that we must respect
2208 * the half-closed timeouts as well.
2209 */
2210 if (!req->analysers && s->be->timeout.tunnel) {
2211 req->rto = req->wto = res->rto = res->wto =
2212 s->be->timeout.tunnel;
2213
2214 if ((req->flags & CF_SHUTR) && tick_isset(sess->fe->timeout.clientfin))
2215 res->wto = sess->fe->timeout.clientfin;
2216 if ((req->flags & CF_SHUTW) && tick_isset(s->be->timeout.serverfin))
2217 res->rto = s->be->timeout.serverfin;
2218 if ((res->flags & CF_SHUTR) && tick_isset(s->be->timeout.serverfin))
2219 req->wto = s->be->timeout.serverfin;
2220 if ((res->flags & CF_SHUTW) && tick_isset(sess->fe->timeout.clientfin))
2221 req->rto = sess->fe->timeout.clientfin;
2222
2223 req->rex = tick_add(now_ms, req->rto);
2224 req->wex = tick_add(now_ms, req->wto);
2225 res->rex = tick_add(now_ms, res->rto);
2226 res->wex = tick_add(now_ms, res->wto);
2227 }
2228 }
2229
2230 /* check if it is wise to enable kernel splicing to forward response data */
2231 if (!(res->flags & (CF_KERN_SPLICING|CF_SHUTR)) &&
2232 res->to_forward &&
2233 (global.tune.options & GTUNE_USE_SPLICE) &&
2234 (objt_cs(si_f->end) && __objt_cs(si_f->end)->conn->xprt && __objt_cs(si_f->end)->conn->xprt->snd_pipe &&
2235 __objt_cs(si_f->end)->conn->mux && __objt_cs(si_f->end)->conn->mux->snd_pipe) &&
2236 (objt_cs(si_b->end) && __objt_cs(si_b->end)->conn->xprt && __objt_cs(si_b->end)->conn->xprt->rcv_pipe &&
2237 __objt_cs(si_b->end)->conn->mux && __objt_cs(si_b->end)->conn->mux->rcv_pipe) &&
2238 (pipes_used < global.maxpipes) &&
2239 (((sess->fe->options2|s->be->options2) & PR_O2_SPLIC_RTR) ||
2240 (((sess->fe->options2|s->be->options2) & PR_O2_SPLIC_AUT) &&
2241 (res->flags & CF_STREAMER_FAST)))) {
2242 res->flags |= CF_KERN_SPLICING;
2243 }
2244
2245 /* reflect what the L7 analysers have seen last */
2246 rpf_last = res->flags;
2247
2248 /* Let's see if we can send the pending response now */
2249 si_sync_send(si_f);
2250
2251 /*
2252 * Now forward all shutdown requests between both sides of the buffer
2253 */
2254
2255 /*
2256 * FIXME: this is probably where we should produce error responses.
2257 */
2258
2259 /* first, let's check if the response buffer needs to shutdown(write) */
2260 if (unlikely((res->flags & (CF_SHUTW|CF_SHUTW_NOW|CF_AUTO_CLOSE|CF_SHUTR)) ==
2261 (CF_AUTO_CLOSE|CF_SHUTR))) {
2262 channel_shutw_now(res);
2263 }
2264
2265 /* shutdown(write) pending */
2266 if (unlikely((res->flags & (CF_SHUTW|CF_SHUTW_NOW)) == CF_SHUTW_NOW &&
2267 channel_is_empty(res))) {
2268 si_shutw(si_f);
2269 }
2270
2271 /* shutdown(write) done on the client side, we must stop the server too */
2272 if (unlikely((res->flags & (CF_SHUTW|CF_SHUTR|CF_SHUTR_NOW)) == CF_SHUTW) &&
2273 !res->analysers)
2274 channel_shutr_now(res);
2275
2276 /* shutdown(read) pending */
2277 if (unlikely((res->flags & (CF_SHUTR|CF_SHUTR_NOW)) == CF_SHUTR_NOW)) {
2278 if (si_b->flags & SI_FL_NOHALF)
2279 si_b->flags |= SI_FL_NOLINGER;
2280 si_shutr(si_b);
2281 }
2282
2283 if (si_f->state == SI_ST_DIS ||
2284 si_state_in(si_b->state, SI_SB_RDY|SI_SB_DIS) ||
2285 (si_f->flags & SI_FL_ERR && si_f->state != SI_ST_CLO) ||
2286 (si_b->flags & SI_FL_ERR && si_b->state != SI_ST_CLO))
2287 goto resync_stream_interface;
2288
2289 if ((req->flags & ~rqf_last) & CF_MASK_ANALYSER)
2290 goto resync_request;
2291
2292 if ((res->flags ^ rpf_last) & CF_MASK_STATIC)
2293 goto resync_response;
2294
2295 if (((req->flags ^ rqf_last) | (res->flags ^ rpf_last)) & CF_MASK_ANALYSER)
2296 goto resync_request;
2297
2298 /* we're interested in getting wakeups again */
2299 si_f->flags &= ~SI_FL_DONT_WAKE;
2300 si_b->flags &= ~SI_FL_DONT_WAKE;
2301
2302 /* This is needed only when debugging is enabled, to indicate
2303 * client-side or server-side close. Please note that in the unlikely
2304 * event where both sides would close at once, the sequence is reported
2305 * on the server side first.
2306 */
2307 if (unlikely((global.mode & MODE_DEBUG) &&
2308 (!(global.mode & MODE_QUIET) ||
2309 (global.mode & MODE_VERBOSE)))) {
2310 if (si_b->state == SI_ST_CLO &&
2311 si_b->prev_state == SI_ST_EST) {
2312 chunk_printf(&trash, "%08x:%s.srvcls[%04x:%04x]\n",
2313 s->uniq_id, s->be->id,
2314 objt_cs(si_f->end) ? (unsigned short)objt_cs(si_f->end)->conn->handle.fd : -1,
2315 objt_cs(si_b->end) ? (unsigned short)objt_cs(si_b->end)->conn->handle.fd : -1);
2316 DISGUISE(write(1, trash.area, trash.data));
2317 }
2318
2319 if (si_f->state == SI_ST_CLO &&
2320 si_f->prev_state == SI_ST_EST) {
2321 chunk_printf(&trash, "%08x:%s.clicls[%04x:%04x]\n",
2322 s->uniq_id, s->be->id,
2323 objt_cs(si_f->end) ? (unsigned short)objt_cs(si_f->end)->conn->handle.fd : -1,
2324 objt_cs(si_b->end) ? (unsigned short)objt_cs(si_b->end)->conn->handle.fd : -1);
2325 DISGUISE(write(1, trash.area, trash.data));
2326 }
2327 }
2328
2329 if (likely((si_f->state != SI_ST_CLO) || !si_state_in(si_b->state, SI_SB_INI|SI_SB_CLO) ||
2330 (req->analysers & AN_REQ_FLT_END) || (res->analysers & AN_RES_FLT_END))) {
2331 if ((sess->fe->options & PR_O_CONTSTATS) && (s->flags & SF_BE_ASSIGNED) && !(s->flags & SF_IGNORE))
2332 stream_process_counters(s);
2333
2334 si_update_both(si_f, si_b);
2335
2336 /* Trick: if a request is being waiting for the server to respond,
2337 * and if we know the server can timeout, we don't want the timeout
2338 * to expire on the client side first, but we're still interested
2339 * in passing data from the client to the server (eg: POST). Thus,
2340 * we can cancel the client's request timeout if the server's
2341 * request timeout is set and the server has not yet sent a response.
2342 */
2343
2344 if ((res->flags & (CF_AUTO_CLOSE|CF_SHUTR)) == 0 &&
2345 (tick_isset(req->wex) || tick_isset(res->rex))) {
2346 req->flags |= CF_READ_NOEXP;
2347 req->rex = TICK_ETERNITY;
2348 }
2349
2350 /* Reset pending events now */
2351 s->pending_events = 0;
2352
2353 update_exp_and_leave:
2354 /* Note: please ensure that if you branch here you disable SI_FL_DONT_WAKE */
2355 t->expire = tick_first((tick_is_expired(t->expire, now_ms) ? 0 : t->expire),
2356 tick_first(tick_first(req->rex, req->wex),
2357 tick_first(res->rex, res->wex)));
2358 if (!req->analysers)
2359 req->analyse_exp = TICK_ETERNITY;
2360
2361 if ((sess->fe->options & PR_O_CONTSTATS) && (s->flags & SF_BE_ASSIGNED) &&
2362 (!tick_isset(req->analyse_exp) || tick_is_expired(req->analyse_exp, now_ms)))
2363 req->analyse_exp = tick_add(now_ms, 5000);
2364
2365 t->expire = tick_first(t->expire, req->analyse_exp);
2366
2367 t->expire = tick_first(t->expire, res->analyse_exp);
2368
2369 if (si_f->exp)
2370 t->expire = tick_first(t->expire, si_f->exp);
2371
2372 if (si_b->exp)
2373 t->expire = tick_first(t->expire, si_b->exp);
2374
2375 s->pending_events &= ~(TASK_WOKEN_TIMER | TASK_WOKEN_RES);
2376 stream_release_buffers(s);
2377
2378 DBG_TRACE_DEVEL("queuing", STRM_EV_STRM_PROC, s);
2379 return t; /* nothing more to do */
2380 }
2381
2382 DBG_TRACE_DEVEL("releasing", STRM_EV_STRM_PROC, s);
2383
2384 if (s->flags & SF_BE_ASSIGNED)
2385 _HA_ATOMIC_SUB(&s->be->beconn, 1);
2386
2387 if (unlikely((global.mode & MODE_DEBUG) &&
2388 (!(global.mode & MODE_QUIET) || (global.mode & MODE_VERBOSE)))) {
2389 chunk_printf(&trash, "%08x:%s.closed[%04x:%04x]\n",
2390 s->uniq_id, s->be->id,
2391 objt_cs(si_f->end) ? (unsigned short)objt_cs(si_f->end)->conn->handle.fd : -1,
2392 objt_cs(si_b->end) ? (unsigned short)objt_cs(si_b->end)->conn->handle.fd : -1);
2393 DISGUISE(write(1, trash.area, trash.data));
2394 }
2395
2396 s->logs.t_close = tv_ms_elapsed(&s->logs.tv_accept, &now);
2397 if (!(s->flags & SF_IGNORE))
2398 stream_process_counters(s);
2399
2400 if (s->txn && s->txn->status) {
2401 int n;
2402
2403 n = s->txn->status / 100;
2404 if (n < 1 || n > 5)
2405 n = 0;
2406
2407 if (sess->fe->mode == PR_MODE_HTTP) {
2408 _HA_ATOMIC_ADD(&sess->fe->fe_counters.p.http.rsp[n], 1);
2409 }
2410 if ((s->flags & SF_BE_ASSIGNED) &&
2411 (s->be->mode == PR_MODE_HTTP)) {
2412 _HA_ATOMIC_ADD(&s->be->be_counters.p.http.rsp[n], 1);
2413 _HA_ATOMIC_ADD(&s->be->be_counters.p.http.cum_req, 1);
2414 }
2415 }
2416
2417 /* let's do a final log if we need it */
2418 if (!LIST_ISEMPTY(&sess->fe->logformat) && s->logs.logwait &&
2419 !(s->flags & SF_MONITOR) &&
2420 (!(sess->fe->options & PR_O_NULLNOLOG) || req->total)) {
2421 /* we may need to know the position in the queue */
2422 pendconn_free(s);
2423 s->do_log(s);
2424 }
2425
2426 /* update time stats for this stream */
2427 stream_update_time_stats(s);
2428
2429 /* the task MUST not be in the run queue anymore */
2430 stream_free(s);
2431 task_destroy(t);
2432 return NULL;
2433 }
2434
2435 /* Update the stream's backend and server time stats */
stream_update_time_stats(struct stream * s)2436 void stream_update_time_stats(struct stream *s)
2437 {
2438 int t_request;
2439 int t_queue;
2440 int t_connect;
2441 int t_data;
2442 int t_close;
2443 struct server *srv;
2444 unsigned int samples_window;
2445
2446 t_request = 0;
2447 t_queue = s->logs.t_queue;
2448 t_connect = s->logs.t_connect;
2449 t_close = s->logs.t_close;
2450 t_data = s->logs.t_data;
2451
2452 if (s->be->mode != PR_MODE_HTTP)
2453 t_data = t_connect;
2454
2455 if (t_connect < 0 || t_data < 0)
2456 return;
2457
2458 if (tv_isge(&s->logs.tv_request, &s->logs.tv_accept))
2459 t_request = tv_ms_elapsed(&s->logs.tv_accept, &s->logs.tv_request);
2460
2461 t_data -= t_connect;
2462 t_connect -= t_queue;
2463 t_queue -= t_request;
2464
2465 srv = objt_server(s->target);
2466 if (srv) {
2467 samples_window = (((s->be->mode == PR_MODE_HTTP) ?
2468 srv->counters.p.http.cum_req : srv->counters.cum_lbconn) > TIME_STATS_SAMPLES) ? TIME_STATS_SAMPLES : 0;
2469 swrate_add_dynamic(&srv->counters.q_time, samples_window, t_queue);
2470 swrate_add_dynamic(&srv->counters.c_time, samples_window, t_connect);
2471 swrate_add_dynamic(&srv->counters.d_time, samples_window, t_data);
2472 swrate_add_dynamic(&srv->counters.t_time, samples_window, t_close);
2473 HA_ATOMIC_UPDATE_MAX(&srv->counters.qtime_max, t_queue);
2474 HA_ATOMIC_UPDATE_MAX(&srv->counters.ctime_max, t_connect);
2475 HA_ATOMIC_UPDATE_MAX(&srv->counters.dtime_max, t_data);
2476 HA_ATOMIC_UPDATE_MAX(&srv->counters.ttime_max, t_close);
2477 }
2478 samples_window = (((s->be->mode == PR_MODE_HTTP) ?
2479 s->be->be_counters.p.http.cum_req : s->be->be_counters.cum_lbconn) > TIME_STATS_SAMPLES) ? TIME_STATS_SAMPLES : 0;
2480 swrate_add_dynamic(&s->be->be_counters.q_time, samples_window, t_queue);
2481 swrate_add_dynamic(&s->be->be_counters.c_time, samples_window, t_connect);
2482 swrate_add_dynamic(&s->be->be_counters.d_time, samples_window, t_data);
2483 swrate_add_dynamic(&s->be->be_counters.t_time, samples_window, t_close);
2484 HA_ATOMIC_UPDATE_MAX(&s->be->be_counters.qtime_max, t_queue);
2485 HA_ATOMIC_UPDATE_MAX(&s->be->be_counters.ctime_max, t_connect);
2486 HA_ATOMIC_UPDATE_MAX(&s->be->be_counters.dtime_max, t_data);
2487 HA_ATOMIC_UPDATE_MAX(&s->be->be_counters.ttime_max, t_close);
2488 }
2489
2490 /*
2491 * This function adjusts sess->srv_conn and maintains the previous and new
2492 * server's served stream counts. Setting newsrv to NULL is enough to release
2493 * current connection slot. This function also notifies any LB algo which might
2494 * expect to be informed about any change in the number of active streams on a
2495 * server.
2496 */
sess_change_server(struct stream * sess,struct server * newsrv)2497 void sess_change_server(struct stream *sess, struct server *newsrv)
2498 {
2499 if (sess->srv_conn == newsrv)
2500 return;
2501
2502 if (sess->srv_conn) {
2503 _HA_ATOMIC_SUB(&sess->srv_conn->served, 1);
2504 _HA_ATOMIC_SUB(&sess->srv_conn->proxy->served, 1);
2505 __ha_barrier_atomic_store();
2506 if (sess->srv_conn->proxy->lbprm.server_drop_conn) {
2507 HA_SPIN_LOCK(SERVER_LOCK, &sess->srv_conn->lock);
2508 sess->srv_conn->proxy->lbprm.server_drop_conn(sess->srv_conn);
2509 HA_SPIN_UNLOCK(SERVER_LOCK, &sess->srv_conn->lock);
2510 }
2511 stream_del_srv_conn(sess);
2512 }
2513
2514 if (newsrv) {
2515 _HA_ATOMIC_ADD(&newsrv->served, 1);
2516 _HA_ATOMIC_ADD(&newsrv->proxy->served, 1);
2517 __ha_barrier_atomic_store();
2518 if (newsrv->proxy->lbprm.server_take_conn) {
2519 HA_SPIN_LOCK(SERVER_LOCK, &newsrv->lock);
2520 newsrv->proxy->lbprm.server_take_conn(newsrv);
2521 HA_SPIN_UNLOCK(SERVER_LOCK, &newsrv->lock);
2522 }
2523 stream_add_srv_conn(sess, newsrv);
2524 }
2525 }
2526
2527 /* Handle server-side errors for default protocols. It is called whenever a a
2528 * connection setup is aborted or a request is aborted in queue. It sets the
2529 * stream termination flags so that the caller does not have to worry about
2530 * them. It's installed as ->srv_error for the server-side stream_interface.
2531 */
default_srv_error(struct stream * s,struct stream_interface * si)2532 void default_srv_error(struct stream *s, struct stream_interface *si)
2533 {
2534 int err_type = si->err_type;
2535 int err = 0, fin = 0;
2536
2537 if (err_type & SI_ET_QUEUE_ABRT) {
2538 err = SF_ERR_CLICL;
2539 fin = SF_FINST_Q;
2540 }
2541 else if (err_type & SI_ET_CONN_ABRT) {
2542 err = SF_ERR_CLICL;
2543 fin = SF_FINST_C;
2544 }
2545 else if (err_type & SI_ET_QUEUE_TO) {
2546 err = SF_ERR_SRVTO;
2547 fin = SF_FINST_Q;
2548 }
2549 else if (err_type & SI_ET_QUEUE_ERR) {
2550 err = SF_ERR_SRVCL;
2551 fin = SF_FINST_Q;
2552 }
2553 else if (err_type & SI_ET_CONN_TO) {
2554 err = SF_ERR_SRVTO;
2555 fin = SF_FINST_C;
2556 }
2557 else if (err_type & SI_ET_CONN_ERR) {
2558 err = SF_ERR_SRVCL;
2559 fin = SF_FINST_C;
2560 }
2561 else if (err_type & SI_ET_CONN_RES) {
2562 err = SF_ERR_RESOURCE;
2563 fin = SF_FINST_C;
2564 }
2565 else /* SI_ET_CONN_OTHER and others */ {
2566 err = SF_ERR_INTERNAL;
2567 fin = SF_FINST_C;
2568 }
2569
2570 if (!(s->flags & SF_ERR_MASK))
2571 s->flags |= err;
2572 if (!(s->flags & SF_FINST_MASK))
2573 s->flags |= fin;
2574 }
2575
2576 /* kill a stream and set the termination flags to <why> (one of SF_ERR_*) */
stream_shutdown(struct stream * stream,int why)2577 void stream_shutdown(struct stream *stream, int why)
2578 {
2579 if (stream->req.flags & (CF_SHUTW|CF_SHUTW_NOW))
2580 return;
2581
2582 channel_shutw_now(&stream->req);
2583 channel_shutr_now(&stream->res);
2584 stream->task->nice = 1024;
2585 if (!(stream->flags & SF_ERR_MASK))
2586 stream->flags |= why;
2587 task_wakeup(stream->task, TASK_WOKEN_OTHER);
2588 }
2589
2590 /* Appends a dump of the state of stream <s> into buffer <buf> which must have
2591 * preliminary be prepared by its caller, with each line prepended by prefix
2592 * <pfx>, and each line terminated by character <eol>.
2593 */
stream_dump(struct buffer * buf,const struct stream * s,const char * pfx,char eol)2594 void stream_dump(struct buffer *buf, const struct stream *s, const char *pfx, char eol)
2595 {
2596 const struct conn_stream *csf, *csb;
2597 const struct connection *cof, *cob;
2598 const struct appctx *acf, *acb;
2599 const struct server *srv;
2600 const char *src = "unknown";
2601 const char *dst = "unknown";
2602 char pn[INET6_ADDRSTRLEN];
2603 const struct channel *req, *res;
2604 const struct stream_interface *si_f, *si_b;
2605
2606 if (!s) {
2607 chunk_appendf(buf, "%sstrm=%p%c", pfx, s, eol);
2608 return;
2609 }
2610
2611 if (s->obj_type != OBJ_TYPE_STREAM) {
2612 chunk_appendf(buf, "%sstrm=%p [invalid type=%d(%s)]%c",
2613 pfx, s, s->obj_type, obj_type_name(&s->obj_type), eol);
2614 return;
2615 }
2616
2617 si_f = &s->si[0];
2618 si_b = &s->si[1];
2619 req = &s->req;
2620 res = &s->res;
2621
2622 csf = objt_cs(si_f->end);
2623 cof = cs_conn(csf);
2624 acf = objt_appctx(si_f->end);
2625 if (cof && cof->src && addr_to_str(cof->src, pn, sizeof(pn)) >= 0)
2626 src = pn;
2627 else if (acf)
2628 src = acf->applet->name;
2629
2630 csb = objt_cs(si_b->end);
2631 cob = cs_conn(csb);
2632 acb = objt_appctx(si_b->end);
2633 srv = objt_server(s->target);
2634 if (srv)
2635 dst = srv->id;
2636 else if (acb)
2637 dst = acb->applet->name;
2638
2639 chunk_appendf(buf,
2640 "%sstrm=%p,%x src=%s fe=%s be=%s dst=%s%c"
2641 "%stxn=%p,%x txn.req=%s,%x txn.rsp=%s,%x%c"
2642 "%srqf=%x rqa=%x rpf=%x rpa=%x sif=%s,%x sib=%s,%x%c"
2643 "%saf=%p,%u csf=%p,%x%c"
2644 "%sab=%p,%u csb=%p,%x%c"
2645 "%scof=%p,%x:%s(%p)/%s(%p)/%s(%d)%c"
2646 "%scob=%p,%x:%s(%p)/%s(%p)/%s(%d)%c"
2647 "",
2648 pfx, s, s->flags, src, s->sess->fe->id, s->be->id, dst, eol,
2649 pfx, s->txn, (s->txn ? s->txn->flags : 0),
2650 (s->txn ? h1_msg_state_str(s->txn->req.msg_state): "-"), (s->txn ? s->txn->req.flags : 0),
2651 (s->txn ? h1_msg_state_str(s->txn->rsp.msg_state): "-"), (s->txn ? s->txn->rsp.flags : 0), eol,
2652 pfx, req->flags, req->analysers, res->flags, res->analysers,
2653 si_state_str(si_f->state), si_f->flags,
2654 si_state_str(si_b->state), si_b->flags, eol,
2655 pfx, acf, acf ? acf->st0 : 0, csf, csf ? csf->flags : 0, eol,
2656 pfx, acb, acb ? acb->st0 : 0, csb, csb ? csb->flags : 0, eol,
2657 pfx, cof, cof ? cof->flags : 0, conn_get_mux_name(cof), cof?cof->ctx:0, conn_get_xprt_name(cof),
2658 cof ? cof->xprt_ctx : 0, conn_get_ctrl_name(cof), cof ? cof->handle.fd : 0, eol,
2659 pfx, cob, cob ? cob->flags : 0, conn_get_mux_name(cob), cob?cob->ctx:0, conn_get_xprt_name(cob),
2660 cob ? cob->xprt_ctx : 0, conn_get_ctrl_name(cob), cob ? cob->handle.fd : 0, eol);
2661 }
2662
2663 /* dumps an error message for type <type> at ptr <ptr> related to stream <s>,
2664 * having reached loop rate <rate>, then aborts hoping to retrieve a core.
2665 */
stream_dump_and_crash(enum obj_type * obj,int rate)2666 void stream_dump_and_crash(enum obj_type *obj, int rate)
2667 {
2668 const struct stream *s;
2669 char *msg = NULL;
2670 const void *ptr;
2671
2672 ptr = s = objt_stream(obj);
2673 if (!s) {
2674 const struct appctx *appctx = objt_appctx(obj);
2675 if (!appctx)
2676 return;
2677 ptr = appctx;
2678 s = si_strm(appctx->owner);
2679 if (!s)
2680 return;
2681 }
2682
2683 chunk_reset(&trash);
2684 stream_dump(&trash, s, "", ' ');
2685
2686 chunk_appendf(&trash, "filters={");
2687 if (HAS_FILTERS(s)) {
2688 struct filter *filter;
2689
2690 list_for_each_entry(filter, &s->strm_flt.filters, list) {
2691 if (filter->list.p != &s->strm_flt.filters)
2692 chunk_appendf(&trash, ", ");
2693 chunk_appendf(&trash, "%p=\"%s\"", filter, FLT_ID(filter));
2694 }
2695 }
2696 chunk_appendf(&trash, "}");
2697
2698 memprintf(&msg,
2699 "A bogus %s [%p] is spinning at %d calls per second and refuses to die, "
2700 "aborting now! Please report this error to developers "
2701 "[%s]\n",
2702 obj_type_name(obj), ptr, rate, trash.area);
2703
2704 ha_alert("%s", msg);
2705 send_log(NULL, LOG_EMERG, "%s", msg);
2706 abort();
2707 }
2708
2709 /* Generates a unique ID based on the given <format>, stores it in the given <strm> and
2710 * returns the unique ID.
2711
2712 * If this function fails to allocate memory IST_NULL is returned.
2713 *
2714 * If an ID is already stored within the stream nothing happens existing unique ID is
2715 * returned.
2716 */
stream_generate_unique_id(struct stream * strm,struct list * format)2717 struct ist stream_generate_unique_id(struct stream *strm, struct list *format)
2718 {
2719 if (isttest(strm->unique_id)) {
2720 return strm->unique_id;
2721 }
2722 else {
2723 char *unique_id;
2724 int length;
2725 if ((unique_id = pool_alloc(pool_head_uniqueid)) == NULL)
2726 return IST_NULL;
2727
2728 length = build_logline(strm, unique_id, UNIQUEID_LEN, format);
2729 strm->unique_id = ist2(unique_id, length);
2730
2731 return strm->unique_id;
2732 }
2733 }
2734
2735 /************************************************************************/
2736 /* All supported ACL keywords must be declared here. */
2737 /************************************************************************/
2738
2739 /* 0=OK, <0=Alert, >0=Warning */
stream_parse_use_service(const char ** args,int * cur_arg,struct proxy * px,struct act_rule * rule,char ** err)2740 static enum act_parse_ret stream_parse_use_service(const char **args, int *cur_arg,
2741 struct proxy *px, struct act_rule *rule,
2742 char **err)
2743 {
2744 struct action_kw *kw;
2745
2746 /* Check if the service name exists. */
2747 if (*(args[*cur_arg]) == 0) {
2748 memprintf(err, "'%s' expects a service name.", args[0]);
2749 return ACT_RET_PRS_ERR;
2750 }
2751
2752 /* lookup for keyword corresponding to a service. */
2753 kw = action_lookup(&service_keywords, args[*cur_arg]);
2754 if (!kw) {
2755 memprintf(err, "'%s' unknown service name.", args[1]);
2756 return ACT_RET_PRS_ERR;
2757 }
2758 (*cur_arg)++;
2759
2760 /* executes specific rule parser. */
2761 rule->kw = kw;
2762 if (kw->parse((const char **)args, cur_arg, px, rule, err) == ACT_RET_PRS_ERR)
2763 return ACT_RET_PRS_ERR;
2764
2765 /* Register processing function. */
2766 rule->action_ptr = process_use_service;
2767 rule->action = ACT_CUSTOM;
2768
2769 return ACT_RET_PRS_OK;
2770 }
2771
service_keywords_register(struct action_kw_list * kw_list)2772 void service_keywords_register(struct action_kw_list *kw_list)
2773 {
2774 LIST_ADDQ(&service_keywords, &kw_list->list);
2775 }
2776
service_find(const char * kw)2777 struct action_kw *service_find(const char *kw)
2778 {
2779 return action_lookup(&service_keywords, kw);
2780 }
2781
2782 /* Lists the known services on <out> */
list_services(FILE * out)2783 void list_services(FILE *out)
2784 {
2785 struct action_kw_list *kw_list;
2786 int found = 0;
2787 int i;
2788
2789 fprintf(out, "Available services :");
2790 list_for_each_entry(kw_list, &service_keywords, list) {
2791 for (i = 0; kw_list->kw[i].kw != NULL; i++) {
2792 found = 1;
2793 fprintf(out, " %s", kw_list->kw[i].kw);
2794 }
2795 }
2796 if (!found)
2797 fprintf(out, " none\n");
2798 }
2799
2800 /* This function dumps a complete stream state onto the stream interface's
2801 * read buffer. The stream has to be set in strm. It returns 0 if the output
2802 * buffer is full and it needs to be called again, otherwise non-zero. It is
2803 * designed to be called from stats_dump_strm_to_buffer() below.
2804 */
stats_dump_full_strm_to_buffer(struct stream_interface * si,struct stream * strm)2805 static int stats_dump_full_strm_to_buffer(struct stream_interface *si, struct stream *strm)
2806 {
2807 struct appctx *appctx = __objt_appctx(si->end);
2808 struct tm tm;
2809 extern const char *monthname[12];
2810 char pn[INET6_ADDRSTRLEN];
2811 struct conn_stream *cs;
2812 struct connection *conn;
2813 struct appctx *tmpctx;
2814
2815 chunk_reset(&trash);
2816
2817 if (appctx->ctx.sess.section > 0 && appctx->ctx.sess.uid != strm->uniq_id) {
2818 /* stream changed, no need to go any further */
2819 chunk_appendf(&trash, " *** session terminated while we were watching it ***\n");
2820 if (ci_putchk(si_ic(si), &trash) == -1)
2821 goto full;
2822 goto done;
2823 }
2824
2825 switch (appctx->ctx.sess.section) {
2826 case 0: /* main status of the stream */
2827 appctx->ctx.sess.uid = strm->uniq_id;
2828 appctx->ctx.sess.section = 1;
2829 /* fall through */
2830
2831 case 1:
2832 get_localtime(strm->logs.accept_date.tv_sec, &tm);
2833 chunk_appendf(&trash,
2834 "%p: [%02d/%s/%04d:%02d:%02d:%02d.%06d] id=%u proto=%s",
2835 strm,
2836 tm.tm_mday, monthname[tm.tm_mon], tm.tm_year+1900,
2837 tm.tm_hour, tm.tm_min, tm.tm_sec, (int)(strm->logs.accept_date.tv_usec),
2838 strm->uniq_id,
2839 strm_li(strm) ? strm_li(strm)->proto->name : "?");
2840
2841 conn = objt_conn(strm_orig(strm));
2842 switch (conn && conn_get_src(conn) ? addr_to_str(conn->src, pn, sizeof(pn)) : AF_UNSPEC) {
2843 case AF_INET:
2844 case AF_INET6:
2845 chunk_appendf(&trash, " source=%s:%d\n",
2846 pn, get_host_port(conn->src));
2847 break;
2848 case AF_UNIX:
2849 chunk_appendf(&trash, " source=unix:%d\n", strm_li(strm)->luid);
2850 break;
2851 default:
2852 /* no more information to print right now */
2853 chunk_appendf(&trash, "\n");
2854 break;
2855 }
2856
2857 chunk_appendf(&trash,
2858 " flags=0x%x, conn_retries=%d, srv_conn=%p, pend_pos=%p waiting=%d\n",
2859 strm->flags, strm->si[1].conn_retries, strm->srv_conn, strm->pend_pos,
2860 MT_LIST_ADDED(&strm->buffer_wait.list));
2861
2862 chunk_appendf(&trash,
2863 " frontend=%s (id=%u mode=%s), listener=%s (id=%u)",
2864 strm_fe(strm)->id, strm_fe(strm)->uuid, strm_fe(strm)->mode ? "http" : "tcp",
2865 strm_li(strm) ? strm_li(strm)->name ? strm_li(strm)->name : "?" : "?",
2866 strm_li(strm) ? strm_li(strm)->luid : 0);
2867
2868 switch (conn && conn_get_dst(conn) ? addr_to_str(conn->dst, pn, sizeof(pn)) : AF_UNSPEC) {
2869 case AF_INET:
2870 case AF_INET6:
2871 chunk_appendf(&trash, " addr=%s:%d\n",
2872 pn, get_host_port(conn->dst));
2873 break;
2874 case AF_UNIX:
2875 chunk_appendf(&trash, " addr=unix:%d\n", strm_li(strm)->luid);
2876 break;
2877 default:
2878 /* no more information to print right now */
2879 chunk_appendf(&trash, "\n");
2880 break;
2881 }
2882
2883 if (strm->be->cap & PR_CAP_BE)
2884 chunk_appendf(&trash,
2885 " backend=%s (id=%u mode=%s)",
2886 strm->be->id,
2887 strm->be->uuid, strm->be->mode ? "http" : "tcp");
2888 else
2889 chunk_appendf(&trash, " backend=<NONE> (id=-1 mode=-)");
2890
2891 cs = objt_cs(strm->si[1].end);
2892 conn = cs_conn(cs);
2893
2894 switch (conn && conn_get_src(conn) ? addr_to_str(conn->src, pn, sizeof(pn)) : AF_UNSPEC) {
2895 case AF_INET:
2896 case AF_INET6:
2897 chunk_appendf(&trash, " addr=%s:%d\n",
2898 pn, get_host_port(conn->src));
2899 break;
2900 case AF_UNIX:
2901 chunk_appendf(&trash, " addr=unix\n");
2902 break;
2903 default:
2904 /* no more information to print right now */
2905 chunk_appendf(&trash, "\n");
2906 break;
2907 }
2908
2909 if (strm->be->cap & PR_CAP_BE)
2910 chunk_appendf(&trash,
2911 " server=%s (id=%u)",
2912 objt_server(strm->target) ? objt_server(strm->target)->id : "<none>",
2913 objt_server(strm->target) ? objt_server(strm->target)->puid : 0);
2914 else
2915 chunk_appendf(&trash, " server=<NONE> (id=-1)");
2916
2917 switch (conn && conn_get_dst(conn) ? addr_to_str(conn->dst, pn, sizeof(pn)) : AF_UNSPEC) {
2918 case AF_INET:
2919 case AF_INET6:
2920 chunk_appendf(&trash, " addr=%s:%d\n",
2921 pn, get_host_port(conn->dst));
2922 break;
2923 case AF_UNIX:
2924 chunk_appendf(&trash, " addr=unix\n");
2925 break;
2926 default:
2927 /* no more information to print right now */
2928 chunk_appendf(&trash, "\n");
2929 break;
2930 }
2931
2932 chunk_appendf(&trash,
2933 " task=%p (state=0x%02x nice=%d calls=%u rate=%u exp=%s tmask=0x%lx%s",
2934 strm->task,
2935 strm->task->state,
2936 strm->task->nice, strm->task->calls, read_freq_ctr(&strm->call_rate),
2937 strm->task->expire ?
2938 tick_is_expired(strm->task->expire, now_ms) ? "<PAST>" :
2939 human_time(TICKS_TO_MS(strm->task->expire - now_ms),
2940 TICKS_TO_MS(1000)) : "<NEVER>",
2941 strm->task->thread_mask,
2942 task_in_rq(strm->task) ? ", running" : "");
2943
2944 chunk_appendf(&trash,
2945 " age=%s)\n",
2946 human_time(now.tv_sec - strm->logs.accept_date.tv_sec, 1));
2947
2948 if (strm->txn)
2949 chunk_appendf(&trash,
2950 " txn=%p flags=0x%x meth=%d status=%d req.st=%s rsp.st=%s req.f=0x%02x rsp.f=0x%02x\n",
2951 strm->txn, strm->txn->flags, strm->txn->meth, strm->txn->status,
2952 h1_msg_state_str(strm->txn->req.msg_state), h1_msg_state_str(strm->txn->rsp.msg_state),
2953 strm->txn->req.flags, strm->txn->rsp.flags);
2954
2955 chunk_appendf(&trash,
2956 " si[0]=%p (state=%s flags=0x%02x endp0=%s:%p exp=%s et=0x%03x sub=%d)\n",
2957 &strm->si[0],
2958 si_state_str(strm->si[0].state),
2959 strm->si[0].flags,
2960 obj_type_name(strm->si[0].end),
2961 obj_base_ptr(strm->si[0].end),
2962 strm->si[0].exp ?
2963 tick_is_expired(strm->si[0].exp, now_ms) ? "<PAST>" :
2964 human_time(TICKS_TO_MS(strm->si[0].exp - now_ms),
2965 TICKS_TO_MS(1000)) : "<NEVER>",
2966 strm->si[0].err_type, strm->si[0].wait_event.events);
2967
2968 chunk_appendf(&trash,
2969 " si[1]=%p (state=%s flags=0x%02x endp1=%s:%p exp=%s et=0x%03x sub=%d)\n",
2970 &strm->si[1],
2971 si_state_str(strm->si[1].state),
2972 strm->si[1].flags,
2973 obj_type_name(strm->si[1].end),
2974 obj_base_ptr(strm->si[1].end),
2975 strm->si[1].exp ?
2976 tick_is_expired(strm->si[1].exp, now_ms) ? "<PAST>" :
2977 human_time(TICKS_TO_MS(strm->si[1].exp - now_ms),
2978 TICKS_TO_MS(1000)) : "<NEVER>",
2979 strm->si[1].err_type, strm->si[1].wait_event.events);
2980
2981 if ((cs = objt_cs(strm->si[0].end)) != NULL) {
2982 conn = cs->conn;
2983
2984 chunk_appendf(&trash,
2985 " co0=%p ctrl=%s xprt=%s mux=%s data=%s target=%s:%p\n",
2986 conn,
2987 conn_get_ctrl_name(conn),
2988 conn_get_xprt_name(conn),
2989 conn_get_mux_name(conn),
2990 cs_get_data_name(cs),
2991 obj_type_name(conn->target),
2992 obj_base_ptr(conn->target));
2993
2994 chunk_appendf(&trash,
2995 " flags=0x%08x fd=%d fd.state=%02x updt=%d fd.tmask=0x%lx\n",
2996 conn->flags,
2997 conn->handle.fd,
2998 conn->handle.fd >= 0 ? fdtab[conn->handle.fd].state : 0,
2999 conn->handle.fd >= 0 ? !!(fdtab[conn->handle.fd].update_mask & tid_bit) : 0,
3000 conn->handle.fd >= 0 ? fdtab[conn->handle.fd].thread_mask: 0);
3001
3002 chunk_appendf(&trash, " cs=%p csf=0x%08x ctx=%p\n", cs, cs->flags, cs->ctx);
3003 }
3004 else if ((tmpctx = objt_appctx(strm->si[0].end)) != NULL) {
3005 chunk_appendf(&trash,
3006 " app0=%p st0=%d st1=%d st2=%d applet=%s tmask=0x%lx nice=%d calls=%u rate=%u cpu=%llu lat=%llu\n",
3007 tmpctx,
3008 tmpctx->st0,
3009 tmpctx->st1,
3010 tmpctx->st2,
3011 tmpctx->applet->name,
3012 tmpctx->thread_mask,
3013 tmpctx->t->nice, tmpctx->t->calls, read_freq_ctr(&tmpctx->call_rate),
3014 (unsigned long long)tmpctx->t->cpu_time, (unsigned long long)tmpctx->t->lat_time);
3015 }
3016
3017 if ((cs = objt_cs(strm->si[1].end)) != NULL) {
3018 conn = cs->conn;
3019
3020 chunk_appendf(&trash,
3021 " co1=%p ctrl=%s xprt=%s mux=%s data=%s target=%s:%p\n",
3022 conn,
3023 conn_get_ctrl_name(conn),
3024 conn_get_xprt_name(conn),
3025 conn_get_mux_name(conn),
3026 cs_get_data_name(cs),
3027 obj_type_name(conn->target),
3028 obj_base_ptr(conn->target));
3029
3030 chunk_appendf(&trash,
3031 " flags=0x%08x fd=%d fd.state=%02x updt=%d fd.tmask=0x%lx\n",
3032 conn->flags,
3033 conn->handle.fd,
3034 conn->handle.fd >= 0 ? fdtab[conn->handle.fd].state : 0,
3035 conn->handle.fd >= 0 ? !!(fdtab[conn->handle.fd].update_mask & tid_bit) : 0,
3036 conn->handle.fd >= 0 ? fdtab[conn->handle.fd].thread_mask: 0);
3037
3038 chunk_appendf(&trash, " cs=%p csf=0x%08x ctx=%p\n", cs, cs->flags, cs->ctx);
3039 }
3040 else if ((tmpctx = objt_appctx(strm->si[1].end)) != NULL) {
3041 chunk_appendf(&trash,
3042 " app1=%p st0=%d st1=%d st2=%d applet=%s tmask=0x%lx nice=%d calls=%u rate=%u cpu=%llu lat=%llu\n",
3043 tmpctx,
3044 tmpctx->st0,
3045 tmpctx->st1,
3046 tmpctx->st2,
3047 tmpctx->applet->name,
3048 tmpctx->thread_mask,
3049 tmpctx->t->nice, tmpctx->t->calls, read_freq_ctr(&tmpctx->call_rate),
3050 (unsigned long long)tmpctx->t->cpu_time, (unsigned long long)tmpctx->t->lat_time);
3051 }
3052
3053 chunk_appendf(&trash,
3054 " req=%p (f=0x%06x an=0x%x pipe=%d tofwd=%d total=%lld)\n"
3055 " an_exp=%s",
3056 &strm->req,
3057 strm->req.flags, strm->req.analysers,
3058 strm->req.pipe ? strm->req.pipe->data : 0,
3059 strm->req.to_forward, strm->req.total,
3060 strm->req.analyse_exp ?
3061 human_time(TICKS_TO_MS(strm->req.analyse_exp - now_ms),
3062 TICKS_TO_MS(1000)) : "<NEVER>");
3063
3064 chunk_appendf(&trash,
3065 " rex=%s",
3066 strm->req.rex ?
3067 human_time(TICKS_TO_MS(strm->req.rex - now_ms),
3068 TICKS_TO_MS(1000)) : "<NEVER>");
3069
3070 chunk_appendf(&trash,
3071 " wex=%s\n"
3072 " buf=%p data=%p o=%u p=%u i=%u size=%u\n",
3073 strm->req.wex ?
3074 human_time(TICKS_TO_MS(strm->req.wex - now_ms),
3075 TICKS_TO_MS(1000)) : "<NEVER>",
3076 &strm->req.buf,
3077 b_orig(&strm->req.buf), (unsigned int)co_data(&strm->req),
3078 (unsigned int)ci_head_ofs(&strm->req), (unsigned int)ci_data(&strm->req),
3079 (unsigned int)strm->req.buf.size);
3080
3081 if (IS_HTX_STRM(strm)) {
3082 struct htx *htx = htxbuf(&strm->req.buf);
3083
3084 chunk_appendf(&trash,
3085 " htx=%p flags=0x%x size=%u data=%u used=%u wrap=%s extra=%llu\n",
3086 htx, htx->flags, htx->size, htx->data, htx_nbblks(htx),
3087 (htx->tail >= htx->head) ? "NO" : "YES",
3088 (unsigned long long)htx->extra);
3089 }
3090
3091 chunk_appendf(&trash,
3092 " res=%p (f=0x%06x an=0x%x pipe=%d tofwd=%d total=%lld)\n"
3093 " an_exp=%s",
3094 &strm->res,
3095 strm->res.flags, strm->res.analysers,
3096 strm->res.pipe ? strm->res.pipe->data : 0,
3097 strm->res.to_forward, strm->res.total,
3098 strm->res.analyse_exp ?
3099 human_time(TICKS_TO_MS(strm->res.analyse_exp - now_ms),
3100 TICKS_TO_MS(1000)) : "<NEVER>");
3101
3102 chunk_appendf(&trash,
3103 " rex=%s",
3104 strm->res.rex ?
3105 human_time(TICKS_TO_MS(strm->res.rex - now_ms),
3106 TICKS_TO_MS(1000)) : "<NEVER>");
3107
3108 chunk_appendf(&trash,
3109 " wex=%s\n"
3110 " buf=%p data=%p o=%u p=%u i=%u size=%u\n",
3111 strm->res.wex ?
3112 human_time(TICKS_TO_MS(strm->res.wex - now_ms),
3113 TICKS_TO_MS(1000)) : "<NEVER>",
3114 &strm->res.buf,
3115 b_orig(&strm->res.buf), (unsigned int)co_data(&strm->res),
3116 (unsigned int)ci_head_ofs(&strm->res), (unsigned int)ci_data(&strm->res),
3117 (unsigned int)strm->res.buf.size);
3118
3119 if (IS_HTX_STRM(strm)) {
3120 struct htx *htx = htxbuf(&strm->res.buf);
3121
3122 chunk_appendf(&trash,
3123 " htx=%p flags=0x%x size=%u data=%u used=%u wrap=%s extra=%llu\n",
3124 htx, htx->flags, htx->size, htx->data, htx_nbblks(htx),
3125 (htx->tail >= htx->head) ? "NO" : "YES",
3126 (unsigned long long)htx->extra);
3127 }
3128
3129 if (ci_putchk(si_ic(si), &trash) == -1)
3130 goto full;
3131
3132 /* use other states to dump the contents */
3133 }
3134 /* end of dump */
3135 done:
3136 appctx->ctx.sess.uid = 0;
3137 appctx->ctx.sess.section = 0;
3138 return 1;
3139 full:
3140 return 0;
3141 }
3142
3143
cli_parse_show_sess(char ** args,char * payload,struct appctx * appctx,void * private)3144 static int cli_parse_show_sess(char **args, char *payload, struct appctx *appctx, void *private)
3145 {
3146 if (!cli_has_level(appctx, ACCESS_LVL_OPER))
3147 return 1;
3148
3149 if (*args[2] && strcmp(args[2], "all") == 0)
3150 appctx->ctx.sess.target = (void *)-1;
3151 else if (*args[2])
3152 appctx->ctx.sess.target = (void *)strtoul(args[2], NULL, 0);
3153 else
3154 appctx->ctx.sess.target = NULL;
3155 appctx->ctx.sess.section = 0; /* start with stream status */
3156 appctx->ctx.sess.pos = 0;
3157
3158 /* we need to put an end marker into the streams list. We're just moving
3159 * ourselves there, so that once we found ourselves we know we've reached
3160 * the end. Without this we can run forever if new streams arrive faster
3161 * than we can dump them.
3162 */
3163 HA_SPIN_LOCK(STRMS_LOCK, &streams_lock);
3164 LIST_DEL(&si_strm(appctx->owner)->list);
3165 LIST_ADDQ(&streams, &si_strm(appctx->owner)->list);
3166 HA_SPIN_UNLOCK(STRMS_LOCK, &streams_lock);
3167 return 0;
3168 }
3169
3170 /* This function dumps all streams' states onto the stream interface's
3171 * read buffer. It returns 0 if the output buffer is full and it needs
3172 * to be called again, otherwise non-zero. It proceeds in an isolated
3173 * thread so there is no thread safety issue here.
3174 */
cli_io_handler_dump_sess(struct appctx * appctx)3175 static int cli_io_handler_dump_sess(struct appctx *appctx)
3176 {
3177 struct stream_interface *si = appctx->owner;
3178 struct connection *conn;
3179
3180 thread_isolate();
3181
3182 if (unlikely(si_ic(si)->flags & (CF_WRITE_ERROR|CF_SHUTW))) {
3183 /* If we're forced to shut down, we might have to remove our
3184 * reference to the last stream being dumped.
3185 */
3186 if (appctx->st2 == STAT_ST_LIST) {
3187 if (!LIST_ISEMPTY(&appctx->ctx.sess.bref.users)) {
3188 LIST_DEL(&appctx->ctx.sess.bref.users);
3189 LIST_INIT(&appctx->ctx.sess.bref.users);
3190 }
3191 }
3192 goto done;
3193 }
3194
3195 chunk_reset(&trash);
3196
3197 switch (appctx->st2) {
3198 case STAT_ST_INIT:
3199 /* the function had not been called yet, let's prepare the
3200 * buffer for a response. We initialize the current stream
3201 * pointer to the first in the global list. When a target
3202 * stream is being destroyed, it is responsible for updating
3203 * this pointer. We know we have reached the end when this
3204 * pointer points back to the head of the streams list.
3205 */
3206 LIST_INIT(&appctx->ctx.sess.bref.users);
3207 appctx->ctx.sess.bref.ref = streams.n;
3208 appctx->st2 = STAT_ST_LIST;
3209 /* fall through */
3210
3211 case STAT_ST_LIST:
3212 /* first, let's detach the back-ref from a possible previous stream */
3213 if (!LIST_ISEMPTY(&appctx->ctx.sess.bref.users)) {
3214 LIST_DEL(&appctx->ctx.sess.bref.users);
3215 LIST_INIT(&appctx->ctx.sess.bref.users);
3216 }
3217
3218 /* and start from where we stopped, never going further than ourselves */
3219 while (appctx->ctx.sess.bref.ref != si_strm(appctx->owner)->list.n) {
3220 char pn[INET6_ADDRSTRLEN];
3221 struct stream *curr_strm;
3222
3223 curr_strm = LIST_ELEM(appctx->ctx.sess.bref.ref, struct stream *, list);
3224
3225 if (appctx->ctx.sess.target) {
3226 if (appctx->ctx.sess.target != (void *)-1 && appctx->ctx.sess.target != curr_strm)
3227 goto next_sess;
3228
3229 LIST_ADDQ(&curr_strm->back_refs, &appctx->ctx.sess.bref.users);
3230 /* call the proper dump() function and return if we're missing space */
3231 if (!stats_dump_full_strm_to_buffer(si, curr_strm))
3232 goto full;
3233
3234 /* stream dump complete */
3235 LIST_DEL(&appctx->ctx.sess.bref.users);
3236 LIST_INIT(&appctx->ctx.sess.bref.users);
3237 if (appctx->ctx.sess.target != (void *)-1) {
3238 appctx->ctx.sess.target = NULL;
3239 break;
3240 }
3241 else
3242 goto next_sess;
3243 }
3244
3245 chunk_appendf(&trash,
3246 "%p: proto=%s",
3247 curr_strm,
3248 strm_li(curr_strm) ? strm_li(curr_strm)->proto->name : "?");
3249
3250 conn = objt_conn(strm_orig(curr_strm));
3251 switch (conn && conn_get_src(conn) ? addr_to_str(conn->src, pn, sizeof(pn)) : AF_UNSPEC) {
3252 case AF_INET:
3253 case AF_INET6:
3254 chunk_appendf(&trash,
3255 " src=%s:%d fe=%s be=%s srv=%s",
3256 pn,
3257 get_host_port(conn->src),
3258 strm_fe(curr_strm)->id,
3259 (curr_strm->be->cap & PR_CAP_BE) ? curr_strm->be->id : "<NONE>",
3260 objt_server(curr_strm->target) ? objt_server(curr_strm->target)->id : "<none>"
3261 );
3262 break;
3263 case AF_UNIX:
3264 chunk_appendf(&trash,
3265 " src=unix:%d fe=%s be=%s srv=%s",
3266 strm_li(curr_strm)->luid,
3267 strm_fe(curr_strm)->id,
3268 (curr_strm->be->cap & PR_CAP_BE) ? curr_strm->be->id : "<NONE>",
3269 objt_server(curr_strm->target) ? objt_server(curr_strm->target)->id : "<none>"
3270 );
3271 break;
3272 }
3273
3274 chunk_appendf(&trash,
3275 " ts=%02x age=%s calls=%u rate=%u cpu=%llu lat=%llu",
3276 curr_strm->task->state,
3277 human_time(now.tv_sec - curr_strm->logs.tv_accept.tv_sec, 1),
3278 curr_strm->task->calls, read_freq_ctr(&curr_strm->call_rate),
3279 (unsigned long long)curr_strm->task->cpu_time, (unsigned long long)curr_strm->task->lat_time);
3280
3281 chunk_appendf(&trash,
3282 " rq[f=%06xh,i=%u,an=%02xh,rx=%s",
3283 curr_strm->req.flags,
3284 (unsigned int)ci_data(&curr_strm->req),
3285 curr_strm->req.analysers,
3286 curr_strm->req.rex ?
3287 human_time(TICKS_TO_MS(curr_strm->req.rex - now_ms),
3288 TICKS_TO_MS(1000)) : "");
3289
3290 chunk_appendf(&trash,
3291 ",wx=%s",
3292 curr_strm->req.wex ?
3293 human_time(TICKS_TO_MS(curr_strm->req.wex - now_ms),
3294 TICKS_TO_MS(1000)) : "");
3295
3296 chunk_appendf(&trash,
3297 ",ax=%s]",
3298 curr_strm->req.analyse_exp ?
3299 human_time(TICKS_TO_MS(curr_strm->req.analyse_exp - now_ms),
3300 TICKS_TO_MS(1000)) : "");
3301
3302 chunk_appendf(&trash,
3303 " rp[f=%06xh,i=%u,an=%02xh,rx=%s",
3304 curr_strm->res.flags,
3305 (unsigned int)ci_data(&curr_strm->res),
3306 curr_strm->res.analysers,
3307 curr_strm->res.rex ?
3308 human_time(TICKS_TO_MS(curr_strm->res.rex - now_ms),
3309 TICKS_TO_MS(1000)) : "");
3310
3311 chunk_appendf(&trash,
3312 ",wx=%s",
3313 curr_strm->res.wex ?
3314 human_time(TICKS_TO_MS(curr_strm->res.wex - now_ms),
3315 TICKS_TO_MS(1000)) : "");
3316
3317 chunk_appendf(&trash,
3318 ",ax=%s]",
3319 curr_strm->res.analyse_exp ?
3320 human_time(TICKS_TO_MS(curr_strm->res.analyse_exp - now_ms),
3321 TICKS_TO_MS(1000)) : "");
3322
3323 conn = cs_conn(objt_cs(curr_strm->si[0].end));
3324 chunk_appendf(&trash,
3325 " s0=[%d,%1xh,fd=%d,ex=%s]",
3326 curr_strm->si[0].state,
3327 curr_strm->si[0].flags,
3328 conn ? conn->handle.fd : -1,
3329 curr_strm->si[0].exp ?
3330 human_time(TICKS_TO_MS(curr_strm->si[0].exp - now_ms),
3331 TICKS_TO_MS(1000)) : "");
3332
3333 conn = cs_conn(objt_cs(curr_strm->si[1].end));
3334 chunk_appendf(&trash,
3335 " s1=[%d,%1xh,fd=%d,ex=%s]",
3336 curr_strm->si[1].state,
3337 curr_strm->si[1].flags,
3338 conn ? conn->handle.fd : -1,
3339 curr_strm->si[1].exp ?
3340 human_time(TICKS_TO_MS(curr_strm->si[1].exp - now_ms),
3341 TICKS_TO_MS(1000)) : "");
3342
3343 chunk_appendf(&trash,
3344 " exp=%s",
3345 curr_strm->task->expire ?
3346 human_time(TICKS_TO_MS(curr_strm->task->expire - now_ms),
3347 TICKS_TO_MS(1000)) : "");
3348 if (task_in_rq(curr_strm->task))
3349 chunk_appendf(&trash, " run(nice=%d)", curr_strm->task->nice);
3350
3351 chunk_appendf(&trash, "\n");
3352
3353 if (ci_putchk(si_ic(si), &trash) == -1) {
3354 /* let's try again later from this stream. We add ourselves into
3355 * this stream's users so that it can remove us upon termination.
3356 */
3357 LIST_ADDQ(&curr_strm->back_refs, &appctx->ctx.sess.bref.users);
3358 goto full;
3359 }
3360
3361 next_sess:
3362 appctx->ctx.sess.bref.ref = curr_strm->list.n;
3363 }
3364
3365 if (appctx->ctx.sess.target && appctx->ctx.sess.target != (void *)-1) {
3366 /* specified stream not found */
3367 if (appctx->ctx.sess.section > 0)
3368 chunk_appendf(&trash, " *** session terminated while we were watching it ***\n");
3369 else
3370 chunk_appendf(&trash, "Session not found.\n");
3371
3372 if (ci_putchk(si_ic(si), &trash) == -1)
3373 goto full;
3374
3375 appctx->ctx.sess.target = NULL;
3376 appctx->ctx.sess.uid = 0;
3377 goto done;
3378 }
3379 /* fall through */
3380
3381 default:
3382 appctx->st2 = STAT_ST_FIN;
3383 goto done;
3384 }
3385 done:
3386 thread_release();
3387 return 1;
3388 full:
3389 thread_release();
3390 si_rx_room_blk(si);
3391 return 0;
3392 }
3393
cli_release_show_sess(struct appctx * appctx)3394 static void cli_release_show_sess(struct appctx *appctx)
3395 {
3396 if (appctx->st2 == STAT_ST_LIST) {
3397 HA_SPIN_LOCK(STRMS_LOCK, &streams_lock);
3398 if (!LIST_ISEMPTY(&appctx->ctx.sess.bref.users))
3399 LIST_DEL(&appctx->ctx.sess.bref.users);
3400 HA_SPIN_UNLOCK(STRMS_LOCK, &streams_lock);
3401 }
3402 }
3403
3404 /* Parses the "shutdown session" directive, it always returns 1 */
cli_parse_shutdown_session(char ** args,char * payload,struct appctx * appctx,void * private)3405 static int cli_parse_shutdown_session(char **args, char *payload, struct appctx *appctx, void *private)
3406 {
3407 struct stream *strm, *ptr;
3408
3409 if (!cli_has_level(appctx, ACCESS_LVL_ADMIN))
3410 return 1;
3411
3412 if (!*args[2])
3413 return cli_err(appctx, "Session pointer expected (use 'show sess').\n");
3414
3415 ptr = (void *)strtoul(args[2], NULL, 0);
3416
3417 thread_isolate();
3418
3419 /* first, look for the requested stream in the stream table */
3420 list_for_each_entry(strm, &streams, list) {
3421 if (strm == ptr) {
3422 stream_shutdown(strm, SF_ERR_KILLED);
3423 break;
3424 }
3425 }
3426
3427 thread_release();
3428
3429 /* do we have the stream ? */
3430 if (strm != ptr)
3431 return cli_err(appctx, "No such session (use 'show sess').\n");
3432
3433 return 1;
3434 }
3435
3436 /* Parses the "shutdown session server" directive, it always returns 1 */
cli_parse_shutdown_sessions_server(char ** args,char * payload,struct appctx * appctx,void * private)3437 static int cli_parse_shutdown_sessions_server(char **args, char *payload, struct appctx *appctx, void *private)
3438 {
3439 struct server *sv;
3440
3441 if (!cli_has_level(appctx, ACCESS_LVL_ADMIN))
3442 return 1;
3443
3444 sv = cli_find_server(appctx, args[3]);
3445 if (!sv)
3446 return 1;
3447
3448 /* kill all the stream that are on this server */
3449 HA_SPIN_LOCK(SERVER_LOCK, &sv->lock);
3450 srv_shutdown_streams(sv, SF_ERR_KILLED);
3451 HA_SPIN_UNLOCK(SERVER_LOCK, &sv->lock);
3452 return 1;
3453 }
3454
3455 /* register cli keywords */
3456 static struct cli_kw_list cli_kws = {{ },{
3457 { { "show", "sess", NULL }, "show sess [id] : report the list of current sessions or dump this session", cli_parse_show_sess, cli_io_handler_dump_sess, cli_release_show_sess },
3458 { { "shutdown", "session", NULL }, "shutdown session : kill a specific session", cli_parse_shutdown_session, NULL, NULL },
3459 { { "shutdown", "sessions", "server" }, "shutdown sessions server : kill sessions on a server", cli_parse_shutdown_sessions_server, NULL, NULL },
3460 {{},}
3461 }};
3462
3463 INITCALL1(STG_REGISTER, cli_register_kw, &cli_kws);
3464
3465 /* main configuration keyword registration. */
3466 static struct action_kw_list stream_tcp_keywords = { ILH, {
3467 { "use-service", stream_parse_use_service },
3468 { /* END */ }
3469 }};
3470
3471 INITCALL1(STG_REGISTER, tcp_req_cont_keywords_register, &stream_tcp_keywords);
3472
3473 static struct action_kw_list stream_http_keywords = { ILH, {
3474 { "use-service", stream_parse_use_service },
3475 { /* END */ }
3476 }};
3477
3478 INITCALL1(STG_REGISTER, http_req_keywords_register, &stream_http_keywords);
3479
3480 /*
3481 * Local variables:
3482 * c-indent-level: 8
3483 * c-basic-offset: 8
3484 * End:
3485 */
3486