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