RepoMirrors/haproxy
1
0
Fork 0
mirror of http://git.haproxy.org/git/haproxy.git/ synced 2025-03-25 04:17:42 +00:00
Code Issues Packages Projects Releases Wiki Activity

MEDIUM: stream/trace: Register a new trace source with its events

Browse Source 
Runtime traces are now supported for the streams, only if compiled with
debug. process_stream() is covered as well as TCP/HTTP analyzers and filters.

In traces, the first argument is always a stream. So it is easy to get the info
about the channels and the stream-interfaces. The second argument, when defined,
is always a HTTP transaction. And the third one is an HTTP message. The trace
message is adapted to report HTTP info when possible.
This commit is contained in:
Christopher Faulet 2019-11-05 16:18:10 +01:00
parent db703b1918
commit eea8fc737b
6 changed files with 496 additions and 202 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

17
include/proto/stream.h
View File

@ -32,6 +32,23 @@
#include <proto/queue.h>
#include <proto/stick_table.h>
#include <proto/task.h>
#include <proto/trace.h>
extern struct trace_source trace_strm;
/* Details about these events are defined in <src/stream.c> */
#define STRM_EV_STRM_NEW (1ULL << 0)
#define STRM_EV_STRM_FREE (1ULL << 1)
#define STRM_EV_STRM_ERR (1ULL << 2)
#define STRM_EV_STRM_ANA (1ULL << 3)
#define STRM_EV_STRM_PROC (1ULL << 4)
#define STRM_EV_SI_ST (1ULL << 5)
#define STRM_EV_HTTP_ANA (1ULL << 6)
#define STRM_EV_HTTP_ERR (1ULL << 7)
#define STRM_EV_TCP_ANA (1ULL << 8)
#define STRM_EV_TCP_ERR (1ULL << 9)
#define STRM_EV_FLT_ANA (1ULL << 10)
#define STRM_EV_FLT_ERR (1ULL << 11)
#define IS_HTX_STRM(strm) ((strm)->flags & SF_HTX)

12
src/backend.c
View File

@ -60,6 +60,8 @@
#include <proto/ssl_sock.h>
#include <proto/task.h>
#define TRACE_SOURCE &trace_strm
int be_lastsession(const struct proxy *be)
{
if (be->be_counters.last_sess)
@ -1707,14 +1709,7 @@ int tcp_persist_rdp_cookie(struct stream *s, struct channel *req, int an_bit)
uint32_t addr;
char *p;
DPRINTF(stderr,"[%u] %s: stream=%p b=%p, exp(r,w)=%u,%u bf=%08x bh=%lu analysers=%02x\n",
now_ms, __FUNCTION__,
s,
req,
req->rex, req->wex,
req->flags,
ci_data(req),
req->analysers);
DBG_TRACE_ENTER(STRM_EV_STRM_ANA|STRM_EV_TCP_ANA, s);
if (s->flags & SF_ASSIGNED)
goto no_cookie;
@ -1757,6 +1752,7 @@ int tcp_persist_rdp_cookie(struct stream *s, struct channel *req, int an_bit)
no_cookie:
req->analysers &= ~an_bit;
req->analyse_exp = TICK_ETERNITY;
DBG_TRACE_LEAVE(STRM_EV_STRM_ANA|STRM_EV_TCP_ANA, s);
return 1;
}

67
src/filters.c
View File

@ -33,6 +33,8 @@
#include <proto/stream.h>
#include <proto/stream_interface.h>
#define TRACE_SOURCE &trace_strm
/* Pool used to allocate filters */
DECLARE_STATIC_POOL(pool_head_filter, "filter", sizeof(struct filter));
@ -542,14 +544,17 @@ flt_http_end(struct stream *s, struct http_msg *msg)
{
int ret = 1;
DBG_TRACE_ENTER(STRM_EV_STRM_ANA|STRM_EV_HTTP_ANA|STRM_EV_FLT_ANA, s, s->txn, msg);
RESUME_FILTER_LOOP(s, msg->chn) {
if (FLT_OPS(filter)->http_end) {
DBG_TRACE_DEVEL(FLT_ID(filter), STRM_EV_HTTP_ANA|STRM_EV_FLT_ANA, s);
ret = FLT_OPS(filter)->http_end(s, filter, msg);
if (ret <= 0)
BREAK_EXECUTION(s, msg->chn, end);
}
} RESUME_FILTER_END;
end:
DBG_TRACE_LEAVE(STRM_EV_STRM_ANA|STRM_EV_HTTP_ANA|STRM_EV_FLT_ANA, s);
return ret;
}
@ -562,10 +567,14 @@ flt_http_reset(struct stream *s, struct http_msg *msg)
{
struct filter *filter;
DBG_TRACE_ENTER(STRM_EV_STRM_ANA|STRM_EV_HTTP_ANA|STRM_EV_FLT_ANA, s, s->txn, msg);
list_for_each_entry(filter, &strm_flt(s)->filters, list) {
if (FLT_OPS(filter)->http_reset)
if (FLT_OPS(filter)->http_reset) {
DBG_TRACE_DEVEL(FLT_ID(filter), STRM_EV_HTTP_ANA|STRM_EV_FLT_ANA, s);
FLT_OPS(filter)->http_reset(s, filter, msg);
}
}
DBG_TRACE_LEAVE(STRM_EV_STRM_ANA|STRM_EV_HTTP_ANA|STRM_EV_FLT_ANA, s);
}
/*
@ -577,10 +586,14 @@ flt_http_reply(struct stream *s, short status, const struct buffer *msg)
{
struct filter *filter;
DBG_TRACE_ENTER(STRM_EV_STRM_ANA|STRM_EV_HTTP_ANA|STRM_EV_FLT_ANA, s, s->txn, msg);
list_for_each_entry(filter, &strm_flt(s)->filters, list) {
if (FLT_OPS(filter)->http_reply)
if (FLT_OPS(filter)->http_reply) {
DBG_TRACE_DEVEL(FLT_ID(filter), STRM_EV_HTTP_ANA|STRM_EV_FLT_ANA, s);
FLT_OPS(filter)->http_reply(s, filter, status, msg);
}
}
DBG_TRACE_LEAVE(STRM_EV_STRM_ANA|STRM_EV_HTTP_ANA|STRM_EV_FLT_ANA, s);
}
/*
@ -600,6 +613,7 @@ flt_http_payload(struct stream *s, struct http_msg *msg, unsigned int len)
unsigned int out = co_data(msg->chn);
int ret = len - out;
DBG_TRACE_ENTER(STRM_EV_STRM_ANA|STRM_EV_HTTP_ANA|STRM_EV_FLT_ANA, s, s->txn, msg);
list_for_each_entry(filter, &strm_flt(s)->filters, list) {
/* Call "data" filters only */
if (!IS_DATA_FILTER(filter, msg->chn))
@ -608,6 +622,7 @@ flt_http_payload(struct stream *s, struct http_msg *msg, unsigned int len)
unsigned long long *flt_off = &FLT_OFF(filter, msg->chn);
unsigned int offset = *flt_off - *strm_off;
DBG_TRACE_DEVEL(FLT_ID(filter), STRM_EV_HTTP_ANA|STRM_EV_FLT_ANA, s);
ret = FLT_OPS(filter)->http_payload(s, filter, msg, out + offset, ret - offset);
if (ret < 0)
goto end;
@ -617,6 +632,7 @@ flt_http_payload(struct stream *s, struct http_msg *msg, unsigned int len)
}
*strm_off += ret;
end:
DBG_TRACE_LEAVE(STRM_EV_STRM_ANA|STRM_EV_HTTP_ANA|STRM_EV_FLT_ANA, s);
return ret;
}
@ -634,6 +650,8 @@ flt_start_analyze(struct stream *s, struct channel *chn, unsigned int an_bit)
{
int ret = 1;
DBG_TRACE_ENTER(STRM_EV_STRM_ANA|STRM_EV_FLT_ANA, s);
/* If this function is called, this means there is at least one filter,
* so we do not need to check the filter list's emptiness. */
@ -656,6 +674,7 @@ flt_start_analyze(struct stream *s, struct channel *chn, unsigned int an_bit)
FLT_FWD(filter, chn) = 0;
if (FLT_OPS(filter)->channel_start_analyze) {
DBG_TRACE_DEVEL(FLT_ID(filter), STRM_EV_FLT_ANA, s);
ret = FLT_OPS(filter)->channel_start_analyze(s, filter, chn);
if (ret <= 0)
BREAK_EXECUTION(s, chn, end);
@ -663,7 +682,9 @@ flt_start_analyze(struct stream *s, struct channel *chn, unsigned int an_bit)
} RESUME_FILTER_END;
end:
return handle_analyzer_result(s, chn, an_bit, ret);
ret = handle_analyzer_result(s, chn, an_bit, ret);
DBG_TRACE_LEAVE(STRM_EV_STRM_ANA|STRM_EV_FLT_ANA, s);
return ret;
}
/*
@ -681,8 +702,11 @@ flt_pre_analyze(struct stream *s, struct channel *chn, unsigned int an_bit)
{
int ret = 1;
DBG_TRACE_ENTER(STRM_EV_STRM_ANA|STRM_EV_FLT_ANA, s);
RESUME_FILTER_LOOP(s, chn) {
if (FLT_OPS(filter)->channel_pre_analyze && (filter->pre_analyzers & an_bit)) {
DBG_TRACE_DEVEL(FLT_ID(filter), STRM_EV_FLT_ANA, s);
ret = FLT_OPS(filter)->channel_pre_analyze(s, filter, chn, an_bit);
if (ret <= 0)
BREAK_EXECUTION(s, chn, check_result);
@ -690,7 +714,9 @@ flt_pre_analyze(struct stream *s, struct channel *chn, unsigned int an_bit)
} RESUME_FILTER_END;
check_result:
return handle_analyzer_result(s, chn, 0, ret);
ret = handle_analyzer_result(s, chn, 0, ret);
DBG_TRACE_LEAVE(STRM_EV_STRM_ANA|STRM_EV_FLT_ANA, s);
return ret;
}
/*
@ -708,14 +734,19 @@ flt_post_analyze(struct stream *s, struct channel *chn, unsigned int an_bit)
struct filter *filter;
int ret = 1;
DBG_TRACE_ENTER(STRM_EV_STRM_ANA|STRM_EV_FLT_ANA, s);
list_for_each_entry(filter, &strm_flt(s)->filters, list) {
if (FLT_OPS(filter)->channel_post_analyze && (filter->post_analyzers & an_bit)) {
DBG_TRACE_DEVEL(FLT_ID(filter), STRM_EV_FLT_ANA, s);
ret = FLT_OPS(filter)->channel_post_analyze(s, filter, chn, an_bit);
if (ret < 0)
break;
}
}
return handle_analyzer_result(s, chn, 0, ret);
ret = handle_analyzer_result(s, chn, 0, ret);
DBG_TRACE_LEAVE(STRM_EV_STRM_ANA|STRM_EV_FLT_ANA, s);
return ret;
}
/*
@ -730,8 +761,11 @@ flt_analyze_http_headers(struct stream *s, struct channel *chn, unsigned int an_
int ret = 1;
msg = ((chn->flags & CF_ISRESP) ? &s->txn->rsp : &s->txn->req);
DBG_TRACE_ENTER(STRM_EV_STRM_ANA|STRM_EV_HTTP_ANA|STRM_EV_FLT_ANA, s, s->txn, msg);
RESUME_FILTER_LOOP(s, chn) {
if (FLT_OPS(filter)->http_headers) {
DBG_TRACE_DEVEL(FLT_ID(filter), STRM_EV_HTTP_ANA|STRM_EV_FLT_ANA, s);
ret = FLT_OPS(filter)->http_headers(s, filter, msg);
if (ret <= 0)
BREAK_EXECUTION(s, chn, check_result);
@ -740,7 +774,9 @@ flt_analyze_http_headers(struct stream *s, struct channel *chn, unsigned int an_
channel_htx_fwd_headers(chn, htxbuf(&chn->buf));
check_result:
return handle_analyzer_result(s, chn, an_bit, ret);
ret = handle_analyzer_result(s, chn, an_bit, ret);
DBG_TRACE_LEAVE(STRM_EV_STRM_ANA|STRM_EV_HTTP_ANA|STRM_EV_FLT_ANA, s);
return ret;
}
/*
@ -755,6 +791,8 @@ flt_end_analyze(struct stream *s, struct channel *chn, unsigned int an_bit)
{
int ret = 1;
DBG_TRACE_ENTER(STRM_EV_STRM_ANA|STRM_EV_FLT_ANA, s);
/* Check if all filters attached on the stream have finished their
* processing on this channel. */
if (!(chn->flags & CF_FLT_ANALYZE))
@ -766,6 +804,7 @@ flt_end_analyze(struct stream *s, struct channel *chn, unsigned int an_bit)
unregister_data_filter(s, chn, filter);
if (FLT_OPS(filter)->channel_end_analyze) {
DBG_TRACE_DEVEL(FLT_ID(filter), STRM_EV_FLT_ANA, s);
ret = FLT_OPS(filter)->channel_end_analyze(s, filter, chn);
if (ret <= 0)
BREAK_EXECUTION(s, chn, end);
@ -801,8 +840,11 @@ flt_end_analyze(struct stream *s, struct channel *chn, unsigned int an_bit)
/* Remove backend filters from the list */
flt_stream_release(s, 1);
DBG_TRACE_LEAVE(STRM_EV_STRM_ANA|STRM_EV_FLT_ANA, s);
}
else {
DBG_TRACE_DEVEL("waiting for sync", STRM_EV_STRM_ANA|STRM_EV_FLT_ANA, s);
}
return ret;
}
@ -836,6 +878,7 @@ flt_data(struct stream *s, struct channel *chn)
if (FLT_OPS(filter)->tcp_data) {
unsigned int i = ci_data(chn);
DBG_TRACE_DEVEL(FLT_ID(filter), STRM_EV_TCP_ANA|STRM_EV_FLT_ANA, s);
ret = FLT_OPS(filter)->tcp_data(s, filter, chn);
if (ret < 0)
break;
@ -891,6 +934,7 @@ flt_forward_data(struct stream *s, struct channel *chn, unsigned int len)
if (FLT_OPS(filter)->tcp_forward_data) {
/* Remove bytes that the current filter considered as
* forwarded */
DBG_TRACE_DEVEL(FLT_ID(filter), STRM_EV_TCP_ANA|STRM_EV_FLT_ANA, s);
ret = FLT_OPS(filter)->tcp_forward_data(s, filter, chn, ret - *fwd);
if (ret < 0)
goto end;
@ -934,6 +978,8 @@ flt_xfer_data(struct stream *s, struct channel *chn, unsigned int an_bit)
{
int ret = 1;
DBG_TRACE_ENTER(STRM_EV_STRM_ANA|STRM_EV_TCP_ANA|STRM_EV_FLT_ANA, s);
/* If there is no "data" filters, we do nothing */
if (!HAS_DATA_FILTERS(s, chn))
goto end;
@ -966,11 +1012,14 @@ flt_xfer_data(struct stream *s, struct channel *chn, unsigned int an_bit)
}
/* Wait for data */
DBG_TRACE_DEVEL("waiting for more data", STRM_EV_STRM_ANA|STRM_EV_TCP_ANA|STRM_EV_FLT_ANA, s);
return 0;
end:
/* Terminate the data filtering. If <ret> is negative, an error was
* encountered during the filtering. */
return handle_analyzer_result(s, chn, an_bit, ret);
ret = handle_analyzer_result(s, chn, an_bit, ret);
DBG_TRACE_LEAVE(STRM_EV_STRM_ANA|STRM_EV_TCP_ANA|STRM_EV_FLT_ANA, s);
return ret;
}
/*
@ -1028,11 +1077,13 @@ handle_analyzer_result(struct stream *s, struct channel *chn,
s->flags |= SF_ERR_PRXCOND;
if (!(s->flags & SF_FINST_MASK))
s->flags |= finst;
DBG_TRACE_DEVEL("leaving on error", STRM_EV_FLT_ANA|STRM_EV_FLT_ERR, s);
return 0;
wait:
if (!(chn->flags & CF_ISRESP))
channel_dont_connect(chn);
DBG_TRACE_DEVEL("wairing for more data", STRM_EV_FLT_ANA, s);
return 0;
}

203
src/http_ana.c
View File

@ -35,6 +35,8 @@
#include <proto/stats.h>
#include <proto/vars.h>
#define TRACE_SOURCE &trace_strm
extern const char *stat_status_codes[];
struct pool_head *pool_head_requri = NULL;
@ -86,14 +88,7 @@ int http_wait_for_request(struct stream *s, struct channel *req, int an_bit)
struct htx *htx;
struct htx_sl *sl;
DPRINTF(stderr,"[%u] %s: stream=%p b=%p, exp(r,w)=%u,%u bf=%08x bh=%lu analysers=%02x\n",
now_ms, __FUNCTION__,
s,
req,
req->rex, req->wex,
req->flags,
ci_data(req),
req->analysers);
DBG_TRACE_ENTER(STRM_EV_STRM_ANA|STRM_EV_HTTP_ANA, s, txn, msg);
htx = htxbuf(&req->buf);
@ -250,6 +245,8 @@ int http_wait_for_request(struct stream *s, struct channel *req, int an_bit)
}
/* we're not ready yet */
DBG_TRACE_DEVEL("waiting for the request",
STRM_EV_STRM_ANA|STRM_EV_HTTP_ANA, s, txn);
return 0;
failed_keep_alive:
@ -265,6 +262,8 @@ int http_wait_for_request(struct stream *s, struct channel *req, int an_bit)
s->logs.level = 0;
s->res.flags &= ~CF_EXPECT_MORE; /* speed up sending a previous response */
http_reply_and_close(s, txn->status, NULL);
DBG_TRACE_DEVEL("leaving by closing K/A connection",
STRM_EV_STRM_ANA|STRM_EV_HTTP_ANA, s, txn);
return 0;
}
@ -424,6 +423,7 @@ int http_wait_for_request(struct stream *s, struct channel *req, int an_bit)
req->analysers &= ~an_bit;
req->analyse_exp = TICK_ETERNITY;
DBG_TRACE_LEAVE(STRM_EV_STRM_ANA|STRM_EV_HTTP_ANA, s, txn);
return 1;
return_int_err:
@ -452,6 +452,8 @@ int http_wait_for_request(struct stream *s, struct channel *req, int an_bit)
req->analysers &= AN_REQ_FLT_END;
req->analyse_exp = TICK_ETERNITY;
DBG_TRACE_DEVEL("leaving on error",
STRM_EV_STRM_ANA|STRM_EV_HTTP_ANA|STRM_EV_HTTP_ERR, s, txn);
return 0;
}
@ -479,14 +481,7 @@ int http_process_req_common(struct stream *s, struct channel *req, int an_bit, s
goto return_prx_yield;
}
DPRINTF(stderr,"[%u] %s: stream=%p b=%p, exp(r,w)=%u,%u bf=%08x bh=%lu analysers=%02x\n",
now_ms, __FUNCTION__,
s,
req,
req->rex, req->wex,
req->flags,
ci_data(req),
req->analysers);
DBG_TRACE_ENTER(STRM_EV_STRM_ANA|STRM_EV_HTTP_ANA, s, txn, msg);
htx = htxbuf(&req->buf);
@ -619,6 +614,7 @@ int http_process_req_common(struct stream *s, struct channel *req, int an_bit, s
req->analyse_exp = TICK_ETERNITY;
done_without_exp: /* done with this analyser, but dont reset the analyse_exp. */
req->analysers &= ~an_bit;
DBG_TRACE_LEAVE(STRM_EV_STRM_ANA|STRM_EV_HTTP_ANA, s, txn);
return 1;
tarpit:
@ -689,10 +685,14 @@ int http_process_req_common(struct stream *s, struct channel *req, int an_bit, s
req->analysers &= AN_REQ_FLT_END;
req->analyse_exp = TICK_ETERNITY;
DBG_TRACE_DEVEL("leaving on error",
STRM_EV_STRM_ANA|STRM_EV_HTTP_ANA|STRM_EV_HTTP_ERR, s, txn);
return 0;
return_prx_yield:
channel_dont_connect(req);
DBG_TRACE_DEVEL("waiting for more data",
STRM_EV_STRM_ANA|STRM_EV_HTTP_ANA, s, txn);
return 0;
}
@ -715,14 +715,7 @@ int http_process_request(struct stream *s, struct channel *req, int an_bit)
return 0;
}
DPRINTF(stderr,"[%u] %s: stream=%p b=%p, exp(r,w)=%u,%u bf=%08x bh=%lu analysers=%02x\n",
now_ms, __FUNCTION__,
s,
req,
req->rex, req->wex,
req->flags,
ci_data(req),
req->analysers);
DBG_TRACE_ENTER(STRM_EV_STRM_ANA|STRM_EV_HTTP_ANA, s, txn, msg);
/*
* Right now, we know that we have processed the entire headers
@ -750,6 +743,8 @@ int http_process_request(struct stream *s, struct channel *req, int an_bit)
if (!(s->flags & SF_FINST_MASK))
s->flags |= SF_FINST_R;
DBG_TRACE_DEVEL("leaving on error",
STRM_EV_STRM_ANA|STRM_EV_HTTP_ANA|STRM_EV_HTTP_ERR, s, txn);
return 0;
}
sl = http_get_stline(htx);
@ -927,6 +922,7 @@ int http_process_request(struct stream *s, struct channel *req, int an_bit)
s->logs.tv_request = now;
/* OK let's go on with the BODY now */
DBG_TRACE_LEAVE(STRM_EV_STRM_ANA|STRM_EV_HTTP_ANA, s, txn);
return 1;
return_bad_req: /* let's centralize all bad requests */
@ -942,6 +938,8 @@ int http_process_request(struct stream *s, struct channel *req, int an_bit)
s->flags |= SF_ERR_PRXCOND;
if (!(s->flags & SF_FINST_MASK))
s->flags |= SF_FINST_R;
DBG_TRACE_DEVEL("leaving on error",
STRM_EV_STRM_ANA|STRM_EV_HTTP_ANA|STRM_EV_HTTP_ERR, s, txn);
return 0;
}
@ -953,6 +951,7 @@ int http_process_tarpit(struct stream *s, struct channel *req, int an_bit)
{
struct http_txn *txn = s->txn;
DBG_TRACE_ENTER(STRM_EV_STRM_ANA|STRM_EV_HTTP_ANA, s, txn, &txn->req);
/* This connection is being tarpitted. The CLIENT side has
* already set the connect expiration date to the right
* timeout. We just have to check that the client is still
@ -960,8 +959,12 @@ int http_process_tarpit(struct stream *s, struct channel *req, int an_bit)
*/
channel_dont_connect(req);
if ((req->flags & (CF_SHUTR|CF_READ_ERROR)) == 0 &&
!tick_is_expired(req->analyse_exp, now_ms))
!tick_is_expired(req->analyse_exp, now_ms)) {
DBG_TRACE_DEVEL("waiting for tarpit timeout expiry",
STRM_EV_STRM_ANA|STRM_EV_HTTP_ANA, s, txn);
return 0;
}
/* We will set the queue timer to the time spent, just for
* logging purposes. We fake a 500 server error, so that the
@ -981,6 +984,8 @@ int http_process_tarpit(struct stream *s, struct channel *req, int an_bit)
s->flags |= SF_ERR_PRXCOND;
if (!(s->flags & SF_FINST_MASK))
s->flags |= SF_FINST_T;
DBG_TRACE_LEAVE(STRM_EV_STRM_ANA|STRM_EV_HTTP_ANA, s, txn);
return 0;
}
@ -1000,15 +1005,7 @@ int http_wait_for_request_body(struct stream *s, struct channel *req, int an_bit
struct http_msg *msg = &s->txn->req;
struct htx *htx;
DPRINTF(stderr,"[%u] %s: stream=%p b=%p, exp(r,w)=%u,%u bf=%08x bh=%lu analysers=%02x\n",
now_ms, __FUNCTION__,
s,
req,
req->rex, req->wex,
req->flags,
ci_data(req),
req->analysers);
DBG_TRACE_ENTER(STRM_EV_STRM_ANA|STRM_EV_HTTP_ANA, s, txn, msg);
htx = htxbuf(&req->buf);
@ -1062,6 +1059,8 @@ int http_wait_for_request_body(struct stream *s, struct channel *req, int an_bit
channel_dont_connect(req);
if (!tick_isset(req->analyse_exp))
req->analyse_exp = tick_add_ifset(now_ms, s->be->timeout.httpreq);
DBG_TRACE_DEVEL("waiting for more data",
STRM_EV_STRM_ANA|STRM_EV_HTTP_ANA, s, txn);
return 0;
}
@ -1070,6 +1069,7 @@ int http_wait_for_request_body(struct stream *s, struct channel *req, int an_bit
s->logs.tv_request = now; /* update the request timer to reflect full request */
req->analysers &= ~an_bit;
req->analyse_exp = TICK_ETERNITY;
DBG_TRACE_LEAVE(STRM_EV_STRM_ANA|STRM_EV_HTTP_ANA, s, txn);
return 1;
return_int_err:
@ -1097,6 +1097,8 @@ int http_wait_for_request_body(struct stream *s, struct channel *req, int an_bit
_HA_ATOMIC_ADD(&sess->fe->fe_counters.failed_req, 1);
if (sess->listener->counters)
_HA_ATOMIC_ADD(&sess->listener->counters->failed_req, 1);
DBG_TRACE_DEVEL("leaving on error",
STRM_EV_STRM_ANA|STRM_EV_HTTP_ANA|STRM_EV_HTTP_ERR, s, txn);
return 0;
}
@ -1119,14 +1121,7 @@ int http_request_forward_body(struct stream *s, struct channel *req, int an_bit)
short status = 0;
int ret;
DPRINTF(stderr,"[%u] %s: stream=%p b=%p, exp(r,w)=%u,%u bf=%08x bh=%lu analysers=%02x\n",
now_ms, __FUNCTION__,
s,
req,
req->rex, req->wex,
req->flags,
ci_data(req),
req->analysers);
DBG_TRACE_ENTER(STRM_EV_STRM_ANA|STRM_EV_HTTP_ANA, s, txn, msg);
htx = htxbuf(&req->buf);
@ -1145,11 +1140,16 @@ int http_request_forward_body(struct stream *s, struct channel *req, int an_bit)
* was a write error, we may recover.
*/
if (!(req->flags & (CF_READ_ERROR | CF_READ_TIMEOUT)) &&
(s->si[1].flags & SI_FL_L7_RETRY))
(s->si[1].flags & SI_FL_L7_RETRY)) {
DBG_TRACE_DEVEL("leaving on L7 retry",
STRM_EV_STRM_ANA|STRM_EV_HTTP_ANA|STRM_EV_HTTP_ERR, s, txn);
return 0;
}
msg->msg_state = HTTP_MSG_ERROR;
http_end_request(s);
http_end_response(s);
DBG_TRACE_DEVEL("leaving on error",
STRM_EV_STRM_ANA|STRM_EV_HTTP_ANA|STRM_EV_HTTP_ERR, s, txn);
return 1;
}
@ -1242,6 +1242,7 @@ int http_request_forward_body(struct stream *s, struct channel *req, int an_bit)
}
goto return_bad_req;
}
DBG_TRACE_LEAVE(STRM_EV_STRM_ANA|STRM_EV_HTTP_ANA, s, txn);
return 1;
}
@ -1262,6 +1263,8 @@ int http_request_forward_body(struct stream *s, struct channel *req, int an_bit)
* on some systems (eg: Linux). */
channel_auto_read(req);
}
DBG_TRACE_DEVEL("waiting for the end of the HTTP txn",
STRM_EV_STRM_ANA|STRM_EV_HTTP_ANA, s, txn);
return 0;
missing_data_or_waiting:
@ -1296,6 +1299,8 @@ int http_request_forward_body(struct stream *s, struct channel *req, int an_bit)
if (msg->flags & HTTP_MSGF_TE_CHNK)
req->flags |= CF_EXPECT_MORE;
DBG_TRACE_DEVEL("waiting for more data to forward",
STRM_EV_STRM_ANA|STRM_EV_HTTP_ANA, s, txn);
return 0;
return_cli_abort:
@ -1344,6 +1349,8 @@ int http_request_forward_body(struct stream *s, struct channel *req, int an_bit)
s->res.analysers &= AN_RES_FLT_END; /* we're in data phase, we want to abort both directions */
if (!(s->flags & SF_FINST_MASK))
s->flags |= ((txn->rsp.msg_state < HTTP_MSG_ERROR) ? SF_FINST_H : SF_FINST_D);
DBG_TRACE_DEVEL("leaving on error ",
STRM_EV_STRM_ANA|STRM_EV_HTTP_ANA|STRM_EV_HTTP_ERR, s, txn);
return 0;
}
@ -1415,14 +1422,7 @@ int http_wait_for_response(struct stream *s, struct channel *rep, int an_bit)
struct htx_sl *sl;
int n;
DPRINTF(stderr,"[%u] %s: stream=%p b=%p, exp(r,w)=%u,%u bf=%08x bh=%lu analysers=%02x\n",
now_ms, __FUNCTION__,
s,
rep,
rep->rex, rep->wex,
rep->flags,
ci_data(rep),
rep->analysers);
DBG_TRACE_ENTER(STRM_EV_STRM_ANA|STRM_EV_HTTP_ANA, s, txn, msg);
htx = htxbuf(&rep->buf);
@ -1483,8 +1483,11 @@ int http_wait_for_response(struct stream *s, struct channel *rep, int an_bit)
if (conn->err_code == CO_ER_SSL_EARLY_FAILED) {
if ((s->be->retry_type & PR_RE_EARLY_ERROR) &&
(si_b->flags & SI_FL_L7_RETRY) &&
do_l7_retry(s, si_b) == 0)
do_l7_retry(s, si_b) == 0) {
DBG_TRACE_DEVEL("leaving on L7 retry",
STRM_EV_STRM_ANA|STRM_EV_HTTP_ANA, s, txn);
return 0;
}
txn->status = 425;
}
@ -1495,6 +1498,8 @@ int http_wait_for_response(struct stream *s, struct channel *rep, int an_bit)
s->flags |= SF_ERR_SRVCL;
if (!(s->flags & SF_FINST_MASK))
s->flags |= SF_FINST_H;
DBG_TRACE_DEVEL("leaving on error",
STRM_EV_STRM_ANA|STRM_EV_HTTP_ANA|STRM_EV_HTTP_ERR, s, txn);
return 0;
}
@ -1502,8 +1507,11 @@ int http_wait_for_response(struct stream *s, struct channel *rep, int an_bit)
else if (rep->flags & CF_READ_TIMEOUT) {
if ((si_b->flags & SI_FL_L7_RETRY) &&
(s->be->retry_type & PR_RE_TIMEOUT)) {
if (co_data(rep) || do_l7_retry(s, si_b) == 0)
if (co_data(rep) || do_l7_retry(s, si_b) == 0) {
DBG_TRACE_DEVEL("leaving on L7 retry",
STRM_EV_STRM_ANA|STRM_EV_HTTP_ANA, s, txn);
return 0;
}
}
_HA_ATOMIC_ADD(&s->be->be_counters.failed_resp, 1);
if (objt_server(s->target)) {
@ -1520,6 +1528,8 @@ int http_wait_for_response(struct stream *s, struct channel *rep, int an_bit)
s->flags |= SF_ERR_SRVTO;
if (!(s->flags & SF_FINST_MASK))
s->flags |= SF_FINST_H;
DBG_TRACE_DEVEL("leaving on error",
STRM_EV_STRM_ANA|STRM_EV_HTTP_ANA|STRM_EV_HTTP_ERR, s, txn);
return 0;
}
@ -1540,6 +1550,8 @@ int http_wait_for_response(struct stream *s, struct channel *rep, int an_bit)
s->flags |= SF_FINST_H;
/* process_stream() will take care of the error */
DBG_TRACE_DEVEL("leaving on error",
STRM_EV_STRM_ANA|STRM_EV_HTTP_ANA|STRM_EV_HTTP_ERR, s, txn);
return 0;
}
@ -1547,8 +1559,11 @@ int http_wait_for_response(struct stream *s, struct channel *rep, int an_bit)
else if (rep->flags & CF_SHUTR) {
if ((si_b->flags & SI_FL_L7_RETRY) &&
(s->be->retry_type & PR_RE_DISCONNECTED)) {
if (co_data(rep) || do_l7_retry(s, si_b) == 0)
if (co_data(rep) || do_l7_retry(s, si_b) == 0) {
DBG_TRACE_DEVEL("leaving on L7 retry",
STRM_EV_STRM_ANA|STRM_EV_HTTP_ANA, s, txn);
return 0;
}
}
if (txn->flags & TX_NOT_FIRST)
@ -1569,6 +1584,8 @@ int http_wait_for_response(struct stream *s, struct channel *rep, int an_bit)
s->flags |= SF_ERR_SRVCL;
if (!(s->flags & SF_FINST_MASK))
s->flags |= SF_FINST_H;
DBG_TRACE_DEVEL("leaving on error",
STRM_EV_STRM_ANA|STRM_EV_HTTP_ANA|STRM_EV_HTTP_ERR, s, txn);
return 0;
}
@ -1586,11 +1603,15 @@ int http_wait_for_response(struct stream *s, struct channel *rep, int an_bit)
s->flags |= SF_FINST_H;
/* process_stream() will take care of the error */
DBG_TRACE_DEVEL("leaving on error",
STRM_EV_STRM_ANA|STRM_EV_HTTP_ANA|STRM_EV_HTTP_ERR, s, txn);
return 0;
}
channel_dont_close(rep);
rep->flags |= CF_READ_DONTWAIT; /* try to get back here ASAP */
DBG_TRACE_DEVEL("waiting for more data",
STRM_EV_STRM_ANA|STRM_EV_HTTP_ANA, s, txn);
return 0;
}
@ -1766,6 +1787,7 @@ int http_wait_for_response(struct stream *s, struct channel *rep, int an_bit)
rep->analysers &= ~an_bit;
rep->analyse_exp = TICK_ETERNITY;
channel_auto_close(rep);
DBG_TRACE_LEAVE(STRM_EV_STRM_ANA|STRM_EV_HTTP_ANA, s, txn);
return 1;
return_int_err:
@ -1783,8 +1805,11 @@ int http_wait_for_response(struct stream *s, struct channel *rep, int an_bit)
}
if ((s->be->retry_type & PR_RE_JUNK_REQUEST) &&
(si_b->flags & SI_FL_L7_RETRY) &&
do_l7_retry(s, si_b) == 0)
do_l7_retry(s, si_b) == 0) {
DBG_TRACE_DEVEL("leaving on L7 retry",
STRM_EV_STRM_ANA|STRM_EV_HTTP_ANA, s, txn);
return 0;
}
txn->status = 502;
/* fall through */
@ -1799,6 +1824,8 @@ int http_wait_for_response(struct stream *s, struct channel *rep, int an_bit)
s->si[1].flags |= SI_FL_NOLINGER;
rep->analysers &= AN_RES_FLT_END;
rep->analyse_exp = TICK_ETERNITY;
DBG_TRACE_DEVEL("leaving on error",
STRM_EV_STRM_ANA|STRM_EV_HTTP_ANA|STRM_EV_HTTP_ERR, s, txn);
return 0;
abort_keep_alive:
@ -1813,6 +1840,8 @@ int http_wait_for_response(struct stream *s, struct channel *rep, int an_bit)
s->logs.level = 0;
s->res.flags &= ~CF_EXPECT_MORE; /* speed up sending a previous response */
http_reply_and_close(s, txn->status, NULL);
DBG_TRACE_DEVEL("leaving by closing K/A connection",
STRM_EV_STRM_ANA|STRM_EV_HTTP_ANA, s, txn);
return 0;
}
@ -1833,14 +1862,7 @@ int http_process_res_common(struct stream *s, struct channel *rep, int an_bit, s
if (unlikely(msg->msg_state < HTTP_MSG_BODY)) /* we need more data */
return 0;
DPRINTF(stderr,"[%u] %s: stream=%p b=%p, exp(r,w)=%u,%u bf=%08x bh=%lu analysers=%02x\n",
now_ms, __FUNCTION__,
s,
rep,
rep->rex, rep->wex,
rep->flags,
ci_data(rep),
rep->analysers);
DBG_TRACE_ENTER(STRM_EV_STRM_ANA|STRM_EV_HTTP_ANA, s, txn, msg);
htx = htxbuf(&rep->buf);
@ -1893,6 +1915,8 @@ int http_process_res_common(struct stream *s, struct channel *rep, int an_bit, s
/* we need to be called again. */
if (ret == HTTP_RULE_RES_YIELD) {
channel_dont_close(rep);
DBG_TRACE_DEVEL("waiting for more data",
STRM_EV_STRM_ANA|STRM_EV_HTTP_ANA, s, txn);
return 0;
}
@ -2063,6 +2087,7 @@ int http_process_res_common(struct stream *s, struct channel *rep, int an_bit, s
s->do_log(s);
s->logs.bytes_out = 0;
}
DBG_TRACE_LEAVE(STRM_EV_STRM_ANA|STRM_EV_HTTP_ANA, s, txn);
return 1;
return_bad_resp:
@ -2082,6 +2107,8 @@ int http_process_res_common(struct stream *s, struct channel *rep, int an_bit, s
s->flags |= SF_ERR_PRXCOND;
if (!(s->flags & SF_FINST_MASK))
s->flags |= SF_FINST_H;
DBG_TRACE_DEVEL("leaving on error",
STRM_EV_STRM_ANA|STRM_EV_HTTP_ANA|STRM_EV_HTTP_ERR, s, txn);
return 0;
}
@ -2122,14 +2149,7 @@ int http_response_forward_body(struct stream *s, struct channel *res, int an_bit
struct htx *htx;
int ret;
DPRINTF(stderr,"[%u] %s: stream=%p b=%p, exp(r,w)=%u,%u bf=%08x bh=%lu analysers=%02x\n",
now_ms, __FUNCTION__,
s,
res,
res->rex, res->wex,
res->flags,
ci_data(res),
res->analysers);
DBG_TRACE_ENTER(STRM_EV_STRM_ANA|STRM_EV_HTTP_ANA, s, txn, msg);
htx = htxbuf(&res->buf);
@ -2146,6 +2166,8 @@ int http_response_forward_body(struct stream *s, struct channel *res, int an_bit
msg->msg_state = HTTP_MSG_ERROR;
http_end_response(s);
http_end_request(s);
DBG_TRACE_DEVEL("leaving on error",
STRM_EV_STRM_ANA|STRM_EV_HTTP_ANA|STRM_EV_HTTP_ERR, s, txn);
return 1;
}
@ -2232,8 +2254,11 @@ int http_response_forward_body(struct stream *s, struct channel *res, int an_bit
}
goto return_bad_res;
}
DBG_TRACE_LEAVE(STRM_EV_STRM_ANA|STRM_EV_HTTP_ANA, s, txn);
return 1;
}
DBG_TRACE_DEVEL("waiting for the end of the HTTP txn",
STRM_EV_STRM_ANA|STRM_EV_HTTP_ANA, s, txn);
return 0;
missing_data_or_waiting:
@ -2274,6 +2299,8 @@ int http_response_forward_body(struct stream *s, struct channel *res, int an_bit
res->flags |= CF_EXPECT_MORE;
/* the stream handler will take care of timeouts and errors */
DBG_TRACE_DEVEL("waiting for more data to forward",
STRM_EV_STRM_ANA|STRM_EV_HTTP_ANA, s, txn);
return 0;
return_srv_abort:
@ -2316,6 +2343,8 @@ int http_response_forward_body(struct stream *s, struct channel *res, int an_bit
s->req.analysers &= AN_REQ_FLT_END; /* we're in data phase, we want to abort both directions */
if (!(s->flags & SF_FINST_MASK))
s->flags |= SF_FINST_D;
DBG_TRACE_DEVEL("leaving on error",
STRM_EV_STRM_ANA|STRM_EV_HTTP_ANA|STRM_EV_HTTP_ERR, s, txn);
return 0;
}
@ -4630,10 +4659,7 @@ static void http_end_request(struct stream *s)
struct channel *chn = &s->req;
struct http_txn *txn = s->txn;
DPRINTF(stderr,"[%u] %s: stream=%p states=%s,%s req->analysers=0x%08x res->analysers=0x%08x\n",
now_ms, __FUNCTION__, s,
h1_msg_state_str(txn->req.msg_state), h1_msg_state_str(txn->rsp.msg_state),
s->req.analysers, s->res.analysers);
DBG_TRACE_ENTER(STRM_EV_HTTP_ANA, s, txn);
if (unlikely(txn->req.msg_state == HTTP_MSG_ERROR ||
txn->rsp.msg_state == HTTP_MSG_ERROR)) {
@ -4642,8 +4668,10 @@ static void http_end_request(struct stream *s)
goto end;
}
if (unlikely(txn->req.msg_state < HTTP_MSG_DONE))
if (unlikely(txn->req.msg_state < HTTP_MSG_DONE)) {
DBG_TRACE_DEVEL("waiting end of the request", STRM_EV_HTTP_ANA, s, txn);
return;
}
if (txn->req.msg_state == HTTP_MSG_DONE) {
/* No need to read anymore, the request was completely parsed.
@ -4673,6 +4701,7 @@ static void http_end_request(struct stream *s)
/* The server has not finished to respond, so we
* don't want to move in order not to upset it.
*/
DBG_TRACE_DEVEL("waiting end of the response", STRM_EV_HTTP_ANA, s, txn);
return;
}
@ -4686,8 +4715,10 @@ static void http_end_request(struct stream *s)
* poll for reads.
*/
channel_auto_read(chn);
if (b_data(&chn->buf))
if (b_data(&chn->buf)) {
DBG_TRACE_DEVEL("waiting to flush the request", STRM_EV_HTTP_ANA, s, txn);
return;
}
txn->req.msg_state = HTTP_MSG_TUNNEL;
}
else {
@ -4724,6 +4755,7 @@ static void http_end_request(struct stream *s)
txn->req.msg_state = HTTP_MSG_ERROR;
goto end;
}
DBG_TRACE_LEAVE(STRM_EV_HTTP_ANA, s, txn);
return;
}
@ -4752,6 +4784,7 @@ static void http_end_request(struct stream *s)
chn->analysers |= AN_REQ_FLT_XFER_DATA;
channel_auto_close(chn);
channel_auto_read(chn);
DBG_TRACE_LEAVE(STRM_EV_HTTP_ANA, s, txn);
}
@ -4763,10 +4796,7 @@ static void http_end_response(struct stream *s)
struct channel *chn = &s->res;
struct http_txn *txn = s->txn;
DPRINTF(stderr,"[%u] %s: stream=%p states=%s,%s req->analysers=0x%08x res->analysers=0x%08x\n",
now_ms, __FUNCTION__, s,
h1_msg_state_str(txn->req.msg_state), h1_msg_state_str(txn->rsp.msg_state),
s->req.analysers, s->res.analysers);
DBG_TRACE_ENTER(STRM_EV_HTTP_ANA, s, txn);
if (unlikely(txn->req.msg_state == HTTP_MSG_ERROR ||
txn->rsp.msg_state == HTTP_MSG_ERROR)) {
@ -4775,8 +4805,10 @@ static void http_end_response(struct stream *s)
goto end;
}
if (unlikely(txn->rsp.msg_state < HTTP_MSG_DONE))
if (unlikely(txn->rsp.msg_state < HTTP_MSG_DONE)) {
DBG_TRACE_DEVEL("waiting end of the response", STRM_EV_HTTP_ANA, s, txn);
return;
}
if (txn->rsp.msg_state == HTTP_MSG_DONE) {
/* In theory, we don't need to read anymore, but we must
@ -4792,6 +4824,7 @@ static void http_end_response(struct stream *s)
* We have the choice of either breaking the connection
* or letting it pass through. Let's do the later.
*/
DBG_TRACE_DEVEL("waiting end of the request", STRM_EV_HTTP_ANA, s, txn);
return;
}
@ -4803,8 +4836,10 @@ static void http_end_response(struct stream *s)
if (txn->flags & TX_CON_WANT_TUN) {
channel_auto_read(chn);
chn->flags |= CF_NEVER_WAIT;
if (b_data(&chn->buf))
if (b_data(&chn->buf)) {
DBG_TRACE_DEVEL("waiting to flush the respone", STRM_EV_HTTP_ANA, s, txn);
return;
}
txn->rsp.msg_state = HTTP_MSG_TUNNEL;
}
else {
@ -4835,6 +4870,7 @@ static void http_end_response(struct stream *s)
_HA_ATOMIC_ADD(&objt_server(s->target)->counters.cli_aborts, 1);
goto end;
}
DBG_TRACE_LEAVE(STRM_EV_HTTP_ANA, s, txn);
return;
}
@ -4860,6 +4896,7 @@ static void http_end_response(struct stream *s)
chn->analysers |= AN_RES_FLT_XFER_DATA;
channel_auto_close(chn);
channel_auto_read(chn);
DBG_TRACE_LEAVE(STRM_EV_HTTP_ANA, s, txn);
}
void http_server_error(struct stream *s, struct stream_interface *si, int err,

372
src/stream.c
View File

@ -74,6 +74,199 @@ __decl_spinlock(streams_lock);
static struct list service_keywords = LIST_HEAD_INIT(service_keywords);
/* trace source and events */
static void strm_trace(enum trace_level level, uint64_t mask,
const struct trace_source *src,
const struct ist where, const struct ist func,
const void *a1, const void *a2, const void *a3, const void *a4);
/* The event representation is split like this :
* strm - stream
* si - stream interface
* http - http analyzis
* tcp - tcp analyzis
*
* STRM_EV_* macros are defined in <proto/stream.h>
*/
static const struct trace_event strm_trace_events[] = {
{ .mask = STRM_EV_STRM_NEW, .name = "strm_new", .desc = "new stream" },
{ .mask = STRM_EV_STRM_FREE, .name = "strm_free", .desc = "release stream" },
{ .mask = STRM_EV_STRM_ERR, .name = "strm_err", .desc = "error during stream processing" },
{ .mask = STRM_EV_STRM_ANA, .name = "strm_ana", .desc = "stream analyzers" },
{ .mask = STRM_EV_STRM_PROC, .name = "strm_proc", .desc = "stream processing" },
{ .mask = STRM_EV_SI_ST, .name = "si_state", .desc = "processing stream-interface states" },
{ .mask = STRM_EV_HTTP_ANA, .name = "http_ana", .desc = "HTTP analyzers" },
{ .mask = STRM_EV_HTTP_ERR, .name = "http_err", .desc = "error during HTTP analyzis" },
{ .mask = STRM_EV_TCP_ANA, .name = "tcp_ana", .desc = "TCP analyzers" },
{ .mask = STRM_EV_TCP_ERR, .name = "tcp_err", .desc = "error during TCP analyzis" },
{}
};
static const struct name_desc strm_trace_lockon_args[4] = {
/* arg1 */ { /* already used by the stream */ },
/* arg2 */ { },
/* arg3 */ { },
/* arg4 */ { }
};
static const struct name_desc strm_trace_decoding[] = {
#define STRM_VERB_CLEAN 1
{ .name="clean", .desc="only user-friendly stuff, generally suitable for level \"user\"" },
#define STRM_VERB_MINIMAL 2
{ .name="minimal", .desc="report info on stream and stream-interfaces" },
#define STRM_VERB_SIMPLE 3
{ .name="simple", .desc="add info on request and response channels" },
#define STRM_VERB_ADVANCED 4
{ .name="advanced", .desc="add info on channel's buffer for data and developer levels only" },
#define STRM_VERB_COMPLETE 5
{ .name="complete", .desc="add info on channel's buffer" },
{ /* end */ }
};
struct trace_source trace_strm = {
.name = IST("stream"),
.desc = "Applicative stream",
.arg_def = TRC_ARG1_STRM, // TRACE()'s first argument is always a stream
.default_cb = strm_trace,
.known_events = strm_trace_events,
.lockon_args = strm_trace_lockon_args,
.decoding = strm_trace_decoding,
.report_events = ~0, // report everything by default
};
#define TRACE_SOURCE &trace_strm
INITCALL1(STG_REGISTER, trace_register_source, TRACE_SOURCE);
/* the stream traces always expect that arg1, if non-null, is of a stream (from
* which we can derive everything), that arg2, if non-null, is an http
* transaction, that arg3, if non-null, is an http message.
*/
static void strm_trace(enum trace_level level, uint64_t mask, const struct trace_source *src,
const struct ist where, const struct ist func,
const void *a1, const void *a2, const void *a3, const void *a4)
{
const struct stream *s = a1;
const struct http_txn *txn = a2;
const struct http_msg *msg = a3;
struct task *task;
const struct stream_interface *si_f, *si_b;
const struct channel *req, *res;
struct htx *htx;
if (!s || src->verbosity < STRM_VERB_CLEAN)
return;
task = s->task;
si_f = &s->si[0];
si_b = &s->si[1];
req = &s->req;
res = &s->res;
htx = (msg ? htxbuf(&msg->chn->buf) : NULL);
/* General info about the stream (htx/tcp, id...) */
chunk_appendf(&trace_buf, " : [%u,%s]",
s->uniq_id, ((s->flags & SF_HTX) ? "HTX" : "TCP"));
if (s->unique_id)
chunk_appendf(&trace_buf, " id=%s", s->unique_id);
/* Front and back stream-int state */
chunk_appendf(&trace_buf, " SI=(%s,%s)",
si_state_str(si_f->state), si_state_str(si_b->state));
/* If txn is defined, HTTP req/rep states */
if (txn)
chunk_appendf(&trace_buf, " HTTP=(%s,%s)",
h1_msg_state_str(txn->req.msg_state), h1_msg_state_str(txn->rsp.msg_state));
if (msg)
chunk_appendf(&trace_buf, " %s", ((msg->chn->flags & CF_ISRESP) ? "RESPONSE" : "REQUEST"));
if (src->verbosity == STRM_VERB_CLEAN)
return;
/* If msg defined, display status-line if possible (verbosity > MINIMAL) */
if (src->verbosity > STRM_VERB_MINIMAL && htx && htx_nbblks(htx)) {
const struct htx_blk *blk = htx_get_head_blk(htx);
const struct htx_sl *sl = htx_get_blk_ptr(htx, blk);
enum htx_blk_type type = htx_get_blk_type(blk);
if (type == HTX_BLK_REQ_SL || type == HTX_BLK_RES_SL)
chunk_appendf(&trace_buf, " - \"%.*s %.*s %.*s\"",
HTX_SL_P1_LEN(sl), HTX_SL_P1_PTR(sl),
HTX_SL_P2_LEN(sl), HTX_SL_P2_PTR(sl),
HTX_SL_P3_LEN(sl), HTX_SL_P3_PTR(sl));
}
/* If txn defined info about HTTP msgs, otherwise info about SI. */
if (txn) {
chunk_appendf(&trace_buf, " - t=%p s=(%p,0x%08x) txn.flags=0x%08x, http.flags=(0x%08x,0x%08x) status=%d",
task, s, s->flags, txn->flags, txn->req.flags, txn->rsp.flags, txn->status);
}
else {
chunk_appendf(&trace_buf, " - t=%p s=(%p,0x%08x) si_f=(%p,0x%08x,0x%x) si_b=(%p,0x%08x,0x%x) retries=%d",
task, s, s->flags, si_f, si_f->flags, si_f->err_type,
si_b, si_b->flags, si_b->err_type, si_b->conn_retries);
}
if (src->verbosity == STRM_VERB_MINIMAL)
return;
/* If txn defined, don't display all channel info */
if (src->verbosity == STRM_VERB_SIMPLE || txn) {
chunk_appendf(&trace_buf, " req=(%p .fl=0x%08x .exp(r,w,a)=(%u,%u,%u))",
req, req->flags, req->rex, req->wex, req->analyse_exp);
chunk_appendf(&trace_buf, " res=(%p .fl=0x%08x .exp(r,w,a)=(%u,%u,%u))",
res, res->flags, res->rex, res->wex, res->analyse_exp);
}
else {
chunk_appendf(&trace_buf, " req=(%p .fl=0x%08x .ana=0x%08x .exp(r,w,a)=(%u,%u,%u) .o=%lu .tot=%llu .to_fwd=%u)",
req, req->flags, req->analysers, req->rex, req->wex, req->analyse_exp,
req->output, req->total, req->to_forward);
chunk_appendf(&trace_buf, " res=(%p .fl=0x%08x .ana=0x%08x .exp(r,w,a)=(%u,%u,%u) .o=%lu .tot=%llu .to_fwd=%u)",
res, res->flags, res->analysers, res->rex, res->wex, res->analyse_exp,
res->output, res->total, res->to_forward);
}
if (src->verbosity == STRM_VERB_SIMPLE ||
(src->verbosity == STRM_VERB_ADVANCED && src->level < TRACE_LEVEL_DATA))
return;
/* channels' buffer info */
if (s->flags & SF_HTX) {
struct htx *rqhtx = htxbuf(&req->buf);
struct htx *rphtx = htxbuf(&res->buf);
chunk_appendf(&trace_buf, " htx=(%u/%u#%u, %u/%u#%u)",
rqhtx->data, rqhtx->size, htx_nbblks(rqhtx),
rphtx->data, rphtx->size, htx_nbblks(rphtx));
}
else {
chunk_appendf(&trace_buf, " buf=(%u@%p+%u/%u, %u@%p+%u/%u)",
(unsigned int)b_data(&req->buf), b_orig(&req->buf),
(unsigned int)b_head_ofs(&req->buf), (unsigned int)b_size(&req->buf),
(unsigned int)b_data(&req->buf), b_orig(&req->buf),
(unsigned int)b_head_ofs(&req->buf), (unsigned int)b_size(&req->buf));
}
/* If msg defined, display htx info if defined (level > USER) */
if (src->level > TRACE_LEVEL_USER && htx && htx_nbblks(htx)) {
int full = 0;
/* Full htx info (level > STATE && verbosity > SIMPLE) */
if (src->level > TRACE_LEVEL_STATE) {
if (src->verbosity == STRM_VERB_COMPLETE)
full = 1;
}
chunk_memcat(&trace_buf, "\n\t", 2);
htx_dump(&trace_buf, htx, full);
}
}
/* Create a new stream for connection <conn>. Return < 0 on error. This is only
* valid right after the handshake, before the connection's data layer is
* initialized, because it relies on the session to be in conn->owner.
@ -132,6 +325,7 @@ struct stream *stream_new(struct session *sess, enum obj_type *origin)
struct appctx *appctx = objt_appctx(origin);
const struct cs_info *csinfo;
DBG_TRACE_ENTER(STRM_EV_STRM_NEW);
if (unlikely((s = pool_alloc(pool_head_stream)) == NULL))
goto out_fail_alloc;
@ -338,6 +532,7 @@ struct stream *stream_new(struct session *sess, enum obj_type *origin)
* stream is fully initialized before calling task_wakeup. So
* the caller must handle the task_wakeup
*/
DBG_TRACE_LEAVE(STRM_EV_STRM_NEW, s);
return s;
/* Error unrolling */
@ -350,6 +545,7 @@ out_fail_alloc_si1:
tasklet_free(s->si[0].wait_event.tasklet);
out_fail_alloc:
pool_free(pool_head_stream, s);
DBG_TRACE_DEVEL("leaving on error", STRM_EV_STRM_NEW|STRM_EV_STRM_ERR);
return NULL;
}
@ -365,6 +561,8 @@ static void stream_free(struct stream *s)
int must_free_sess;
int i;
DBG_TRACE_POINT(STRM_EV_STRM_FREE, s);
/* detach the stream from its own task before even releasing it so
* that walking over a task list never exhibits a dying stream.
*/
@ -663,6 +861,8 @@ static void sess_update_st_con_tcp(struct stream *s)
struct conn_stream *srv_cs = objt_cs(si->end);
struct connection *conn = srv_cs ? srv_cs->conn : objt_conn(si->end);
DBG_TRACE_ENTER(STRM_EV_STRM_PROC|STRM_EV_SI_ST, s);
/* the client might want to abort */
if ((rep->flags & CF_SHUTW) ||
((req->flags & CF_SHUTW_NOW) &&
@ -673,11 +873,11 @@ static void sess_update_st_con_tcp(struct stream *s)
if (s->srv_error)
s->srv_error(s, si);
/* Note: state = SI_ST_DIS now */
return;
DBG_TRACE_STATE("client abort during connection attempt", STRM_EV_STRM_PROC|STRM_EV_SI_ST|STRM_EV_STRM_ERR, s);
}
/* retryable error ? */
if (si->flags & (SI_FL_EXP|SI_FL_ERR)) {
else if (si->flags & (SI_FL_EXP|SI_FL_ERR)) {
if (!(s->flags & SF_SRV_REUSED) && conn) {
conn_stop_tracking(conn);
conn_full_close(conn);
@ -691,8 +891,10 @@ static void sess_update_st_con_tcp(struct stream *s)
}
si->state = SI_ST_CER;
return;
DBG_TRACE_STATE("connection failed, retry", STRM_EV_STRM_PROC|STRM_EV_SI_ST|STRM_EV_STRM_ERR, s);
}
DBG_TRACE_LEAVE(STRM_EV_STRM_PROC|STRM_EV_SI_ST, s);
}
/* This function is called with (si->state == SI_ST_CER) meaning that a
@ -710,6 +912,8 @@ static void sess_update_st_cer(struct stream *s)
struct conn_stream *cs = objt_cs(si->end);
struct connection *conn = cs_conn(cs);
DBG_TRACE_ENTER(STRM_EV_STRM_PROC|STRM_EV_SI_ST, s);
si->exp = TICK_ETERNITY;
si->flags &= ~SI_FL_EXP;
@ -741,6 +945,7 @@ static void sess_update_st_cer(struct stream *s)
* client provoke retries.
*/
si->conn_retries = 0;
DBG_TRACE_DEVEL("Bad SSL cert, disable connection retries", STRM_EV_STRM_PROC|STRM_EV_SI_ST|STRM_EV_STRM_ERR, s);
}
}
@ -766,7 +971,9 @@ static void sess_update_st_cer(struct stream *s)
si->state = SI_ST_CLO;
if (s->srv_error)
s->srv_error(s, si);
return;
DBG_TRACE_STATE("connection failed", STRM_EV_STRM_PROC|STRM_EV_SI_ST|STRM_EV_STRM_ERR, s);
goto end;
}
stream_choose_redispatch(s);
@ -796,7 +1003,11 @@ static void sess_update_st_cer(struct stream *s)
si->exp = tick_add(now_ms, MS_TO_TICKS(delay));
}
si->flags &= ~SI_FL_ERR;
DBG_TRACE_STATE("retry a new connection", STRM_EV_STRM_PROC|STRM_EV_SI_ST, s);
}
end:
DBG_TRACE_LEAVE(STRM_EV_STRM_PROC|STRM_EV_SI_ST, s);
}
/* This function is called with (si->state == SI_ST_RDY) meaning that a
@ -814,6 +1025,7 @@ static void sess_update_st_rdy_tcp(struct stream *s)
struct conn_stream *srv_cs = objt_cs(si->end);
struct connection *conn = srv_cs ? srv_cs->conn : objt_conn(si->end);
DBG_TRACE_ENTER(STRM_EV_STRM_PROC|STRM_EV_SI_ST, s);
/* We know the connection at least succeeded, though it could have
* since met an error for any other reason. At least it didn't time out
* eventhough the timeout might have been reported right after success.
@ -843,7 +1055,8 @@ static void sess_update_st_rdy_tcp(struct stream *s)
si->err_type |= SI_ET_CONN_ABRT;
if (s->srv_error)
s->srv_error(s, si);
return;
DBG_TRACE_STATE("client abort during connection attempt", STRM_EV_STRM_PROC|STRM_EV_SI_ST|STRM_EV_STRM_ERR, s);
goto end;
}
/* retryable error ? */
@ -856,15 +1069,20 @@ static void sess_update_st_rdy_tcp(struct stream *s)
if (!si->err_type)
si->err_type = SI_ET_CONN_ERR;
si->state = SI_ST_CER;
return;
DBG_TRACE_STATE("connection failed, retry", STRM_EV_STRM_PROC|STRM_EV_SI_ST|STRM_EV_STRM_ERR, s);
goto end;
}
}
/* data were sent and/or we had no error, sess_establish() will
* now take over.
*/
DBG_TRACE_STATE("connection established", STRM_EV_STRM_PROC|STRM_EV_SI_ST, s);
si->err_type = SI_ET_NONE;
si->state = SI_ST_EST;
end:
DBG_TRACE_LEAVE(STRM_EV_STRM_PROC|STRM_EV_SI_ST, s);
}
/*
@ -887,6 +1105,7 @@ static void sess_establish(struct stream *s)
struct channel *req = &s->req;
struct channel *rep = &s->res;
DBG_TRACE_ENTER(STRM_EV_STRM_PROC|STRM_EV_SI_ST, s);
/* First, centralize the timers information, and clear any irrelevant
* timeout.
*/
@ -905,6 +1124,7 @@ static void sess_establish(struct stream *s)
req->flags |= CF_WRITE_ERROR;
rep->flags |= CF_READ_ERROR;
si->err_type = SI_ET_DATA_ERR;
DBG_TRACE_STATE("read/write error", STRM_EV_STRM_PROC|STRM_EV_SI_ST|STRM_EV_STRM_ERR, s);
}
if (objt_server(s->target))
@ -946,8 +1166,12 @@ static void sess_establish(struct stream *s)
req->wex = TICK_ETERNITY;
/* If we managed to get the whole response, and we don't have anything
* left to send, or can't, switch to SI_ST_DIS now. */
if (rep->flags & (CF_SHUTR | CF_SHUTW))
if (rep->flags & (CF_SHUTR | CF_SHUTW)) {
si->state = SI_ST_DIS;
DBG_TRACE_STATE("response channel shutdwn for read/write", STRM_EV_STRM_PROC|STRM_EV_SI_ST|STRM_EV_STRM_ERR, s);
}
DBG_TRACE_LEAVE(STRM_EV_STRM_PROC|STRM_EV_SI_ST, s);
}
/* Check if the connection request is in such a state that it can be aborted. */
@ -970,13 +1194,7 @@ static void sess_update_stream_int(struct stream *s)
struct stream_interface *si = &s->si[1];
struct channel *req = &s->req;
DPRINTF(stderr,"[%u] %s: sess=%p rq=%p, rp=%p, exp(r,w)=%u,%u rqf=%08x rpf=%08x rqh=%lu rqt=%lu rph=%lu rpt=%lu cs=%d ss=%d\n",
now_ms, __FUNCTION__,
s,
req, &s->res,
req->rex, s->res.wex,
req->flags, s->res.flags,
ci_data(req), co_data(req), ci_data(&s->res), co_data(&s->res), s->si[0].state, s->si[1].state);
DBG_TRACE_ENTER(STRM_EV_STRM_PROC|STRM_EV_SI_ST, s);
if (si->state == SI_ST_ASS) {
/* Server assigned to connection request, we have to try to connect now */
@ -987,6 +1205,7 @@ static void sess_update_stream_int(struct stream *s)
*/
if (check_req_may_abort(req, s)) {
si->err_type |= SI_ET_CONN_ABRT;
DBG_TRACE_STATE("connection aborted", STRM_EV_STRM_PROC|STRM_EV_SI_ST|STRM_EV_STRM_ERR, s);
goto abort_connection;
}
@ -999,7 +1218,8 @@ static void sess_update_stream_int(struct stream *s)
srv_inc_sess_ctr(srv);
if (srv)
srv_set_sess_last(srv);
return;
DBG_TRACE_STATE("connection attempt", STRM_EV_STRM_PROC|STRM_EV_SI_ST, s);
goto end;
}
/* We have received a synchronous error. We might have to
@ -1037,7 +1257,8 @@ static void sess_update_stream_int(struct stream *s)
si->state = SI_ST_CLO;
if (s->srv_error)
s->srv_error(s, si);
return;
DBG_TRACE_STATE("internal error during connection", STRM_EV_STRM_PROC|STRM_EV_SI_ST|STRM_EV_STRM_ERR, s);
goto end;
}
/* We are facing a retryable error, but we don't want to run a
@ -1047,8 +1268,9 @@ static void sess_update_stream_int(struct stream *s)
si->state = SI_ST_CER;
si->flags &= ~SI_FL_ERR;
sess_update_st_cer(s);
DBG_TRACE_STATE("connection error, retry", STRM_EV_STRM_PROC|STRM_EV_SI_ST|STRM_EV_STRM_ERR, s);
/* now si->state is one of SI_ST_CLO, SI_ST_TAR, SI_ST_ASS, SI_ST_REQ */
return;
}
else if (si->state == SI_ST_QUE) {
/* connection request was queued, check for any update */
@ -1065,7 +1287,8 @@ static void sess_update_stream_int(struct stream *s)
s->logs.t_queue = tv_ms_elapsed(&s->logs.tv_accept, &now);
si->state = SI_ST_ASS;
}
return;
DBG_TRACE_STATE("dequeue connection request", STRM_EV_STRM_PROC|STRM_EV_SI_ST, s);
goto end;
}
/* Connection request still in queue... */
@ -1089,7 +1312,8 @@ static void sess_update_stream_int(struct stream *s)
si->state = SI_ST_CLO;
if (s->srv_error)
s->srv_error(s, si);
return;
DBG_TRACE_STATE("connection request still queued", STRM_EV_STRM_PROC|STRM_EV_SI_ST, s);
goto end;
}
/* Connection remains in queue, check if we have to abort it */
@ -1100,16 +1324,17 @@ static void sess_update_stream_int(struct stream *s)
pendconn_cond_unlink(s->pend_pos);
si->err_type |= SI_ET_QUEUE_ABRT;
DBG_TRACE_STATE("abort queued connection request", STRM_EV_STRM_PROC|STRM_EV_SI_ST|STRM_EV_STRM_ERR, s);
goto abort_connection;
}
/* Nothing changed */
return;
}
else if (si->state == SI_ST_TAR) {
/* Connection request might be aborted */
if (check_req_may_abort(req, s)) {
si->err_type |= SI_ET_CONN_ABRT;
DBG_TRACE_STATE("connection aborted", STRM_EV_STRM_PROC|STRM_EV_SI_ST|STRM_EV_STRM_ERR, s);
goto abort_connection;
}
@ -1127,8 +1352,12 @@ static void sess_update_stream_int(struct stream *s)
si->state = SI_ST_ASS;
else
si->state = SI_ST_REQ;
return;
DBG_TRACE_STATE("retry connection now", STRM_EV_STRM_PROC|STRM_EV_SI_ST, s);
}
end:
DBG_TRACE_LEAVE(STRM_EV_STRM_PROC|STRM_EV_SI_ST, s);
return;
abort_connection:
@ -1140,6 +1369,7 @@ abort_connection:
si->state = SI_ST_CLO;
if (s->srv_error)
s->srv_error(s, si);
DBG_TRACE_DEVEL("leaving on error", STRM_EV_STRM_PROC|STRM_EV_SI_ST|STRM_EV_STRM_ERR, s);
return;
}
@ -1179,17 +1409,11 @@ static void sess_prepare_conn_req(struct stream *s)
{
struct stream_interface *si = &s->si[1];
DPRINTF(stderr,"[%u] %s: sess=%p rq=%p, rp=%p, exp(r,w)=%u,%u rqf=%08x rpf=%08x rqh=%lu rqt=%lu rph=%lu rpt=%lu cs=%d ss=%d\n",
now_ms, __FUNCTION__,
s,
&s->req, &s->res,
s->req.rex, s->res.wex,
s->req.flags, s->res.flags,
ci_data(&s->req), co_data(&s->req), ci_data(&s->res), co_data(&s->res), s->si[0].state, s->si[1].state);
if (si->state != SI_ST_REQ)
return;
DBG_TRACE_ENTER(STRM_EV_STRM_PROC|STRM_EV_SI_ST, s);
if (unlikely(obj_type(s->target) == OBJ_TYPE_APPLET)) {
/* the applet directly goes to the EST state */
struct appctx *appctx = objt_appctx(si->end);
@ -1211,7 +1435,8 @@ static void sess_prepare_conn_req(struct stream *s)
si->state = SI_ST_CLO;
if (s->srv_error)
s->srv_error(s, si);
return;
DBG_TRACE_STATE("failed to register applet", STRM_EV_STRM_PROC|STRM_EV_SI_ST|STRM_EV_STRM_ERR, s);
goto end;
}
if (tv_iszero(&s->logs.tv_request))
@ -1220,8 +1445,10 @@ static void sess_prepare_conn_req(struct stream *s)
si->state = SI_ST_EST;
si->err_type = SI_ET_NONE;
be_set_sess_last(s->be);
DBG_TRACE_STATE("applet registered", STRM_EV_STRM_PROC|STRM_EV_SI_ST, s);
/* let sess_establish() finish the job */
return;
goto end;
}
/* Try to assign a server */
@ -1229,8 +1456,10 @@ static void sess_prepare_conn_req(struct stream *s)
/* We did not get a server. Either we queued the
* connection request, or we encountered an error.
*/
if (si->state == SI_ST_QUE)
return;
if (si->state == SI_ST_QUE) {
DBG_TRACE_STATE("connection request queued", STRM_EV_STRM_PROC|STRM_EV_SI_ST, s);
goto end;
}
/* we did not get any server, let's check the cause */
si_shutr(si);
@ -1241,13 +1470,18 @@ static void sess_prepare_conn_req(struct stream *s)
si->state = SI_ST_CLO;
if (s->srv_error)
s->srv_error(s, si);
return;
DBG_TRACE_STATE("connection request failed", STRM_EV_STRM_PROC|STRM_EV_SI_ST|STRM_EV_STRM_ERR, s);
goto end;
}
/* The server is assigned */
s->logs.t_queue = tv_ms_elapsed(&s->logs.tv_accept, &now);
si->state = SI_ST_ASS;
be_set_sess_last(s->be);
DBG_TRACE_STATE("connection request assigned to a server", STRM_EV_STRM_PROC|STRM_EV_SI_ST, s);
end:
DBG_TRACE_LEAVE(STRM_EV_STRM_PROC|STRM_EV_SI_ST, s);
}
/* This function parses the use-service action ruleset. It executes
@ -1320,14 +1554,7 @@ static int process_switching_rules(struct stream *s, struct channel *req, int an
req->analysers &= ~an_bit;
req->analyse_exp = TICK_ETERNITY;
DPRINTF(stderr,"[%u] %s: stream=%p b=%p, exp(r,w)=%u,%u bf=%08x bh=%lu analysers=%02x\n",
now_ms, __FUNCTION__,
s,
req,
req->rex, req->wex,
req->flags,
ci_data(req),
req->analysers);
DBG_TRACE_ENTER(STRM_EV_STRM_ANA, s);
/* now check whether we have some switching rules for this request */
if (!(s->flags & SF_BE_ASSIGNED)) {
@ -1416,6 +1643,7 @@ static int process_switching_rules(struct stream *s, struct channel *req, int an
}
}
DBG_TRACE_LEAVE(STRM_EV_STRM_ANA, s);
return 1;
sw_failed:
@ -1432,6 +1660,7 @@ static int process_switching_rules(struct stream *s, struct channel *req, int an
s->txn->status = 500;
s->req.analysers &= AN_REQ_FLT_END;
s->req.analyse_exp = TICK_ETERNITY;
DBG_TRACE_DEVEL("leaving on error", STRM_EV_STRM_ANA|STRM_EV_STRM_ERR, s);
return 0;
}
@ -1445,14 +1674,7 @@ static int process_server_rules(struct stream *s, struct channel *req, int an_bi
struct session *sess = s->sess;
struct server_rule *rule;
DPRINTF(stderr,"[%u] %s: stream=%p b=%p, exp(r,w)=%u,%u bf=%08x bl=%lu analysers=%02x\n",
now_ms, __FUNCTION__,
s,
req,
req->rex, req->wex,
req->flags,
c_data(req),
req->analysers);
DBG_TRACE_ENTER(STRM_EV_STRM_ANA, s);
if (!(s->flags & SF_ASSIGNED)) {
list_for_each_entry(rule, &px->server_rules, list) {
@ -1482,6 +1704,7 @@ static int process_server_rules(struct stream *s, struct channel *req, int an_bi
req->analysers &= ~an_bit;
req->analyse_exp = TICK_ETERNITY;
DBG_TRACE_LEAVE(STRM_EV_STRM_ANA, s);
return 1;
}
@ -1538,14 +1761,7 @@ static int process_sticking_rules(struct stream *s, struct channel *req, int an_
struct session *sess = s->sess;
struct sticking_rule *rule;
DPRINTF(stderr,"[%u] %s: stream=%p b=%p, exp(r,w)=%u,%u bf=%08x bh=%lu analysers=%02x\n",
now_ms, __FUNCTION__,
s,
req,
req->rex, req->wex,
req->flags,
ci_data(req),
req->analysers);
DBG_TRACE_ENTER(STRM_EV_STRM_ANA, s);
list_for_each_entry(rule, &px->sticking_rules, list) {
int ret = 1 ;
@ -1605,6 +1821,7 @@ static int process_sticking_rules(struct stream *s, struct channel *req, int an_
req->analysers &= ~an_bit;
req->analyse_exp = TICK_ETERNITY;
DBG_TRACE_LEAVE(STRM_EV_STRM_ANA, s);
return 1;
}
@ -1620,14 +1837,7 @@ static int process_store_rules(struct stream *s, struct channel *rep, int an_bit
int i;
int nbreq = s->store_count;
DPRINTF(stderr,"[%u] %s: stream=%p b=%p, exp(r,w)=%u,%u bf=%08x bh=%lu analysers=%02x\n",
now_ms, __FUNCTION__,
s,
rep,
rep->rex, rep->wex,
rep->flags,
ci_data(rep),
rep->analysers);
DBG_TRACE_ENTER(STRM_EV_STRM_ANA, s);
list_for_each_entry(rule, &px->storersp_rules, list) {
int ret = 1 ;
@ -1717,6 +1927,8 @@ static int process_store_rules(struct stream *s, struct channel *rep, int an_bit
rep->analysers &= ~an_bit;
rep->analyse_exp = TICK_ETERNITY;
DBG_TRACE_LEAVE(STRM_EV_STRM_ANA, s);
return 1;
}
@ -1788,6 +2000,8 @@ struct task *process_stream(struct task *t, void *context, unsigned short state)
struct stream_interface *si_f, *si_b;
unsigned int rate;
DBG_TRACE_ENTER(STRM_EV_STRM_PROC, s);
activity[tid].stream++;
req = &s->req;
@ -1805,9 +2019,6 @@ struct task *process_stream(struct task *t, void *context, unsigned short state)
stream_dump_and_crash(&s->obj_type, read_freq_ctr(&s->call_rate));
}
//DPRINTF(stderr, "%s:%d: cs=%d ss=%d(%d) rqf=0x%08x rpf=0x%08x\n", __FUNCTION__, __LINE__,
// si_f->state, si_b->state, si_b->err_type, req->flags, res->flags);
/* this data may be no longer valid, clear it */
if (s->txn)
memset(&s->txn->auth, 0, sizeof(s->txn->auth));
@ -1972,18 +2183,7 @@ struct task *process_stream(struct task *t, void *context, unsigned short state)
rp_prod_last = si_b->state;
/* Check for connection closure */
DPRINTF(stderr,
"[%u] %s:%d: task=%p s=%p, sfl=0x%08x, rq=%p, rp=%p, exp(r,w)=%u,%u rqf=%08x rpf=%08x rqh=%lu rqt=%lu rph=%lu rpt=%lu cs=%d ss=%d, cet=0x%x set=0x%x retr=%d\n",
now_ms, __FUNCTION__, __LINE__,
t,
s, s->flags,
req, res,
req->rex, res->wex,
req->flags, res->flags,
ci_data(req), co_data(req), ci_data(res), co_data(res), si_f->state, si_b->state,
si_f->err_type, si_b->err_type,
si_b->conn_retries);
DBG_TRACE_POINT(STRM_EV_STRM_PROC, s);
/* nothing special to be done on client side */
if (unlikely(si_f->state == SI_ST_DIS))
@ -2650,17 +2850,15 @@ struct task *process_stream(struct task *t, void *context, unsigned short state)
if (si_b->exp)
t->expire = tick_first(t->expire, si_b->exp);
DPRINTF(stderr,
"[%u] queuing with exp=%u req->rex=%u req->wex=%u req->ana_exp=%u"
" rep->rex=%u rep->wex=%u, si[0].exp=%u, si[1].exp=%u, cs=%d, ss=%d\n",
now_ms, t->expire, req->rex, req->wex, req->analyse_exp,
res->rex, res->wex, si_f->exp, si_b->exp, si_f->state, si_b->state);
s->pending_events &= ~(TASK_WOKEN_TIMER | TASK_WOKEN_RES);
stream_release_buffers(s);
DBG_TRACE_DEVEL("queuing", STRM_EV_STRM_PROC, s);
return t; /* nothing more to do */
}
DBG_TRACE_DEVEL("releasing", STRM_EV_STRM_PROC, s);
if (s->flags & SF_BE_ASSIGNED)
_HA_ATOMIC_SUB(&s->be->beconn, 1);

27
src/tcp_rules.c
View File

@ -36,6 +36,8 @@
#include <proto/stream_interface.h>
#include <proto/tcp_rules.h>
#define TRACE_SOURCE &trace_strm
/* List head of all known action keywords for "tcp-request connection" */
struct list tcp_req_conn_keywords = LIST_HEAD_INIT(tcp_req_conn_keywords);
struct list tcp_req_sess_keywords = LIST_HEAD_INIT(tcp_req_sess_keywords);
@ -104,14 +106,7 @@ int tcp_inspect_request(struct stream *s, struct channel *req, int an_bit)
int partial;
int act_flags = 0;
DPRINTF(stderr,"[%u] %s: stream=%p b=%p, exp(r,w)=%u,%u bf=%08x bh=%lu analysers=%02x\n",
now_ms, __FUNCTION__,
s,
req,
req->rex, req->wex,
req->flags,
ci_data(req),
req->analysers);
DBG_TRACE_ENTER(STRM_EV_STRM_ANA|STRM_EV_TCP_ANA, s);
/* We don't know whether we have enough data, so must proceed
* this way :
@ -177,6 +172,7 @@ resume_execution:
s->flags |= SF_ERR_PRXCOND;
if (!(s->flags & SF_FINST_MASK))
s->flags |= SF_FINST_R;
DBG_TRACE_DEVEL("leaving on error", STRM_EV_STRM_ANA|STRM_EV_TCP_ANA|STRM_EV_TCP_ERR, s);
return 0;
}
else if (rule->action >= ACT_ACTION_TRK_SC0 && rule->action <= ACT_ACTION_TRK_SCMAX) {
@ -258,6 +254,7 @@ resume_execution:
*/
req->analysers &= ~an_bit;
req->analyse_exp = TICK_ETERNITY;
DBG_TRACE_LEAVE(STRM_EV_STRM_ANA|STRM_EV_TCP_ANA, s);
return 1;
missing_data:
@ -265,6 +262,7 @@ resume_execution:
/* just set the request timeout once at the beginning of the request */
if (!tick_isset(req->analyse_exp) && s->be->tcp_req.inspect_delay)
req->analyse_exp = tick_add(now_ms, s->be->tcp_req.inspect_delay);
DBG_TRACE_DEVEL("waiting for more data", STRM_EV_STRM_ANA|STRM_EV_TCP_ANA, s);
return 0;
}
@ -282,14 +280,7 @@ int tcp_inspect_response(struct stream *s, struct channel *rep, int an_bit)
int partial;
int act_flags = 0;
DPRINTF(stderr,"[%u] %s: stream=%p b=%p, exp(r,w)=%u,%u bf=%08x bh=%lu analysers=%02x\n",
now_ms, __FUNCTION__,
s,
rep,
rep->rex, rep->wex,
rep->flags,
ci_data(rep),
rep->analysers);
DBG_TRACE_ENTER(STRM_EV_STRM_ANA|STRM_EV_TCP_ANA, s);
/* We don't know whether we have enough data, so must proceed
* this way :
@ -328,6 +319,7 @@ int tcp_inspect_response(struct stream *s, struct channel *rep, int an_bit)
/* just set the analyser timeout once at the beginning of the response */
if (!tick_isset(rep->analyse_exp) && s->be->tcp_rep.inspect_delay)
rep->analyse_exp = tick_add(now_ms, s->be->tcp_rep.inspect_delay);
DBG_TRACE_DEVEL("waiting for more data", STRM_EV_STRM_ANA|STRM_EV_TCP_ANA, s);
return 0;
}
@ -358,6 +350,7 @@ resume_execution:
s->flags |= SF_ERR_PRXCOND;
if (!(s->flags & SF_FINST_MASK))
s->flags |= SF_FINST_D;
DBG_TRACE_DEVEL("leaving on error", STRM_EV_STRM_ANA|STRM_EV_TCP_ANA|STRM_EV_TCP_ERR, s);
return 0;
}
else if (rule->action == ACT_TCP_CLOSE) {
@ -385,6 +378,7 @@ resume_execution:
case ACT_RET_YIELD:
channel_dont_close(rep);
s->current_rule = rule;
DBG_TRACE_DEVEL("waiting for more data", STRM_EV_STRM_ANA|STRM_EV_TCP_ANA, s);
return 0;
}
break; /* ACT_RET_STOP/DONE */
@ -397,6 +391,7 @@ resume_execution:
*/
rep->analysers &= ~an_bit;
rep->analyse_exp = TICK_ETERNITY;
DBG_TRACE_LEAVE(STRM_EV_STRM_ANA|STRM_EV_TCP_ANA, s);
return 1;
}