MEDIUM: tree-wide: Use CS util functions instead of SI ones
At many places, we now use the new CS functions to get a stream or a channel
from a conn-stream instead of using the stream-interface API. It is the
first step to reduce the scope of the stream-interfaces. The main change
here is about the applet I/O callback functions. Before the refactoring, the
stream-interface was the appctx owner. Thus, it was heavily used. Now, as
far as possible,the conn-stream is used. Of course, it remains many calls to
the stream-interface API.
diff --git a/addons/promex/service-prometheus.c b/addons/promex/service-prometheus.c
index e02da75..b241922 100644
--- a/addons/promex/service-prometheus.c
+++ b/addons/promex/service-prometheus.c
@@ -19,6 +19,8 @@
#include <haproxy/backend.h>
#include <haproxy/cfgparse.h>
#include <haproxy/check.h>
+#include <haproxy/conn_stream.h>
+#include <haproxy/cs_utils.h>
#include <haproxy/frontend.h>
#include <haproxy/global.h>
#include <haproxy/http.h>
@@ -543,7 +545,7 @@
{
static struct ist prefix = IST("haproxy_process_");
struct field val;
- struct channel *chn = si_ic(cs_si(appctx->owner));
+ struct channel *chn = cs_ic(appctx->owner);
struct ist out = ist2(trash.area, 0);
size_t max = htx_get_max_blksz(htx, channel_htx_recv_max(chn, htx));
int ret = 1;
@@ -594,7 +596,7 @@
static struct ist prefix = IST("haproxy_frontend_");
struct proxy *px;
struct field val;
- struct channel *chn = si_ic(cs_si(appctx->owner));
+ struct channel *chn = cs_ic(appctx->owner);
struct ist out = ist2(trash.area, 0);
size_t max = htx_get_max_blksz(htx, channel_htx_recv_max(chn, htx));
struct field *stats = stat_l[STATS_DOMAIN_PROXY];
@@ -694,7 +696,7 @@
static struct ist prefix = IST("haproxy_listener_");
struct proxy *px;
struct field val;
- struct channel *chn = si_ic(cs_si(appctx->owner));
+ struct channel *chn = cs_ic(appctx->owner);
struct ist out = ist2(trash.area, 0);
size_t max = htx_get_max_blksz(htx, channel_htx_recv_max(chn, htx));
struct field *stats = stat_l[STATS_DOMAIN_PROXY];
@@ -785,7 +787,7 @@
struct proxy *px;
struct server *sv;
struct field val;
- struct channel *chn = si_ic(cs_si(appctx->owner));
+ struct channel *chn = cs_ic(appctx->owner);
struct ist out = ist2(trash.area, 0);
size_t max = htx_get_max_blksz(htx, channel_htx_recv_max(chn, htx));
struct field *stats = stat_l[STATS_DOMAIN_PROXY];
@@ -938,7 +940,7 @@
struct proxy *px;
struct server *sv;
struct field val;
- struct channel *chn = si_ic(cs_si(appctx->owner));
+ struct channel *chn = cs_ic(appctx->owner);
struct ist out = ist2(trash.area, 0);
size_t max = htx_get_max_blksz(htx, channel_htx_recv_max(chn, htx));
struct field *stats = stat_l[STATS_DOMAIN_PROXY];
@@ -1107,7 +1109,7 @@
{
static struct ist prefix = IST("haproxy_sticktable_");
struct field val;
- struct channel *chn = si_ic(cs_si(appctx->owner));
+ struct channel *chn = cs_ic(appctx->owner);
struct ist out = ist2(trash.area, 0);
size_t max = htx_get_max_blksz(htx, channel_htx_recv_max(chn, htx));
int ret = 1;
@@ -1168,7 +1170,7 @@
* -1 in case of any error.
* Uses <appctx.ctx.stats.obj1> as a pointer to the current proxy and <obj2> as
* a pointer to the current server/listener. */
-static int promex_dump_metrics(struct appctx *appctx, struct stream_interface *si, struct htx *htx)
+static int promex_dump_metrics(struct appctx *appctx, struct conn_stream *cs, struct htx *htx)
{
int ret;
@@ -1301,7 +1303,7 @@
return 1;
full:
- si_rx_room_blk(si);
+ si_rx_room_blk(cs->si);
return 0;
error:
/* unrecoverable error */
@@ -1315,10 +1317,10 @@
/* Parse the query string of request URI to filter the metrics. It returns 1 on
* success and -1 on error. */
-static int promex_parse_uri(struct appctx *appctx, struct stream_interface *si)
+static int promex_parse_uri(struct appctx *appctx, struct conn_stream *cs)
{
- struct channel *req = si_oc(si);
- struct channel *res = si_ic(si);
+ struct channel *req = cs_oc(cs);
+ struct channel *res = cs_ic(cs);
struct htx *req_htx, *res_htx;
struct htx_sl *sl;
char *p, *key, *value;
@@ -1425,9 +1427,9 @@
/* Send HTTP headers of the response. It returns 1 on success and 0 if <htx> is
* full. */
-static int promex_send_headers(struct appctx *appctx, struct stream_interface *si, struct htx *htx)
+static int promex_send_headers(struct appctx *appctx, struct conn_stream *cs, struct htx *htx)
{
- struct channel *chn = si_ic(cs_si(appctx->owner));
+ struct channel *chn = cs_ic(cs);
struct htx_sl *sl;
unsigned int flags;
@@ -1446,7 +1448,7 @@
return 1;
full:
htx_reset(htx);
- si_rx_room_blk(si);
+ si_rx_room_blk(cs->si);
return 0;
}
@@ -1463,26 +1465,26 @@
/* The main I/O handler for the promex applet. */
static void promex_appctx_handle_io(struct appctx *appctx)
{
- struct stream_interface *si = cs_si(appctx->owner);
- struct stream *s = si_strm(si);
- struct channel *req = si_oc(si);
- struct channel *res = si_ic(si);
+ struct conn_stream *cs = appctx->owner;
+ struct stream *s = __cs_strm(cs);
+ struct channel *req = cs_oc(cs);
+ struct channel *res = cs_ic(cs);
struct htx *req_htx, *res_htx;
int ret;
res_htx = htx_from_buf(&res->buf);
- if (unlikely(si->state == SI_ST_DIS || si->state == SI_ST_CLO))
+ if (unlikely(cs->si->state == SI_ST_DIS || cs->si->state == SI_ST_CLO))
goto out;
/* Check if the input buffer is available. */
if (!b_size(&res->buf)) {
- si_rx_room_blk(si);
+ si_rx_room_blk(cs->si);
goto out;
}
switch (appctx->st0) {
case PROMEX_ST_INIT:
- ret = promex_parse_uri(appctx, si);
+ ret = promex_parse_uri(appctx, cs);
if (ret <= 0) {
if (ret == -1)
goto error;
@@ -1493,13 +1495,13 @@
/* fall through */
case PROMEX_ST_HEAD:
- if (!promex_send_headers(appctx, si, res_htx))
+ if (!promex_send_headers(appctx, cs, res_htx))
goto out;
appctx->st0 = ((s->txn->meth == HTTP_METH_HEAD) ? PROMEX_ST_DONE : PROMEX_ST_DUMP);
/* fall through */
case PROMEX_ST_DUMP:
- ret = promex_dump_metrics(appctx, si, res_htx);
+ ret = promex_dump_metrics(appctx, cs, res_htx);
if (ret <= 0) {
if (ret == -1)
goto error;
@@ -1517,13 +1519,13 @@
*/
if (htx_is_empty(res_htx)) {
if (!htx_add_endof(res_htx, HTX_BLK_EOT)) {
- si_rx_room_blk(si);
+ si_rx_room_blk(cs->si);
goto out;
}
channel_add_input(res, 1);
}
res_htx->flags |= HTX_FL_EOM;
- si->cs->endp->flags |= CS_EP_EOI;
+ cs->endp->flags |= CS_EP_EOI;
res->flags |= CF_EOI;
appctx->st0 = PROMEX_ST_END;
/* fall through */
@@ -1531,7 +1533,7 @@
case PROMEX_ST_END:
if (!(res->flags & CF_SHUTR)) {
res->flags |= CF_READ_NULL;
- si_shutr(si);
+ si_shutr(cs->si);
}
}
@@ -1547,8 +1549,8 @@
error:
res->flags |= CF_READ_NULL;
- si_shutr(si);
- si_shutw(si);
+ si_shutr(cs->si);
+ si_shutw(cs->si);
}
struct applet promex_applet = {
diff --git a/include/haproxy/applet-t.h b/include/haproxy/applet-t.h
index bbc6b02..6bcabec 100644
--- a/include/haproxy/applet-t.h
+++ b/include/haproxy/applet-t.h
@@ -33,8 +33,9 @@
#define APPLET_WANT_DIE 0x01 /* applet was running and requested to die */
struct appctx;
-struct stream;
struct proxy;
+struct conn_stream;
+struct cs_endpoint;
/* Applet descriptor */
struct applet {
diff --git a/include/haproxy/resolvers.h b/include/haproxy/resolvers.h
index 4058f9f..975f9d4 100644
--- a/include/haproxy/resolvers.h
+++ b/include/haproxy/resolvers.h
@@ -26,7 +26,7 @@
struct proxy;
struct server;
-struct stream_interface;
+struct conn_stream;
struct act_rule;
struct list;
@@ -55,7 +55,7 @@
enum act_parse_ret resolv_parse_do_resolve(const char **args, int *orig_arg, struct proxy *px, struct act_rule *rule, char **err);
int check_action_do_resolve(struct act_rule *rule, struct proxy *px, char **err);
-int stats_dump_resolvers(struct stream_interface *si,
+int stats_dump_resolvers(struct conn_stream *cs,
struct field *stats, size_t stats_count,
struct list *stat_modules);
void resolv_stats_clear_counters(int clrall, struct list *stat_modules);
diff --git a/src/activity.c b/src/activity.c
index 7386247..e72b9c7 100644
--- a/src/activity.c
+++ b/src/activity.c
@@ -16,6 +16,8 @@
#include <haproxy/clock.h>
#include <haproxy/channel.h>
#include <haproxy/cli.h>
+#include <haproxy/conn_stream.h>
+#include <haproxy/cs_utils.h>
#include <haproxy/freq_ctr.h>
#include <haproxy/stream_interface.h>
#include <haproxy/tools.h>
@@ -610,13 +612,13 @@
unsigned long long tot_alloc_calls, tot_free_calls;
unsigned long long tot_alloc_bytes, tot_free_bytes;
#endif
- struct stream_interface *si = cs_si(appctx->owner);
+ struct conn_stream *cs = appctx->owner;
struct buffer *name_buffer = get_trash_chunk();
const char *str;
int max_lines;
int i, max;
- if (unlikely(si_ic(si)->flags & (CF_WRITE_ERROR|CF_SHUTW)))
+ if (unlikely(cs_ic(cs)->flags & (CF_WRITE_ERROR|CF_SHUTW)))
return 1;
chunk_reset(&trash);
@@ -636,9 +638,9 @@
"Memory usage profiling : %-8s # set profiling memory {on|off}\n",
str, (profiling & HA_PROF_MEMORY) ? "on" : "off");
- if (ci_putchk(si_ic(si), &trash) == -1) {
+ if (ci_putchk(cs_ic(cs), &trash) == -1) {
/* failed, try again */
- si_rx_room_blk(si);
+ si_rx_room_blk(cs->si);
return 0;
}
@@ -686,16 +688,16 @@
print_time_short(&trash, " ", tmp_activity[i].lat_time, "");
print_time_short(&trash, " ", tmp_activity[i].lat_time / tmp_activity[i].calls, "\n");
- if (ci_putchk(si_ic(si), &trash) == -1) {
+ if (ci_putchk(cs_ic(cs), &trash) == -1) {
/* failed, try again */
- si_rx_room_blk(si);
+ si_rx_room_blk(cs->si);
return 0;
}
}
- if (ci_putchk(si_ic(si), &trash) == -1) {
+ if (ci_putchk(cs_ic(cs), &trash) == -1) {
/* failed, try again */
- si_rx_room_blk(si);
+ si_rx_room_blk(cs->si);
return 0;
}
@@ -751,14 +753,14 @@
chunk_appendf(&trash, "\n");
- if (ci_putchk(si_ic(si), &trash) == -1) {
- si_rx_room_blk(si);
+ if (ci_putchk(cs_ic(cs), &trash) == -1) {
+ si_rx_room_blk(cs->si);
return 0;
}
}
- if (ci_putchk(si_ic(si), &trash) == -1) {
- si_rx_room_blk(si);
+ if (ci_putchk(cs_ic(cs), &trash) == -1) {
+ si_rx_room_blk(cs->si);
return 0;
}
@@ -778,8 +780,8 @@
tot_alloc_calls - tot_free_calls,
tot_alloc_bytes - tot_free_bytes);
- if (ci_putchk(si_ic(si), &trash) == -1) {
- si_rx_room_blk(si);
+ if (ci_putchk(cs_ic(cs), &trash) == -1) {
+ si_rx_room_blk(cs->si);
return 0;
}
@@ -837,7 +839,7 @@
static int cli_io_handler_show_tasks(struct appctx *appctx)
{
struct sched_activity tmp_activity[256] __attribute__((aligned(64)));
- struct stream_interface *si = cs_si(appctx->owner);
+ struct conn_stream *cs = appctx->owner;
struct buffer *name_buffer = get_trash_chunk();
struct sched_activity *entry;
const struct tasklet *tl;
@@ -848,7 +850,7 @@
int thr, queue;
int i, max;
- if (unlikely(si_ic(si)->flags & (CF_WRITE_ERROR|CF_SHUTW)))
+ if (unlikely(cs_ic(cs)->flags & (CF_WRITE_ERROR|CF_SHUTW)))
return 1;
/* It's not possible to scan queues in small chunks and yield in the
@@ -964,9 +966,9 @@
print_time_short(&trash, " ", tmp_activity[i].lat_time / tmp_activity[i].calls, "\n");
}
- if (ci_putchk(si_ic(si), &trash) == -1) {
+ if (ci_putchk(cs_ic(cs), &trash) == -1) {
/* failed, try again */
- si_rx_room_blk(si);
+ si_rx_room_blk(cs->si);
return 0;
}
return 1;
diff --git a/src/applet.c b/src/applet.c
index 663a9df..65e4337 100644
--- a/src/applet.c
+++ b/src/applet.c
@@ -16,6 +16,8 @@
#include <haproxy/api.h>
#include <haproxy/applet.h>
#include <haproxy/channel.h>
+#include <haproxy/conn_stream.h>
+#include <haproxy/cs_utils.h>
#include <haproxy/list.h>
#include <haproxy/stream.h>
#include <haproxy/stream_interface.h>
@@ -100,21 +102,21 @@
int appctx_buf_available(void *arg)
{
struct appctx *appctx = arg;
- struct stream_interface *si = cs_si(appctx->owner);
+ struct conn_stream *cs = appctx->owner;
/* allocation requested ? */
- if (!(si->flags & SI_FL_RXBLK_BUFF))
+ if (!(cs->si->flags & SI_FL_RXBLK_BUFF))
return 0;
- si_rx_buff_rdy(si);
+ si_rx_buff_rdy(cs->si);
/* was already allocated another way ? if so, don't take this one */
- if (c_size(si_ic(si)) || si_ic(si)->pipe)
+ if (c_size(cs_ic(cs)) || cs_ic(cs)->pipe)
return 0;
/* allocation possible now ? */
- if (!b_alloc(&si_ic(si)->buf)) {
- si_rx_buff_blk(si);
+ if (!b_alloc(&cs_ic(cs)->buf)) {
+ si_rx_buff_blk(cs->si);
return 0;
}
@@ -126,7 +128,7 @@
struct task *task_run_applet(struct task *t, void *context, unsigned int state)
{
struct appctx *app = context;
- struct stream_interface *si;
+ struct conn_stream *cs = app->owner;
unsigned int rate;
size_t count;
@@ -135,14 +137,12 @@
return NULL;
}
- si = cs_si(app->owner);
-
/* We always pretend the applet can't get and doesn't want to
* put, it's up to it to change this if needed. This ensures
* that one applet which ignores any event will not spin.
*/
- si_cant_get(si);
- si_rx_endp_done(si);
+ si_cant_get(cs->si);
+ si_rx_endp_done(cs->si);
/* Now we'll try to allocate the input buffer. We wake up the applet in
* all cases. So this is the applet's responsibility to check if this
@@ -150,32 +150,32 @@
* some other processing if needed. The applet doesn't have anything to
* do if it needs the buffer, it will be called again upon readiness.
*/
- if (!si_alloc_ibuf(si, &app->buffer_wait))
- si_rx_endp_more(si);
+ if (!si_alloc_ibuf(cs->si, &app->buffer_wait))
+ si_rx_endp_more(cs->si);
- count = co_data(si_oc(si));
+ count = co_data(cs_oc(cs));
app->applet->fct(app);
/* now check if the applet has released some room and forgot to
* notify the other side about it.
*/
- if (count != co_data(si_oc(si))) {
- si_oc(si)->flags |= CF_WRITE_PARTIAL | CF_WROTE_DATA;
- si_rx_room_rdy(si_opposite(si));
+ if (count != co_data(cs_oc(cs))) {
+ cs_oc(cs)->flags |= CF_WRITE_PARTIAL | CF_WROTE_DATA;
+ si_rx_room_rdy(cs_opposite(cs)->si);
}
/* measure the call rate and check for anomalies when too high */
rate = update_freq_ctr(&app->call_rate, 1);
if (rate >= 100000 && app->call_rate.prev_ctr && // looped more than 100k times over last second
- ((b_size(si_ib(si)) && si->flags & SI_FL_RXBLK_BUFF) || // asks for a buffer which is present
- (b_size(si_ib(si)) && !b_data(si_ib(si)) && si->flags & SI_FL_RXBLK_ROOM) || // asks for room in an empty buffer
- (b_data(si_ob(si)) && si_tx_endp_ready(si) && !si_tx_blocked(si)) || // asks for data already present
- (!b_data(si_ib(si)) && b_data(si_ob(si)) && // didn't return anything ...
- (si_oc(si)->flags & (CF_WRITE_PARTIAL|CF_SHUTW_NOW)) == CF_SHUTW_NOW))) { // ... and left data pending after a shut
+ ((b_size(cs_ib(cs)) && cs->si->flags & SI_FL_RXBLK_BUFF) || // asks for a buffer which is present
+ (b_size(cs_ib(cs)) && !b_data(cs_ib(cs)) && cs->si->flags & SI_FL_RXBLK_ROOM) || // asks for room in an empty buffer
+ (b_data(cs_ob(cs)) && si_tx_endp_ready(cs->si) && !si_tx_blocked(cs->si)) || // asks for data already present
+ (!b_data(cs_ib(cs)) && b_data(cs_ob(cs)) && // didn't return anything ...
+ (cs_oc(cs)->flags & (CF_WRITE_PARTIAL|CF_SHUTW_NOW)) == CF_SHUTW_NOW))) { // ... and left data pending after a shut
stream_dump_and_crash(&app->obj_type, read_freq_ctr(&app->call_rate));
}
- si_applet_wake_cb(si);
- channel_release_buffer(si_ic(si), &app->buffer_wait);
+ si_applet_wake_cb(cs->si);
+ channel_release_buffer(cs_ic(cs), &app->buffer_wait);
return t;
}
diff --git a/src/backend.c b/src/backend.c
index 09f10e8..a23aecf 100644
--- a/src/backend.c
+++ b/src/backend.c
@@ -28,6 +28,8 @@
#include <haproxy/backend.h>
#include <haproxy/channel.h>
#include <haproxy/check.h>
+#include <haproxy/conn_stream.h>
+#include <haproxy/cs_utils.h>
#include <haproxy/frontend.h>
#include <haproxy/global.h>
#include <haproxy/hash.h>
@@ -1720,7 +1722,7 @@
((cli_conn->flags & CO_FL_EARLY_DATA) ||
((s->be->retry_type & PR_RE_EARLY_ERROR) &&
cs_si(s->csb)->conn_retries == s->be->conn_retries)) &&
- !channel_is_empty(si_oc(cs_si(s->csb))) &&
+ !channel_is_empty(cs_oc(s->csb)) &&
srv_conn->flags & CO_FL_SSL_WAIT_HS)
srv_conn->flags &= ~(CO_FL_SSL_WAIT_HS | CO_FL_WAIT_L6_CONN);
#endif
@@ -1757,7 +1759,7 @@
if (!si_state_in(cs_si(s->csb)->state, SI_SB_EST|SI_SB_DIS|SI_SB_CLO) &&
(srv_conn->flags & CO_FL_WAIT_XPRT) == 0) {
cs_si(s->csb)->exp = TICK_ETERNITY;
- si_oc(cs_si(s->csb))->flags |= CF_WRITE_NULL;
+ cs_oc(s->csb)->flags |= CF_WRITE_NULL;
if (cs_si(s->csb)->state == SI_ST_CON)
cs_si(s->csb)->state = SI_ST_RDY;
}
@@ -1769,8 +1771,8 @@
* wake callback. Otherwise si_cs_recv()/si_cs_send() already take
* care of it.
*/
- if ((s->csb->endp->flags & CS_EP_EOI) && !(si_ic(cs_si(s->csb))->flags & CF_EOI))
- si_ic(cs_si(s->csb))->flags |= (CF_EOI|CF_READ_PARTIAL);
+ if ((s->csb->endp->flags & CS_EP_EOI) && !(cs_ic(s->csb)->flags & CF_EOI))
+ cs_ic(s->csb)->flags |= (CF_EOI|CF_READ_PARTIAL);
/* catch all sync connect while the mux is not already installed */
if (!srv_conn->mux && !(srv_conn->flags & CO_FL_WAIT_XPRT)) {
@@ -1888,12 +1890,12 @@
void back_try_conn_req(struct stream *s)
{
struct server *srv = objt_server(s->target);
- struct stream_interface *si = cs_si(s->csb);
+ struct conn_stream *cs = s->csb;
struct channel *req = &s->req;
DBG_TRACE_ENTER(STRM_EV_STRM_PROC|STRM_EV_SI_ST, s);
- if (si->state == SI_ST_ASS) {
+ if (cs->si->state == SI_ST_ASS) {
/* Server assigned to connection request, we have to try to connect now */
int conn_err;
@@ -1901,7 +1903,7 @@
* request may be aborted instead.
*/
if (back_may_abort_req(req, s)) {
- si->err_type |= SI_ET_CONN_ABRT;
+ cs->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;
}
@@ -1923,8 +1925,8 @@
* abort, retry immediately or redispatch.
*/
if (conn_err == SF_ERR_INTERNAL) {
- if (!si->err_type) {
- si->err_type = SI_ET_CONN_OTHER;
+ if (!cs->si->err_type) {
+ cs->si->err_type = SI_ET_CONN_OTHER;
}
if (srv)
@@ -1941,8 +1943,8 @@
process_srv_queue(srv);
/* Failed and not retryable. */
- si_shutr(si);
- si_shutw(si);
+ si_shutr(cs->si);
+ si_shutw(cs->si);
req->flags |= CF_WRITE_ERROR;
s->logs.t_queue = tv_ms_elapsed(&s->logs.tv_accept, &now);
@@ -1951,9 +1953,9 @@
pendconn_cond_unlink(s->pend_pos);
/* no stream was ever accounted for this server */
- si->state = SI_ST_CLO;
+ cs->si->state = SI_ST_CLO;
if (s->srv_error)
- s->srv_error(s, si);
+ s->srv_error(s, cs->si);
DBG_TRACE_STATE("internal error during connection", STRM_EV_STRM_PROC|STRM_EV_SI_ST|STRM_EV_STRM_ERR, s);
goto end;
}
@@ -1962,14 +1964,14 @@
* turn-around now, as the problem is likely a source port
* allocation problem, so we want to retry now.
*/
- si->state = SI_ST_CER;
- si->flags &= ~SI_FL_ERR;
+ cs->si->state = SI_ST_CER;
+ cs->si->flags &= ~SI_FL_ERR;
back_handle_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 */
}
- else if (si->state == SI_ST_QUE) {
+ else if (cs->si->state == SI_ST_QUE) {
/* connection request was queued, check for any update */
if (!pendconn_dequeue(s)) {
/* The connection is not in the queue anymore. Either
@@ -1977,22 +1979,22 @@
* go directly to the assigned state, or we need to
* load-balance first and go to the INI state.
*/
- si->exp = TICK_ETERNITY;
+ cs->si->exp = TICK_ETERNITY;
if (unlikely(!(s->flags & SF_ASSIGNED)))
- si->state = SI_ST_REQ;
+ cs->si->state = SI_ST_REQ;
else {
s->logs.t_queue = tv_ms_elapsed(&s->logs.tv_accept, &now);
- si->state = SI_ST_ASS;
+ cs->si->state = SI_ST_ASS;
}
DBG_TRACE_STATE("dequeue connection request", STRM_EV_STRM_PROC|STRM_EV_SI_ST, s);
goto end;
}
/* Connection request still in queue... */
- if (si->flags & SI_FL_EXP) {
+ if (cs->si->flags & SI_FL_EXP) {
/* ... and timeout expired */
- si->exp = TICK_ETERNITY;
- si->flags &= ~SI_FL_EXP;
+ cs->si->exp = TICK_ETERNITY;
+ cs->si->flags &= ~SI_FL_EXP;
s->logs.t_queue = tv_ms_elapsed(&s->logs.tv_accept, &now);
/* we may need to know the position in the queue for logging */
@@ -2001,14 +2003,14 @@
if (srv)
_HA_ATOMIC_INC(&srv->counters.failed_conns);
_HA_ATOMIC_INC(&s->be->be_counters.failed_conns);
- si_shutr(si);
- si_shutw(si);
+ si_shutr(cs->si);
+ si_shutw(cs->si);
req->flags |= CF_WRITE_TIMEOUT;
- if (!si->err_type)
- si->err_type = SI_ET_QUEUE_TO;
- si->state = SI_ST_CLO;
+ if (!cs->si->err_type)
+ cs->si->err_type = SI_ET_QUEUE_TO;
+ cs->si->state = SI_ST_CLO;
if (s->srv_error)
- s->srv_error(s, si);
+ s->srv_error(s, cs->si);
DBG_TRACE_STATE("connection request still queued", STRM_EV_STRM_PROC|STRM_EV_SI_ST, s);
goto end;
}
@@ -2020,35 +2022,35 @@
/* we may need to know the position in the queue for logging */
pendconn_cond_unlink(s->pend_pos);
- si->err_type |= SI_ET_QUEUE_ABRT;
+ cs->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 */
}
- else if (si->state == SI_ST_TAR) {
+ else if (cs->si->state == SI_ST_TAR) {
/* Connection request might be aborted */
if (back_may_abort_req(req, s)) {
- si->err_type |= SI_ET_CONN_ABRT;
+ cs->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;
}
- if (!(si->flags & SI_FL_EXP))
+ if (!(cs->si->flags & SI_FL_EXP))
return; /* still in turn-around */
- si->flags &= ~SI_FL_EXP;
- si->exp = TICK_ETERNITY;
+ cs->si->flags &= ~SI_FL_EXP;
+ cs->si->exp = TICK_ETERNITY;
/* we keep trying on the same server as long as the stream is
* marked "assigned".
* FIXME: Should we force a redispatch attempt when the server is down ?
*/
if (s->flags & SF_ASSIGNED)
- si->state = SI_ST_ASS;
+ cs->si->state = SI_ST_ASS;
else
- si->state = SI_ST_REQ;
+ cs->si->state = SI_ST_REQ;
DBG_TRACE_STATE("retry connection now", STRM_EV_STRM_PROC|STRM_EV_SI_ST, s);
}
@@ -2059,13 +2061,13 @@
abort_connection:
/* give up */
- si->exp = TICK_ETERNITY;
- si->flags &= ~SI_FL_EXP;
- si_shutr(si);
- si_shutw(si);
- si->state = SI_ST_CLO;
+ cs->si->exp = TICK_ETERNITY;
+ cs->si->flags &= ~SI_FL_EXP;
+ si_shutr(cs->si);
+ si_shutw(cs->si);
+ cs->si->state = SI_ST_CLO;
if (s->srv_error)
- s->srv_error(s, si);
+ s->srv_error(s, cs->si);
DBG_TRACE_DEVEL("leaving on error", STRM_EV_STRM_PROC|STRM_EV_SI_ST|STRM_EV_STRM_ERR, s);
return;
}
@@ -2078,9 +2080,9 @@
*/
void back_handle_st_req(struct stream *s)
{
- struct stream_interface *si = cs_si(s->csb);
+ struct conn_stream *cs = s->csb;
- if (si->state != SI_ST_REQ)
+ if (cs->si->state != SI_ST_REQ)
return;
DBG_TRACE_ENTER(STRM_EV_STRM_PROC|STRM_EV_SI_ST, s);
@@ -2090,7 +2092,7 @@
struct appctx *appctx = cs_appctx(s->csb);
if (!appctx || appctx->applet != __objt_applet(s->target))
- appctx = si_register_handler(si, objt_applet(s->target));
+ appctx = si_register_handler(cs->si, objt_applet(s->target));
if (!appctx) {
/* No more memory, let's immediately abort. Force the
@@ -2099,13 +2101,13 @@
*/
s->flags &= ~(SF_ERR_MASK | SF_FINST_MASK);
- si_shutr(si);
- si_shutw(si);
+ si_shutr(cs->si);
+ si_shutw(cs->si);
s->req.flags |= CF_WRITE_ERROR;
- si->err_type = SI_ET_CONN_RES;
- si->state = SI_ST_CLO;
+ cs->si->err_type = SI_ET_CONN_RES;
+ cs->si->state = SI_ST_CLO;
if (s->srv_error)
- s->srv_error(s, si);
+ s->srv_error(s, cs->si);
DBG_TRACE_STATE("failed to register applet", STRM_EV_STRM_PROC|STRM_EV_SI_ST|STRM_EV_STRM_ERR, s);
goto end;
}
@@ -2113,8 +2115,8 @@
if (tv_iszero(&s->logs.tv_request))
s->logs.tv_request = now;
s->logs.t_queue = tv_ms_elapsed(&s->logs.tv_accept, &now);
- si->state = SI_ST_EST;
- si->err_type = SI_ET_NONE;
+ cs->si->state = SI_ST_EST;
+ cs->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);
@@ -2127,27 +2129,27 @@
/* We did not get a server. Either we queued the
* connection request, or we encountered an error.
*/
- if (si->state == SI_ST_QUE) {
+ if (cs->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);
- si_shutw(si);
+ si_shutr(cs->si);
+ si_shutw(cs->si);
s->req.flags |= CF_WRITE_ERROR;
- if (!si->err_type)
- si->err_type = SI_ET_CONN_OTHER;
- si->state = SI_ST_CLO;
+ if (!cs->si->err_type)
+ cs->si->err_type = SI_ET_CONN_OTHER;
+ cs->si->state = SI_ST_CLO;
if (s->srv_error)
- s->srv_error(s, si);
+ s->srv_error(s, cs->si);
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;
+ cs->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);
@@ -2164,7 +2166,7 @@
*/
void back_handle_st_con(struct stream *s)
{
- struct stream_interface *si = cs_si(s->csb);
+ struct conn_stream *cs = s->csb;
struct channel *req = &s->req;
struct channel *rep = &s->res;
@@ -2174,11 +2176,11 @@
if ((rep->flags & CF_SHUTW) ||
((req->flags & CF_SHUTW_NOW) &&
(channel_is_empty(req) || (s->be->options & PR_O_ABRT_CLOSE)))) {
- si->flags |= SI_FL_NOLINGER;
- si_shutw(si);
- si->err_type |= SI_ET_CONN_ABRT;
+ cs->si->flags |= SI_FL_NOLINGER;
+ si_shutw(cs->si);
+ cs->si->err_type |= SI_ET_CONN_ABRT;
if (s->srv_error)
- s->srv_error(s, si);
+ s->srv_error(s, cs->si);
/* Note: state = SI_ST_DIS now */
DBG_TRACE_STATE("client abort during connection attempt", STRM_EV_STRM_PROC|STRM_EV_SI_ST|STRM_EV_STRM_ERR, s);
goto end;
@@ -2186,15 +2188,15 @@
done:
/* retryable error ? */
- if (si->flags & (SI_FL_EXP|SI_FL_ERR)) {
- if (!si->err_type) {
- if (si->flags & SI_FL_ERR)
- si->err_type = SI_ET_CONN_ERR;
+ if (cs->si->flags & (SI_FL_EXP|SI_FL_ERR)) {
+ if (!cs->si->err_type) {
+ if (cs->si->flags & SI_FL_ERR)
+ cs->si->err_type = SI_ET_CONN_ERR;
else
- si->err_type = SI_ET_CONN_TO;
+ cs->si->err_type = SI_ET_CONN_TO;
}
- si->state = SI_ST_CER;
+ cs->si->state = SI_ST_CER;
DBG_TRACE_STATE("connection failed, retry", STRM_EV_STRM_PROC|STRM_EV_SI_ST|STRM_EV_STRM_ERR, s);
}
@@ -2213,16 +2215,16 @@
*/
void back_handle_st_cer(struct stream *s)
{
- struct stream_interface *si = cs_si(s->csb);
+ struct conn_stream *cs = s->csb;
DBG_TRACE_ENTER(STRM_EV_STRM_PROC|STRM_EV_SI_ST, s);
- si->exp = TICK_ETERNITY;
- si->flags &= ~SI_FL_EXP;
+ cs->si->exp = TICK_ETERNITY;
+ cs->si->flags &= ~SI_FL_EXP;
/* we probably have to release last stream from the server */
if (objt_server(s->target)) {
- struct connection *conn = cs_conn(s->csb);
+ struct connection *conn = cs_conn(cs);
health_adjust(__objt_server(s->target), HANA_STATUS_L4_ERR);
@@ -2231,7 +2233,7 @@
_HA_ATOMIC_DEC(&__objt_server(s->target)->cur_sess);
}
- if ((si->flags & SI_FL_ERR) &&
+ if ((cs->si->flags & SI_FL_ERR) &&
conn && conn->err_code == CO_ER_SSL_MISMATCH_SNI) {
/* We tried to connect to a server which is configured
* with "verify required" and which doesn't have the
@@ -2249,16 +2251,16 @@
* provided by the client and we don't want to let the
* client provoke retries.
*/
- si->conn_retries = 0;
+ cs->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);
}
}
/* ensure that we have enough retries left */
- si->conn_retries--;
- if (si->conn_retries < 0 || !(s->be->retry_type & PR_RE_CONN_FAILED)) {
- if (!si->err_type) {
- si->err_type = SI_ET_CONN_ERR;
+ cs->si->conn_retries--;
+ if (cs->si->conn_retries < 0 || !(s->be->retry_type & PR_RE_CONN_FAILED)) {
+ if (!cs->si->err_type) {
+ cs->si->err_type = SI_ET_CONN_ERR;
}
if (objt_server(s->target))
@@ -2269,13 +2271,13 @@
process_srv_queue(objt_server(s->target));
/* shutw is enough so stop a connecting socket */
- si_shutw(si);
+ si_shutw(cs->si);
s->req.flags |= CF_WRITE_ERROR;
s->res.flags |= CF_READ_ERROR;
- si->state = SI_ST_CLO;
+ cs->si->state = SI_ST_CLO;
if (s->srv_error)
- s->srv_error(s, si);
+ s->srv_error(s, cs->si);
DBG_TRACE_STATE("connection failed", STRM_EV_STRM_PROC|STRM_EV_SI_ST|STRM_EV_STRM_ERR, s);
goto end;
@@ -2291,9 +2293,9 @@
* Note: the stream-interface will be switched to ST_REQ, ST_ASS or
* ST_TAR and SI_FL_ERR and SI_FL_EXP flags will be unset.
*/
- if (cs_reset_endp(s->csb) < 0) {
- if (!si->err_type)
- si->err_type = SI_ET_CONN_OTHER;
+ if (cs_reset_endp(cs) < 0) {
+ if (!cs->si->err_type)
+ cs->si->err_type = SI_ET_CONN_OTHER;
if (objt_server(s->target))
_HA_ATOMIC_INC(&objt_server(s->target)->counters.internal_errors);
@@ -2303,13 +2305,13 @@
process_srv_queue(objt_server(s->target));
/* shutw is enough so stop a connecting socket */
- si_shutw(si);
+ si_shutw(cs->si);
s->req.flags |= CF_WRITE_ERROR;
s->res.flags |= CF_READ_ERROR;
- si->state = SI_ST_CLO;
+ cs->si->state = SI_ST_CLO;
if (s->srv_error)
- s->srv_error(s, si);
+ s->srv_error(s, cs->si);
DBG_TRACE_STATE("error resetting endpoint", STRM_EV_STRM_PROC|STRM_EV_SI_ST|STRM_EV_STRM_ERR, s);
goto end;
@@ -2317,7 +2319,7 @@
stream_choose_redispatch(s);
- if (si->flags & SI_FL_ERR) {
+ if (cs->si->flags & SI_FL_ERR) {
/* The error was an asynchronous connection error, and we will
* likely have to retry connecting to the same server, most
* likely leading to the same result. To avoid this, we wait
@@ -2333,17 +2335,17 @@
if (s->be->timeout.connect && s->be->timeout.connect < delay)
delay = s->be->timeout.connect;
- if (!si->err_type)
- si->err_type = SI_ET_CONN_ERR;
+ if (!cs->si->err_type)
+ cs->si->err_type = SI_ET_CONN_ERR;
/* only wait when we're retrying on the same server */
- if ((si->state == SI_ST_ASS ||
+ if ((cs->si->state == SI_ST_ASS ||
(s->be->lbprm.algo & BE_LB_KIND) != BE_LB_KIND_RR ||
(s->be->srv_act <= 1)) && !reused) {
- si->state = SI_ST_TAR;
- si->exp = tick_add(now_ms, MS_TO_TICKS(delay));
+ cs->si->state = SI_ST_TAR;
+ cs->si->exp = tick_add(now_ms, MS_TO_TICKS(delay));
}
- si->flags &= ~SI_FL_ERR;
+ cs->si->flags &= ~SI_FL_ERR;
DBG_TRACE_STATE("retry a new connection", STRM_EV_STRM_PROC|STRM_EV_SI_ST, s);
}
@@ -2360,7 +2362,7 @@
*/
void back_handle_st_rdy(struct stream *s)
{
- struct stream_interface *si = cs_si(s->csb);
+ struct conn_stream *cs = s->csb;
struct channel *req = &s->req;
struct channel *rep = &s->res;
@@ -2389,20 +2391,20 @@
((req->flags & CF_SHUTW_NOW) &&
(channel_is_empty(req) || (s->be->options & PR_O_ABRT_CLOSE)))) {
/* give up */
- si->flags |= SI_FL_NOLINGER;
- si_shutw(si);
- si->err_type |= SI_ET_CONN_ABRT;
+ cs->si->flags |= SI_FL_NOLINGER;
+ si_shutw(cs->si);
+ cs->si->err_type |= SI_ET_CONN_ABRT;
if (s->srv_error)
- s->srv_error(s, si);
+ s->srv_error(s, cs->si);
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 ? */
- if (si->flags & SI_FL_ERR) {
- if (!si->err_type)
- si->err_type = SI_ET_CONN_ERR;
- si->state = SI_ST_CER;
+ if (cs->si->flags & SI_FL_ERR) {
+ if (!cs->si->err_type)
+ cs->si->err_type = SI_ET_CONN_ERR;
+ cs->si->state = SI_ST_CER;
DBG_TRACE_STATE("connection failed, retry", STRM_EV_STRM_PROC|STRM_EV_SI_ST|STRM_EV_STRM_ERR, s);
goto end;
}
@@ -2412,8 +2414,8 @@
* 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;
+ cs->si->err_type = SI_ET_NONE;
+ cs->si->state = SI_ST_EST;
end:
DBG_TRACE_LEAVE(STRM_EV_STRM_PROC|STRM_EV_SI_ST, s);
diff --git a/src/cache.c b/src/cache.c
index ce6c7ea..5358b8b 100644
--- a/src/cache.c
+++ b/src/cache.c
@@ -18,6 +18,8 @@
#include <haproxy/cfgparse.h>
#include <haproxy/channel.h>
#include <haproxy/cli.h>
+#include <haproxy/conn_stream.h>
+#include <haproxy/cs_utils.h>
#include <haproxy/errors.h>
#include <haproxy/filters.h>
#include <haproxy/hash.h>
@@ -1272,7 +1274,7 @@
unsigned int max, total;
uint32_t blksz;
- max = htx_get_max_blksz(htx, channel_htx_recv_max(si_ic(cs_si(appctx->owner)), htx));
+ max = htx_get_max_blksz(htx, channel_htx_recv_max(cs_ic(appctx->owner), htx));
if (!max)
return 0;
blksz = ((type == HTX_BLK_HDR || type == HTX_BLK_TLR)
@@ -1315,7 +1317,7 @@
unsigned int max, total, rem_data;
uint32_t blksz;
- max = htx_get_max_blksz(htx, channel_htx_recv_max(si_ic(cs_si(appctx->owner)), htx));
+ max = htx_get_max_blksz(htx, channel_htx_recv_max(cs_ic(appctx->owner), htx));
if (!max)
return 0;
@@ -1429,9 +1431,9 @@
{
struct cache_entry *cache_ptr = appctx->ctx.cache.entry;
struct shared_block *first = block_ptr(cache_ptr);
- struct stream_interface *si = cs_si(appctx->owner);
- struct channel *req = si_oc(si);
- struct channel *res = si_ic(si);
+ struct conn_stream *cs = appctx->owner;
+ struct channel *req = cs_oc(cs);
+ struct channel *res = cs_ic(cs);
struct htx *req_htx, *res_htx;
struct buffer *errmsg;
unsigned int len;
@@ -1440,12 +1442,12 @@
res_htx = htx_from_buf(&res->buf);
total = res_htx->data;
- if (unlikely(si->state == SI_ST_DIS || si->state == SI_ST_CLO))
+ if (unlikely(cs->si->state == SI_ST_DIS || cs->si->state == SI_ST_CLO))
goto out;
/* Check if the input buffer is available. */
if (!b_size(&res->buf)) {
- si_rx_room_blk(si);
+ si_rx_room_blk(cs->si);
goto out;
}
@@ -1479,7 +1481,7 @@
/* Skip response body for HEAD requests or in case of "304 Not
* Modified" response. */
- if (si_strm(si)->txn->meth == HTTP_METH_HEAD || appctx->ctx.cache.send_notmodified)
+ if (__cs_strm(cs)->txn->meth == HTTP_METH_HEAD || appctx->ctx.cache.send_notmodified)
appctx->st0 = HTX_CACHE_EOM;
else
appctx->st0 = HTX_CACHE_DATA;
@@ -1490,7 +1492,7 @@
if (len) {
ret = htx_cache_dump_msg(appctx, res_htx, len, HTX_BLK_UNUSED);
if (ret < len) {
- si_rx_room_blk(si);
+ si_rx_room_blk(cs->si);
goto out;
}
}
@@ -1500,7 +1502,7 @@
if (appctx->st0 == HTX_CACHE_EOM) {
/* no more data are expected. */
res_htx->flags |= HTX_FL_EOM;
- si->cs->endp->flags |= CS_EP_EOI;
+ cs->endp->flags |= CS_EP_EOI;
res->flags |= CF_EOI;
appctx->st0 = HTX_CACHE_END;
}
@@ -1508,7 +1510,7 @@
end:
if (!(res->flags & CF_SHUTR) && appctx->st0 == HTX_CACHE_END) {
res->flags |= CF_READ_NULL;
- si_shutr(si);
+ si_shutr(cs->si);
}
out:
@@ -2565,7 +2567,7 @@
static int cli_io_handler_show_cache(struct appctx *appctx)
{
struct cache* cache = appctx->ctx.cli.p0;
- struct stream_interface *si = cs_si(appctx->owner);
+ struct conn_stream *cs = appctx->owner;
if (cache == NULL) {
cache = LIST_ELEM((caches).n, typeof(struct cache *), list);
@@ -2580,8 +2582,8 @@
next_key = appctx->ctx.cli.i0;
if (!next_key) {
chunk_printf(&trash, "%p: %s (shctx:%p, available blocks:%d)\n", cache, cache->id, shctx_ptr(cache), shctx_ptr(cache)->nbav);
- if (ci_putchk(si_ic(si), &trash) == -1) {
- si_rx_room_blk(si);
+ if (ci_putchk(cs_ic(cs), &trash) == -1) {
+ si_rx_room_blk(cs->si);
return 0;
}
}
@@ -2618,8 +2620,8 @@
shctx_unlock(shctx_ptr(cache));
- if (ci_putchk(si_ic(si), &trash) == -1) {
- si_rx_room_blk(si);
+ if (ci_putchk(cs_ic(cs), &trash) == -1) {
+ si_rx_room_blk(cs->si);
return 0;
}
}
@@ -2659,7 +2661,7 @@
if (!smp->strm || smp->strm->target != &http_cache_applet.obj_type)
return 0;
- /* Get appctx from the stream_interface. */
+ /* Get appctx from the conn-stream. */
appctx = cs_appctx(smp->strm->csb);
if (appctx && appctx->rule) {
cconf = appctx->rule->arg.act.p[0];
diff --git a/src/cli.c b/src/cli.c
index 665631f..8ebd896 100644
--- a/src/cli.c
+++ b/src/cli.c
@@ -34,6 +34,8 @@
#include <haproxy/channel.h>
#include <haproxy/check.h>
#include <haproxy/cli.h>
+#include <haproxy/conn_stream.h>
+#include <haproxy/cs_utils.h>
#include <haproxy/compression.h>
#include <haproxy/dns-t.h>
#include <haproxy/errors.h>
@@ -706,7 +708,7 @@
{
if (appctx->cli_severity_output)
return appctx->cli_severity_output;
- return strm_li(si_strm(cs_si(appctx->owner)))->bind_conf->severity_output;
+ return strm_li(__cs_strm(appctx->owner))->bind_conf->severity_output;
}
/* Processes the CLI interpreter on the stats socket. This function is called
@@ -874,14 +876,14 @@
*/
static void cli_io_handler(struct appctx *appctx)
{
- struct stream_interface *si = cs_si(appctx->owner);
- struct channel *req = si_oc(si);
- struct channel *res = si_ic(si);
- struct bind_conf *bind_conf = strm_li(si_strm(si))->bind_conf;
+ struct conn_stream *cs = appctx->owner;
+ struct channel *req = cs_oc(cs);
+ struct channel *res = cs_ic(cs);
+ struct bind_conf *bind_conf = strm_li(__cs_strm(cs))->bind_conf;
int reql;
int len;
- if (unlikely(si->state == SI_ST_DIS || si->state == SI_ST_CLO))
+ if (unlikely(cs->si->state == SI_ST_DIS || cs->si->state == SI_ST_CLO))
goto out;
/* Check if the input buffer is available. */
@@ -905,7 +907,7 @@
/* Let's close for real now. We just close the request
* side, the conditions below will complete if needed.
*/
- si_shutw(si);
+ si_shutw(cs->si);
free_trash_chunk(appctx->chunk);
appctx->chunk = NULL;
break;
@@ -927,8 +929,8 @@
/* ensure we have some output room left in the event we
* would want to return some info right after parsing.
*/
- if (buffer_almost_full(si_ib(si))) {
- si_rx_room_blk(si);
+ if (buffer_almost_full(cs_ib(cs))) {
+ si_rx_room_blk(cs->si);
break;
}
@@ -939,10 +941,10 @@
*/
if (appctx->st1 & APPCTX_CLI_ST1_PAYLOAD)
- reql = co_getline(si_oc(si), str,
+ reql = co_getline(cs_oc(cs), str,
appctx->chunk->size - appctx->chunk->data - 1);
else
- reql = co_getdelim(si_oc(si), str,
+ reql = co_getdelim(cs_oc(cs), str,
appctx->chunk->size - appctx->chunk->data - 1,
"\n;", '\\');
@@ -1027,7 +1029,7 @@
}
/* re-adjust req buffer */
- co_skip(si_oc(si), reql);
+ co_skip(cs_oc(cs), reql);
req->flags |= CF_READ_DONTWAIT; /* we plan to read small requests */
}
else { /* output functions */
@@ -1068,7 +1070,7 @@
appctx->st0 = CLI_ST_PROMPT;
}
else
- si_rx_room_blk(si);
+ si_rx_room_blk(cs->si);
break;
case CLI_ST_CALLBACK: /* use custom pointer */
@@ -1082,7 +1084,7 @@
}
break;
default: /* abnormal state */
- si->flags |= SI_FL_ERR;
+ cs->si->flags |= SI_FL_ERR;
break;
}
@@ -1105,10 +1107,10 @@
prompt = "\n";
}
- if (ci_putstr(si_ic(si), prompt) != -1)
+ if (ci_putstr(cs_ic(cs), prompt) != -1)
appctx->st0 = CLI_ST_GETREQ;
else
- si_rx_room_blk(si);
+ si_rx_room_blk(cs->si);
}
/* If the output functions are still there, it means they require more room. */
@@ -1137,39 +1139,39 @@
* refills the buffer with new bytes in non-interactive
* mode, avoiding to close on apparently empty commands.
*/
- if (co_data(si_oc(si))) {
+ if (co_data(cs_oc(cs))) {
appctx_wakeup(appctx);
goto out;
}
}
}
- if ((res->flags & CF_SHUTR) && (si->state == SI_ST_EST)) {
+ if ((res->flags & CF_SHUTR) && (cs->si->state == SI_ST_EST)) {
DPRINTF(stderr, "%s@%d: si to buf closed. req=%08x, res=%08x, st=%d\n",
- __FUNCTION__, __LINE__, req->flags, res->flags, si->state);
+ __FUNCTION__, __LINE__, req->flags, res->flags, cs->si->state);
/* Other side has closed, let's abort if we have no more processing to do
* and nothing more to consume. This is comparable to a broken pipe, so
* we forward the close to the request side so that it flows upstream to
* the client.
*/
- si_shutw(si);
+ si_shutw(cs->si);
}
- if ((req->flags & CF_SHUTW) && (si->state == SI_ST_EST) && (appctx->st0 < CLI_ST_OUTPUT)) {
+ if ((req->flags & CF_SHUTW) && (cs->si->state == SI_ST_EST) && (appctx->st0 < CLI_ST_OUTPUT)) {
DPRINTF(stderr, "%s@%d: buf to si closed. req=%08x, res=%08x, st=%d\n",
- __FUNCTION__, __LINE__, req->flags, res->flags, si->state);
+ __FUNCTION__, __LINE__, req->flags, res->flags, cs->si->state);
/* We have no more processing to do, and nothing more to send, and
* the client side has closed. So we'll forward this state downstream
* on the response buffer.
*/
- si_shutr(si);
+ si_shutr(cs->si);
res->flags |= CF_READ_NULL;
}
out:
DPRINTF(stderr, "%s@%d: st=%d, rqf=%x, rpf=%x, rqh=%lu, rqs=%lu, rh=%lu, rs=%lu\n",
__FUNCTION__, __LINE__,
- si->state, req->flags, res->flags, ci_data(req), co_data(req), ci_data(res), co_data(res));
+ cs->si->state, req->flags, res->flags, ci_data(req), co_data(req), ci_data(res), co_data(res));
}
/* This is called when the stream interface is closed. For instance, upon an
@@ -1197,10 +1199,10 @@
*/
static int cli_io_handler_show_env(struct appctx *appctx)
{
- struct stream_interface *si = cs_si(appctx->owner);
+ struct conn_stream *cs = appctx->owner;
char **var = appctx->ctx.cli.p0;
- if (unlikely(si_ic(si)->flags & (CF_WRITE_ERROR|CF_SHUTW)))
+ if (unlikely(cs_ic(cs)->flags & (CF_WRITE_ERROR|CF_SHUTW)))
return 1;
chunk_reset(&trash);
@@ -1211,8 +1213,8 @@
while (*var) {
chunk_printf(&trash, "%s\n", *var);
- if (ci_putchk(si_ic(si), &trash) == -1) {
- si_rx_room_blk(si);
+ if (ci_putchk(cs_ic(cs), &trash) == -1) {
+ si_rx_room_blk(cs->si);
return 0;
}
if (appctx->st2 == STAT_ST_END)
@@ -1232,11 +1234,11 @@
*/
static int cli_io_handler_show_fd(struct appctx *appctx)
{
- struct stream_interface *si = cs_si(appctx->owner);
+ struct conn_stream *cs = appctx->owner;
int fd = appctx->ctx.cli.i0;
int ret = 1;
- if (unlikely(si_ic(si)->flags & (CF_WRITE_ERROR|CF_SHUTW)))
+ if (unlikely(cs_ic(cs)->flags & (CF_WRITE_ERROR|CF_SHUTW)))
goto end;
chunk_reset(&trash);
@@ -1403,8 +1405,8 @@
#endif
chunk_appendf(&trash, "%s\n", suspicious ? " !" : "");
- if (ci_putchk(si_ic(si), &trash) == -1) {
- si_rx_room_blk(si);
+ if (ci_putchk(cs_ic(cs), &trash) == -1) {
+ si_rx_room_blk(cs->si);
appctx->ctx.cli.i0 = fd;
ret = 0;
break;
@@ -1431,10 +1433,10 @@
*/
static int cli_io_handler_show_activity(struct appctx *appctx)
{
- struct stream_interface *si = cs_si(appctx->owner);
+ struct conn_stream *cs = appctx->owner;
int thr;
- if (unlikely(si_ic(si)->flags & (CF_WRITE_ERROR|CF_SHUTW)))
+ if (unlikely(cs_ic(cs)->flags & (CF_WRITE_ERROR|CF_SHUTW)))
return 1;
chunk_reset(&trash);
@@ -1515,10 +1517,10 @@
chunk_appendf(&trash, "ctr2:"); SHOW_TOT(thr, activity[thr].ctr2);
#endif
- if (ci_putchk(si_ic(si), &trash) == -1) {
+ if (ci_putchk(cs_ic(cs), &trash) == -1) {
chunk_reset(&trash);
chunk_printf(&trash, "[output too large, cannot dump]\n");
- si_rx_room_blk(si);
+ si_rx_room_blk(cs->si);
}
#undef SHOW_AVG
@@ -1534,15 +1536,15 @@
static int cli_io_handler_show_cli_sock(struct appctx *appctx)
{
struct bind_conf *bind_conf;
- struct stream_interface *si = cs_si(appctx->owner);
+ struct conn_stream *cs = appctx->owner;
chunk_reset(&trash);
switch (appctx->st2) {
case STAT_ST_INIT:
chunk_printf(&trash, "# socket lvl processes\n");
- if (ci_putchk(si_ic(si), &trash) == -1) {
- si_rx_room_blk(si);
+ if (ci_putchk(cs_ic(cs), &trash) == -1) {
+ si_rx_room_blk(cs->si);
return 0;
}
appctx->st2 = STAT_ST_LIST;
@@ -1603,8 +1605,8 @@
chunk_appendf(&trash, "all\n");
- if (ci_putchk(si_ic(si), &trash) == -1) {
- si_rx_room_blk(si);
+ if (ci_putchk(cs_ic(cs), &trash) == -1) {
+ si_rx_room_blk(cs->si);
return 0;
}
}
@@ -1671,8 +1673,7 @@
/* parse a "set timeout" CLI request. It always returns 1. */
static int cli_parse_set_timeout(char **args, char *payload, struct appctx *appctx, void *private)
{
- struct stream_interface *si = cs_si(appctx->owner);
- struct stream *s = si_strm(si);
+ struct stream *s = __cs_strm(appctx->owner);
if (strcmp(args[2], "cli") == 0) {
unsigned timeout;
@@ -1955,9 +1956,9 @@
char *cmsgbuf = NULL;
unsigned char *tmpbuf = NULL;
struct cmsghdr *cmsg;
- struct stream_interface *si = cs_si(appctx->owner);
- struct stream *s = si_strm(si);
- struct connection *remote = cs_conn(si_opposite(si)->cs);
+ struct conn_stream *cs = appctx->owner;
+ struct stream *s = __cs_strm(cs);
+ struct connection *remote = cs_conn(cs_opposite(cs));
struct msghdr msghdr;
struct iovec iov;
struct timeval tv = { .tv_sec = 1, .tv_usec = 0 };
@@ -2155,7 +2156,7 @@
void pcli_write_prompt(struct stream *s)
{
struct buffer *msg = get_trash_chunk();
- struct channel *oc = si_oc(cs_si(s->csf));
+ struct channel *oc = cs_oc(s->csf);
if (!(s->pcli_flags & PCLI_F_PROMPT))
return;
diff --git a/src/debug.c b/src/debug.c
index 73fe159..9af8aed 100644
--- a/src/debug.c
+++ b/src/debug.c
@@ -29,6 +29,8 @@
#include <haproxy/api.h>
#include <haproxy/buf.h>
#include <haproxy/cli.h>
+#include <haproxy/conn_stream.h>
+#include <haproxy/cs_utils.h>
#include <haproxy/clock.h>
#include <haproxy/debug.h>
#include <haproxy/fd.h>
@@ -249,9 +251,9 @@
if (task->process == process_stream && task->context)
s = (struct stream *)task->context;
else if (task->process == task_run_applet && task->context)
- s = si_strm(cs_si(((struct appctx *)task->context)->owner));
+ s = cs_strm(((struct appctx *)task->context)->owner);
else if (task->process == si_cs_io_cb && task->context)
- s = si_strm((struct stream_interface *)task->context);
+ s = cs_strm(((struct stream_interface *)task->context)->cs);
if (s)
stream_dump(buf, s, pfx, '\n');
@@ -288,10 +290,10 @@
*/
static int cli_io_handler_show_threads(struct appctx *appctx)
{
- struct stream_interface *si = cs_si(appctx->owner);
+ struct conn_stream *cs = appctx->owner;
int thr;
- if (unlikely(si_ic(si)->flags & (CF_WRITE_ERROR|CF_SHUTW)))
+ if (unlikely(cs_ic(cs)->flags & (CF_WRITE_ERROR|CF_SHUTW)))
return 1;
if (appctx->st0)
@@ -302,9 +304,9 @@
chunk_reset(&trash);
ha_thread_dump_all_to_trash();
- if (ci_putchk(si_ic(si), &trash) == -1) {
+ if (ci_putchk(cs_ic(cs), &trash) == -1) {
/* failed, try again */
- si_rx_room_blk(si);
+ si_rx_room_blk(cs->si);
appctx->st1 = thr;
return 0;
}
@@ -674,7 +676,7 @@
*/
static int debug_parse_cli_stream(char **args, char *payload, struct appctx *appctx, void *private)
{
- struct stream *s = si_strm(cs_si(appctx->owner));
+ struct stream *s = __cs_strm(appctx->owner);
int arg;
void *ptr;
int size;
@@ -1040,7 +1042,7 @@
*/
static int debug_iohandler_fd(struct appctx *appctx)
{
- struct stream_interface *si = cs_si(appctx->owner);
+ struct conn_stream *cs = appctx->owner;
struct sockaddr_storage sa;
struct stat statbuf;
socklen_t salen, vlen;
@@ -1049,7 +1051,7 @@
int ret = 1;
int i, fd;
- if (unlikely(si_ic(si)->flags & (CF_WRITE_ERROR|CF_SHUTW)))
+ if (unlikely(cs_ic(cs)->flags & (CF_WRITE_ERROR|CF_SHUTW)))
goto end;
chunk_reset(&trash);
@@ -1173,8 +1175,8 @@
chunk_appendf(&trash, "\n");
- if (ci_putchk(si_ic(si), &trash) == -1) {
- si_rx_room_blk(si);
+ if (ci_putchk(cs_ic(cs), &trash) == -1) {
+ si_rx_room_blk(cs->si);
appctx->ctx.cli.i0 = fd;
ret = 0;
break;
@@ -1224,11 +1226,11 @@
*/
static int debug_iohandler_memstats(struct appctx *appctx)
{
- struct stream_interface *si = cs_si(appctx->owner);
+ struct conn_stream *cs = appctx->owner;
struct mem_stats *ptr = appctx->ctx.cli.p0;
int ret = 1;
- if (unlikely(si_ic(si)->flags & (CF_WRITE_ERROR|CF_SHUTW)))
+ if (unlikely(cs_ic(cs)->flags & (CF_WRITE_ERROR|CF_SHUTW)))
goto end;
chunk_reset(&trash);
@@ -1273,8 +1275,8 @@
(unsigned long)ptr->size, (unsigned long)ptr->calls,
(unsigned long)(ptr->calls ? (ptr->size / ptr->calls) : 0));
- if (ci_putchk(si_ic(si), &trash) == -1) {
- si_rx_room_blk(si);
+ if (ci_putchk(cs_ic(cs), &trash) == -1) {
+ si_rx_room_blk(cs->si);
appctx->ctx.cli.p0 = ptr;
ret = 0;
break;
diff --git a/src/dns.c b/src/dns.c
index 7228581..755053a 100644
--- a/src/dns.c
+++ b/src/dns.c
@@ -25,6 +25,8 @@
#include <haproxy/channel.h>
#include <haproxy/check.h>
#include <haproxy/cli.h>
+#include <haproxy/conn_stream.h>
+#include <haproxy/cs_utils.h>
#include <haproxy/dgram.h>
#include <haproxy/dns.h>
#include <haproxy/errors.h>
@@ -407,7 +409,7 @@
*/
static void dns_session_io_handler(struct appctx *appctx)
{
- struct stream_interface *si = cs_si(appctx->owner);
+ struct conn_stream *cs = appctx->owner;
struct dns_session *ds = appctx->ctx.sft.ptr;
struct ring *ring = &ds->ring;
struct buffer *buf = &ring->buf;
@@ -429,21 +431,21 @@
goto close;
/* an error was detected */
- if (unlikely(si_ic(si)->flags & (CF_WRITE_ERROR|CF_SHUTW)))
+ if (unlikely(cs_ic(cs)->flags & (CF_WRITE_ERROR|CF_SHUTW)))
goto close;
/* con closed by server side, we will skip data write and drain data from channel */
- if ((si_oc(si)->flags & CF_SHUTW)) {
+ if ((cs_oc(cs)->flags & CF_SHUTW)) {
goto read;
}
/* if the connection is not established, inform the stream that we want
* to be notified whenever the connection completes.
*/
- if (si_opposite(si)->state < SI_ST_EST) {
- si_cant_get(si);
- si_rx_conn_blk(si);
- si_rx_endp_more(si);
+ if (cs_opposite(cs)->si->state < SI_ST_EST) {
+ si_cant_get(cs->si);
+ si_rx_conn_blk(cs->si);
+ si_rx_endp_more(cs->si);
return;
}
@@ -475,7 +477,7 @@
* the message so that we can take our reference there if we have to
* stop before the end (ret=0).
*/
- if (si_opposite(si)->state == SI_ST_EST) {
+ if (cs_opposite(cs)->si->state == SI_ST_EST) {
/* we were already there, adjust the offset to be relative to
* the buffer's head and remove us from the counter.
*/
@@ -497,7 +499,7 @@
BUG_ON(msg_len + ofs + cnt + 1 > b_data(buf));
/* retrieve available room on output channel */
- available_room = channel_recv_max(si_ic(si));
+ available_room = channel_recv_max(cs_ic(cs));
/* tx_msg_offset null means we are at the start of a new message */
if (!ds->tx_msg_offset) {
@@ -505,7 +507,7 @@
/* check if there is enough room to put message len and query id */
if (available_room < sizeof(slen) + sizeof(new_qid)) {
- si_rx_room_blk(si);
+ si_rx_room_blk(cs->si);
ret = 0;
break;
}
@@ -513,7 +515,7 @@
/* put msg len into then channel */
slen = (uint16_t)msg_len;
slen = htons(slen);
- ci_putblk(si_ic(si), (char *)&slen, sizeof(slen));
+ ci_putblk(cs_ic(cs), (char *)&slen, sizeof(slen));
available_room -= sizeof(slen);
/* backup original query id */
@@ -531,7 +533,7 @@
new_qid = htons(new_qid);
/* put new query id into the channel */
- ci_putblk(si_ic(si), (char *)&new_qid, sizeof(new_qid));
+ ci_putblk(cs_ic(cs), (char *)&new_qid, sizeof(new_qid));
available_room -= sizeof(new_qid);
/* keep query id mapping */
@@ -563,7 +565,7 @@
/* check if it remains available room on output chan */
if (unlikely(!available_room)) {
- si_rx_room_blk(si);
+ si_rx_room_blk(cs->si);
ret = 0;
break;
}
@@ -586,12 +588,12 @@
}
trash.data += len;
- if (ci_putchk(si_ic(si), &trash) == -1) {
+ if (ci_putchk(cs_ic(cs), &trash) == -1) {
/* should never happen since we
* check available_room is large
* enough here.
*/
- si_rx_room_blk(si);
+ si_rx_room_blk(cs->si);
ret = 0;
break;
}
@@ -599,7 +601,7 @@
if (ds->tx_msg_offset) {
/* msg was not fully processed, we must be awake to drain pending data */
- si_rx_room_blk(si);
+ si_rx_room_blk(cs->si);
ret = 0;
break;
}
@@ -619,7 +621,7 @@
BUG_ON(LIST_INLIST(&appctx->wait_entry));
LIST_APPEND(&ring->waiters, &appctx->wait_entry);
HA_RWLOCK_WRUNLOCK(DNS_LOCK, &ring->lock);
- si_rx_endp_done(si);
+ si_rx_endp_done(cs->si);
}
read:
@@ -639,35 +641,35 @@
if (!ds->rx_msg.len) {
/* next message len is not fully available into the channel */
- if (co_data(si_oc(si)) < 2)
+ if (co_data(cs_oc(cs)) < 2)
break;
/* retrieve message len */
- co_getblk(si_oc(si), (char *)&msg_len, 2, 0);
+ co_getblk(cs_oc(cs), (char *)&msg_len, 2, 0);
/* mark as consumed */
- co_skip(si_oc(si), 2);
+ co_skip(cs_oc(cs), 2);
/* store message len */
ds->rx_msg.len = ntohs(msg_len);
}
- if (!co_data(si_oc(si))) {
+ if (!co_data(cs_oc(cs))) {
/* we need more data but nothing is available */
break;
}
- if (co_data(si_oc(si)) + ds->rx_msg.offset < ds->rx_msg.len) {
+ if (co_data(cs_oc(cs)) + ds->rx_msg.offset < ds->rx_msg.len) {
/* message only partially available */
/* read available data */
- co_getblk(si_oc(si), ds->rx_msg.area + ds->rx_msg.offset, co_data(si_oc(si)), 0);
+ co_getblk(cs_oc(cs), ds->rx_msg.area + ds->rx_msg.offset, co_data(cs_oc(cs)), 0);
/* update message offset */
- ds->rx_msg.offset += co_data(si_oc(si));
+ ds->rx_msg.offset += co_data(cs_oc(cs));
/* consume all pending data from the channel */
- co_skip(si_oc(si), co_data(si_oc(si)));
+ co_skip(cs_oc(cs), co_data(cs_oc(cs)));
/* we need to wait for more data */
break;
@@ -676,10 +678,10 @@
/* enough data is available into the channel to read the message until the end */
/* read from the channel until the end of the message */
- co_getblk(si_oc(si), ds->rx_msg.area + ds->rx_msg.offset, ds->rx_msg.len - ds->rx_msg.offset, 0);
+ co_getblk(cs_oc(cs), ds->rx_msg.area + ds->rx_msg.offset, ds->rx_msg.len - ds->rx_msg.offset, 0);
/* consume all data until the end of the message from the channel */
- co_skip(si_oc(si), ds->rx_msg.len - ds->rx_msg.offset);
+ co_skip(cs_oc(cs), ds->rx_msg.len - ds->rx_msg.offset);
/* reset reader offset to 0 for next message reand */
ds->rx_msg.offset = 0;
@@ -725,7 +727,7 @@
if (!LIST_INLIST(&ds->waiter)) {
/* there is no more pending data to read and the con was closed by the server side */
- if (!co_data(si_oc(si)) && (si_oc(si)->flags & CF_SHUTW)) {
+ if (!co_data(cs_oc(cs)) && (cs_oc(cs)->flags & CF_SHUTW)) {
goto close;
}
}
@@ -734,9 +736,9 @@
return;
close:
- si_shutw(si);
- si_shutr(si);
- si_ic(si)->flags |= CF_READ_NULL;
+ si_shutw(cs->si);
+ si_shutr(cs->si);
+ cs_ic(cs)->flags |= CF_READ_NULL;
}
void dns_queries_flush(struct dns_session *ds)
diff --git a/src/flt_spoe.c b/src/flt_spoe.c
index 1e731b1..619043b 100644
--- a/src/flt_spoe.c
+++ b/src/flt_spoe.c
@@ -18,6 +18,8 @@
#include <haproxy/arg.h>
#include <haproxy/cfgparse.h>
#include <haproxy/check.h>
+#include <haproxy/conn_stream.h>
+#include <haproxy/cs_utils.h>
#include <haproxy/filters.h>
#include <haproxy/freq_ctr.h>
#include <haproxy/frontend.h>
@@ -1135,7 +1137,7 @@
static int
spoe_send_frame(struct appctx *appctx, char *buf, size_t framesz)
{
- struct stream_interface *si = cs_si(appctx->owner);
+ struct conn_stream *cs = appctx->owner;
int ret;
uint32_t netint;
@@ -1143,10 +1145,10 @@
* length. */
netint = htonl(framesz);
memcpy(buf, (char *)&netint, 4);
- ret = ci_putblk(si_ic(si), buf, framesz+4);
+ ret = ci_putblk(cs_ic(cs), buf, framesz+4);
if (ret <= 0) {
- if ((ret == -3 && b_is_null(&si_ic(si)->buf)) || ret == -1) {
- si_rx_room_blk(si);
+ if ((ret == -3 && b_is_null(&cs_ic(cs)->buf)) || ret == -1) {
+ si_rx_room_blk(cs->si);
return 1; /* retry */
}
SPOE_APPCTX(appctx)->status_code = SPOE_FRM_ERR_IO;
@@ -1161,18 +1163,18 @@
static int
spoe_recv_frame(struct appctx *appctx, char *buf, size_t framesz)
{
- struct stream_interface *si = cs_si(appctx->owner);
+ struct conn_stream *cs = appctx->owner;
int ret;
uint32_t netint;
- ret = co_getblk(si_oc(si), (char *)&netint, 4, 0);
+ ret = co_getblk(cs_oc(cs), (char *)&netint, 4, 0);
if (ret > 0) {
framesz = ntohl(netint);
if (framesz > SPOE_APPCTX(appctx)->max_frame_size) {
SPOE_APPCTX(appctx)->status_code = SPOE_FRM_ERR_TOO_BIG;
return -1;
}
- ret = co_getblk(si_oc(si), buf, framesz, 4);
+ ret = co_getblk(cs_oc(cs), buf, framesz, 4);
}
if (ret <= 0) {
if (ret == 0) {
@@ -1217,10 +1219,10 @@
static void
spoe_release_appctx(struct appctx *appctx)
{
- struct stream_interface *si = cs_si(appctx->owner);
- struct spoe_appctx *spoe_appctx = SPOE_APPCTX(appctx);
- struct spoe_agent *agent;
- struct spoe_context *ctx, *back;
+ struct conn_stream *cs = appctx->owner;
+ struct spoe_appctx *spoe_appctx = SPOE_APPCTX(appctx);
+ struct spoe_agent *agent;
+ struct spoe_context *ctx, *back;
if (spoe_appctx == NULL)
return;
@@ -1252,9 +1254,9 @@
if (spoe_appctx->status_code == SPOE_FRM_ERR_NONE)
spoe_appctx->status_code = SPOE_FRM_ERR_IO;
- si_shutw(si);
- si_shutr(si);
- si_ic(si)->flags |= CF_READ_NULL;
+ si_shutw(cs->si);
+ si_shutr(cs->si);
+ cs_ic(cs)->flags |= CF_READ_NULL;
}
/* Destroy the task attached to this applet */
@@ -1337,21 +1339,21 @@
static int
spoe_handle_connect_appctx(struct appctx *appctx)
{
- struct stream_interface *si = cs_si(appctx->owner);
- struct spoe_agent *agent = SPOE_APPCTX(appctx)->agent;
+ struct conn_stream *cs = appctx->owner;
+ struct spoe_agent *agent = SPOE_APPCTX(appctx)->agent;
char *frame, *buf;
int ret;
- if (si_state_in(si->state, SI_SB_CER|SI_SB_DIS|SI_SB_CLO)) {
+ if (si_state_in(cs->si->state, SI_SB_CER|SI_SB_DIS|SI_SB_CLO)) {
/* closed */
SPOE_APPCTX(appctx)->status_code = SPOE_FRM_ERR_IO;
goto exit;
}
- if (!si_state_in(si->state, SI_SB_RDY|SI_SB_EST)) {
+ if (!si_state_in(cs->si->state, SI_SB_RDY|SI_SB_EST)) {
/* not connected yet */
- si_rx_endp_more(si);
- task_wakeup(si_strm(si)->task, TASK_WOKEN_MSG);
+ si_rx_endp_more(cs->si);
+ task_wakeup(__cs_strm(cs)->task, TASK_WOKEN_MSG);
goto stop;
}
@@ -1403,13 +1405,13 @@
static int
spoe_handle_connecting_appctx(struct appctx *appctx)
{
- struct stream_interface *si = cs_si(appctx->owner);
- struct spoe_agent *agent = SPOE_APPCTX(appctx)->agent;
+ struct conn_stream *cs = appctx->owner;
+ struct spoe_agent *agent = SPOE_APPCTX(appctx)->agent;
char *frame;
int ret;
- if (si->state == SI_ST_CLO || si_opposite(si)->state == SI_ST_CLO) {
+ if (cs->si->state == SI_ST_CLO || cs_opposite(cs)->si->state == SI_ST_CLO) {
SPOE_APPCTX(appctx)->status_code = SPOE_FRM_ERR_IO;
goto exit;
}
@@ -1458,7 +1460,7 @@
next:
/* Do not forget to remove processed frame from the output buffer */
if (trash.data)
- co_skip(si_oc(si), trash.data);
+ co_skip(cs_oc(cs), trash.data);
SPOE_APPCTX(appctx)->task->expire =
tick_add_ifset(now_ms, agent->timeout.idle);
@@ -1648,7 +1650,7 @@
/* Do not forget to remove processed frame from the output buffer */
if (trash.data)
- co_skip(si_oc(cs_si(appctx->owner)), trash.data);
+ co_skip(cs_oc(appctx->owner), trash.data);
end:
return ret;
}
@@ -1656,12 +1658,12 @@
static int
spoe_handle_processing_appctx(struct appctx *appctx)
{
- struct stream_interface *si = cs_si(appctx->owner);
- struct server *srv = objt_server(si_strm(si)->target);
+ struct conn_stream *cs = appctx->owner;
+ struct server *srv = objt_server(__cs_strm(cs)->target);
struct spoe_agent *agent = SPOE_APPCTX(appctx)->agent;
int ret, skip_sending = 0, skip_receiving = 0, active_s = 0, active_r = 0, close_asap = 0;
- if (si->state == SI_ST_CLO || si_opposite(si)->state == SI_ST_CLO) {
+ if (cs->si->state == SI_ST_CLO || cs_opposite(cs)->si->state == SI_ST_CLO) {
SPOE_APPCTX(appctx)->status_code = SPOE_FRM_ERR_IO;
goto exit;
}
@@ -1779,12 +1781,12 @@
static int
spoe_handle_disconnect_appctx(struct appctx *appctx)
{
- struct stream_interface *si = cs_si(appctx->owner);
- struct spoe_agent *agent = SPOE_APPCTX(appctx)->agent;
+ struct conn_stream *cs = appctx->owner;
+ struct spoe_agent *agent = SPOE_APPCTX(appctx)->agent;
char *frame, *buf;
int ret;
- if (si->state == SI_ST_CLO || si_opposite(si)->state == SI_ST_CLO)
+ if (cs->si->state == SI_ST_CLO || cs_opposite(cs)->si->state == SI_ST_CLO)
goto exit;
if (appctx->st1 == SPOE_APPCTX_ERR_TOUT)
@@ -1832,11 +1834,11 @@
static int
spoe_handle_disconnecting_appctx(struct appctx *appctx)
{
- struct stream_interface *si = cs_si(appctx->owner);
+ struct conn_stream *cs = appctx->owner;
char *frame;
int ret;
- if (si->state == SI_ST_CLO || si_opposite(si)->state == SI_ST_CLO) {
+ if (cs->si->state == SI_ST_CLO || cs_opposite(cs)->si->state == SI_ST_CLO) {
SPOE_APPCTX(appctx)->status_code = SPOE_FRM_ERR_IO;
goto exit;
}
@@ -1883,7 +1885,7 @@
next:
/* Do not forget to remove processed frame from the output buffer */
if (trash.data)
- co_skip(si_oc(cs_si(appctx->owner)), trash.data);
+ co_skip(cs_oc(cs), trash.data);
return 0;
stop:
@@ -1897,8 +1899,8 @@
static void
spoe_handle_appctx(struct appctx *appctx)
{
- struct stream_interface *si = cs_si(appctx->owner);
- struct spoe_agent *agent;
+ struct conn_stream *cs = appctx->owner;
+ struct spoe_agent *agent;
if (SPOE_APPCTX(appctx) == NULL)
return;
@@ -1958,9 +1960,9 @@
appctx->st0 = SPOE_APPCTX_ST_END;
SPOE_APPCTX(appctx)->task->expire = TICK_ETERNITY;
- si_shutw(si);
- si_shutr(si);
- si_ic(si)->flags |= CF_READ_NULL;
+ si_shutw(cs->si);
+ si_shutr(cs->si);
+ cs_ic(cs)->flags |= CF_READ_NULL;
/* fall through */
case SPOE_APPCTX_ST_END:
diff --git a/src/hlua.c b/src/hlua.c
index 931e741..380e61b 100644
--- a/src/hlua.c
+++ b/src/hlua.c
@@ -34,6 +34,8 @@
#include <haproxy/cli.h>
#include <haproxy/clock.h>
#include <haproxy/connection.h>
+#include <haproxy/conn_stream.h>
+#include <haproxy/cs_utils.h>
#include <haproxy/filters.h>
#include <haproxy/h1.h>
#include <haproxy/hlua.h>
@@ -1907,32 +1909,32 @@
*/
static void hlua_socket_handler(struct appctx *appctx)
{
- struct stream_interface *si = cs_si(appctx->owner);
+ struct conn_stream *cs = appctx->owner;
if (appctx->ctx.hlua_cosocket.die) {
- si_shutw(si);
- si_shutr(si);
- si_ic(si)->flags |= CF_READ_NULL;
+ si_shutw(cs->si);
+ si_shutr(cs->si);
+ cs_ic(cs)->flags |= CF_READ_NULL;
notification_wake(&appctx->ctx.hlua_cosocket.wake_on_read);
notification_wake(&appctx->ctx.hlua_cosocket.wake_on_write);
- stream_shutdown(si_strm(si), SF_ERR_KILLED);
+ stream_shutdown(__cs_strm(cs), SF_ERR_KILLED);
}
/* If we can't write, wakeup the pending write signals. */
- if (channel_output_closed(si_ic(si)))
+ if (channel_output_closed(cs_ic(cs)))
notification_wake(&appctx->ctx.hlua_cosocket.wake_on_write);
/* If we can't read, wakeup the pending read signals. */
- if (channel_input_closed(si_oc(si)))
+ if (channel_input_closed(cs_oc(cs)))
notification_wake(&appctx->ctx.hlua_cosocket.wake_on_read);
/* if the connection is not established, inform the stream that we want
* to be notified whenever the connection completes.
*/
- if (si_opposite(si)->state < SI_ST_EST) {
- si_cant_get(si);
- si_rx_conn_blk(si);
- si_rx_endp_more(si);
+ if (cs_opposite(cs)->si->state < SI_ST_EST) {
+ si_cant_get(cs->si);
+ si_rx_conn_blk(cs->si);
+ si_rx_endp_more(cs->si);
return;
}
@@ -1940,24 +1942,24 @@
appctx->ctx.hlua_cosocket.connected = 1;
/* Wake the tasks which wants to write if the buffer have available space. */
- if (channel_may_recv(si_ic(si)))
+ if (channel_may_recv(cs_ic(cs)))
notification_wake(&appctx->ctx.hlua_cosocket.wake_on_write);
/* Wake the tasks which wants to read if the buffer contains data. */
- if (!channel_is_empty(si_oc(si)))
+ if (!channel_is_empty(cs_oc(cs)))
notification_wake(&appctx->ctx.hlua_cosocket.wake_on_read);
/* Some data were injected in the buffer, notify the stream
* interface.
*/
- if (!channel_is_empty(si_ic(si)))
- si_update(si);
+ if (!channel_is_empty(cs_ic(cs)))
+ si_update(cs->si);
/* If write notifications are registered, we considers we want
* to write, so we clear the blocking flag.
*/
if (notification_registered(&appctx->ctx.hlua_cosocket.wake_on_write))
- si_rx_endp_more(si);
+ si_rx_endp_more(cs->si);
}
/* This function is called when the "struct stream" is destroyed.
@@ -2079,7 +2081,6 @@
size_t len2;
int skip_at_end = 0;
struct channel *oc;
- struct stream_interface *si;
struct stream *s;
struct xref *peer;
int missing_bytes;
@@ -2103,8 +2104,7 @@
if (!peer)
goto no_peer;
appctx = container_of(peer, struct appctx, ctx.hlua_cosocket.xref);
- si = cs_si(appctx->owner);
- s = si_strm(si);
+ s = __cs_strm(appctx->owner);
oc = &s->res;
if (wanted == HLSR_READ_LINE) {
@@ -2313,8 +2313,8 @@
int send_len;
int sent;
struct xref *peer;
- struct stream_interface *si;
struct stream *s;
+ struct conn_stream *cs;
/* Get hlua struct, or NULL if we execute from main lua state */
hlua = hlua_gethlua(L);
@@ -2342,8 +2342,8 @@
return 1;
}
appctx = container_of(peer, struct appctx, ctx.hlua_cosocket.xref);
- si = cs_si(appctx->owner);
- s = si_strm(si);
+ cs = appctx->owner;
+ s = __cs_strm(cs);
/* Check for connection close. */
if (channel_output_closed(&s->req)) {
@@ -2366,7 +2366,7 @@
* the request buffer if its not required.
*/
if (s->req.buf.size == 0) {
- if (!si_alloc_ibuf(si, &appctx->buffer_wait))
+ if (!si_alloc_ibuf(cs->si, &appctx->buffer_wait))
goto hlua_socket_write_yield_return;
}
@@ -2554,7 +2554,7 @@
struct hlua_socket *socket;
struct xref *peer;
struct appctx *appctx;
- struct stream_interface *si;
+ struct conn_stream *cs;
const struct sockaddr_storage *dst;
int ret;
@@ -2575,8 +2575,8 @@
return 1;
}
appctx = container_of(peer, struct appctx, ctx.hlua_cosocket.xref);
- si = cs_si(appctx->owner);
- dst = si_dst(si_opposite(si));
+ cs = appctx->owner;
+ dst = si_dst(cs_opposite(cs)->si);
if (!dst) {
xref_unlock(&socket->xref, peer);
lua_pushnil(L);
@@ -2595,7 +2595,6 @@
struct connection *conn;
struct appctx *appctx;
struct xref *peer;
- struct stream_interface *si;
struct stream *s;
int ret;
@@ -2616,8 +2615,7 @@
return 1;
}
appctx = container_of(peer, struct appctx, ctx.hlua_cosocket.xref);
- si = cs_si(appctx->owner);
- s = si_strm(si);
+ s = __cs_strm(appctx->owner);
conn = cs_conn(s->csb);
if (!conn || !conn_get_src(conn)) {
@@ -2645,7 +2643,6 @@
struct hlua *hlua;
struct xref *peer;
struct appctx *appctx;
- struct stream_interface *si;
struct stream *s;
/* Get hlua struct, or NULL if we execute from main lua state */
@@ -2667,8 +2664,7 @@
return 2;
}
appctx = container_of(peer, struct appctx, ctx.hlua_cosocket.xref);
- si = cs_si(appctx->owner);
- s = si_strm(si);
+ s = __cs_strm(appctx->owner);
/* Check if we run on the same thread than the xreator thread.
* We cannot access to the socket if the thread is different.
@@ -2715,7 +2711,7 @@
int low, high;
struct sockaddr_storage *addr;
struct xref *peer;
- struct stream_interface *si;
+ struct conn_stream *cs;
struct stream *s;
if (lua_gettop(L) < 2)
@@ -2779,10 +2775,10 @@
}
appctx = container_of(peer, struct appctx, ctx.hlua_cosocket.xref);
- si = cs_si(appctx->owner);
- s = si_strm(si);
+ cs = appctx->owner;
+ s = __cs_strm(cs);
- if (!sockaddr_alloc(&si_opposite(si)->dst, addr, sizeof(*addr))) {
+ if (!sockaddr_alloc(&cs_opposite(cs)->si->dst, addr, sizeof(*addr))) {
xref_unlock(&socket->xref, peer);
WILL_LJMP(luaL_error(L, "connect: internal error"));
}
@@ -2822,7 +2818,6 @@
struct hlua_socket *socket;
struct xref *peer;
struct appctx *appctx;
- struct stream_interface *si;
struct stream *s;
MAY_LJMP(check_args(L, 3, "connect_ssl"));
@@ -2835,8 +2830,7 @@
return 1;
}
appctx = container_of(peer, struct appctx, ctx.hlua_cosocket.xref);
- si = cs_si(appctx->owner);
- s = si_strm(si);
+ s = __cs_strm(appctx->owner);
s->target = &socket_ssl->obj_type;
xref_unlock(&socket->xref, peer);
@@ -2856,7 +2850,6 @@
double dtmout;
struct xref *peer;
struct appctx *appctx;
- struct stream_interface *si;
struct stream *s;
MAY_LJMP(check_args(L, 2, "settimeout"));
@@ -2891,8 +2884,7 @@
return 0;
}
appctx = container_of(peer, struct appctx, ctx.hlua_cosocket.xref);
- si = cs_si(appctx->owner);
- s = si_strm(si);
+ s = __cs_strm(appctx->owner);
s->sess->fe->timeout.connect = tmout;
s->req.rto = tmout;
@@ -4265,8 +4257,7 @@
static int hlua_applet_tcp_new(lua_State *L, struct appctx *ctx)
{
struct hlua_appctx *luactx;
- struct stream_interface *si = cs_si(ctx->owner);
- struct stream *s = si_strm(si);
+ struct stream *s = __cs_strm(ctx->owner);
struct proxy *p;
ALREADY_CHECKED(s);
@@ -4449,7 +4440,7 @@
__LJMP static int hlua_applet_tcp_getline_yield(lua_State *L, int status, lua_KContext ctx)
{
struct hlua_appctx *luactx = MAY_LJMP(hlua_checkapplet_tcp(L, 1));
- struct stream_interface *si = cs_si(luactx->appctx->owner);
+ struct conn_stream *cs = luactx->appctx->owner;
int ret;
const char *blk1;
size_t len1;
@@ -4457,11 +4448,11 @@
size_t len2;
/* Read the maximum amount of data available. */
- ret = co_getline_nc(si_oc(si), &blk1, &len1, &blk2, &len2);
+ ret = co_getline_nc(cs_oc(cs), &blk1, &len1, &blk2, &len2);
/* Data not yet available. return yield. */
if (ret == 0) {
- si_cant_get(si);
+ si_cant_get(cs->si);
MAY_LJMP(hlua_yieldk(L, 0, 0, hlua_applet_tcp_getline_yield, TICK_ETERNITY, 0));
}
@@ -4480,7 +4471,7 @@
luaL_addlstring(&luactx->b, blk2, len2);
/* Consume input channel output buffer data. */
- co_skip(si_oc(si), len1 + len2);
+ co_skip(cs_oc(cs), len1 + len2);
luaL_pushresult(&luactx->b);
return 1;
}
@@ -4503,7 +4494,7 @@
__LJMP static int hlua_applet_tcp_recv_yield(lua_State *L, int status, lua_KContext ctx)
{
struct hlua_appctx *luactx = MAY_LJMP(hlua_checkapplet_tcp(L, 1));
- struct stream_interface *si = cs_si(luactx->appctx->owner);
+ struct conn_stream *cs = luactx->appctx->owner;
size_t len = MAY_LJMP(luaL_checkinteger(L, 2));
int ret;
const char *blk1;
@@ -4512,11 +4503,11 @@
size_t len2;
/* Read the maximum amount of data available. */
- ret = co_getblk_nc(si_oc(si), &blk1, &len1, &blk2, &len2);
+ ret = co_getblk_nc(cs_oc(cs), &blk1, &len1, &blk2, &len2);
/* Data not yet available. return yield. */
if (ret == 0) {
- si_cant_get(si);
+ si_cant_get(cs->si);
MAY_LJMP(hlua_yieldk(L, 0, 0, hlua_applet_tcp_recv_yield, TICK_ETERNITY, 0));
}
@@ -4538,8 +4529,8 @@
*/
luaL_addlstring(&luactx->b, blk1, len1);
luaL_addlstring(&luactx->b, blk2, len2);
- co_skip(si_oc(si), len1 + len2);
- si_cant_get(si);
+ co_skip(cs_oc(cs), len1 + len2);
+ si_cant_get(cs->si);
MAY_LJMP(hlua_yieldk(L, 0, 0, hlua_applet_tcp_recv_yield, TICK_ETERNITY, 0));
} else {
@@ -4557,13 +4548,13 @@
len -= len2;
/* Consume input channel output buffer data. */
- co_skip(si_oc(si), len1 + len2);
+ co_skip(cs_oc(cs), len1 + len2);
/* If there is no other data available, yield waiting for new data. */
if (len > 0) {
lua_pushinteger(L, len);
lua_replace(L, 2);
- si_cant_get(si);
+ si_cant_get(cs->si);
MAY_LJMP(hlua_yieldk(L, 0, 0, hlua_applet_tcp_recv_yield, TICK_ETERNITY, 0));
}
@@ -4611,8 +4602,8 @@
struct hlua_appctx *luactx = MAY_LJMP(hlua_checkapplet_tcp(L, 1));
const char *str = MAY_LJMP(luaL_checklstring(L, 2, &len));
int l = MAY_LJMP(luaL_checkinteger(L, 3));
- struct stream_interface *si = cs_si(luactx->appctx->owner);
- struct channel *chn = si_ic(si);
+ struct conn_stream *cs = luactx->appctx->owner;
+ struct channel *chn = cs_ic(cs);
int max;
/* Get the max amount of data which can write as input in the channel. */
@@ -4632,7 +4623,7 @@
* applet, and returns a yield.
*/
if (l < len) {
- si_rx_room_blk(si);
+ si_rx_room_blk(cs->si);
MAY_LJMP(hlua_yieldk(L, 0, 0, hlua_applet_tcp_send_yield, TICK_ETERNITY, 0));
}
@@ -4674,8 +4665,7 @@
{
struct hlua_appctx *luactx;
struct hlua_txn htxn;
- struct stream_interface *si = cs_si(ctx->owner);
- struct stream *s = si_strm(si);
+ struct stream *s = __cs_strm(ctx->owner);
struct proxy *px = s->be;
struct htx *htx;
struct htx_blk *blk;
@@ -4936,8 +4926,8 @@
__LJMP static int hlua_applet_http_getline_yield(lua_State *L, int status, lua_KContext ctx)
{
struct hlua_appctx *luactx = MAY_LJMP(hlua_checkapplet_http(L, 1));
- struct stream_interface *si = cs_si(luactx->appctx->owner);
- struct channel *req = si_oc(si);
+ struct conn_stream *cs = luactx->appctx->owner;
+ struct channel *req = cs_oc(cs);
struct htx *htx;
struct htx_blk *blk;
size_t count;
@@ -5003,7 +4993,7 @@
htx_to_buf(htx, &req->buf);
if (!stop) {
- si_cant_get(si);
+ si_cant_get(cs->si);
MAY_LJMP(hlua_yieldk(L, 0, 0, hlua_applet_http_getline_yield, TICK_ETERNITY, 0));
}
@@ -5031,8 +5021,8 @@
__LJMP static int hlua_applet_http_recv_yield(lua_State *L, int status, lua_KContext ctx)
{
struct hlua_appctx *luactx = MAY_LJMP(hlua_checkapplet_http(L, 1));
- struct stream_interface *si = cs_si(luactx->appctx->owner);
- struct channel *req = si_oc(si);
+ struct conn_stream *cs = luactx->appctx->owner;
+ struct channel *req = cs_oc(cs);
struct htx *htx;
struct htx_blk *blk;
size_t count;
@@ -5101,7 +5091,7 @@
lua_pushinteger(L, len);
lua_replace(L, 2);
}
- si_cant_get(si);
+ si_cant_get(cs->si);
MAY_LJMP(hlua_yieldk(L, 0, 0, hlua_applet_http_recv_yield, TICK_ETERNITY, 0));
}
@@ -5140,8 +5130,8 @@
__LJMP static int hlua_applet_http_send_yield(lua_State *L, int status, lua_KContext ctx)
{
struct hlua_appctx *luactx = MAY_LJMP(hlua_checkapplet_http(L, 1));
- struct stream_interface *si = cs_si(luactx->appctx->owner);
- struct channel *res = si_ic(si);
+ struct conn_stream *cs = luactx->appctx->owner;
+ struct channel *res = cs_ic(cs);
struct htx *htx = htx_from_buf(&res->buf);
const char *data;
size_t len;
@@ -5173,7 +5163,7 @@
if (l < len) {
snd_yield:
htx_to_buf(htx, &res->buf);
- si_rx_room_blk(si);
+ si_rx_room_blk(cs->si);
MAY_LJMP(hlua_yieldk(L, 0, 0, hlua_applet_http_send_yield, TICK_ETERNITY, 0));
}
@@ -5274,8 +5264,8 @@
__LJMP static int hlua_applet_http_send_response(lua_State *L)
{
struct hlua_appctx *luactx = MAY_LJMP(hlua_checkapplet_http(L, 1));
- struct stream_interface *si = cs_si(luactx->appctx->owner);
- struct channel *res = si_ic(si);
+ struct conn_stream *cs = luactx->appctx->owner;
+ struct channel *res = cs_ic(cs);
struct htx *htx;
struct htx_sl *sl;
struct h1m h1m;
@@ -5473,11 +5463,11 @@
__LJMP static int hlua_applet_http_start_response_yield(lua_State *L, int status, lua_KContext ctx)
{
struct hlua_appctx *luactx = MAY_LJMP(hlua_checkapplet_http(L, 1));
- struct stream_interface *si = cs_si(luactx->appctx->owner);
- struct channel *res = si_ic(si);
+ struct conn_stream *cs = luactx->appctx->owner;
+ struct channel *res = cs_ic(cs);
if (co_data(res)) {
- si_rx_room_blk(si);
+ si_rx_room_blk(cs->si);
MAY_LJMP(hlua_yieldk(L, 0, 0, hlua_applet_http_start_response_yield, TICK_ETERNITY, 0));
}
return MAY_LJMP(hlua_applet_http_send_response(L));
@@ -9182,8 +9172,8 @@
static int hlua_applet_tcp_init(struct appctx *ctx)
{
- struct stream_interface *si = cs_si(ctx->owner);
- struct stream *strm = si_strm(si);
+ struct conn_stream *cs = ctx->owner;
+ struct stream *strm = __cs_strm(cs);
struct hlua *hlua;
struct task *task;
char **arg;
@@ -9271,17 +9261,17 @@
RESET_SAFE_LJMP(hlua);
/* Wakeup the applet ASAP. */
- si_cant_get(si);
- si_rx_endp_more(si);
+ si_cant_get(cs->si);
+ si_rx_endp_more(cs->si);
return 1;
}
void hlua_applet_tcp_fct(struct appctx *ctx)
{
- struct stream_interface *si = cs_si(ctx->owner);
- struct stream *strm = si_strm(si);
- struct channel *res = si_ic(si);
+ struct conn_stream *cs = ctx->owner;
+ struct stream *strm = __cs_strm(cs);
+ struct channel *res = cs_ic(cs);
struct act_rule *rule = ctx->rule;
struct proxy *px = strm->be;
struct hlua *hlua = ctx->ctx.hlua_apptcp.hlua;
@@ -9289,12 +9279,12 @@
/* The applet execution is already done. */
if (ctx->ctx.hlua_apptcp.flags & APPLET_DONE) {
/* eat the whole request */
- co_skip(si_oc(si), co_data(si_oc(si)));
+ co_skip(cs_oc(cs), co_data(cs_oc(cs)));
return;
}
/* If the stream is disconnect or closed, ldo nothing. */
- if (unlikely(si->state == SI_ST_DIS || si->state == SI_ST_CLO))
+ if (unlikely(cs->si->state == SI_ST_DIS || cs->si->state == SI_ST_CLO))
return;
/* Execute the function. */
@@ -9304,9 +9294,9 @@
ctx->ctx.hlua_apptcp.flags |= APPLET_DONE;
/* eat the whole request */
- co_skip(si_oc(si), co_data(si_oc(si)));
+ co_skip(cs_oc(cs), co_data(cs_oc(cs)));
res->flags |= CF_READ_NULL;
- si_shutr(si);
+ si_shutr(cs->si);
return;
/* yield. */
@@ -9351,8 +9341,8 @@
error:
/* For all other cases, just close the stream. */
- si_shutw(si);
- si_shutr(si);
+ si_shutw(cs->si);
+ si_shutr(cs->si);
ctx->ctx.hlua_apptcp.flags |= APPLET_DONE;
}
@@ -9370,8 +9360,8 @@
*/
static int hlua_applet_http_init(struct appctx *ctx)
{
- struct stream_interface *si = cs_si(ctx->owner);
- struct stream *strm = si_strm(si);
+ struct conn_stream *cs = ctx->owner;
+ struct stream *strm = __cs_strm(cs);
struct http_txn *txn;
struct hlua *hlua;
char **arg;
@@ -9465,17 +9455,17 @@
RESET_SAFE_LJMP(hlua);
/* Wakeup the applet when data is ready for read. */
- si_cant_get(si);
+ si_cant_get(cs->si);
return 1;
}
void hlua_applet_http_fct(struct appctx *ctx)
{
- struct stream_interface *si = cs_si(ctx->owner);
- struct stream *strm = si_strm(si);
- struct channel *req = si_oc(si);
- struct channel *res = si_ic(si);
+ struct conn_stream *cs = ctx->owner;
+ struct stream *strm = __cs_strm(cs);
+ struct channel *req = cs_oc(cs);
+ struct channel *res = cs_ic(cs);
struct act_rule *rule = ctx->rule;
struct proxy *px = strm->be;
struct hlua *hlua = ctx->ctx.hlua_apphttp.hlua;
@@ -9484,12 +9474,12 @@
res_htx = htx_from_buf(&res->buf);
/* If the stream is disconnect or closed, ldo nothing. */
- if (unlikely(si->state == SI_ST_DIS || si->state == SI_ST_CLO))
+ if (unlikely(cs->si->state == SI_ST_DIS || cs->si->state == SI_ST_CLO))
goto out;
/* Check if the input buffer is available. */
if (!b_size(&res->buf)) {
- si_rx_room_blk(si);
+ si_rx_room_blk(cs->si);
goto out;
}
/* check that the output is not closed */
@@ -9500,7 +9490,7 @@
if (!HLUA_IS_RUNNING(hlua) &&
!(ctx->ctx.hlua_apphttp.flags & APPLET_DONE)) {
if (!co_data(req)) {
- si_cant_get(si);
+ si_cant_get(cs->si);
goto out;
}
}
@@ -9569,14 +9559,14 @@
*/
if (htx_is_empty(res_htx) && (strm->txn->rsp.flags & (HTTP_MSGF_XFER_LEN|HTTP_MSGF_CNT_LEN)) == HTTP_MSGF_XFER_LEN) {
if (!htx_add_endof(res_htx, HTX_BLK_EOT)) {
- si_rx_room_blk(si);
+ si_rx_room_blk(cs->si);
goto out;
}
channel_add_input(res, 1);
}
res_htx->flags |= HTX_FL_EOM;
- si->cs->endp->flags |= CS_EP_EOI;
+ cs->endp->flags |= CS_EP_EOI;
res->flags |= CF_EOI;
strm->txn->status = ctx->ctx.hlua_apphttp.status;
ctx->ctx.hlua_apphttp.flags |= APPLET_RSP_SENT;
@@ -9586,7 +9576,7 @@
if (ctx->ctx.hlua_apphttp.flags & APPLET_DONE) {
if (!(res->flags & CF_SHUTR)) {
res->flags |= CF_READ_NULL;
- si_shutr(si);
+ si_shutr(cs->si);
}
/* eat the whole request */
@@ -10101,15 +10091,15 @@
static int hlua_cli_io_handler_fct(struct appctx *appctx)
{
struct hlua *hlua;
- struct stream_interface *si;
+ struct conn_stream *cs;
struct hlua_function *fcn;
hlua = appctx->ctx.hlua_cli.hlua;
- si = cs_si(appctx->owner);
+ cs = appctx->owner;
fcn = appctx->ctx.hlua_cli.fcn;
/* If the stream is disconnect or closed, ldo nothing. */
- if (unlikely(si->state == SI_ST_DIS || si->state == SI_ST_CLO))
+ if (unlikely(cs->si->state == SI_ST_DIS || cs->si->state == SI_ST_CLO))
return 1;
/* Execute the function. */
@@ -10123,7 +10113,7 @@
case HLUA_E_AGAIN:
/* We want write. */
if (HLUA_IS_WAKERESWR(hlua))
- si_rx_room_blk(si);
+ si_rx_room_blk(cs->si);
/* Set the timeout. */
if (hlua->wake_time != TICK_ETERNITY)
task_schedule(hlua->task, hlua->wake_time);
diff --git a/src/http_client.c b/src/http_client.c
index c067736..965dc9d 100644
--- a/src/http_client.c
+++ b/src/http_client.c
@@ -17,7 +17,8 @@
#include <haproxy/cli.h>
#include <haproxy/dynbuf.h>
#include <haproxy/cfgparse.h>
-#include <haproxy/connection.h>
+#include <haproxy/conn_stream.h>
+#include <haproxy/cs_utils.h>
#include <haproxy/global.h>
#include <haproxy/istbuf.h>
#include <haproxy/h1_htx.h>
@@ -166,7 +167,7 @@
*/
static int hc_cli_io_handler(struct appctx *appctx)
{
- struct stream_interface *si = cs_si(appctx->owner);
+ struct conn_stream *cs = appctx->owner;
struct buffer *trash = alloc_trash_chunk();
struct httpclient *hc = appctx->ctx.cli.p0;
struct http_hdr *hdrs, *hdr;
@@ -176,8 +177,8 @@
if (appctx->ctx.cli.i0 & HC_CLI_F_RES_STLINE) {
chunk_appendf(trash, "%.*s %d %.*s\n", (unsigned int)istlen(hc->res.vsn), istptr(hc->res.vsn),
hc->res.status, (unsigned int)istlen(hc->res.reason), istptr(hc->res.reason));
- if (ci_putchk(si_ic(si), trash) == -1)
- si_rx_room_blk(si);
+ if (ci_putchk(cs_ic(cs), trash) == -1)
+ si_rx_room_blk(cs->si);
appctx->ctx.cli.i0 &= ~HC_CLI_F_RES_STLINE;
goto out;
}
@@ -190,8 +191,8 @@
}
if (!chunk_memcat(trash, "\r\n", 2))
goto out;
- if (ci_putchk(si_ic(si), trash) == -1)
- si_rx_room_blk(si);
+ if (ci_putchk(cs_ic(cs), trash) == -1)
+ si_rx_room_blk(cs->si);
appctx->ctx.cli.i0 &= ~HC_CLI_F_RES_HDR;
goto out;
}
@@ -199,8 +200,8 @@
if (appctx->ctx.cli.i0 & HC_CLI_F_RES_BODY) {
int ret;
- ret = httpclient_res_xfer(hc, &si_ic(si)->buf);
- channel_add_input(si_ic(si), ret); /* forward what we put in the buffer channel */
+ ret = httpclient_res_xfer(hc, cs_ib(cs));
+ channel_add_input(cs_ic(cs), ret); /* forward what we put in the buffer channel */
if (!httpclient_data(hc)) {/* remove the flag if the buffer was emptied */
appctx->ctx.cli.i0 &= ~HC_CLI_F_RES_BODY;
@@ -210,8 +211,8 @@
/* we must close only if F_END is the last flag */
if (appctx->ctx.cli.i0 == HC_CLI_F_RES_END) {
- si_shutw(si);
- si_shutr(si);
+ si_shutw(cs->si);
+ si_shutr(cs->si);
appctx->ctx.cli.i0 &= ~HC_CLI_F_RES_END;
goto out;
}
@@ -219,7 +220,7 @@
out:
/* we didn't clear every flags, we should come back to finish things */
if (appctx->ctx.cli.i0)
- si_rx_room_blk(si);
+ si_rx_room_blk(cs->si);
free_trash_chunk(trash);
return 0;
@@ -640,8 +641,8 @@
static void httpclient_applet_io_handler(struct appctx *appctx)
{
struct httpclient *hc = appctx->ctx.httpclient.ptr;
- struct stream_interface *si = cs_si(appctx->owner);
- struct stream *s = si_strm(si);
+ struct conn_stream *cs = appctx->owner;
+ struct stream *s = __cs_strm(cs);
struct channel *req = &s->req;
struct channel *res = &s->res;
struct htx_blk *blk = NULL;
@@ -735,7 +736,7 @@
/* if the request contains the HTX_FL_EOM, we finished the request part. */
if (htx->flags & HTX_FL_EOM) {
- si->cs->endp->flags |= CS_EP_EOI;
+ cs->endp->flags |= CS_EP_EOI;
req->flags |= CF_EOI;
appctx->st0 = HTTPCLIENT_S_RES_STLINE;
}
@@ -925,13 +926,13 @@
process_data:
- si_rx_chan_rdy(si);
+ si_rx_chan_rdy(cs->si);
return;
more:
/* There was not enough data in the response channel */
- si_rx_room_blk(si);
+ si_rx_room_blk(cs->si);
if (appctx->st0 == HTTPCLIENT_S_RES_END)
goto end;
@@ -947,8 +948,8 @@
return;
end:
- si_shutw(si);
- si_shutr(si);
+ si_shutw(cs->si);
+ si_shutr(cs->si);
return;
}
diff --git a/src/log.c b/src/log.c
index 80db27a..cbb86e5 100644
--- a/src/log.c
+++ b/src/log.c
@@ -28,6 +28,8 @@
#include <haproxy/applet-t.h>
#include <haproxy/cfgparse.h>
#include <haproxy/clock.h>
+#include <haproxy/conn_stream.h>
+#include <haproxy/cs_utils.h>
#include <haproxy/fd.h>
#include <haproxy/frontend.h>
#include <haproxy/global.h>
@@ -3562,8 +3564,8 @@
static void syslog_io_handler(struct appctx *appctx)
{
static THREAD_LOCAL struct ist metadata[LOG_META_FIELDS];
- struct stream_interface *si = cs_si(appctx->owner);
- struct stream *s = si_strm(si);
+ struct conn_stream *cs = appctx->owner;
+ struct stream *s = __cs_strm(cs);
struct proxy *frontend = strm_fe(s);
struct listener *l = strm_li(s);
struct buffer *buf = get_trash_chunk();
@@ -3575,14 +3577,14 @@
size_t size;
max_accept = l->maxaccept ? l->maxaccept : 1;
- while (co_data(si_oc(si))) {
+ while (co_data(cs_oc(cs))) {
char c;
if (max_accept <= 0)
goto missing_budget;
max_accept--;
- to_skip = co_getchar(si_oc(si), &c);
+ to_skip = co_getchar(cs_oc(cs), &c);
if (!to_skip)
goto missing_data;
else if (to_skip < 0)
@@ -3592,7 +3594,7 @@
/* rfc-6587, Non-Transparent-Framing: messages separated by
* a trailing LF or CR LF
*/
- to_skip = co_getline(si_oc(si), buf->area, buf->size);
+ to_skip = co_getline(cs_oc(cs), buf->area, buf->size);
if (!to_skip)
goto missing_data;
else if (to_skip < 0)
@@ -3616,7 +3618,7 @@
char *p = NULL;
int msglen;
- to_skip = co_getword(si_oc(si), buf->area, buf->size, ' ');
+ to_skip = co_getword(cs_oc(cs), buf->area, buf->size, ' ');
if (!to_skip)
goto missing_data;
else if (to_skip < 0)
@@ -3633,7 +3635,7 @@
if (msglen > buf->size)
goto parse_error;
- msglen = co_getblk(si_oc(si), buf->area, msglen, to_skip);
+ msglen = co_getblk(cs_oc(cs), buf->area, msglen, to_skip);
if (!msglen)
goto missing_data;
else if (msglen < 0)
@@ -3646,7 +3648,7 @@
else
goto parse_error;
- co_skip(si_oc(si), to_skip);
+ co_skip(cs_oc(cs), to_skip);
/* update counters */
_HA_ATOMIC_INC(&cum_log_messages);
@@ -3660,7 +3662,7 @@
missing_data:
/* we need more data to read */
- si_oc(si)->flags |= CF_READ_DONTWAIT;
+ cs_oc(cs)->flags |= CF_READ_DONTWAIT;
return;
@@ -3683,10 +3685,10 @@
_HA_ATOMIC_INC(&frontend->fe_counters.cli_aborts);
close:
- si_shutw(si);
- si_shutr(si);
+ si_shutw(cs->si);
+ si_shutr(cs->si);
- si_ic(si)->flags |= CF_READ_NULL;
+ cs_ic(cs)->flags |= CF_READ_NULL;
return;
}
diff --git a/src/map.c b/src/map.c
index 870d924..54c78b6 100644
--- a/src/map.c
+++ b/src/map.c
@@ -17,6 +17,8 @@
#include <haproxy/applet-t.h>
#include <haproxy/arg.h>
#include <haproxy/cli.h>
+#include <haproxy/conn_stream.h>
+#include <haproxy/cs_utils.h>
#include <haproxy/map.h>
#include <haproxy/pattern.h>
#include <haproxy/regex.h>
@@ -323,10 +325,10 @@
/* expects the current generation ID in appctx->cli.cli.i0 */
static int cli_io_handler_pat_list(struct appctx *appctx)
{
- struct stream_interface *si = cs_si(appctx->owner);
+ struct conn_stream *cs = appctx->owner;
struct pat_ref_elt *elt;
- if (unlikely(si_ic(si)->flags & (CF_WRITE_ERROR|CF_SHUTW))) {
+ if (unlikely(cs_ic(cs)->flags & (CF_WRITE_ERROR|CF_SHUTW))) {
/* If we're forced to shut down, we might have to remove our
* reference to the last ref_elt being dumped.
*/
@@ -382,13 +384,13 @@
chunk_appendf(&trash, "%p %s\n",
elt, elt->pattern);
- if (ci_putchk(si_ic(si), &trash) == -1) {
+ if (ci_putchk(cs_ic(cs), &trash) == -1) {
/* let's try again later from this stream. We add ourselves into
* this stream's users so that it can remove us upon termination.
*/
LIST_APPEND(&elt->back_refs, &appctx->ctx.map.bref.users);
HA_SPIN_UNLOCK(PATREF_LOCK, &appctx->ctx.map.ref->lock);
- si_rx_room_blk(si);
+ si_rx_room_blk(cs->si);
return 0;
}
skip:
@@ -406,7 +408,7 @@
static int cli_io_handler_pats_list(struct appctx *appctx)
{
- struct stream_interface *si = cs_si(appctx->owner);
+ struct conn_stream *cs = appctx->owner;
switch (appctx->st2) {
case STAT_ST_INIT:
@@ -416,8 +418,8 @@
*/
chunk_reset(&trash);
chunk_appendf(&trash, "# id (file) description\n");
- if (ci_putchk(si_ic(si), &trash) == -1) {
- si_rx_room_blk(si);
+ if (ci_putchk(cs_ic(cs), &trash) == -1) {
+ si_rx_room_blk(cs->si);
return 0;
}
@@ -444,11 +446,11 @@
appctx->ctx.map.ref->display, appctx->ctx.map.ref->curr_gen, appctx->ctx.map.ref->next_gen,
appctx->ctx.map.ref->entry_cnt);
- if (ci_putchk(si_ic(si), &trash) == -1) {
+ if (ci_putchk(cs_ic(cs), &trash) == -1) {
/* let's try again later from this stream. We add ourselves into
* this stream's users so that it can remove us upon termination.
*/
- si_rx_room_blk(si);
+ si_rx_room_blk(cs->si);
return 0;
}
@@ -468,7 +470,7 @@
static int cli_io_handler_map_lookup(struct appctx *appctx)
{
- struct stream_interface *si = cs_si(appctx->owner);
+ struct conn_stream *cs = appctx->owner;
struct sample sample;
struct pattern *pat;
int match_method;
@@ -564,12 +566,12 @@
chunk_appendf(&trash, "\n");
/* display response */
- if (ci_putchk(si_ic(si), &trash) == -1) {
+ if (ci_putchk(cs_ic(cs), &trash) == -1) {
/* let's try again later from this stream. We add ourselves into
* this stream's users so that it can remove us upon termination.
*/
HA_SPIN_UNLOCK(PATREF_LOCK, &appctx->ctx.map.ref->lock);
- si_rx_room_blk(si);
+ si_rx_room_blk(cs->si);
return 0;
}
@@ -993,7 +995,6 @@
*/
static int cli_io_handler_clear_map(struct appctx *appctx)
{
- struct stream_interface *si = cs_si(appctx->owner);
int finished;
HA_SPIN_LOCK(PATREF_LOCK, &appctx->ctx.map.ref->lock);
@@ -1002,7 +1003,7 @@
if (!finished) {
/* let's come back later */
- si_rx_endp_more(si);
+ si_rx_endp_more(cs_si(appctx->owner));
return 0;
}
return 1;
diff --git a/src/mux_fcgi.c b/src/mux_fcgi.c
index 9af915d..50f271f 100644
--- a/src/mux_fcgi.c
+++ b/src/mux_fcgi.c
@@ -18,6 +18,7 @@
#include <haproxy/cfgparse.h>
#include <haproxy/connection.h>
#include <haproxy/conn_stream.h>
+#include <haproxy/cs_utils.h>
#include <haproxy/errors.h>
#include <haproxy/fcgi-app.h>
#include <haproxy/fcgi.h>
@@ -1234,8 +1235,8 @@
struct fcgi_strm_params *params)
{
struct connection *cli_conn = objt_conn(fstrm->sess->origin);
- const struct sockaddr_storage *src = (cs_check(fstrm->cs) ? conn_src(fconn->conn) : si_src(si_opposite(cs_si(fstrm->cs))));
- const struct sockaddr_storage *dst = (cs_check(fstrm->cs) ? conn_dst(fconn->conn) : si_dst(si_opposite(cs_si(fstrm->cs))));
+ const struct sockaddr_storage *src = (cs_check(fstrm->cs) ? conn_src(fconn->conn) : si_src(cs_opposite(fstrm->cs)->si));
+ const struct sockaddr_storage *dst = (cs_check(fstrm->cs) ? conn_dst(fconn->conn) : si_dst(cs_opposite(fstrm->cs)->si));
struct ist p;
if (!sl)
diff --git a/src/mworker.c b/src/mworker.c
index e690f8b..cac4a5b 100644
--- a/src/mworker.c
+++ b/src/mworker.c
@@ -26,6 +26,8 @@
#include <haproxy/api.h>
#include <haproxy/cfgparse.h>
#include <haproxy/cli.h>
+#include <haproxy/conn_stream.h>
+#include <haproxy/cs_utils.h>
#include <haproxy/errors.h>
#include <haproxy/fd.h>
#include <haproxy/global.h>
@@ -509,14 +511,14 @@
/* Displays workers and processes */
static int cli_io_handler_show_proc(struct appctx *appctx)
{
- struct stream_interface *si = cs_si(appctx->owner);
+ struct conn_stream *cs = appctx->owner;
struct mworker_proc *child;
int old = 0;
int up = now.tv_sec - proc_self->timestamp;
char *uptime = NULL;
char *reloadtxt = NULL;
- if (unlikely(si_ic(si)->flags & (CF_WRITE_ERROR|CF_SHUTW)))
+ if (unlikely(cs_ic(cs)->flags & (CF_WRITE_ERROR|CF_SHUTW)))
return 1;
chunk_reset(&trash);
@@ -603,8 +605,8 @@
- if (ci_putchk(si_ic(si), &trash) == -1) {
- si_rx_room_blk(si);
+ if (ci_putchk(cs_ic(cs), &trash) == -1) {
+ si_rx_room_blk(cs->si);
return 0;
}
diff --git a/src/peers.c b/src/peers.c
index c31aa81..9846612 100644
--- a/src/peers.c
+++ b/src/peers.c
@@ -28,6 +28,8 @@
#include <haproxy/applet.h>
#include <haproxy/channel.h>
#include <haproxy/cli.h>
+#include <haproxy/conn_stream.h>
+#include <haproxy/cs_utils.h>
#include <haproxy/dict.h>
#include <haproxy/errors.h>
#include <haproxy/fd.h>
@@ -444,14 +446,9 @@
struct peers *peers = NULL;
if (peer->appctx) {
- struct stream_interface *si;
+ struct stream *s = __cs_strm(peer->appctx->owner);
- si = cs_si(peer->appctx->owner);
- if (si) {
- struct stream *s = si_strm(si);
-
- peers = strm_fe(s)->parent;
- }
+ peers = strm_fe(s)->parent;
}
if (peers)
@@ -1042,20 +1039,13 @@
*/
void __peer_session_deinit(struct peer *peer)
{
- struct stream_interface *si;
struct stream *s;
struct peers *peers;
if (!peer->appctx)
return;
- si = cs_si(peer->appctx->owner);
- if (!si)
- return;
-
- s = si_strm(si);
- if (!s)
- return;
+ s = __cs_strm(peer->appctx->owner);
peers = strm_fe(s)->parent;
if (!peers)
@@ -1148,10 +1138,10 @@
*/
static inline int peer_getline(struct appctx *appctx)
{
+ struct conn_stream *cs = appctx->owner;
int n;
- struct stream_interface *si = cs_si(appctx->owner);
- n = co_getline(si_oc(si), trash.area, trash.size);
+ n = co_getline(cs_oc(cs), trash.area, trash.size);
if (!n)
return 0;
@@ -1165,7 +1155,7 @@
else
trash.area[n - 1] = 0;
- co_skip(si_oc(si), n);
+ co_skip(cs_oc(cs), n);
return n;
}
@@ -1182,7 +1172,7 @@
struct peer_prep_params *params)
{
int ret, msglen;
- struct stream_interface *si = cs_si(appctx->owner);
+ struct conn_stream *cs = appctx->owner;
msglen = peer_prepare_msg(trash.area, trash.size, params);
if (!msglen) {
@@ -1192,11 +1182,11 @@
}
/* message to buffer */
- ret = ci_putblk(si_ic(si), trash.area, msglen);
+ ret = ci_putblk(cs_ic(cs), trash.area, msglen);
if (ret <= 0) {
if (ret == -1) {
/* No more write possible */
- si_rx_room_blk(si);
+ si_rx_room_blk(cs->si);
return -1;
}
appctx->st0 = PEER_SESS_ST_END;
@@ -1662,7 +1652,7 @@
static int peer_treat_updatemsg(struct appctx *appctx, struct peer *p, int updt, int exp,
char **msg_cur, char *msg_end, int msg_len, int totl)
{
- struct stream_interface *si = cs_si(appctx->owner);
+ struct conn_stream *cs = appctx->owner;
struct shared_table *st = p->remote_table;
struct stksess *ts, *newts;
uint32_t update;
@@ -2005,7 +1995,7 @@
ignore_msg:
/* skip consumed message */
- co_skip(si_oc(si), totl);
+ co_skip(cs_oc(cs), totl);
TRACE_DEVEL("leaving in error", PEERS_EV_UPDTMSG);
return 0;
@@ -2114,7 +2104,7 @@
static inline int peer_treat_definemsg(struct appctx *appctx, struct peer *p,
char **msg_cur, char *msg_end, int totl)
{
- struct stream_interface *si = cs_si(appctx->owner);
+ struct conn_stream *cs = appctx->owner;
int table_id_len;
struct shared_table *st;
int table_type;
@@ -2288,7 +2278,7 @@
return 1;
ignore_msg:
- co_skip(si_oc(si), totl);
+ co_skip(cs_oc(cs), totl);
return 0;
malformed_exit:
@@ -2313,10 +2303,10 @@
uint32_t *msg_len, int *totl)
{
int reql;
- struct stream_interface *si = cs_si(appctx->owner);
+ struct conn_stream *cs = appctx->owner;
char *cur;
- reql = co_getblk(si_oc(si), msg_head, 2 * sizeof(char), *totl);
+ reql = co_getblk(cs_oc(cs), msg_head, 2 * sizeof(char), *totl);
if (reql <= 0) /* closed or EOL not found */
goto incomplete;
@@ -2330,11 +2320,11 @@
/* Read and Decode message length */
msg_head += *totl;
msg_head_sz -= *totl;
- reql = co_data(si_oc(si)) - *totl;
+ reql = co_data(cs_oc(cs)) - *totl;
if (reql > msg_head_sz)
reql = msg_head_sz;
- reql = co_getblk(si_oc(si), msg_head, reql, *totl);
+ reql = co_getblk(cs_oc(cs), msg_head, reql, *totl);
if (reql <= 0) /* closed */
goto incomplete;
@@ -2360,7 +2350,7 @@
return -1;
}
- reql = co_getblk(si_oc(si), trash.area, *msg_len, *totl);
+ reql = co_getblk(cs_oc(cs), trash.area, *msg_len, *totl);
if (reql <= 0) /* closed */
goto incomplete;
*totl += reql;
@@ -2369,7 +2359,7 @@
return 1;
incomplete:
- if (reql < 0 || (si_oc(si)->flags & (CF_SHUTW|CF_SHUTW_NOW))) {
+ if (reql < 0 || (cs_oc(cs)->flags & (CF_SHUTW|CF_SHUTW_NOW))) {
/* there was an error or the message was truncated */
appctx->st0 = PEER_SESS_ST_END;
return -1;
@@ -2385,8 +2375,7 @@
static inline int peer_treat_awaited_msg(struct appctx *appctx, struct peer *peer, unsigned char *msg_head,
char **msg_cur, char *msg_end, int msg_len, int totl)
{
- struct stream_interface *si = cs_si(appctx->owner);
- struct stream *s = si_strm(si);
+ struct stream *s = __cs_strm(appctx->owner);
struct peers *peers = strm_fe(s)->parent;
if (msg_head[0] == PEER_MSG_CLASS_CONTROL) {
@@ -2673,8 +2662,7 @@
char *p;
int reql;
struct peer *peer;
- struct stream_interface *si = cs_si(appctx->owner);
- struct stream *s = si_strm(si);
+ struct stream *s = __cs_strm(appctx->owner);
struct peers *peers = strm_fe(s)->parent;
reql = peer_getline(appctx);
@@ -2834,8 +2822,8 @@
*/
static void peer_io_handler(struct appctx *appctx)
{
- struct stream_interface *si = cs_si(appctx->owner);
- struct stream *s = si_strm(si);
+ struct conn_stream *cs = appctx->owner;
+ struct stream *s = __cs_strm(cs);
struct peers *curpeers = strm_fe(s)->parent;
struct peer *curpeer = NULL;
int reql = 0;
@@ -2844,8 +2832,8 @@
int prev_state;
/* Check if the input buffer is available. */
- if (si_ic(si)->buf.size == 0) {
- si_rx_room_blk(si);
+ if (cs_ib(cs)->size == 0) {
+ si_rx_room_blk(cs->si);
goto out;
}
@@ -2981,7 +2969,7 @@
}
}
- if (si_ic(si)->flags & CF_WRITE_PARTIAL)
+ if (cs_ic(cs)->flags & CF_WRITE_PARTIAL)
curpeer->statuscode = PEER_SESS_SC_CONNECTEDCODE;
reql = peer_getline(appctx);
@@ -3044,7 +3032,7 @@
curpeer->flags |= PEER_F_ALIVE;
/* skip consumed message */
- co_skip(si_oc(si), totl);
+ co_skip(cs_oc(cs), totl);
/* loop on that state to peek next message */
goto switchstate;
@@ -3111,15 +3099,15 @@
HA_SPIN_UNLOCK(PEER_LOCK, &curpeer->lock);
curpeer = NULL;
}
- si_shutw(si);
- si_shutr(si);
- si_ic(si)->flags |= CF_READ_NULL;
+ si_shutw(cs->si);
+ si_shutr(cs->si);
+ cs_ic(cs)->flags |= CF_READ_NULL;
goto out;
}
}
}
out:
- si_oc(si)->flags |= CF_READ_DONTWAIT;
+ cs_oc(cs)->flags |= CF_READ_DONTWAIT;
if (curpeer)
HA_SPIN_UNLOCK(PEER_LOCK, &curpeer->lock);
@@ -3751,7 +3739,7 @@
* Returns 0 if the output buffer is full and needs to be called again, non-zero if not.
* Dedicated to be called by cli_io_handler_show_peers() cli I/O handler.
*/
-static int peers_dump_head(struct buffer *msg, struct stream_interface *si, struct peers *peers)
+static int peers_dump_head(struct buffer *msg, struct conn_stream *cs, struct peers *peers)
{
struct tm tm;
@@ -3767,8 +3755,8 @@
TICKS_TO_MS(1000)) : "<NEVER>",
peers->sync_task ? peers->sync_task->calls : 0);
- if (ci_putchk(si_ic(si), msg) == -1) {
- si_rx_room_blk(si);
+ if (ci_putchk(cs_ic(cs), msg) == -1) {
+ si_rx_room_blk(cs->si);
return 0;
}
@@ -3780,11 +3768,11 @@
* Returns 0 if the output buffer is full and needs to be called again, non-zero
* if not. Dedicated to be called by cli_io_handler_show_peers() cli I/O handler.
*/
-static int peers_dump_peer(struct buffer *msg, struct stream_interface *si, struct peer *peer, int flags)
+static int peers_dump_peer(struct buffer *msg, struct conn_stream *cs, struct peer *peer, int flags)
{
struct connection *conn;
char pn[INET6_ADDRSTRLEN];
- struct stream_interface *peer_si;
+ struct conn_stream *peer_cs;
struct stream *peer_s;
struct appctx *appctx;
struct shared_table *st;
@@ -3826,15 +3814,10 @@
chunk_appendf(&trash, " appctx:%p st0=%d st1=%d task_calls=%u", appctx, appctx->st0, appctx->st1,
appctx->t ? appctx->t->calls : 0);
- peer_si = cs_si(peer->appctx->owner);
- if (!peer_si)
- goto table_info;
-
- peer_s = si_strm(peer_si);
- if (!peer_s)
- goto table_info;
+ peer_cs = peer->appctx->owner;
+ peer_s = __cs_strm(peer_cs);
- chunk_appendf(&trash, " state=%s", si_state_str(si_opposite(peer_si)->state));
+ chunk_appendf(&trash, " state=%s", si_state_str(cs_opposite(peer_cs)->si->state));
conn = objt_conn(strm_orig(peer_s));
if (conn)
@@ -3931,8 +3914,8 @@
end:
chunk_appendf(&trash, "\n");
- if (ci_putchk(si_ic(si), msg) == -1) {
- si_rx_room_blk(si);
+ if (ci_putchk(cs_ic(cs), msg) == -1) {
+ si_rx_room_blk(cs->si);
return 0;
}
@@ -3949,7 +3932,6 @@
{
int show_all;
int ret = 0, first_peers = 1;
- struct stream_interface *si = cs_si(appctx->owner);
thread_isolate();
@@ -3978,7 +3960,7 @@
chunk_appendf(&trash, "\n");
else
first_peers = 0;
- if (!peers_dump_head(&trash, si, appctx->ctx.cfgpeers.peers))
+ if (!peers_dump_head(&trash, appctx->owner, appctx->ctx.cfgpeers.peers))
goto out;
appctx->ctx.cfgpeers.peer = appctx->ctx.cfgpeers.peers->remote;
@@ -3996,7 +3978,7 @@
appctx->st2 = STAT_ST_END;
}
else {
- if (!peers_dump_peer(&trash, si, appctx->ctx.cfgpeers.peer, appctx->ctx.cfgpeers.flags))
+ if (!peers_dump_peer(&trash, appctx->owner, appctx->ctx.cfgpeers.peer, appctx->ctx.cfgpeers.flags))
goto out;
appctx->ctx.cfgpeers.peer = appctx->ctx.cfgpeers.peer->next;
diff --git a/src/pool.c b/src/pool.c
index fe10bad..25a3c7d 100644
--- a/src/pool.c
+++ b/src/pool.c
@@ -19,6 +19,8 @@
#include <haproxy/cfgparse.h>
#include <haproxy/channel.h>
#include <haproxy/cli.h>
+#include <haproxy/conn_stream.h>
+#include <haproxy/cs_utils.h>
#include <haproxy/errors.h>
#include <haproxy/global.h>
#include <haproxy/list.h>
@@ -1004,11 +1006,11 @@
*/
static int cli_io_handler_dump_pools(struct appctx *appctx)
{
- struct stream_interface *si = cs_si(appctx->owner);
+ struct conn_stream *cs = appctx->owner;
dump_pools_to_trash();
- if (ci_putchk(si_ic(si), &trash) == -1) {
- si_rx_room_blk(si);
+ if (ci_putchk(cs_ic(cs), &trash) == -1) {
+ si_rx_room_blk(cs->si);
return 0;
}
return 1;
diff --git a/src/proxy.c b/src/proxy.c
index dc02863..5024178 100644
--- a/src/proxy.c
+++ b/src/proxy.c
@@ -26,6 +26,8 @@
#include <haproxy/capture-t.h>
#include <haproxy/cfgparse.h>
#include <haproxy/cli.h>
+#include <haproxy/conn_stream.h>
+#include <haproxy/cs_utils.h>
#include <haproxy/errors.h>
#include <haproxy/fd.h>
#include <haproxy/filters.h>
@@ -2649,9 +2651,9 @@
* It uses the proxy pointer from cli.p0, the proxy's id from cli.i0 and the server's
* pointer from cli.p1.
*/
-static int dump_servers_state(struct stream_interface *si)
+static int dump_servers_state(struct conn_stream *cs)
{
- struct appctx *appctx = __cs_appctx(si->cs);
+ struct appctx *appctx = __cs_appctx(cs);
struct proxy *px = appctx->ctx.cli.p0;
struct server *srv;
char srv_addr[INET6_ADDRSTRLEN + 1];
@@ -2714,8 +2716,8 @@
chunk_appendf(&trash, "\n");
}
- if (ci_putchk(si_ic(si), &trash) == -1) {
- si_rx_room_blk(si);
+ if (ci_putchk(cs_ic(cs), &trash) == -1) {
+ si_rx_room_blk(cs->si);
return 0;
}
}
@@ -2728,7 +2730,7 @@
*/
static int cli_io_handler_servers_state(struct appctx *appctx)
{
- struct stream_interface *si = cs_si(appctx->owner);
+ struct conn_stream *cs = appctx->owner;
struct proxy *curproxy;
chunk_reset(&trash);
@@ -2747,8 +2749,8 @@
"# bkname/svname bkid/svid addr port - purge_delay used_cur used_max need_est unsafe_nb safe_nb idle_lim idle_cur idle_per_thr[%d]\n",
global.nbthread);
- if (ci_putchk(si_ic(si), &trash) == -1) {
- si_rx_room_blk(si);
+ if (ci_putchk(cs_ic(cs), &trash) == -1) {
+ si_rx_room_blk(cs->si);
return 0;
}
appctx->st2 = STAT_ST_INFO;
@@ -2759,7 +2761,7 @@
curproxy = appctx->ctx.cli.p0;
/* servers are only in backends */
if ((curproxy->cap & PR_CAP_BE) && !(curproxy->cap & PR_CAP_INT)) {
- if (!dump_servers_state(si))
+ if (!dump_servers_state(cs))
return 0;
}
/* only the selected proxy is dumped */
@@ -2775,15 +2777,15 @@
*/
static int cli_io_handler_show_backend(struct appctx *appctx)
{
- struct stream_interface *si = cs_si(appctx->owner);
+ struct conn_stream *cs = appctx->owner;
struct proxy *curproxy;
chunk_reset(&trash);
if (!appctx->ctx.cli.p0) {
chunk_printf(&trash, "# name\n");
- if (ci_putchk(si_ic(si), &trash) == -1) {
- si_rx_room_blk(si);
+ if (ci_putchk(cs_ic(cs), &trash) == -1) {
+ si_rx_room_blk(cs->si);
return 0;
}
appctx->ctx.cli.p0 = proxies_list;
@@ -2797,8 +2799,8 @@
continue;
chunk_appendf(&trash, "%s\n", curproxy->id);
- if (ci_putchk(si_ic(si), &trash) == -1) {
- si_rx_room_blk(si);
+ if (ci_putchk(cs_ic(cs), &trash) == -1) {
+ si_rx_room_blk(cs->si);
return 0;
}
}
@@ -3078,10 +3080,10 @@
*/
static int cli_io_handler_show_errors(struct appctx *appctx)
{
- struct stream_interface *si = cs_si(appctx->owner);
+ struct conn_stream *cs = appctx->owner;
extern const char *monthname[12];
- if (unlikely(si_ic(si)->flags & (CF_WRITE_ERROR|CF_SHUTW)))
+ if (unlikely(cs_ic(cs)->flags & (CF_WRITE_ERROR|CF_SHUTW)))
return 1;
chunk_reset(&trash);
@@ -3098,7 +3100,7 @@
tm.tm_hour, tm.tm_min, tm.tm_sec, (int)(date.tv_usec/1000),
error_snapshot_id);
- if (ci_putchk(si_ic(si), &trash) == -1)
+ if (ci_putchk(cs_ic(cs), &trash) == -1)
goto cant_send;
appctx->ctx.errors.px = proxies_list;
@@ -3188,7 +3190,7 @@
chunk_appendf(&trash, " \n");
- if (ci_putchk(si_ic(si), &trash) == -1)
+ if (ci_putchk(cs_ic(cs), &trash) == -1)
goto cant_send_unlock;
appctx->ctx.errors.ptr = 0;
@@ -3199,7 +3201,7 @@
/* the snapshot changed while we were dumping it */
chunk_appendf(&trash,
" WARNING! update detected on this snapshot, dump interrupted. Please re-check!\n");
- if (ci_putchk(si_ic(si), &trash) == -1)
+ if (ci_putchk(cs_ic(cs), &trash) == -1)
goto cant_send_unlock;
goto next;
@@ -3215,7 +3217,7 @@
if (newptr == appctx->ctx.errors.ptr)
goto cant_send_unlock;
- if (ci_putchk(si_ic(si), &trash) == -1)
+ if (ci_putchk(cs_ic(cs), &trash) == -1)
goto cant_send_unlock;
appctx->ctx.errors.ptr = newptr;
@@ -3236,7 +3238,7 @@
cant_send_unlock:
HA_RWLOCK_RDUNLOCK(PROXY_LOCK, &appctx->ctx.errors.px->lock);
cant_send:
- si_rx_room_blk(si);
+ si_rx_room_blk(cs->si);
return 0;
}
diff --git a/src/resolvers.c b/src/resolvers.c
index e8117d8..85373a2 100644
--- a/src/resolvers.c
+++ b/src/resolvers.c
@@ -27,6 +27,8 @@
#include <haproxy/channel.h>
#include <haproxy/check.h>
#include <haproxy/cli.h>
+#include <haproxy/conn_stream.h>
+#include <haproxy/cs_utils.h>
#include <haproxy/dns.h>
#include <haproxy/errors.h>
#include <haproxy/fd.h>
@@ -2581,13 +2583,13 @@
}
-static int stats_dump_resolv_to_buffer(struct stream_interface *si,
+static int stats_dump_resolv_to_buffer(struct conn_stream *cs,
struct dns_nameserver *ns,
struct field *stats, size_t stats_count,
struct list *stat_modules)
{
- struct appctx *appctx = __cs_appctx(si->cs);
- struct channel *rep = si_ic(si);
+ struct appctx *appctx = __cs_appctx(cs);
+ struct channel *rep = cs_ic(cs);
struct stats_module *mod;
size_t idx = 0;
@@ -2609,19 +2611,19 @@
return 1;
full:
- si_rx_room_rdy(si);
+ si_rx_room_rdy(cs->si);
return 0;
}
/* Uses <appctx.ctx.stats.obj1> as a pointer to the current resolver and <obj2>
* as a pointer to the current nameserver.
*/
-int stats_dump_resolvers(struct stream_interface *si,
+int stats_dump_resolvers(struct conn_stream *cs,
struct field *stats, size_t stats_count,
struct list *stat_modules)
{
- struct appctx *appctx = __cs_appctx(si->cs);
- struct channel *rep = si_ic(si);
+ struct appctx *appctx = __cs_appctx(cs);
+ struct channel *rep = cs_ic(cs);
struct resolvers *resolver = appctx->ctx.stats.obj1;
struct dns_nameserver *ns = appctx->ctx.stats.obj2;
@@ -2642,7 +2644,7 @@
if (buffer_almost_full(&rep->buf))
goto full;
- if (!stats_dump_resolv_to_buffer(si, ns,
+ if (!stats_dump_resolv_to_buffer(cs, ns,
stats, stats_count,
stat_modules)) {
return 0;
@@ -2655,7 +2657,7 @@
return 1;
full:
- si_rx_room_blk(si);
+ si_rx_room_blk(cs->si);
return 0;
}
@@ -2745,7 +2747,7 @@
*/
static int cli_io_handler_dump_resolvers_to_buffer(struct appctx *appctx)
{
- struct stream_interface *si = cs_si(appctx->owner);
+ struct conn_stream *cs = appctx->owner;
struct resolvers *resolvers;
struct dns_nameserver *ns;
@@ -2789,11 +2791,11 @@
}
/* display response */
- if (ci_putchk(si_ic(si), &trash) == -1) {
+ if (ci_putchk(cs_ic(cs), &trash) == -1) {
/* let's try again later from this session. We add ourselves into
* this session's users so that it can remove us upon termination.
*/
- si_rx_room_blk(si);
+ si_rx_room_blk(cs->si);
return 0;
}
/* fall through */
diff --git a/src/ring.c b/src/ring.c
index e5d4fb5..1668ee3 100644
--- a/src/ring.c
+++ b/src/ring.c
@@ -23,6 +23,8 @@
#include <haproxy/applet.h>
#include <haproxy/buf.h>
#include <haproxy/cli.h>
+#include <haproxy/conn_stream.h>
+#include <haproxy/cs_utils.h>
#include <haproxy/ring.h>
#include <haproxy/stream_interface.h>
#include <haproxy/thread.h>
@@ -276,7 +278,7 @@
*/
int cli_io_handler_show_ring(struct appctx *appctx)
{
- struct stream_interface *si = cs_si(appctx->owner);
+ struct conn_stream *cs = appctx->owner;
struct ring *ring = appctx->ctx.cli.p0;
struct buffer *buf = &ring->buf;
size_t ofs = appctx->ctx.cli.o0;
@@ -284,7 +286,7 @@
size_t len, cnt;
int ret;
- if (unlikely(si_ic(si)->flags & (CF_WRITE_ERROR|CF_SHUTW)))
+ if (unlikely(cs_ic(cs)->flags & (CF_WRITE_ERROR|CF_SHUTW)))
return 1;
HA_RWLOCK_WRLOCK(LOGSRV_LOCK, &ring->lock);
@@ -343,8 +345,8 @@
trash.data += len;
trash.area[trash.data++] = '\n';
- if (ci_putchk(si_ic(si), &trash) == -1) {
- si_rx_room_blk(si);
+ if (ci_putchk(cs_ic(cs), &trash) == -1) {
+ si_rx_room_blk(cs->si);
ret = 0;
break;
}
@@ -360,16 +362,16 @@
/* we've drained everything and are configured to wait for more
* data or an event (keypress, close)
*/
- if (!si_oc(si)->output && !(si_oc(si)->flags & CF_SHUTW)) {
+ if (!cs_oc(cs)->output && !(cs_oc(cs)->flags & CF_SHUTW)) {
/* let's be woken up once new data arrive */
HA_RWLOCK_WRLOCK(LOGSRV_LOCK, &ring->lock);
LIST_APPEND(&ring->waiters, &appctx->wait_entry);
HA_RWLOCK_WRUNLOCK(LOGSRV_LOCK, &ring->lock);
- si_rx_endp_done(si);
+ si_rx_endp_done(cs->si);
ret = 0;
}
/* always drain all the request */
- co_skip(si_oc(si), si_oc(si)->output);
+ co_skip(cs_oc(cs), cs_oc(cs)->output);
}
return ret;
}
diff --git a/src/server.c b/src/server.c
index a87f864..5918819 100644
--- a/src/server.c
+++ b/src/server.c
@@ -25,6 +25,8 @@
#include <haproxy/check.h>
#include <haproxy/cli.h>
#include <haproxy/connection.h>
+#include <haproxy/conn_stream.h>
+#include <haproxy/cs_utils.h>
#include <haproxy/dict-t.h>
#include <haproxy/errors.h>
#include <haproxy/global.h>
@@ -4314,7 +4316,7 @@
static int cli_parse_get_weight(char **args, char *payload, struct appctx *appctx, void *private)
{
- struct stream_interface *si = cs_si(appctx->owner);
+ struct conn_stream *cs = appctx->owner;
struct proxy *px;
struct server *sv;
char *line;
@@ -4336,8 +4338,8 @@
/* return server's effective weight at the moment */
snprintf(trash.area, trash.size, "%d (initial %d)\n", sv->uweight,
sv->iweight);
- if (ci_putstr(si_ic(si), trash.area) == -1) {
- si_rx_room_blk(si);
+ if (ci_putstr(cs_ic(cs), trash.area) == -1) {
+ si_rx_room_blk(cs->si);
return 0;
}
return 1;
diff --git a/src/sink.c b/src/sink.c
index c5cd1d7..91ee67f 100644
--- a/src/sink.c
+++ b/src/sink.c
@@ -22,6 +22,8 @@
#include <haproxy/api.h>
#include <haproxy/cfgparse.h>
#include <haproxy/cli.h>
+#include <haproxy/conn_stream.h>
+#include <haproxy/cs_utils.h>
#include <haproxy/errors.h>
#include <haproxy/list.h>
#include <haproxy/log.h>
@@ -294,8 +296,8 @@
*/
static void sink_forward_io_handler(struct appctx *appctx)
{
- struct stream_interface *si = cs_si(appctx->owner);
- struct stream *s = si_strm(si);
+ struct conn_stream *cs = appctx->owner;
+ struct stream *s = __cs_strm(cs);
struct sink *sink = strm_fe(s)->parent;
struct sink_forward_target *sft = appctx->ctx.sft.ptr;
struct ring *ring = sink->ctx.ring;
@@ -312,25 +314,25 @@
* and we don't want expire on this case
* with a syslog server
*/
- si_oc(si)->rex = TICK_ETERNITY;
+ cs_oc(cs)->rex = TICK_ETERNITY;
/* rto should not change but it seems the case */
- si_oc(si)->rto = TICK_ETERNITY;
+ cs_oc(cs)->rto = TICK_ETERNITY;
/* an error was detected */
- if (unlikely(si_ic(si)->flags & (CF_WRITE_ERROR|CF_SHUTW)))
+ if (unlikely(cs_ic(cs)->flags & (CF_WRITE_ERROR|CF_SHUTW)))
goto close;
/* con closed by server side */
- if ((si_oc(si)->flags & CF_SHUTW))
+ if ((cs_oc(cs)->flags & CF_SHUTW))
goto close;
/* if the connection is not established, inform the stream that we want
* to be notified whenever the connection completes.
*/
- if (si_opposite(si)->state < SI_ST_EST) {
- si_cant_get(si);
- si_rx_conn_blk(si);
- si_rx_endp_more(si);
+ if (cs_opposite(cs)->si->state < SI_ST_EST) {
+ si_cant_get(cs->si);
+ si_rx_conn_blk(cs->si);
+ si_rx_endp_more(cs->si);
return;
}
@@ -366,7 +368,7 @@
* the message so that we can take our reference there if we have to
* stop before the end (ret=0).
*/
- if (si_opposite(si)->state == SI_ST_EST) {
+ if (cs_opposite(cs)->si->state == SI_ST_EST) {
/* we were already there, adjust the offset to be relative to
* the buffer's head and remove us from the counter.
*/
@@ -394,8 +396,8 @@
trash.data += len;
trash.area[trash.data++] = '\n';
- if (ci_putchk(si_ic(si), &trash) == -1) {
- si_rx_room_blk(si);
+ if (ci_putchk(cs_ic(cs), &trash) == -1) {
+ si_rx_room_blk(cs->si);
ret = 0;
break;
}
@@ -413,18 +415,18 @@
HA_RWLOCK_WRLOCK(LOGSRV_LOCK, &ring->lock);
LIST_APPEND(&ring->waiters, &appctx->wait_entry);
HA_RWLOCK_WRUNLOCK(LOGSRV_LOCK, &ring->lock);
- si_rx_endp_done(si);
+ si_rx_endp_done(cs->si);
}
HA_SPIN_UNLOCK(SFT_LOCK, &sft->lock);
/* always drain data from server */
- co_skip(si_oc(si), si_oc(si)->output);
+ co_skip(cs_oc(cs), cs_oc(cs)->output);
return;
close:
- si_shutw(si);
- si_shutr(si);
- si_ic(si)->flags |= CF_READ_NULL;
+ si_shutw(cs->si);
+ si_shutr(cs->si);
+ cs_ic(cs)->flags |= CF_READ_NULL;
}
/*
@@ -433,8 +435,8 @@
*/
static void sink_forward_oc_io_handler(struct appctx *appctx)
{
- struct stream_interface *si = cs_si(appctx->owner);
- struct stream *s = si_strm(si);
+ struct conn_stream *cs = appctx->owner;
+ struct stream *s = __cs_strm(cs);
struct sink *sink = strm_fe(s)->parent;
struct sink_forward_target *sft = appctx->ctx.sft.ptr;
struct ring *ring = sink->ctx.ring;
@@ -452,25 +454,25 @@
* and we don't want expire on this case
* with a syslog server
*/
- si_oc(si)->rex = TICK_ETERNITY;
+ cs_oc(cs)->rex = TICK_ETERNITY;
/* rto should not change but it seems the case */
- si_oc(si)->rto = TICK_ETERNITY;
+ cs_oc(cs)->rto = TICK_ETERNITY;
/* an error was detected */
- if (unlikely(si_ic(si)->flags & (CF_WRITE_ERROR|CF_SHUTW)))
+ if (unlikely(cs_ic(cs)->flags & (CF_WRITE_ERROR|CF_SHUTW)))
goto close;
/* con closed by server side */
- if ((si_oc(si)->flags & CF_SHUTW))
+ if ((cs_oc(cs)->flags & CF_SHUTW))
goto close;
/* if the connection is not established, inform the stream that we want
* to be notified whenever the connection completes.
*/
- if (si_opposite(si)->state < SI_ST_EST) {
- si_cant_get(si);
- si_rx_conn_blk(si);
- si_rx_endp_more(si);
+ if (cs_opposite(cs)->si->state < SI_ST_EST) {
+ si_cant_get(cs->si);
+ si_rx_conn_blk(cs->si);
+ si_rx_endp_more(cs->si);
return;
}
@@ -506,7 +508,7 @@
* the message so that we can take our reference there if we have to
* stop before the end (ret=0).
*/
- if (si_opposite(si)->state == SI_ST_EST) {
+ if (cs_opposite(cs)->si->state == SI_ST_EST) {
/* we were already there, adjust the offset to be relative to
* the buffer's head and remove us from the counter.
*/
@@ -538,8 +540,8 @@
trash.data += b_getblk(buf, p + 1, msg_len, ofs + cnt);
- if (ci_putchk(si_ic(si), &trash) == -1) {
- si_rx_room_blk(si);
+ if (ci_putchk(cs_ic(cs), &trash) == -1) {
+ si_rx_room_blk(cs->si);
ret = 0;
break;
}
@@ -557,37 +559,25 @@
HA_RWLOCK_WRLOCK(LOGSRV_LOCK, &ring->lock);
LIST_APPEND(&ring->waiters, &appctx->wait_entry);
HA_RWLOCK_WRUNLOCK(LOGSRV_LOCK, &ring->lock);
- si_rx_endp_done(si);
+ si_rx_endp_done(cs->si);
}
HA_SPIN_UNLOCK(SFT_LOCK, &sft->lock);
/* always drain data from server */
- co_skip(si_oc(si), si_oc(si)->output);
+ co_skip(cs_oc(cs), cs_oc(cs)->output);
return;
close:
- si_shutw(si);
- si_shutr(si);
- si_ic(si)->flags |= CF_READ_NULL;
+ si_shutw(cs->si);
+ si_shutr(cs->si);
+ cs_ic(cs)->flags |= CF_READ_NULL;
}
void __sink_forward_session_deinit(struct sink_forward_target *sft)
{
- struct stream_interface *si;
- struct stream *s;
+ struct stream *s = __cs_strm(sft->appctx->owner);
struct sink *sink;
- if (!sft->appctx)
- return;
-
- si = cs_si(sft->appctx->owner);
- if (!si)
- return;
-
- s = si_strm(si);
- if (!s)
- return;
-
sink = strm_fe(s)->parent;
if (!sink)
return;
diff --git a/src/ssl_ckch.c b/src/ssl_ckch.c
index 3bd0610..f8dc3cb 100644
--- a/src/ssl_ckch.c
+++ b/src/ssl_ckch.c
@@ -29,6 +29,8 @@
#include <haproxy/base64.h>
#include <haproxy/channel.h>
#include <haproxy/cli.h>
+#include <haproxy/conn_stream.h>
+#include <haproxy/cs_utils.h>
#include <haproxy/errors.h>
#include <haproxy/ssl_ckch.h>
#include <haproxy/ssl_sock.h>
@@ -1233,7 +1235,7 @@
{
struct buffer *trash = alloc_trash_chunk();
struct ebmb_node *node;
- struct stream_interface *si = cs_si(appctx->owner);
+ struct conn_stream *cs = appctx->owner;
struct ckch_store *ckchs;
if (trash == NULL)
@@ -1258,8 +1260,8 @@
chunk_appendf(trash, "%s\n", ckchs->path);
node = ebmb_next(node);
- if (ci_putchk(si_ic(si), trash) == -1) {
- si_rx_room_blk(si);
+ if (ci_putchk(cs_ic(cs), trash) == -1) {
+ si_rx_room_blk(cs->si);
goto yield;
}
}
@@ -1632,7 +1634,7 @@
/* IO handler of the details "show ssl cert <filename>" */
static int cli_io_handler_show_cert_detail(struct appctx *appctx)
{
- struct stream_interface *si = cs_si(appctx->owner);
+ struct conn_stream *cs = appctx->owner;
struct ckch_store *ckchs = appctx->ctx.cli.p0;
struct buffer *out = alloc_trash_chunk();
int retval = 0;
@@ -1662,8 +1664,8 @@
ckch_store_show_ocsp_certid(ckchs, out);
end:
- if (ci_putchk(si_ic(si), out) == -1) {
- si_rx_room_blk(si);
+ if (ci_putchk(cs_ic(cs), out) == -1) {
+ si_rx_room_blk(cs->si);
goto yield;
}
@@ -1680,7 +1682,7 @@
static int cli_io_handler_show_cert_ocsp_detail(struct appctx *appctx)
{
#if ((defined SSL_CTRL_SET_TLSEXT_STATUS_REQ_CB && !defined OPENSSL_NO_OCSP) && !defined OPENSSL_IS_BORINGSSL)
- struct stream_interface *si = cs_si(appctx->owner);
+ struct conn_stream *cs = appctx->owner;
struct ckch_store *ckchs = appctx->ctx.cli.p0;
struct buffer *out = alloc_trash_chunk();
int from_transaction = appctx->ctx.cli.i0;
@@ -1707,8 +1709,8 @@
goto end_no_putchk;
}
- if (ci_putchk(si_ic(si), out) == -1) {
- si_rx_room_blk(si);
+ if (ci_putchk(cs_ic(cs), out) == -1) {
+ si_rx_room_blk(cs->si);
goto yield;
}
@@ -1958,7 +1960,7 @@
*/
static int cli_io_handler_commit_cert(struct appctx *appctx)
{
- struct stream_interface *si = cs_si(appctx->owner);
+ struct conn_stream *cs = appctx->owner;
int y = 0;
char *err = NULL;
struct ckch_store *old_ckchs, *new_ckchs = NULL;
@@ -1968,7 +1970,7 @@
if (trash == NULL)
goto error;
- if (unlikely(si_ic(si)->flags & (CF_WRITE_ERROR|CF_SHUTW)))
+ if (unlikely(cs_ic(cs)->flags & (CF_WRITE_ERROR|CF_SHUTW)))
goto error;
while (1) {
@@ -1976,8 +1978,8 @@
case SETCERT_ST_INIT:
/* This state just print the update message */
chunk_printf(trash, "Committing %s", ckchs_transaction.path);
- if (ci_putchk(si_ic(si), trash) == -1) {
- si_rx_room_blk(si);
+ if (ci_putchk(cs_ic(cs), trash) == -1) {
+ si_rx_room_blk(cs->si);
goto yield;
}
appctx->st2 = SETCERT_ST_GEN;
@@ -2051,25 +2053,25 @@
chunk_appendf(trash, "\n");
chunk_appendf(trash, "Success!\n");
- if (ci_putchk(si_ic(si), trash) == -1)
- si_rx_room_blk(si);
+ if (ci_putchk(cs_ic(cs), trash) == -1)
+ si_rx_room_blk(cs->si);
free_trash_chunk(trash);
/* success: call the release function and don't come back */
return 1;
yield:
/* store the state */
- if (ci_putchk(si_ic(si), trash) == -1)
- si_rx_room_blk(si);
+ if (ci_putchk(cs_ic(cs), trash) == -1)
+ si_rx_room_blk(cs->si);
free_trash_chunk(trash);
- si_rx_endp_more(si); /* let's come back later */
+ si_rx_endp_more(cs->si); /* let's come back later */
return 0; /* should come back */
error:
/* spin unlock and free are done in the release function */
if (trash) {
chunk_appendf(trash, "\n%sFailed!\n", err);
- if (ci_putchk(si_ic(si), trash) == -1)
- si_rx_room_blk(si);
+ if (ci_putchk(cs_ic(cs), trash) == -1)
+ si_rx_room_blk(cs->si);
free_trash_chunk(trash);
}
/* error: call the release function and don't come back */
@@ -2698,7 +2700,7 @@
*/
static int cli_io_handler_commit_cafile_crlfile(struct appctx *appctx)
{
- struct stream_interface *si = cs_si(appctx->owner);
+ struct conn_stream *cs = appctx->owner;
int y = 0;
char *err = NULL;
struct cafile_entry *old_cafile_entry = NULL, *new_cafile_entry = NULL;
@@ -2708,7 +2710,7 @@
if (trash == NULL)
goto error;
- if (unlikely(si_ic(si)->flags & (CF_WRITE_ERROR|CF_SHUTW)))
+ if (unlikely(cs_ic(cs)->flags & (CF_WRITE_ERROR|CF_SHUTW)))
goto error;
while (1) {
@@ -2725,8 +2727,8 @@
default:
goto error;
}
- if (ci_putchk(si_ic(si), trash) == -1) {
- si_rx_room_blk(si);
+ if (ci_putchk(cs_ic(cs), trash) == -1) {
+ si_rx_room_blk(cs->si);
goto yield;
}
appctx->st2 = SETCERT_ST_GEN;
@@ -2836,25 +2838,25 @@
chunk_appendf(trash, "\n");
chunk_appendf(trash, "Success!\n");
- if (ci_putchk(si_ic(si), trash) == -1)
- si_rx_room_blk(si);
+ if (ci_putchk(cs_ic(cs), trash) == -1)
+ si_rx_room_blk(cs->si);
free_trash_chunk(trash);
/* success: call the release function and don't come back */
return 1;
yield:
/* store the state */
- if (ci_putchk(si_ic(si), trash) == -1)
- si_rx_room_blk(si);
+ if (ci_putchk(cs_ic(cs), trash) == -1)
+ si_rx_room_blk(cs->si);
free_trash_chunk(trash);
- si_rx_endp_more(si); /* let's come back later */
+ si_rx_endp_more(cs->si); /* let's come back later */
return 0; /* should come back */
error:
/* spin unlock and free are done in the release function */
if (trash) {
chunk_appendf(trash, "\n%sFailed!\n", err);
- if (ci_putchk(si_ic(si), trash) == -1)
- si_rx_room_blk(si);
+ if (ci_putchk(cs_ic(cs), trash) == -1)
+ si_rx_room_blk(cs->si);
free_trash_chunk(trash);
}
/* error: call the release function and don't come back */
@@ -2922,7 +2924,7 @@
/* IO handler of details "show ssl ca-file <filename[:index]>" */
static int cli_io_handler_show_cafile_detail(struct appctx *appctx)
{
- struct stream_interface *si = cs_si(appctx->owner);
+ struct conn_stream *cs = appctx->owner;
struct cafile_entry *cafile_entry = appctx->ctx.cli.p0;
struct buffer *out = alloc_trash_chunk();
int i;
@@ -2969,8 +2971,8 @@
}
end:
- if (ci_putchk(si_ic(si), out) == -1) {
- si_rx_room_blk(si);
+ if (ci_putchk(cs_ic(cs), out) == -1) {
+ si_rx_room_blk(cs->si);
goto yield;
}
@@ -3075,7 +3077,7 @@
{
struct buffer *trash = alloc_trash_chunk();
struct ebmb_node *node;
- struct stream_interface *si = cs_si(appctx->owner);
+ struct conn_stream *cs = appctx->owner;
struct cafile_entry *cafile_entry;
if (trash == NULL)
@@ -3108,8 +3110,8 @@
}
node = ebmb_next(node);
- if (ci_putchk(si_ic(si), trash) == -1) {
- si_rx_room_blk(si);
+ if (ci_putchk(cs_ic(cs), trash) == -1) {
+ si_rx_room_blk(cs->si);
goto yield;
}
}
@@ -3581,7 +3583,7 @@
/* IO handler of details "show ssl crl-file <filename[:index]>" */
static int cli_io_handler_show_crlfile_detail(struct appctx *appctx)
{
- struct stream_interface *si = cs_si(appctx->owner);
+ struct conn_stream *cs = appctx->owner;
struct cafile_entry *cafile_entry = appctx->ctx.cli.p0;
struct buffer *out = alloc_trash_chunk();
int i;
@@ -3628,8 +3630,8 @@
}
end:
- if (ci_putchk(si_ic(si), out) == -1) {
- si_rx_room_blk(si);
+ if (ci_putchk(cs_ic(cs), out) == -1) {
+ si_rx_room_blk(cs->si);
goto yield;
}
@@ -3711,7 +3713,7 @@
{
struct buffer *trash = alloc_trash_chunk();
struct ebmb_node *node;
- struct stream_interface *si = cs_si(appctx->owner);
+ struct conn_stream *cs = appctx->owner;
struct cafile_entry *cafile_entry;
if (trash == NULL)
@@ -3740,8 +3742,8 @@
}
node = ebmb_next(node);
- if (ci_putchk(si_ic(si), trash) == -1) {
- si_rx_room_blk(si);
+ if (ci_putchk(cs_ic(cs), trash) == -1) {
+ si_rx_room_blk(cs->si);
goto yield;
}
}
diff --git a/src/ssl_crtlist.c b/src/ssl_crtlist.c
index c1675fe..f2d6496 100644
--- a/src/ssl_crtlist.c
+++ b/src/ssl_crtlist.c
@@ -22,6 +22,8 @@
#include <haproxy/channel.h>
#include <haproxy/cli.h>
+#include <haproxy/conn_stream.h>
+#include <haproxy/cs_utils.h>
#include <haproxy/errors.h>
#include <haproxy/ssl_ckch.h>
#include <haproxy/ssl_crtlist.h>
@@ -887,7 +889,7 @@
static int cli_io_handler_dump_crtlist(struct appctx *appctx)
{
struct buffer *trash = alloc_trash_chunk();
- struct stream_interface *si = cs_si(appctx->owner);
+ struct conn_stream *cs = appctx->owner;
struct ebmb_node *lnode;
if (trash == NULL)
@@ -899,8 +901,8 @@
lnode = ebmb_first(&crtlists_tree);
while (lnode) {
chunk_appendf(trash, "%s\n", lnode->key);
- if (ci_putchk(si_ic(si), trash) == -1) {
- si_rx_room_blk(si);
+ if (ci_putchk(cs_ic(cs), trash) == -1) {
+ si_rx_room_blk(cs->si);
goto yield;
}
lnode = ebmb_next(lnode);
@@ -918,7 +920,7 @@
{
struct buffer *trash = alloc_trash_chunk();
struct crtlist *crtlist;
- struct stream_interface *si = cs_si(appctx->owner);
+ struct conn_stream *cs = appctx->owner;
struct crtlist_entry *entry;
if (trash == NULL)
@@ -930,8 +932,8 @@
if (entry == NULL) {
entry = LIST_ELEM((crtlist->ord_entries).n, typeof(entry), by_crtlist);
chunk_appendf(trash, "# %s\n", crtlist->node.key);
- if (ci_putchk(si_ic(si), trash) == -1) {
- si_rx_room_blk(si);
+ if (ci_putchk(cs_ic(cs), trash) == -1) {
+ si_rx_room_blk(cs->si);
goto yield;
}
}
@@ -949,8 +951,8 @@
dump_crtlist_filters(trash, entry);
chunk_appendf(trash, "\n");
- if (ci_putchk(si_ic(si), trash) == -1) {
- si_rx_room_blk(si);
+ if (ci_putchk(cs_ic(cs), trash) == -1) {
+ si_rx_room_blk(cs->si);
goto yield;
}
}
@@ -1041,7 +1043,7 @@
static int cli_io_handler_add_crtlist(struct appctx *appctx)
{
struct bind_conf_list *bind_conf_node;
- struct stream_interface *si = cs_si(appctx->owner);
+ struct conn_stream *cs = appctx->owner;
struct crtlist *crtlist = appctx->ctx.cli.p0;
struct crtlist_entry *entry = appctx->ctx.cli.p1;
struct ckch_store *store = entry->node.key;
@@ -1057,7 +1059,7 @@
/* for each bind_conf which use the crt-list, a new ckch_inst must be
* created.
*/
- if (unlikely(si_ic(si)->flags & (CF_WRITE_ERROR|CF_SHUTW)))
+ if (unlikely(cs_ic(cs)->flags & (CF_WRITE_ERROR|CF_SHUTW)))
goto error;
while (1) {
@@ -1065,8 +1067,8 @@
case SETCERT_ST_INIT:
/* This state just print the update message */
chunk_printf(trash, "Inserting certificate '%s' in crt-list '%s'", store->path, crtlist->node.key);
- if (ci_putchk(si_ic(si), trash) == -1) {
- si_rx_room_blk(si);
+ if (ci_putchk(cs_ic(cs), trash) == -1) {
+ si_rx_room_blk(cs->si);
goto yield;
}
appctx->st2 = SETCERT_ST_GEN;
@@ -1126,25 +1128,25 @@
if (errcode & ERR_WARN)
chunk_appendf(trash, "%s", err);
chunk_appendf(trash, "Success!\n");
- if (ci_putchk(si_ic(si), trash) == -1)
- si_rx_room_blk(si);
+ if (ci_putchk(cs_ic(cs), trash) == -1)
+ si_rx_room_blk(cs->si);
free_trash_chunk(trash);
/* success: call the release function and don't come back */
return 1;
yield:
/* store the state */
- if (ci_putchk(si_ic(si), trash) == -1)
- si_rx_room_blk(si);
+ if (ci_putchk(cs_ic(cs), trash) == -1)
+ si_rx_room_blk(cs->si);
free_trash_chunk(trash);
- si_rx_endp_more(si); /* let's come back later */
+ si_rx_endp_more(cs->si); /* let's come back later */
return 0; /* should come back */
error:
/* spin unlock and free are done in the release function */
if (trash) {
chunk_appendf(trash, "\n%sFailed!\n", err);
- if (ci_putchk(si_ic(si), trash) == -1)
- si_rx_room_blk(si);
+ if (ci_putchk(cs_ic(cs), trash) == -1)
+ si_rx_room_blk(cs->si);
free_trash_chunk(trash);
}
/* error: call the release function and don't come back */
diff --git a/src/ssl_sock.c b/src/ssl_sock.c
index a6314f5..bdd73e2 100644
--- a/src/ssl_sock.c
+++ b/src/ssl_sock.c
@@ -51,6 +51,8 @@
#include <haproxy/chunk.h>
#include <haproxy/cli.h>
#include <haproxy/connection.h>
+#include <haproxy/conn_stream.h>
+#include <haproxy/cs_utils.h>
#include <haproxy/dynbuf.h>
#include <haproxy/errors.h>
#include <haproxy/fd.h>
@@ -7194,7 +7196,7 @@
*/
static int cli_io_handler_tlskeys_files(struct appctx *appctx) {
- struct stream_interface *si = cs_si(appctx->owner);
+ struct conn_stream *cs = appctx->owner;
switch (appctx->st2) {
case STAT_ST_INIT:
@@ -7209,8 +7211,8 @@
else
chunk_appendf(&trash, "# id (file)\n");
- if (ci_putchk(si_ic(si), &trash) == -1) {
- si_rx_room_blk(si);
+ if (ci_putchk(cs_ic(cs), &trash) == -1) {
+ si_rx_room_blk(cs->si);
return 0;
}
@@ -7265,12 +7267,12 @@
chunk_appendf(&trash, "%d.%d <unknown>\n", ref->unique_id, appctx->ctx.cli.i1);
}
- if (ci_putchk(si_ic(si), &trash) == -1) {
+ if (ci_putchk(cs_ic(cs), &trash) == -1) {
/* let's try again later from this stream. We add ourselves into
* this stream's users so that it can remove us upon termination.
*/
HA_RWLOCK_RDUNLOCK(TLSKEYS_REF_LOCK, &ref->lock);
- si_rx_room_blk(si);
+ si_rx_room_blk(cs->si);
return 0;
}
appctx->ctx.cli.i1++;
@@ -7278,11 +7280,11 @@
HA_RWLOCK_RDUNLOCK(TLSKEYS_REF_LOCK, &ref->lock);
appctx->ctx.cli.i1 = 0;
}
- if (ci_putchk(si_ic(si), &trash) == -1) {
+ if (ci_putchk(cs_ic(cs), &trash) == -1) {
/* let's try again later from this stream. We add ourselves into
* this stream's users so that it can remove us upon termination.
*/
- si_rx_room_blk(si);
+ si_rx_room_blk(cs->si);
return 0;
}
@@ -7470,7 +7472,7 @@
struct buffer *trash = alloc_trash_chunk();
struct buffer *tmp = NULL;
struct ebmb_node *node;
- struct stream_interface *si = cs_si(appctx->owner);
+ struct conn_stream *cs = appctx->owner;
struct certificate_ocsp *ocsp = NULL;
BIO *bio = NULL;
int write = -1;
@@ -7525,8 +7527,8 @@
chunk_appendf(trash, "%s\n", tmp->area);
node = ebmb_next(node);
- if (ci_putchk(si_ic(si), trash) == -1) {
- si_rx_room_blk(si);
+ if (ci_putchk(cs_ic(cs), trash) == -1) {
+ si_rx_room_blk(cs->si);
goto yield;
}
}
@@ -7657,7 +7659,7 @@
{
struct buffer *trash = alloc_trash_chunk();
struct certificate_ocsp *ocsp = NULL;
- struct stream_interface *si = cs_si(appctx->owner);
+ struct conn_stream *cs = appctx->owner;
ocsp = appctx->ctx.cli.p0;
@@ -7669,8 +7671,8 @@
return 1;
}
- if (ci_putchk(si_ic(si), trash) == -1) {
- si_rx_room_blk(si);
+ if (ci_putchk(cs_ic(cs), trash) == -1) {
+ si_rx_room_blk(cs->si);
goto yield;
}
appctx->ctx.cli.p0 = NULL;
diff --git a/src/stats.c b/src/stats.c
index e18a516..a613a12 100644
--- a/src/stats.c
+++ b/src/stats.c
@@ -35,6 +35,8 @@
#include <haproxy/cli.h>
#include <haproxy/clock.h>
#include <haproxy/compression.h>
+#include <haproxy/conn_stream.h>
+#include <haproxy/cs_utils.h>
#include <haproxy/debug.h>
#include <haproxy/errors.h>
#include <haproxy/fd.h>
@@ -309,9 +311,9 @@
return 1;
}
-static const char *stats_scope_ptr(struct appctx *appctx, struct stream_interface *si)
+static const char *stats_scope_ptr(struct appctx *appctx, struct conn_stream *cs)
{
- struct channel *req = si_oc(si);
+ struct channel *req = cs_oc(cs);
struct htx *htx = htxbuf(&req->buf);
struct htx_blk *blk;
struct ist uri;
@@ -1807,12 +1809,12 @@
}
/* Dumps a frontend's line to the trash for the current proxy <px> and uses
- * the state from stream interface <si>. The caller is responsible for clearing
+ * the state from conn-stream <cs>. The caller is responsible for clearing
* the trash if needed. Returns non-zero if it emits anything, zero otherwise.
*/
-static int stats_dump_fe_stats(struct stream_interface *si, struct proxy *px)
+static int stats_dump_fe_stats(struct conn_stream *cs, struct proxy *px)
{
- struct appctx *appctx = __cs_appctx(si->cs);
+ struct appctx *appctx = __cs_appctx(cs);
struct field *stats = stat_l[STATS_DOMAIN_PROXY];
struct stats_module *mod;
size_t stats_count = ST_F_TOTAL_FIELDS;
@@ -1974,12 +1976,12 @@
}
/* Dumps a line for listener <l> and proxy <px> to the trash and uses the state
- * from stream interface <si>. The caller is responsible for clearing the trash
+ * from conn-stream <cs>. The caller is responsible for clearing the trash
* if needed. Returns non-zero if it emits anything, zero otherwise.
*/
-static int stats_dump_li_stats(struct stream_interface *si, struct proxy *px, struct listener *l)
+static int stats_dump_li_stats(struct conn_stream *cs, struct proxy *px, struct listener *l)
{
- struct appctx *appctx = __cs_appctx(si->cs);
+ struct appctx *appctx = __cs_appctx(cs);
struct field *stats = stat_l[STATS_DOMAIN_PROXY];
struct stats_module *mod;
size_t stats_count = ST_F_TOTAL_FIELDS;
@@ -2484,13 +2486,13 @@
}
/* Dumps a line for server <sv> and proxy <px> to the trash and uses the state
- * from stream interface <si>, and server state <state>. The caller is
+ * from conn-stream <cs>, and server state <state>. The caller is
* responsible for clearing the trash if needed. Returns non-zero if it emits
* anything, zero otherwise.
*/
-static int stats_dump_sv_stats(struct stream_interface *si, struct proxy *px, struct server *sv)
+static int stats_dump_sv_stats(struct conn_stream *cs, struct proxy *px, struct server *sv)
{
- struct appctx *appctx = __cs_appctx(si->cs);
+ struct appctx *appctx = __cs_appctx(cs);
struct stats_module *mod;
struct field *stats = stat_l[STATS_DOMAIN_PROXY];
size_t stats_count = ST_F_TOTAL_FIELDS;
@@ -2813,9 +2815,9 @@
* interface <si>. The caller is responsible for clearing the trash if needed.
* Returns non-zero if it emits anything, zero otherwise.
*/
-static int stats_dump_be_stats(struct stream_interface *si, struct proxy *px)
+static int stats_dump_be_stats(struct conn_stream *cs, struct proxy *px)
{
- struct appctx *appctx = __cs_appctx(si->cs);
+ struct appctx *appctx = __cs_appctx(cs);
struct field *stats = stat_l[STATS_DOMAIN_PROXY];
struct stats_module *mod;
size_t stats_count = ST_F_TOTAL_FIELDS;
@@ -2851,12 +2853,12 @@
}
/* Dumps the HTML table header for proxy <px> to the trash for and uses the state from
- * stream interface <si> and per-uri parameters <uri>. The caller is responsible
+ * conn-stream <cs> and per-uri parameters <uri>. The caller is responsible
* for clearing the trash if needed.
*/
-static void stats_dump_html_px_hdr(struct stream_interface *si, struct proxy *px)
+static void stats_dump_html_px_hdr(struct conn_stream *cs, struct proxy *px)
{
- struct appctx *appctx = __cs_appctx(si->cs);
+ struct appctx *appctx = __cs_appctx(cs);
char scope_txt[STAT_SCOPE_TXT_MAXLEN + sizeof STAT_SCOPE_PATTERN];
struct stats_module *mod;
int stats_module_len = 0;
@@ -2867,7 +2869,7 @@
/* scope_txt = search pattern + search query, appctx->ctx.stats.scope_len is always <= STAT_SCOPE_TXT_MAXLEN */
scope_txt[0] = 0;
if (appctx->ctx.stats.scope_len) {
- const char *scope_ptr = stats_scope_ptr(appctx, si);
+ const char *scope_ptr = stats_scope_ptr(appctx, cs);
strcpy(scope_txt, STAT_SCOPE_PATTERN);
memcpy(scope_txt + strlen(STAT_SCOPE_PATTERN), scope_ptr, appctx->ctx.stats.scope_len);
@@ -2961,11 +2963,11 @@
}
/* Dumps the HTML table trailer for proxy <px> to the trash for and uses the state from
- * stream interface <si>. The caller is responsible for clearing the trash if needed.
+ * conn_stream <cs>. The caller is responsible for clearing the trash if needed.
*/
-static void stats_dump_html_px_end(struct stream_interface *si, struct proxy *px)
+static void stats_dump_html_px_end(struct conn_stream *cs, struct proxy *px)
{
- struct appctx *appctx = __cs_appctx(si->cs);
+ struct appctx *appctx = __cs_appctx(cs);
chunk_appendf(&trash, "</table>");
@@ -2999,18 +3001,18 @@
}
/*
- * Dumps statistics for a proxy. The output is sent to the stream interface's
+ * Dumps statistics for a proxy. The output is sent to the conn-stream's
* input buffer. Returns 0 if it had to stop dumping data because of lack of
* buffer space, or non-zero if everything completed. This function is used
* both by the CLI and the HTTP entry points, and is able to dump the output
* in HTML or CSV formats. If the later, <uri> must be NULL.
*/
-int stats_dump_proxy_to_buffer(struct stream_interface *si, struct htx *htx,
+int stats_dump_proxy_to_buffer(struct conn_stream *cs, struct htx *htx,
struct proxy *px, struct uri_auth *uri)
{
- struct appctx *appctx = __cs_appctx(si->cs);
- struct stream *s = si_strm(si);
- struct channel *rep = si_ic(si);
+ struct appctx *appctx = __cs_appctx(cs);
+ struct stream *s = __cs_strm(cs);
+ struct channel *rep = cs_ic(cs);
struct server *sv, *svs; /* server and server-state, server-state=server or server->track */
struct listener *l;
@@ -3047,7 +3049,7 @@
* name does not match, skip it.
*/
if (appctx->ctx.stats.scope_len) {
- const char *scope_ptr = stats_scope_ptr(appctx, si);
+ const char *scope_ptr = stats_scope_ptr(appctx, cs);
if (strnistr(px->id, strlen(px->id), scope_ptr, appctx->ctx.stats.scope_len) == NULL)
return 1;
@@ -3063,7 +3065,7 @@
case STAT_PX_ST_TH:
if (appctx->ctx.stats.flags & STAT_FMT_HTML) {
- stats_dump_html_px_hdr(si, px);
+ stats_dump_html_px_hdr(cs, px);
if (!stats_putchk(rep, htx, &trash))
goto full;
}
@@ -3073,7 +3075,7 @@
case STAT_PX_ST_FE:
/* print the frontend */
- if (stats_dump_fe_stats(si, px)) {
+ if (stats_dump_fe_stats(cs, px)) {
if (!stats_putchk(rep, htx, &trash))
goto full;
}
@@ -3107,7 +3109,7 @@
}
/* print the frontend */
- if (stats_dump_li_stats(si, px, l)) {
+ if (stats_dump_li_stats(cs, px, l)) {
if (!stats_putchk(rep, htx, &trash))
goto full;
}
@@ -3169,7 +3171,7 @@
continue;
}
- if (stats_dump_sv_stats(si, px, sv)) {
+ if (stats_dump_sv_stats(cs, px, sv)) {
if (!stats_putchk(rep, htx, &trash))
goto full;
}
@@ -3180,7 +3182,7 @@
case STAT_PX_ST_BE:
/* print the backend */
- if (stats_dump_be_stats(si, px)) {
+ if (stats_dump_be_stats(cs, px)) {
if (!stats_putchk(rep, htx, &trash))
goto full;
}
@@ -3190,7 +3192,7 @@
case STAT_PX_ST_END:
if (appctx->ctx.stats.flags & STAT_FMT_HTML) {
- stats_dump_html_px_end(si, px);
+ stats_dump_html_px_end(cs, px);
if (!stats_putchk(rep, htx, &trash))
goto full;
}
@@ -3207,7 +3209,7 @@
}
full:
- si_rx_room_blk(si);
+ si_rx_room_blk(cs->si);
return 0;
}
@@ -3378,15 +3380,15 @@
}
/* Dumps the HTML stats information block to the trash for and uses the state from
- * stream interface <si> and per-uri parameters <uri>. The caller is responsible
+ * conn-stream <cs> and per-uri parameters <uri>. The caller is responsible
* for clearing the trash if needed.
*/
-static void stats_dump_html_info(struct stream_interface *si, struct uri_auth *uri)
+static void stats_dump_html_info(struct conn_stream *cs, struct uri_auth *uri)
{
- struct appctx *appctx = __cs_appctx(si->cs);
+ struct appctx *appctx = __cs_appctx(cs);
unsigned int up = (now.tv_sec - start_date.tv_sec);
char scope_txt[STAT_SCOPE_TXT_MAXLEN + sizeof STAT_SCOPE_PATTERN];
- const char *scope_ptr = stats_scope_ptr(appctx, si);
+ const char *scope_ptr = stats_scope_ptr(appctx, cs);
unsigned long long bps = (unsigned long long)read_freq_ctr(&global.out_32bps) * 32;
/* Turn the bytes per second to bits per second and take care of the
@@ -3661,12 +3663,12 @@
/* Uses <appctx.ctx.stats.obj1> as a pointer to the current proxy and <obj2> as
* a pointer to the current server/listener.
*/
-static int stats_dump_proxies(struct stream_interface *si,
+static int stats_dump_proxies(struct conn_stream *cs,
struct htx *htx,
struct uri_auth *uri)
{
- struct appctx *appctx = __cs_appctx(si->cs);
- struct channel *rep = si_ic(si);
+ struct appctx *appctx = __cs_appctx(cs);
+ struct channel *rep = cs_ic(cs);
struct proxy *px;
/* dump proxies */
@@ -3687,7 +3689,7 @@
*/
if (!(px->flags & PR_FL_DISABLED) && px->uuid > 0 &&
(px->cap & (PR_CAP_FE | PR_CAP_BE)) && !(px->cap & PR_CAP_INT)) {
- if (stats_dump_proxy_to_buffer(si, htx, px, uri) == 0)
+ if (stats_dump_proxy_to_buffer(cs, htx, px, uri) == 0)
return 0;
}
@@ -3698,22 +3700,22 @@
return 1;
full:
- si_rx_room_blk(si);
+ si_rx_room_blk(cs->si);
return 0;
}
-/* This function dumps statistics onto the stream interface's read buffer in
+/* This function dumps statistics onto the conn-stream's read buffer in
* either CSV or HTML format. <uri> contains some HTML-specific parameters that
* are ignored for CSV format (hence <uri> may be NULL there). It returns 0 if
* it had to stop writing data and an I/O is needed, 1 if the dump is finished
* and the stream must be closed, or -1 in case of any error. This function is
* used by both the CLI and the HTTP handlers.
*/
-static int stats_dump_stat_to_buffer(struct stream_interface *si, struct htx *htx,
+static int stats_dump_stat_to_buffer(struct conn_stream *cs, struct htx *htx,
struct uri_auth *uri)
{
- struct appctx *appctx = __cs_appctx(si->cs);
- struct channel *rep = si_ic(si);
+ struct appctx *appctx = __cs_appctx(cs);
+ struct channel *rep = cs_ic(cs);
enum stats_domain domain = appctx->ctx.stats.domain;
chunk_reset(&trash);
@@ -3745,7 +3747,7 @@
case STAT_ST_INFO:
if (appctx->ctx.stats.flags & STAT_FMT_HTML) {
- stats_dump_html_info(si, uri);
+ stats_dump_html_info(cs, uri);
if (!stats_putchk(rep, htx, &trash))
goto full;
}
@@ -3760,7 +3762,7 @@
case STAT_ST_LIST:
switch (domain) {
case STATS_DOMAIN_RESOLVERS:
- if (!stats_dump_resolvers(si, stat_l[domain],
+ if (!stats_dump_resolvers(cs, stat_l[domain],
stat_count[domain],
&stats_module_list[domain])) {
return 0;
@@ -3770,7 +3772,7 @@
case STATS_DOMAIN_PROXY:
default:
/* dump proxies */
- if (!stats_dump_proxies(si, htx, uri))
+ if (!stats_dump_proxies(cs, htx, uri))
return 0;
break;
}
@@ -3801,7 +3803,7 @@
}
full:
- si_rx_room_blk(si);
+ si_rx_room_blk(cs->si);
return 0;
}
@@ -3811,10 +3813,10 @@
* Parse the posted data and enable/disable servers if necessary.
* Returns 1 if request was parsed or zero if it needs more data.
*/
-static int stats_process_http_post(struct stream_interface *si)
+static int stats_process_http_post(struct conn_stream *cs)
{
- struct stream *s = si_strm(si);
- struct appctx *appctx = __cs_appctx(si->cs);
+ struct stream *s = __cs_strm(cs);
+ struct appctx *appctx = __cs_appctx(cs);
struct proxy *px = NULL;
struct server *sv = NULL;
@@ -4146,11 +4148,11 @@
}
-static int stats_send_http_headers(struct stream_interface *si, struct htx *htx)
+static int stats_send_http_headers(struct conn_stream *cs, struct htx *htx)
{
- struct stream *s = si_strm(si);
+ struct stream *s = __cs_strm(cs);
struct uri_auth *uri = s->be->uri_auth;
- struct appctx *appctx = __cs_appctx(si->cs);
+ struct appctx *appctx = __cs_appctx(cs);
struct htx_sl *sl;
unsigned int flags;
@@ -4194,24 +4196,24 @@
full:
htx_reset(htx);
- si_rx_room_blk(si);
+ si_rx_room_blk(cs->si);
return 0;
}
-static int stats_send_http_redirect(struct stream_interface *si, struct htx *htx)
+static int stats_send_http_redirect(struct conn_stream *cs, struct htx *htx)
{
char scope_txt[STAT_SCOPE_TXT_MAXLEN + sizeof STAT_SCOPE_PATTERN];
- struct stream *s = si_strm(si);
+ struct stream *s = __cs_strm(cs);
struct uri_auth *uri = s->be->uri_auth;
- struct appctx *appctx = __cs_appctx(si->cs);
+ struct appctx *appctx = __cs_appctx(cs);
struct htx_sl *sl;
unsigned int flags;
/* scope_txt = search pattern + search query, appctx->ctx.stats.scope_len is always <= STAT_SCOPE_TXT_MAXLEN */
scope_txt[0] = 0;
if (appctx->ctx.stats.scope_len) {
- const char *scope_ptr = stats_scope_ptr(appctx, si);
+ const char *scope_ptr = stats_scope_ptr(appctx, cs);
strcpy(scope_txt, STAT_SCOPE_PATTERN);
memcpy(scope_txt + strlen(STAT_SCOPE_PATTERN), scope_ptr, appctx->ctx.stats.scope_len);
@@ -4253,7 +4255,7 @@
full:
htx_reset(htx);
- si_rx_room_blk(si);
+ si_rx_room_blk(cs->si);
return 0;
}
@@ -4265,10 +4267,10 @@
*/
static void http_stats_io_handler(struct appctx *appctx)
{
- struct stream_interface *si = cs_si(appctx->owner);
- struct stream *s = si_strm(si);
- struct channel *req = si_oc(si);
- struct channel *res = si_ic(si);
+ struct conn_stream *cs = appctx->owner;
+ struct stream *s = __cs_strm(cs);
+ struct channel *req = cs_oc(cs);
+ struct channel *res = cs_ic(cs);
struct htx *req_htx, *res_htx;
/* only proxy stats are available via http */
@@ -4276,12 +4278,12 @@
res_htx = htx_from_buf(&res->buf);
- if (unlikely(si->state == SI_ST_DIS || si->state == SI_ST_CLO))
+ if (unlikely(cs->si->state == SI_ST_DIS || cs->si->state == SI_ST_CLO))
goto out;
/* Check if the input buffer is available. */
if (!b_size(&res->buf)) {
- si_rx_room_blk(si);
+ si_rx_room_blk(cs->si);
goto out;
}
@@ -4291,7 +4293,7 @@
/* all states are processed in sequence */
if (appctx->st0 == STAT_HTTP_HEAD) {
- if (stats_send_http_headers(si, res_htx)) {
+ if (stats_send_http_headers(cs, res_htx)) {
if (s->txn->meth == HTTP_METH_HEAD)
appctx->st0 = STAT_HTTP_DONE;
else
@@ -4300,19 +4302,19 @@
}
if (appctx->st0 == STAT_HTTP_DUMP) {
- if (stats_dump_stat_to_buffer(si, res_htx, s->be->uri_auth))
+ if (stats_dump_stat_to_buffer(cs, res_htx, s->be->uri_auth))
appctx->st0 = STAT_HTTP_DONE;
}
if (appctx->st0 == STAT_HTTP_POST) {
- if (stats_process_http_post(si))
+ if (stats_process_http_post(cs))
appctx->st0 = STAT_HTTP_LAST;
else if (req->flags & CF_SHUTR)
appctx->st0 = STAT_HTTP_DONE;
}
if (appctx->st0 == STAT_HTTP_LAST) {
- if (stats_send_http_redirect(si, res_htx))
+ if (stats_send_http_redirect(cs, res_htx))
appctx->st0 = STAT_HTTP_DONE;
}
@@ -4324,13 +4326,13 @@
*/
if (htx_is_empty(res_htx)) {
if (!htx_add_endof(res_htx, HTX_BLK_EOT)) {
- si_rx_room_blk(si);
+ si_rx_room_blk(cs->si);
goto out;
}
channel_add_input(res, 1);
}
res_htx->flags |= HTX_FL_EOM;
- si->cs->endp->flags |= CS_EP_EOI;
+ cs->endp->flags |= CS_EP_EOI;
res->flags |= CF_EOI;
appctx->st0 = STAT_HTTP_END;
}
@@ -4338,7 +4340,7 @@
if (appctx->st0 == STAT_HTTP_END) {
if (!(res->flags & CF_SHUTR)) {
res->flags |= CF_READ_NULL;
- si_shutr(si);
+ si_shutr(cs->si);
}
/* eat the whole request */
@@ -4359,7 +4361,7 @@
*/
htx_to_buf(res_htx, &res->buf);
if (!channel_is_empty(res))
- si_stop_get(si);
+ si_stop_get(cs->si);
}
/* Dump all fields from <info> into <out> using the "show info" format (name: value) */
@@ -4526,13 +4528,13 @@
return 1;
}
-/* This function dumps information onto the stream interface's read buffer.
+/* This function dumps information onto the conn-stream's read buffer.
* It returns 0 as long as it does not complete, non-zero upon completion.
* No state is used.
*/
-static int stats_dump_info_to_buffer(struct stream_interface *si)
+static int stats_dump_info_to_buffer(struct conn_stream *cs)
{
- struct appctx *appctx = __cs_appctx(si->cs);
+ struct appctx *appctx = __cs_appctx(cs);
if (!stats_fill_info(info, INF_TOTAL_FIELDS, appctx->ctx.stats.flags))
return 0;
@@ -4546,8 +4548,8 @@
else
stats_dump_info_fields(&trash, info, appctx->ctx.stats.flags);
- if (ci_putchk(si_ic(si), &trash) == -1) {
- si_rx_room_blk(si);
+ if (ci_putchk(cs_ic(cs), &trash) == -1) {
+ si_rx_room_blk(cs->si);
return 0;
}
@@ -4768,14 +4770,14 @@
* It returns 0 as long as it does not complete, non-zero upon completion.
* No state is used.
*/
-static int stats_dump_json_schema_to_buffer(struct stream_interface *si)
+static int stats_dump_json_schema_to_buffer(struct conn_stream *cs)
{
chunk_reset(&trash);
stats_dump_json_schema(&trash);
- if (ci_putchk(si_ic(si), &trash) == -1) {
- si_rx_room_blk(si);
+ if (ci_putchk(cs_ic(cs), &trash) == -1) {
+ si_rx_room_blk(cs->si);
return 0;
}
@@ -4927,7 +4929,7 @@
appctx->ctx.stats.scope_len = 0;
appctx->ctx.stats.flags = STAT_SHNODE | STAT_SHDESC;
- if ((strm_li(si_strm(cs_si(appctx->owner)))->bind_conf->level & ACCESS_LVL_MASK) >= ACCESS_LVL_OPER)
+ if ((strm_li(__cs_strm(appctx->owner))->bind_conf->level & ACCESS_LVL_MASK) >= ACCESS_LVL_OPER)
appctx->ctx.stats.flags |= STAT_SHLGNDS;
/* proxy is the default domain */
@@ -4983,20 +4985,20 @@
static int cli_io_handler_dump_info(struct appctx *appctx)
{
- return stats_dump_info_to_buffer(cs_si(appctx->owner));
+ return stats_dump_info_to_buffer(appctx->owner);
}
-/* This I/O handler runs as an applet embedded in a stream interface. It is
+/* This I/O handler runs as an applet embedded in a conn-stream. It is
* used to send raw stats over a socket.
*/
static int cli_io_handler_dump_stat(struct appctx *appctx)
{
- return stats_dump_stat_to_buffer(cs_si(appctx->owner), NULL, NULL);
+ return stats_dump_stat_to_buffer(appctx->owner, NULL, NULL);
}
static int cli_io_handler_dump_json_schema(struct appctx *appctx)
{
- return stats_dump_json_schema_to_buffer(cs_si(appctx->owner));
+ return stats_dump_json_schema_to_buffer(appctx->owner);
}
int stats_allocate_proxy_counters_internal(struct extra_counters **counters,
diff --git a/src/stick_table.c b/src/stick_table.c
index 667123c..664de04 100644
--- a/src/stick_table.c
+++ b/src/stick_table.c
@@ -22,6 +22,8 @@
#include <haproxy/arg.h>
#include <haproxy/cfgparse.h>
#include <haproxy/cli.h>
+#include <haproxy/conn_stream.h>
+#include <haproxy/cs_utils.h>
#include <haproxy/dict.h>
#include <haproxy/errors.h>
#include <haproxy/global.h>
@@ -4237,10 +4239,10 @@
* and needs to be called again, otherwise non-zero.
*/
static int table_dump_head_to_buffer(struct buffer *msg,
- struct stream_interface *si,
+ struct conn_stream *cs,
struct stktable *t, struct stktable *target)
{
- struct stream *s = si_strm(si);
+ struct stream *s = __cs_strm(cs);
chunk_appendf(msg, "# table: %s, type: %s, size:%d, used:%d\n",
t->id, stktable_types[t->type].kw, t->size, t->current);
@@ -4250,8 +4252,8 @@
if (target && (strm_li(s)->bind_conf->level & ACCESS_LVL_MASK) < ACCESS_LVL_OPER)
chunk_appendf(msg, "# contents not dumped due to insufficient privileges\n");
- if (ci_putchk(si_ic(si), msg) == -1) {
- si_rx_room_blk(si);
+ if (ci_putchk(cs_ic(cs), msg) == -1) {
+ si_rx_room_blk(cs->si);
return 0;
}
@@ -4263,7 +4265,7 @@
* and needs to be called again, otherwise non-zero.
*/
static int table_dump_entry_to_buffer(struct buffer *msg,
- struct stream_interface *si,
+ struct conn_stream *cs,
struct stktable *t, struct stksess *entry)
{
int dt;
@@ -4377,8 +4379,8 @@
}
chunk_appendf(msg, "\n");
- if (ci_putchk(si_ic(si), msg) == -1) {
- si_rx_room_blk(si);
+ if (ci_putchk(cs_ic(cs), msg) == -1) {
+ si_rx_room_blk(cs->si);
return 0;
}
@@ -4391,7 +4393,7 @@
*/
static int table_process_entry_per_key(struct appctx *appctx, char **args)
{
- struct stream_interface *si = cs_si(appctx->owner);
+ struct conn_stream *cs = appctx->owner;
struct stktable *t = appctx->ctx.table.target;
struct stksess *ts;
uint32_t uint32_key;
@@ -4457,12 +4459,12 @@
if (!ts)
return 1;
chunk_reset(&trash);
- if (!table_dump_head_to_buffer(&trash, si, t, t)) {
+ if (!table_dump_head_to_buffer(&trash, cs, t, t)) {
stktable_release(t, ts);
return 0;
}
HA_RWLOCK_RDLOCK(STK_SESS_LOCK, &ts->lock);
- if (!table_dump_entry_to_buffer(&trash, si, t, ts)) {
+ if (!table_dump_entry_to_buffer(&trash, cs, t, ts)) {
HA_RWLOCK_RDUNLOCK(STK_SESS_LOCK, &ts->lock);
stktable_release(t, ts);
return 0;
@@ -4646,8 +4648,8 @@
*/
static int cli_io_handler_table(struct appctx *appctx)
{
- struct stream_interface *si = cs_si(appctx->owner);
- struct stream *s = si_strm(si);
+ struct conn_stream *cs = appctx->owner;
+ struct stream *s = __cs_strm(cs);
struct ebmb_node *eb;
int skip_entry;
int show = appctx->ctx.table.action == STK_CLI_ACT_SHOW;
@@ -4664,7 +4666,7 @@
* data though.
*/
- if (unlikely(si_ic(si)->flags & (CF_WRITE_ERROR|CF_SHUTW))) {
+ if (unlikely(cs_ic(cs)->flags & (CF_WRITE_ERROR|CF_SHUTW))) {
/* in case of abort, remove any refcount we might have set on an entry */
if (appctx->st2 == STAT_ST_LIST) {
stksess_kill_if_expired(appctx->ctx.table.t, appctx->ctx.table.entry, 1);
@@ -4694,7 +4696,7 @@
}
if (appctx->ctx.table.t->size) {
- if (show && !table_dump_head_to_buffer(&trash, si, appctx->ctx.table.t, appctx->ctx.table.target))
+ if (show && !table_dump_head_to_buffer(&trash, cs, appctx->ctx.table.t, appctx->ctx.table.target))
return 0;
if (appctx->ctx.table.target &&
@@ -4770,7 +4772,7 @@
}
if (show && !skip_entry &&
- !table_dump_entry_to_buffer(&trash, si, appctx->ctx.table.t, appctx->ctx.table.entry)) {
+ !table_dump_entry_to_buffer(&trash, cs, appctx->ctx.table.t, appctx->ctx.table.entry)) {
HA_RWLOCK_RDUNLOCK(STK_SESS_LOCK, &appctx->ctx.table.entry->lock);
return 0;
}
diff --git a/src/stream.c b/src/stream.c
index 535e5fa..833be0e 100644
--- a/src/stream.c
+++ b/src/stream.c
@@ -29,6 +29,8 @@
#include <haproxy/check.h>
#include <haproxy/cli.h>
#include <haproxy/connection.h>
+#include <haproxy/conn_stream.h>
+#include <haproxy/cs_utils.h>
#include <haproxy/dict.h>
#include <haproxy/dynbuf.h>
#include <haproxy/fd.h>
@@ -2766,7 +2768,7 @@
if (!appctx)
return;
ptr = appctx;
- s = si_strm(cs_si(appctx->owner));
+ s = __cs_strm(appctx->owner);
if (!s)
return;
}
@@ -3098,13 +3100,13 @@
* buffer is full and it needs to be called again, otherwise non-zero. It is
* designed to be called from stats_dump_strm_to_buffer() below.
*/
-static int stats_dump_full_strm_to_buffer(struct stream_interface *si, struct stream *strm)
+static int stats_dump_full_strm_to_buffer(struct conn_stream *cs, struct stream *strm)
{
- struct appctx *appctx = __cs_appctx(si->cs);
+ struct appctx *appctx = __cs_appctx(cs);
+ struct conn_stream *csf, *csb;
struct tm tm;
extern const char *monthname[12];
char pn[INET6_ADDRSTRLEN];
- struct conn_stream *cs;
struct connection *conn;
struct appctx *tmpctx;
@@ -3113,7 +3115,7 @@
if (appctx->ctx.sess.section > 0 && appctx->ctx.sess.uid != strm->uniq_id) {
/* stream changed, no need to go any further */
chunk_appendf(&trash, " *** session terminated while we were watching it ***\n");
- if (ci_putchk(si_ic(si), &trash) == -1)
+ if (ci_putchk(cs_ic(cs), &trash) == -1)
goto full;
goto done;
}
@@ -3272,17 +3274,17 @@
TICKS_TO_MS(1000)) : "<NEVER>",
strm->csb->si->err_type, strm->csb->si->wait_event.events);
- cs = strm->csf;
- chunk_appendf(&trash, " cs=%p csf=0x%08x endp=%p,0x%08x\n", cs, cs->flags, cs->endp->target, cs->endp->flags);
+ csf = strm->csf;
+ chunk_appendf(&trash, " cs=%p csf=0x%08x endp=%p,0x%08x\n", csf, csf->flags, csf->endp->target, csf->endp->flags);
- if ((conn = cs_conn(cs)) != NULL) {
+ if ((conn = cs_conn(csf)) != NULL) {
chunk_appendf(&trash,
" co0=%p ctrl=%s xprt=%s mux=%s data=%s target=%s:%p\n",
conn,
conn_get_ctrl_name(conn),
conn_get_xprt_name(conn),
conn_get_mux_name(conn),
- cs_get_data_name(cs),
+ cs_get_data_name(csf),
obj_type_name(conn->target),
obj_base_ptr(conn->target));
@@ -3295,7 +3297,7 @@
conn_fd(conn) >= 0 ? fdtab[conn->handle.fd].thread_mask: 0);
}
- else if ((tmpctx = cs_appctx(cs)) != NULL) {
+ else if ((tmpctx = cs_appctx(csf)) != NULL) {
chunk_appendf(&trash,
" app0=%p st0=%d st1=%d st2=%d applet=%s tmask=0x%lx nice=%d calls=%u rate=%u cpu=%llu lat=%llu\n",
tmpctx,
@@ -3308,16 +3310,16 @@
(unsigned long long)tmpctx->t->cpu_time, (unsigned long long)tmpctx->t->lat_time);
}
- cs = strm->csb;
- chunk_appendf(&trash, " cs=%p csb=0x%08x endp=%p,0x%08x\n", cs, cs->flags, cs->endp->target, cs->endp->flags);
- if ((conn = cs_conn(cs)) != NULL) {
+ csb = strm->csb;
+ chunk_appendf(&trash, " cs=%p csb=0x%08x endp=%p,0x%08x\n", csb, csb->flags, csb->endp->target, csb->endp->flags);
+ if ((conn = cs_conn(csb)) != NULL) {
chunk_appendf(&trash,
" co1=%p ctrl=%s xprt=%s mux=%s data=%s target=%s:%p\n",
conn,
conn_get_ctrl_name(conn),
conn_get_xprt_name(conn),
conn_get_mux_name(conn),
- cs_get_data_name(cs),
+ cs_get_data_name(csb),
obj_type_name(conn->target),
obj_base_ptr(conn->target));
@@ -3330,7 +3332,7 @@
conn_fd(conn) >= 0 ? fdtab[conn->handle.fd].thread_mask: 0);
}
- else if ((tmpctx = cs_appctx(cs)) != NULL) {
+ else if ((tmpctx = cs_appctx(csb)) != NULL) {
chunk_appendf(&trash,
" app1=%p st0=%d st1=%d st2=%d applet=%s tmask=0x%lx nice=%d calls=%u rate=%u cpu=%llu lat=%llu\n",
tmpctx,
@@ -3436,7 +3438,7 @@
chunk_appendf(&trash, " current_rule=\"%s\" [%s:%d]\n", rule->kw->kw, rule->conf.file, rule->conf.line);
}
- if (ci_putchk(si_ic(si), &trash) == -1)
+ if (ci_putchk(cs_ic(cs), &trash) == -1)
goto full;
/* use other states to dump the contents */
@@ -3469,7 +3471,7 @@
/* let's set our own stream's epoch to the current one and increment
* it so that we know which streams were already there before us.
*/
- si_strm(cs_si(appctx->owner))->stream_epoch = _HA_ATOMIC_FETCH_ADD(&stream_epoch, 1);
+ __cs_strm(appctx->owner)->stream_epoch = _HA_ATOMIC_FETCH_ADD(&stream_epoch, 1);
return 0;
}
@@ -3480,12 +3482,12 @@
*/
static int cli_io_handler_dump_sess(struct appctx *appctx)
{
- struct stream_interface *si = cs_si(appctx->owner);
+ struct conn_stream *cs = appctx->owner;
struct connection *conn;
thread_isolate();
- if (unlikely(si_ic(si)->flags & (CF_WRITE_ERROR|CF_SHUTW))) {
+ if (unlikely(cs_ic(cs)->flags & (CF_WRITE_ERROR|CF_SHUTW))) {
/* If we're forced to shut down, we might have to remove our
* reference to the last stream being dumped.
*/
@@ -3532,7 +3534,7 @@
else {
/* check if we've found a stream created after issuing the "show sess" */
curr_strm = LIST_ELEM(appctx->ctx.sess.bref.ref, struct stream *, list);
- if ((int)(curr_strm->stream_epoch - si_strm(cs_si(appctx->owner))->stream_epoch) > 0)
+ if ((int)(curr_strm->stream_epoch - __cs_strm(appctx->owner)->stream_epoch) > 0)
done = 1;
}
@@ -3550,7 +3552,7 @@
LIST_APPEND(&curr_strm->back_refs, &appctx->ctx.sess.bref.users);
/* call the proper dump() function and return if we're missing space */
- if (!stats_dump_full_strm_to_buffer(si, curr_strm))
+ if (!stats_dump_full_strm_to_buffer(cs, curr_strm))
goto full;
/* stream dump complete */
@@ -3672,7 +3674,7 @@
chunk_appendf(&trash, "\n");
- if (ci_putchk(si_ic(si), &trash) == -1) {
+ if (ci_putchk(cs_ic(cs), &trash) == -1) {
/* let's try again later from this stream. We add ourselves into
* this stream's users so that it can remove us upon termination.
*/
@@ -3691,7 +3693,7 @@
else
chunk_appendf(&trash, "Session not found.\n");
- if (ci_putchk(si_ic(si), &trash) == -1)
+ if (ci_putchk(cs_ic(cs), &trash) == -1)
goto full;
appctx->ctx.sess.target = NULL;
@@ -3709,7 +3711,7 @@
return 1;
full:
thread_release();
- si_rx_room_blk(si);
+ si_rx_room_blk(cs->si);
return 0;
}