* include/proto/stream_interface.h
* This file contains stream_interface function prototypes
*
- * Copyright (C) 2000-2012 Willy Tarreau - w@1wt.eu
+ * Copyright (C) 2000-2014 Willy Tarreau - w@1wt.eu
*
* This library is free software; you can redistribute it and/or
* modify it under the terms of the GNU Lesser General Public
struct appctx *stream_int_register_handler(struct stream_interface *si, struct si_applet *app);
void stream_int_unregister_handler(struct stream_interface *si);
+/* returns the channel which receives data from this stream interface (input channel) */
+static inline struct channel *si_ic(struct stream_interface *si)
+{
+ return si->ib;
+}
+
+/* returns the channel which feeds data to this stream interface (output channel) */
+static inline struct channel *si_oc(struct stream_interface *si)
+{
+ return si->ob;
+}
+
/* Initializes all required fields for a new appctx. Note that it does the
* minimum acceptable initialization for an appctx. This means only the
* 3 integer states st0, st1, st2 are zeroed.
return SN_ERR_INTERNAL;
if (!conn_ctrl_ready(conn) || !conn_xprt_ready(conn)) {
- ret = conn->ctrl->connect(conn, !channel_is_empty(si->ob), 0);
+ ret = conn->ctrl->connect(conn, !channel_is_empty(si_oc(si)), 0);
if (ret != SN_ERR_NONE)
return ret;
/* we're in the process of establishing a connection */
si->state = SI_ST_CON;
}
- else if (!channel_is_empty(si->ob)) {
+ else if (!channel_is_empty(si_oc(si))) {
/* reuse the existing connection, we'll have to send a
* request there.
*/
if (target && s->listener->bind_conf->level < ACCESS_LVL_OPER)
chunk_appendf(msg, "# contents not dumped due to insufficient privileges\n");
- if (bi_putchk(si->ib, msg) == -1)
+ if (bi_putchk(si_ic(si), msg) == -1)
return 0;
return 1;
}
chunk_appendf(msg, "\n");
- if (bi_putchk(si->ib, msg) == -1)
+ if (bi_putchk(si_ic(si), msg) == -1)
return 0;
return 1;
/* return server's effective weight at the moment */
snprintf(trash.str, trash.size, "%d (initial %d)\n", sv->uweight, sv->iweight);
- bi_putstr(si->ib, trash.str);
+ bi_putstr(si_ic(si), trash.str);
return 1;
}
else if (strcmp(args[1], "map") == 0 || strcmp(args[1], "acl") == 0) {
static void cli_io_handler(struct stream_interface *si)
{
struct appctx *appctx = __objt_appctx(si->end);
- struct channel *req = si->ob;
- struct channel *res = si->ib;
+ struct channel *req = si_oc(si);
+ struct channel *res = si_ic(si);
int reql;
int len;
/* 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->buf)) {
- si->ib->flags |= CF_WAKE_WRITE;
+ if (buffer_almost_full(si_ic(si)->buf)) {
+ si_ic(si)->flags |= CF_WAKE_WRITE;
break;
}
- reql = bo_getline(si->ob, trash.str, trash.size);
+ reql = bo_getline(si_oc(si), trash.str, trash.size);
if (reql <= 0) { /* closed or EOL not found */
if (reql == 0)
break;
}
/* re-adjust req buffer */
- bo_skip(si->ob, reql);
+ bo_skip(si_oc(si), reql);
req->flags |= CF_READ_DONTWAIT; /* we plan to read small requests */
}
else { /* output functions: first check if the output buffer is closed then abort */
switch (appctx->st0) {
case STAT_CLI_PRINT:
- if (bi_putstr(si->ib, appctx->ctx.cli.msg) != -1)
+ if (bi_putstr(si_ic(si), appctx->ctx.cli.msg) != -1)
appctx->st0 = STAT_CLI_PROMPT;
break;
case STAT_CLI_PRINT_FREE:
- if (bi_putstr(si->ib, appctx->ctx.cli.err) != -1) {
+ if (bi_putstr(si_ic(si), appctx->ctx.cli.err) != -1) {
free(appctx->ctx.cli.err);
appctx->st0 = STAT_CLI_PROMPT;
}
/* The post-command prompt is either LF alone or LF + '> ' in interactive mode */
if (appctx->st0 == STAT_CLI_PROMPT) {
- if (bi_putstr(si->ib, appctx->st1 ? "\n> " : "\n") != -1)
+ if (bi_putstr(si_ic(si), appctx->st1 ? "\n> " : "\n") != -1)
appctx->st0 = STAT_CLI_GETREQ;
}
si_update(si);
/* we don't want to expire timeouts while we're processing requests */
- si->ib->rex = TICK_ETERNITY;
- si->ob->wex = TICK_ETERNITY;
+ si_ic(si)->rex = TICK_ETERNITY;
+ si_oc(si)->wex = TICK_ETERNITY;
out:
DPRINTF(stderr, "%s@%d: st=%d, rqf=%x, rpf=%x, rqh=%d, rqs=%d, rh=%d, rs=%d\n",
global.node, global.desc ? global.desc : ""
);
- if (bi_putchk(si->ib, &trash) == -1)
+ if (bi_putchk(si_ic(si), &trash) == -1)
return 0;
return 1;
static int stats_dump_pools_to_buffer(struct stream_interface *si)
{
dump_pools_to_trash();
- if (bi_putchk(si->ib, &trash) == -1)
+ if (bi_putchk(si_ic(si), &trash) == -1)
return 0;
return 1;
}
scope_txt[0] = 0;
if (appctx->ctx.stats.scope_len) {
strcpy(scope_txt, STAT_SCOPE_PATTERN);
- memcpy(scope_txt + strlen(STAT_SCOPE_PATTERN), bo_ptr(si->ob->buf) + appctx->ctx.stats.scope_str, appctx->ctx.stats.scope_len);
+ memcpy(scope_txt + strlen(STAT_SCOPE_PATTERN), bo_ptr(si_oc(si)->buf) + appctx->ctx.stats.scope_str, appctx->ctx.stats.scope_len);
scope_txt[strlen(STAT_SCOPE_PATTERN) + appctx->ctx.stats.scope_len] = 0;
}
{
struct appctx *appctx = __objt_appctx(si->end);
struct session *s = si_sess(si);
- struct channel *rep = si->ib;
+ struct channel *rep = si_ic(si);
struct server *sv, *svs; /* server and server-state, server-state=server or server->track */
struct listener *l;
* name does not match, skip it.
*/
if (appctx->ctx.stats.scope_len &&
- strnistr(px->id, strlen(px->id), bo_ptr(si->ob->buf) + appctx->ctx.stats.scope_str, appctx->ctx.stats.scope_len) == NULL)
+ strnistr(px->id, strlen(px->id), bo_ptr(si_oc(si)->buf) + appctx->ctx.stats.scope_str, appctx->ctx.stats.scope_len) == NULL)
return 1;
if ((appctx->ctx.stats.flags & STAT_BOUND) &&
);
/* scope_txt = search query, appctx->ctx.stats.scope_len is always <= STAT_SCOPE_TXT_MAXLEN */
- memcpy(scope_txt, bo_ptr(si->ob->buf) + appctx->ctx.stats.scope_str, appctx->ctx.stats.scope_len);
+ memcpy(scope_txt, bo_ptr(si_oc(si)->buf) + appctx->ctx.stats.scope_str, appctx->ctx.stats.scope_len);
scope_txt[appctx->ctx.stats.scope_len] = '\0';
chunk_appendf(&trash,
scope_txt[0] = 0;
if (appctx->ctx.stats.scope_len) {
strcpy(scope_txt, STAT_SCOPE_PATTERN);
- memcpy(scope_txt + strlen(STAT_SCOPE_PATTERN), bo_ptr(si->ob->buf) + appctx->ctx.stats.scope_str, appctx->ctx.stats.scope_len);
+ memcpy(scope_txt + strlen(STAT_SCOPE_PATTERN), bo_ptr(si_oc(si)->buf) + appctx->ctx.stats.scope_str, appctx->ctx.stats.scope_len);
scope_txt[strlen(STAT_SCOPE_PATTERN) + appctx->ctx.stats.scope_len] = 0;
}
static int stats_dump_stat_to_buffer(struct stream_interface *si, struct uri_auth *uri)
{
struct appctx *appctx = __objt_appctx(si->end);
- struct channel *rep = si->ib;
+ struct channel *rep = si_ic(si);
struct proxy *px;
chunk_reset(&trash);
goto out;
}
- reql = bo_getblk(si->ob, temp->str, s->txn.req.body_len, s->txn.req.eoh + 2);
+ reql = bo_getblk(si_oc(si), temp->str, s->txn.req.body_len, s->txn.req.eoh + 2);
if (reql <= 0) {
/* we need more data */
appctx->ctx.stats.st_code = STAT_STATUS_NONE;
s->txn.status = 200;
s->logs.tv_request = now;
- if (bi_putchk(si->ib, &trash) == -1)
+ if (bi_putchk(si_ic(si), &trash) == -1)
return 0;
return 1;
scope_txt[0] = 0;
if (appctx->ctx.stats.scope_len) {
strcpy(scope_txt, STAT_SCOPE_PATTERN);
- memcpy(scope_txt + strlen(STAT_SCOPE_PATTERN), bo_ptr(si->ob->buf) + appctx->ctx.stats.scope_str, appctx->ctx.stats.scope_len);
+ memcpy(scope_txt + strlen(STAT_SCOPE_PATTERN), bo_ptr(si_oc(si)->buf) + appctx->ctx.stats.scope_str, appctx->ctx.stats.scope_len);
scope_txt[strlen(STAT_SCOPE_PATTERN) + appctx->ctx.stats.scope_len] = 0;
}
s->txn.status = 303;
s->logs.tv_request = now;
- if (bi_putchk(si->ib, &trash) == -1)
+ if (bi_putchk(si_ic(si), &trash) == -1)
return 0;
return 1;
{
struct appctx *appctx = __objt_appctx(si->end);
struct session *s = si_sess(si);
- struct channel *req = si->ob;
- struct channel *res = si->ib;
+ struct channel *req = si_oc(si);
+ struct channel *res = si_ic(si);
if (unlikely(si->state == SI_ST_DIS || si->state == SI_ST_CLO))
goto out;
}
if (appctx->st0 == STAT_HTTP_DUMP) {
- unsigned int prev_len = si->ib->buf->i;
+ unsigned int prev_len = si_ic(si)->buf->i;
unsigned int data_len;
unsigned int last_len;
unsigned int last_fwd = 0;
* the output area. For this, we temporarily disable
* forwarding on the channel.
*/
- last_fwd = si->ib->to_forward;
- si->ib->to_forward = 0;
+ last_fwd = si_ic(si)->to_forward;
+ si_ic(si)->to_forward = 0;
chunk_printf(&trash, "\r\n000000\r\n");
- if (bi_putchk(si->ib, &trash) == -1) {
- si->ib->to_forward = last_fwd;
+ if (bi_putchk(si_ic(si), &trash) == -1) {
+ si_ic(si)->to_forward = last_fwd;
goto fail;
}
}
- data_len = si->ib->buf->i;
+ data_len = si_ic(si)->buf->i;
if (stats_dump_stat_to_buffer(si, s->be->uri_auth))
appctx->st0 = STAT_HTTP_DONE;
- last_len = si->ib->buf->i;
+ last_len = si_ic(si)->buf->i;
/* Now we must either adjust or remove the chunk size. This is
* not easy because the chunk size might wrap at the end of the
* applet.
*/
if (appctx->ctx.stats.flags & STAT_CHUNKED) {
- si->ib->total -= (last_len - prev_len);
- si->ib->buf->i -= (last_len - prev_len);
+ si_ic(si)->total -= (last_len - prev_len);
+ si_ic(si)->buf->i -= (last_len - prev_len);
if (last_len != data_len) {
chunk_printf(&trash, "\r\n%06x\r\n", (last_len - data_len));
- bi_putchk(si->ib, &trash);
+ bi_putchk(si_ic(si), &trash);
- si->ib->total += (last_len - data_len);
- si->ib->buf->i += (last_len - data_len);
+ si_ic(si)->total += (last_len - data_len);
+ si_ic(si)->buf->i += (last_len - data_len);
}
/* now re-enable forwarding */
- channel_forward(si->ib, last_fwd);
+ channel_forward(si_ic(si), last_fwd);
}
}
if (appctx->st0 == STAT_HTTP_POST) {
if (stats_process_http_post(si))
appctx->st0 = STAT_HTTP_LAST;
- else if (si->ob->flags & CF_SHUTR)
+ else if (si_oc(si)->flags & CF_SHUTR)
appctx->st0 = STAT_HTTP_DONE;
}
if (appctx->st0 == STAT_HTTP_DONE) {
if (appctx->ctx.stats.flags & STAT_CHUNKED) {
chunk_printf(&trash, "\r\n0\r\n\r\n");
- if (bi_putchk(si->ib, &trash) == -1)
+ if (bi_putchk(si_ic(si), &trash) == -1)
goto fail;
}
/* eat the whole request */
- bo_skip(si->ob, si->ob->buf->o);
+ bo_skip(si_oc(si), si_oc(si)->buf->o);
res->flags |= CF_READ_NULL;
si_shutr(si);
}
si_update(si);
/* we don't want to expire timeouts while we're processing requests */
- si->ib->rex = TICK_ETERNITY;
- si->ob->wex = TICK_ETERNITY;
+ si_ic(si)->rex = TICK_ETERNITY;
+ si_oc(si)->wex = TICK_ETERNITY;
out:
if (unlikely(si->state == SI_ST_DIS || si->state == SI_ST_CLO)) {
if (appctx->ctx.sess.section > 0 && appctx->ctx.sess.uid != sess->uniq_id) {
/* session changed, no need to go any further */
chunk_appendf(&trash, " *** session terminated while we were watching it ***\n");
- if (bi_putchk(si->ib, &trash) == -1)
+ if (bi_putchk(si_ic(si), &trash) == -1)
return 0;
appctx->ctx.sess.uid = 0;
appctx->ctx.sess.section = 0;
sess->txn.rsp.next, sess->res.buf->i,
sess->res.buf->size);
- if (bi_putchk(si->ib, &trash) == -1)
+ if (bi_putchk(si_ic(si), &trash) == -1)
return 0;
/* use other states to dump the contents */
*/
chunk_reset(&trash);
chunk_appendf(&trash, "# id (file) description\n");
- if (bi_putchk(si->ib, &trash) == -1)
+ if (bi_putchk(si_ic(si), &trash) == -1)
return 0;
/* Now, we start the browsing of the references lists.
appctx->ctx.map.ref->reference ? appctx->ctx.map.ref->reference : "",
appctx->ctx.map.ref->display);
- if (bi_putchk(si->ib, &trash) == -1) {
+ if (bi_putchk(si_ic(si), &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.
*/
chunk_appendf(&trash, "\n");
/* display response */
- if (bi_putchk(si->ib, &trash) == -1) {
+ if (bi_putchk(si_ic(si), &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.
*/
chunk_appendf(&trash, "%p %s\n",
appctx->ctx.map.elt, appctx->ctx.map.elt->pattern);
- if (bi_putchk(si->ib, &trash) == -1) {
+ if (bi_putchk(si_ic(si), &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.
*/
struct appctx *appctx = __objt_appctx(si->end);
struct connection *conn;
- if (unlikely(si->ib->flags & (CF_WRITE_ERROR|CF_SHUTW))) {
+ if (unlikely(si_ic(si)->flags & (CF_WRITE_ERROR|CF_SHUTW))) {
/* If we're forced to shut down, we might have to remove our
* reference to the last session being dumped.
*/
chunk_appendf(&trash, "\n");
- if (bi_putchk(si->ib, &trash) == -1) {
+ if (bi_putchk(si_ic(si), &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.
*/
else
chunk_appendf(&trash, "Session not found.\n");
- if (bi_putchk(si->ib, &trash) == -1)
+ if (bi_putchk(si_ic(si), &trash) == -1)
return 0;
appctx->ctx.sess.target = NULL;
* data though.
*/
- if (unlikely(si->ib->flags & (CF_WRITE_ERROR|CF_SHUTW))) {
+ if (unlikely(si_ic(si)->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) {
appctx->ctx.table.entry->ref_cnt--;
struct appctx *appctx = __objt_appctx(si->end);
extern const char *monthname[12];
- if (unlikely(si->ib->flags & (CF_WRITE_ERROR|CF_SHUTW)))
+ if (unlikely(si_ic(si)->flags & (CF_WRITE_ERROR|CF_SHUTW)))
return 1;
chunk_reset(&trash);
tm.tm_hour, tm.tm_min, tm.tm_sec, (int)(date.tv_usec/1000),
error_snapshot_id);
- if (bi_putchk(si->ib, &trash) == -1) {
+ if (bi_putchk(si_ic(si), &trash) == -1) {
/* Socket buffer full. Let's try again later from the same point */
return 0;
}
es->b_flags, es->b_out, es->b_tot,
es->len, es->b_wrap, es->pos);
- if (bi_putchk(si->ib, &trash) == -1) {
+ if (bi_putchk(si_ic(si), &trash) == -1) {
/* Socket buffer full. Let's try again later from the same point */
return 0;
}
/* the snapshot changed while we were dumping it */
chunk_appendf(&trash,
" WARNING! update detected on this snapshot, dump interrupted. Please re-check!\n");
- if (bi_putchk(si->ib, &trash) == -1)
+ if (bi_putchk(si_ic(si), &trash) == -1)
return 0;
goto next;
}
if (newptr == appctx->ctx.errors.ptr)
return 0;
- if (bi_putchk(si->ib, &trash) == -1) {
+ if (bi_putchk(si_ic(si), &trash) == -1) {
/* Socket buffer full. Let's try again later from the same point */
return 0;
}
static void hlua_socket_handler(struct stream_interface *si)
{
struct appctx *appctx = objt_appctx(si->end);
- struct connection *c = objt_conn(si->ib->cons->end);
+ struct connection *c = objt_conn(si_ic(si)->cons->end);
/* Wakeup the main session if the client connection is closed. */
- if (!c || channel_output_closed(si->ib) || channel_input_closed(si->ob)) {
+ if (!c || channel_output_closed(si_ic(si)) || channel_input_closed(si_oc(si))) {
if (appctx->ctx.hlua.socket) {
appctx->ctx.hlua.socket->s = NULL;
appctx->ctx.hlua.socket = NULL;
}
si_shutw(si);
si_shutr(si);
- si->ib->flags |= CF_READ_NULL;
+ si_ic(si)->flags |= CF_READ_NULL;
hlua_com_wake(&appctx->ctx.hlua.wake_on_read);
hlua_com_wake(&appctx->ctx.hlua.wake_on_write);
return;
appctx->ctx.hlua.connected = 1;
/* Wake the tasks which wants to write if the buffer have avalaible space. */
- if (channel_may_recv(si->ob))
+ if (channel_may_recv(si_oc(si)))
hlua_com_wake(&appctx->ctx.hlua.wake_on_write);
/* Wake the tasks which wants to read if the buffer contains data. */
- if (channel_is_empty(si->ib))
+ if (channel_is_empty(si_ic(si)))
hlua_com_wake(&appctx->ctx.hlua.wake_on_read);
}
goto connection_closed;
if (wanted == HLSR_READ_LINE) {
-
/* Read line. */
- nblk = bo_getline_nc(socket->s->si[0].ob, &blk1, &len1, &blk2, &len2);
+ nblk = bo_getline_nc(si_oc(&socket->s->si[0]), &blk1, &len1, &blk2, &len2);
if (nblk < 0) /* Connection close. */
goto connection_closed;
if (nblk == 0) /* No data avalaible. */
}
else if (wanted == HLSR_READ_ALL) {
-
/* Read all the available data. */
- nblk = bo_getblk_nc(socket->s->si[0].ob, &blk1, &len1, &blk2, &len2);
+ nblk = bo_getblk_nc(si_oc(&socket->s->si[0]), &blk1, &len1, &blk2, &len2);
if (nblk < 0) /* Connection close. */
goto connection_closed;
if (nblk == 0) /* No data avalaible. */
}
else {
-
/* Read a block of data. */
- nblk = bo_getblk_nc(socket->s->si[0].ob, &blk1, &len1, &blk2, &len2);
+ nblk = bo_getblk_nc(si_oc(&socket->s->si[0]), &blk1, &len1, &blk2, &len2);
if (nblk < 0) /* Connection close. */
goto connection_closed;
if (nblk == 0) /* No data avalaible. */
}
/* Consume data. */
- bo_skip(socket->s->si[0].ob, len + skip_at_end);
+ bo_skip(si_oc(&socket->s->si[0]), len + skip_at_end);
/* Don't wait anything. */
si_update(&socket->s->si[0]);
}
/* Check for avalaible space. */
- len = buffer_total_space(socket->s->si[0].ib->buf);
+ len = buffer_total_space(si_ic(&socket->s->si[0])->buf);
if (len <= 0)
goto hlua_socket_write_yield_return;
/* send data */
if (len < send_len)
send_len = len;
- len = bi_putblk(socket->s->si[0].ib, buf+sent, send_len);
+ len = bi_putblk(si_ic(&socket->s->si[0]), buf+sent, send_len);
/* "Not enough space" (-1), "Buffer too little to contain
* the data" (-2) are not expected because the available length
/* update buffers. */
si_update(&socket->s->si[0]);
- socket->s->si[0].ib->rex = TICK_ETERNITY;
- socket->s->si[0].ob->wex = TICK_ETERNITY;
+ si_ic(&socket->s->si[0])->rex = TICK_ETERNITY;
+ si_oc(&socket->s->si[0])->wex = TICK_ETERNITY;
/* Update length sent. */
lua_pop(L, 1);
MAY_LJMP(check_args(L, 1, "close"));
s = MAY_LJMP(hlua_checktxn(L, 1));
- channel_abort(s->s->si[0].ib);
- channel_auto_close(s->s->si[0].ib);
- channel_erase(s->s->si[0].ib);
- channel_auto_read(s->s->si[0].ob);
- channel_auto_close(s->s->si[0].ob);
- channel_shutr_now(s->s->si[0].ob);
+ channel_abort(si_ic(&s->s->si[0]));
+ channel_auto_close(si_ic(&s->s->si[0]));
+ channel_erase(si_ic(&s->s->si[0]));
+ channel_auto_read(si_oc(&s->s->si[0]));
+ channel_auto_close(si_oc(&s->s->si[0]));
+ channel_shutr_now(si_oc(&s->s->si[0]));
return 0;
}
appctx->st0 = PEER_SESS_ST_GETVERSION;
/* fall through */
case PEER_SESS_ST_GETVERSION:
- reql = bo_getline(si->ob, trash.str, trash.size);
+ reql = bo_getline(si_oc(si), trash.str, trash.size);
if (reql <= 0) { /* closed or EOL not found */
if (reql == 0)
goto out;
else
trash.str[reql-1] = 0;
- bo_skip(si->ob, reql);
+ bo_skip(si_oc(si), reql);
/* test version */
if (strcmp(PEER_SESSION_PROTO_NAME " 1.0", trash.str) != 0) {
appctx->st0 = PEER_SESS_ST_GETHOST;
/* fall through */
case PEER_SESS_ST_GETHOST:
- reql = bo_getline(si->ob, trash.str, trash.size);
+ reql = bo_getline(si_oc(si), trash.str, trash.size);
if (reql <= 0) { /* closed or EOL not found */
if (reql == 0)
goto out;
else
trash.str[reql-1] = 0;
- bo_skip(si->ob, reql);
+ bo_skip(si_oc(si), reql);
/* test hostname match */
if (strcmp(localpeer, trash.str) != 0) {
case PEER_SESS_ST_GETPEER: {
struct peer *curpeer;
char *p;
- reql = bo_getline(si->ob, trash.str, trash.size);
+ reql = bo_getline(si_oc(si), trash.str, trash.size);
if (reql <= 0) { /* closed or EOL not found */
if (reql == 0)
goto out;
else
trash.str[reql-1] = 0;
- bo_skip(si->ob, reql);
+ bo_skip(si_oc(si), reql);
/* parse line "<peer name> <pid>" */
p = strchr(trash.str, ' ');
size_t key_size;
char *p;
- reql = bo_getline(si->ob, trash.str, trash.size);
+ reql = bo_getline(si_oc(si), trash.str, trash.size);
if (reql <= 0) { /* closed or EOL not found */
if (reql == 0)
goto out;
else
trash.str[reql-1] = 0;
- bo_skip(si->ob, reql);
+ bo_skip(si_oc(si), reql);
/* Parse line "<table name> <type> <size>" */
p = strchr(trash.str, ' ');
struct peer_session *ps = (struct peer_session *)appctx->ctx.peers.ptr;
repl = snprintf(trash.str, trash.size, "%d\n", PEER_SESS_SC_SUCCESSCODE);
- repl = bi_putblk(si->ib, trash.str, repl);
+ repl = bi_putblk(si_ic(si), trash.str, repl);
if (repl <= 0) {
if (repl == -1)
goto out;
goto switchstate;
}
- repl = bi_putblk(si->ib, trash.str, repl);
+ repl = bi_putblk(si_ic(si), trash.str, repl);
if (repl <= 0) {
if (repl == -1)
goto out;
case PEER_SESS_ST_GETSTATUS: {
struct peer_session *ps = (struct peer_session *)appctx->ctx.peers.ptr;
- if (si->ib->flags & CF_WRITE_PARTIAL)
+ if (si_ic(si)->flags & CF_WRITE_PARTIAL)
ps->statuscode = PEER_SESS_SC_CONNECTEDCODE;
- reql = bo_getline(si->ob, trash.str, trash.size);
+ reql = bo_getline(si_oc(si), trash.str, trash.size);
if (reql <= 0) { /* closed or EOL not found */
if (reql == 0)
goto out;
else
trash.str[reql-1] = 0;
- bo_skip(si->ob, reql);
+ bo_skip(si_oc(si), reql);
/* Register status code */
ps->statuscode = atoi(trash.str);
char c;
int totl = 0;
- reql = bo_getblk(si->ob, (char *)&c, sizeof(c), totl);
+ reql = bo_getblk(si_oc(si), (char *)&c, sizeof(c), totl);
if (reql <= 0) /* closed or EOL not found */
goto incomplete;
pushack = ps->pushack + (unsigned int)(c & 0x7F);
}
else {
- reql = bo_getblk(si->ob, (char *)&netinteger, sizeof(netinteger), totl);
+ reql = bo_getblk(si_oc(si), (char *)&netinteger, sizeof(netinteger), totl);
if (reql <= 0) /* closed or EOL not found */
goto incomplete;
unsigned int to_read, to_store;
/* read size first */
- reql = bo_getblk(si->ob, (char *)&netinteger, sizeof(netinteger), totl);
+ reql = bo_getblk(si_oc(si), (char *)&netinteger, sizeof(netinteger), totl);
if (reql <= 0) /* closed or EOL not found */
goto incomplete;
to_store = 0;
to_read = ntohl(netinteger);
- if (to_read + totl > si->ob->buf->size) {
+ if (to_read + totl > si_oc(si)->buf->size) {
/* impossible to read a key this large, abort */
reql = -1;
goto incomplete;
* the rest is drained into the trash.
*/
if (to_store) {
- reql = bo_getblk(si->ob, (char *)newts->key.key, to_store, totl);
+ reql = bo_getblk(si_oc(si), (char *)newts->key.key, to_store, totl);
if (reql <= 0) /* closed or incomplete */
goto incomplete;
newts->key.key[reql] = 0;
to_read -= reql;
}
if (to_read) {
- reql = bo_getblk(si->ob, trash.str, to_read, totl);
+ reql = bo_getblk(si_oc(si), trash.str, to_read, totl);
if (reql <= 0) /* closed or incomplete */
goto incomplete;
totl += reql;
}
}
else if (ps->table->table->type == STKTABLE_TYPE_INTEGER) {
- reql = bo_getblk(si->ob, (char *)&netinteger, sizeof(netinteger), totl);
+ reql = bo_getblk(si_oc(si), (char *)&netinteger, sizeof(netinteger), totl);
if (reql <= 0) /* closed or EOL not found */
goto incomplete;
newts = stksess_new(ps->table->table, NULL);
else {
/* type ip or binary */
newts = stksess_new(ps->table->table, NULL);
- reql = bo_getblk(si->ob, newts ? (char *)newts->key.key : trash.str, ps->table->table->key_size, totl);
+ reql = bo_getblk(si_oc(si), newts ? (char *)newts->key.key : trash.str, ps->table->table->key_size, totl);
if (reql <= 0) /* closed or EOL not found */
goto incomplete;
totl += reql;
}
/* read server id */
- reql = bo_getblk(si->ob, (char *)&netinteger, sizeof(netinteger), totl);
+ reql = bo_getblk(si_oc(si), (char *)&netinteger, sizeof(netinteger), totl);
if (reql <= 0) /* closed or EOL not found */
goto incomplete;
/* ack message */
uint32_t netinteger;
- reql = bo_getblk(si->ob, (char *)&netinteger, sizeof(netinteger), totl);
+ reql = bo_getblk(si_oc(si), (char *)&netinteger, sizeof(netinteger), totl);
if (reql <= 0) /* closed or EOL not found */
goto incomplete;
}
/* skip consumed message */
- bo_skip(si->ob, totl);
+ bo_skip(si_oc(si), totl);
/* loop on that state to peek next message */
goto switchstate;
/* Confirm finished or partial messages */
while (ps->confirm) {
/* There is a confirm messages to send */
- repl = bi_putchr(si->ib, 'c');
+ repl = bi_putchr(si_ic(si), 'c');
if (repl <= 0) {
/* no more write possible */
if (repl == -1)
!(ps->table->flags & SHTABLE_F_RESYNC_PROCESS)) {
/* Current peer was elected to request a resync */
- repl = bi_putchr(si->ib, 'R');
+ repl = bi_putchr(si_ic(si), 'R');
if (repl <= 0) {
/* no more write possible */
if (repl == -1)
netinteger = htonl(ps->pushack);
memcpy(&trash.str[1], &netinteger, sizeof(netinteger));
- repl = bi_putblk(si->ib, trash.str, 1+sizeof(netinteger));
+ repl = bi_putblk(si_ic(si), trash.str, 1+sizeof(netinteger));
if (repl <= 0) {
/* no more write possible */
if (repl == -1)
msglen = peer_prepare_datamsg(ts, ps, trash.str, trash.size);
if (msglen) {
/* message to buffer */
- repl = bi_putblk(si->ib, trash.str, msglen);
+ repl = bi_putblk(si_ic(si), trash.str, msglen);
if (repl <= 0) {
/* no more write possible */
if (repl == -1)
msglen = peer_prepare_datamsg(ts, ps, trash.str, trash.size);
if (msglen) {
/* message to buffer */
- repl = bi_putblk(si->ib, trash.str, msglen);
+ repl = bi_putblk(si_ic(si), trash.str, msglen);
if (repl <= 0) {
/* no more write possible */
if (repl == -1)
if (!(ps->flags & PEER_F_TEACH_FINISHED)) {
/* process final lesson message */
- repl = bi_putchr(si->ib, ((ps->table->flags & SHTABLE_RESYNC_STATEMASK) == SHTABLE_RESYNC_FINISHED) ? 'F' : 'C');
+ repl = bi_putchr(si_ic(si), ((ps->table->flags & SHTABLE_RESYNC_STATEMASK) == SHTABLE_RESYNC_FINISHED) ? 'F' : 'C');
if (repl <= 0) {
/* no more write possible */
if (repl == -1)
msglen = peer_prepare_datamsg(ts, ps, trash.str, trash.size);
if (msglen) {
/* message to buffer */
- repl = bi_putblk(si->ib, trash.str, msglen);
+ repl = bi_putblk(si_ic(si), trash.str, msglen);
if (repl <= 0) {
/* no more write possible */
if (repl == -1)
case PEER_SESS_ST_EXIT:
repl = snprintf(trash.str, trash.size, "%d\n", appctx->st1);
- if (bi_putblk(si->ib, trash.str, repl) == -1)
+ if (bi_putblk(si_ic(si), trash.str, repl) == -1)
goto out;
appctx->st0 = PEER_SESS_ST_END;
/* fall through */
case PEER_SESS_ST_END: {
si_shutw(si);
si_shutr(si);
- si->ib->flags |= CF_READ_NULL;
+ si_ic(si)->flags |= CF_READ_NULL;
goto quit;
}
}
}
out:
si_update(si);
- si->ob->flags |= CF_READ_DONTWAIT;
+ si_oc(si)->flags |= CF_READ_DONTWAIT;
/* we don't want to expire timeouts while we're processing requests */
- si->ib->rex = TICK_ETERNITY;
- si->ob->wex = TICK_ETERNITY;
+ si_ic(si)->rex = TICK_ETERNITY;
+ si_oc(si)->wex = TICK_ETERNITY;
quit:
return;
}
static void http_server_error(struct session *s, struct stream_interface *si,
int err, int finst, int status, const struct chunk *msg)
{
- channel_auto_read(si->ob);
- channel_abort(si->ob);
- channel_auto_close(si->ob);
- channel_erase(si->ob);
- channel_auto_close(si->ib);
- channel_auto_read(si->ib);
+ channel_auto_read(si_oc(si));
+ channel_abort(si_oc(si));
+ channel_auto_close(si_oc(si));
+ channel_erase(si_oc(si));
+ channel_auto_close(si_ic(si));
+ channel_auto_read(si_ic(si));
if (status > 0 && msg) {
s->txn.status = status;
- bo_inject(si->ib, msg->str, msg->len);
+ bo_inject(si_ic(si), msg->str, msg->len);
}
if (!(s->flags & SN_ERR_MASK))
s->flags |= err;
*/
static int sess_update_st_con_tcp(struct session *s, struct stream_interface *si)
{
- struct channel *req = si->ob;
- struct channel *rep = si->ib;
+ struct channel *req = si_oc(si);
+ struct channel *rep = si_ic(si);
struct connection *srv_conn = __objt_conn(si->end);
/* If we got an error, or if nothing happened and the connection timed
* attempts and error reports.
*/
if (unlikely(si->flags & (SI_FL_EXP|SI_FL_ERR))) {
- if (unlikely(si->ob->flags & CF_WRITE_PARTIAL)) {
+ if (unlikely(si_oc(si)->flags & CF_WRITE_PARTIAL)) {
/* Some data were sent past the connection establishment,
* so we need to pretend we're established to log correctly
* and let later states handle the failure.
*/
si->state = SI_ST_EST;
si->err_type = SI_ET_DATA_ERR;
- si->ib->flags |= CF_READ_ERROR | CF_WRITE_ERROR;
+ si_ic(si)->flags |= CF_READ_ERROR | CF_WRITE_ERROR;
return 1;
}
si->exp = TICK_ETERNITY;
/* shutw is enough so stop a connecting socket */
si_shutw(si);
- si->ob->flags |= CF_WRITE_ERROR;
- si->ib->flags |= CF_READ_ERROR;
+ si_oc(si)->flags |= CF_WRITE_ERROR;
+ si_ic(si)->flags |= CF_READ_ERROR;
si->state = SI_ST_CLO;
if (s->srv_error)
*/
static void sess_establish(struct session *s, struct stream_interface *si)
{
- struct channel *req = si->ob;
- struct channel *rep = si->ib;
+ struct channel *req = si_oc(si);
+ struct channel *rep = si_ic(si);
/* First, centralize the timers information */
s->logs.t_connect = tv_ms_elapsed(&s->logs.tv_accept, &now);
/* Failed and not retryable. */
si_shutr(si);
si_shutw(si);
- si->ob->flags |= CF_WRITE_ERROR;
+ si_oc(si)->flags |= CF_WRITE_ERROR;
s->logs.t_queue = tv_ms_elapsed(&s->logs.tv_accept, &now);
s->be->be_counters.failed_conns++;
si_shutr(si);
si_shutw(si);
- si->ob->flags |= CF_WRITE_TIMEOUT;
+ si_oc(si)->flags |= CF_WRITE_TIMEOUT;
if (!si->err_type)
si->err_type = SI_ET_QUEUE_TO;
si->state = SI_ST_CLO;
}
/* Connection remains in queue, check if we have to abort it */
- if ((si->ob->flags & (CF_READ_ERROR)) ||
- ((si->ob->flags & CF_SHUTW_NOW) && /* empty and client aborted */
- (channel_is_empty(si->ob) || s->be->options & PR_O_ABRT_CLOSE))) {
+ if ((si_oc(si)->flags & (CF_READ_ERROR)) ||
+ ((si_oc(si)->flags & CF_SHUTW_NOW) && /* empty and client aborted */
+ (channel_is_empty(si_oc(si)) || s->be->options & PR_O_ABRT_CLOSE))) {
/* give up */
si->exp = TICK_ETERNITY;
s->logs.t_queue = tv_ms_elapsed(&s->logs.tv_accept, &now);
}
else if (si->state == SI_ST_TAR) {
/* Connection request might be aborted */
- if ((si->ob->flags & (CF_READ_ERROR)) ||
- ((si->ob->flags & CF_SHUTW_NOW) && /* empty and client aborted */
- (channel_is_empty(si->ob) || s->be->options & PR_O_ABRT_CLOSE))) {
+ if ((si_oc(si)->flags & (CF_READ_ERROR)) ||
+ ((si_oc(si)->flags & CF_SHUTW_NOW) && /* empty and client aborted */
+ (channel_is_empty(si_oc(si)) || s->be->options & PR_O_ABRT_CLOSE))) {
/* give up */
si->exp = TICK_ETERNITY;
si_shutr(si);
si_shutr(si);
si_shutw(si);
- si->ob->flags |= CF_WRITE_ERROR;
+ si_oc(si)->flags |= CF_WRITE_ERROR;
si->err_type = SI_ET_CONN_RES;
si->state = SI_ST_CLO;
if (s->srv_error)
/* we did not get any server, let's check the cause */
si_shutr(si);
si_shutw(si);
- si->ob->flags |= CF_WRITE_ERROR;
+ si_oc(si)->flags |= CF_WRITE_ERROR;
if (!si->err_type)
si->err_type = SI_ET_CONN_OTHER;
si->state = SI_ST_CLO;
if (!si->err_type)
si->err_type = SI_ET_DATA_ERR;
- si->ob->flags |= CF_WRITE_ERROR;
- si->ib->flags |= CF_READ_ERROR;
+ si_oc(si)->flags |= CF_WRITE_ERROR;
+ si_ic(si)->flags |= CF_READ_ERROR;
}
/*
*/
void stream_int_retnclose(struct stream_interface *si, const struct chunk *msg)
{
- channel_auto_read(si->ib);
- channel_abort(si->ib);
- channel_auto_close(si->ib);
- channel_erase(si->ib);
- channel_truncate(si->ob);
+ channel_auto_read(si_ic(si));
+ channel_abort(si_ic(si));
+ channel_auto_close(si_ic(si));
+ channel_erase(si_ic(si));
+ channel_truncate(si_oc(si));
if (likely(msg && msg->len))
- bo_inject(si->ob, msg->str, msg->len);
+ bo_inject(si_oc(si), msg->str, msg->len);
- si->ob->wex = tick_add_ifset(now_ms, si->ob->wto);
- channel_auto_read(si->ob);
- channel_auto_close(si->ob);
- channel_shutr_now(si->ob);
+ si_oc(si)->wex = tick_add_ifset(now_ms, si_oc(si)->wto);
+ channel_auto_read(si_oc(si));
+ channel_auto_close(si_oc(si));
+ channel_shutr_now(si_oc(si));
}
/* default update function for embedded tasks, to be used at the end of the i/o handler */
DPRINTF(stderr, "%s: si=%p, si->state=%d ib->flags=%08x ob->flags=%08x\n",
__FUNCTION__,
- si, si->state, si->ib->flags, si->ob->flags);
+ si, si->state, si_ic(si)->flags, si_oc(si)->flags);
if (si->state != SI_ST_EST)
return;
- if ((si->ob->flags & (CF_SHUTW|CF_SHUTW_NOW)) == CF_SHUTW_NOW &&
- channel_is_empty(si->ob))
+ if ((si_oc(si)->flags & (CF_SHUTW|CF_SHUTW_NOW)) == CF_SHUTW_NOW &&
+ channel_is_empty(si_oc(si)))
si_shutw(si);
- if ((si->ob->flags & (CF_SHUTW|CF_SHUTW_NOW)) == 0 && channel_may_recv(si->ob))
+ if ((si_oc(si)->flags & (CF_SHUTW|CF_SHUTW_NOW)) == 0 && channel_may_recv(si_oc(si)))
si->flags |= SI_FL_WAIT_DATA;
/* we're almost sure that we need some space if the buffer is not
* empty, even if it's not full, because the applets can't fill it.
*/
- if ((si->ib->flags & (CF_SHUTR|CF_DONT_READ)) == 0 && !channel_is_empty(si->ib))
+ if ((si_ic(si)->flags & (CF_SHUTR|CF_DONT_READ)) == 0 && !channel_is_empty(si_ic(si)))
si->flags |= SI_FL_WAIT_ROOM;
- if (si->ob->flags & CF_WRITE_ACTIVITY) {
- if (tick_isset(si->ob->wex))
- si->ob->wex = tick_add_ifset(now_ms, si->ob->wto);
+ if (si_oc(si)->flags & CF_WRITE_ACTIVITY) {
+ if (tick_isset(si_oc(si)->wex))
+ si_oc(si)->wex = tick_add_ifset(now_ms, si_oc(si)->wto);
}
- if (si->ib->flags & CF_READ_ACTIVITY ||
- (si->ob->flags & CF_WRITE_ACTIVITY && !(si->flags & SI_FL_INDEP_STR))) {
- if (tick_isset(si->ib->rex))
- si->ib->rex = tick_add_ifset(now_ms, si->ib->rto);
+ if (si_ic(si)->flags & CF_READ_ACTIVITY ||
+ (si_oc(si)->flags & CF_WRITE_ACTIVITY && !(si->flags & SI_FL_INDEP_STR))) {
+ if (tick_isset(si_ic(si)->rex))
+ si_ic(si)->rex = tick_add_ifset(now_ms, si_ic(si)->rto);
}
/* save flags to detect changes */
old_flags = si->flags;
- if (likely((si->ob->flags & (CF_SHUTW|CF_WRITE_PARTIAL|CF_DONT_READ)) == CF_WRITE_PARTIAL &&
- channel_may_recv(si->ob) &&
- (si->ob->prod->flags & SI_FL_WAIT_ROOM)))
- si_chk_rcv(si->ob->prod);
-
- if (((si->ib->flags & CF_READ_PARTIAL) && !channel_is_empty(si->ib)) &&
- (si->ib->pipe /* always try to send spliced data */ ||
- (si->ib->buf->i == 0 && (si->ib->cons->flags & SI_FL_WAIT_DATA)))) {
- si_chk_snd(si->ib->cons);
+ if (likely((si_oc(si)->flags & (CF_SHUTW|CF_WRITE_PARTIAL|CF_DONT_READ)) == CF_WRITE_PARTIAL &&
+ channel_may_recv(si_oc(si)) &&
+ (si_oc(si)->prod->flags & SI_FL_WAIT_ROOM)))
+ si_chk_rcv(si_oc(si)->prod);
+
+ if (((si_ic(si)->flags & CF_READ_PARTIAL) && !channel_is_empty(si_ic(si))) &&
+ (si_ic(si)->pipe /* always try to send spliced data */ ||
+ (si_ic(si)->buf->i == 0 && (si_ic(si)->cons->flags & SI_FL_WAIT_DATA)))) {
+ si_chk_snd(si_ic(si)->cons);
/* check if the consumer has freed some space */
- if (channel_may_recv(si->ib) && !si->ib->pipe)
+ if (channel_may_recv(si_ic(si)) && !si_ic(si)->pipe)
si->flags &= ~SI_FL_WAIT_ROOM;
}
((old_flags & ~si->flags) & (SI_FL_WAIT_ROOM|SI_FL_WAIT_DATA)) ||
/* changes on the production side */
- (si->ib->flags & (CF_READ_NULL|CF_READ_ERROR)) ||
+ (si_ic(si)->flags & (CF_READ_NULL|CF_READ_ERROR)) ||
si->state != SI_ST_EST ||
(si->flags & SI_FL_ERR) ||
- ((si->ib->flags & CF_READ_PARTIAL) &&
- (!si->ib->to_forward || si->ib->cons->state != SI_ST_EST)) ||
+ ((si_ic(si)->flags & CF_READ_PARTIAL) &&
+ (!si_ic(si)->to_forward || si_ic(si)->cons->state != SI_ST_EST)) ||
/* changes on the consumption side */
- (si->ob->flags & (CF_WRITE_NULL|CF_WRITE_ERROR)) ||
- ((si->ob->flags & CF_WRITE_ACTIVITY) &&
- ((si->ob->flags & CF_SHUTW) ||
- ((si->ob->flags & CF_WAKE_WRITE) &&
- (si->ob->prod->state != SI_ST_EST ||
- (channel_is_empty(si->ob) && !si->ob->to_forward)))))) {
+ (si_oc(si)->flags & (CF_WRITE_NULL|CF_WRITE_ERROR)) ||
+ ((si_oc(si)->flags & CF_WRITE_ACTIVITY) &&
+ ((si_oc(si)->flags & CF_SHUTW) ||
+ ((si_oc(si)->flags & CF_WAKE_WRITE) &&
+ (si_oc(si)->prod->state != SI_ST_EST ||
+ (channel_is_empty(si_oc(si)) && !si_oc(si)->to_forward)))))) {
if (!(si->flags & SI_FL_DONT_WAKE) && si->owner)
task_wakeup(si->owner, TASK_WOKEN_IO);
}
- if (si->ib->flags & CF_READ_ACTIVITY)
- si->ib->flags &= ~CF_READ_DONTWAIT;
+ if (si_ic(si)->flags & CF_READ_ACTIVITY)
+ si_ic(si)->flags &= ~CF_READ_DONTWAIT;
}
/*
*/
static void stream_int_shutr(struct stream_interface *si)
{
- si->ib->flags &= ~CF_SHUTR_NOW;
- if (si->ib->flags & CF_SHUTR)
+ si_ic(si)->flags &= ~CF_SHUTR_NOW;
+ if (si_ic(si)->flags & CF_SHUTR)
return;
- si->ib->flags |= CF_SHUTR;
- si->ib->rex = TICK_ETERNITY;
+ si_ic(si)->flags |= CF_SHUTR;
+ si_ic(si)->rex = TICK_ETERNITY;
si->flags &= ~SI_FL_WAIT_ROOM;
if (si->state != SI_ST_EST && si->state != SI_ST_CON)
return;
- if (si->ob->flags & CF_SHUTW) {
+ if (si_oc(si)->flags & CF_SHUTW) {
si->state = SI_ST_DIS;
si->exp = TICK_ETERNITY;
si_applet_release(si);
*/
static void stream_int_shutw(struct stream_interface *si)
{
- si->ob->flags &= ~CF_SHUTW_NOW;
- if (si->ob->flags & CF_SHUTW)
+ si_oc(si)->flags &= ~CF_SHUTW_NOW;
+ if (si_oc(si)->flags & CF_SHUTW)
return;
- si->ob->flags |= CF_SHUTW;
- si->ob->wex = TICK_ETERNITY;
+ si_oc(si)->flags |= CF_SHUTW;
+ si_oc(si)->wex = TICK_ETERNITY;
si->flags &= ~SI_FL_WAIT_DATA;
switch (si->state) {
* no risk so we close both sides immediately.
*/
if (!(si->flags & (SI_FL_ERR | SI_FL_NOLINGER)) &&
- !(si->ib->flags & (CF_SHUTR|CF_DONT_READ)))
+ !(si_ic(si)->flags & (CF_SHUTR|CF_DONT_READ)))
return;
/* fall through */
si_applet_release(si);
default:
si->flags &= ~(SI_FL_WAIT_ROOM | SI_FL_NOLINGER);
- si->ib->flags &= ~CF_SHUTR_NOW;
- si->ib->flags |= CF_SHUTR;
- si->ib->rex = TICK_ETERNITY;
+ si_ic(si)->flags &= ~CF_SHUTR_NOW;
+ si_ic(si)->flags |= CF_SHUTR;
+ si_ic(si)->rex = TICK_ETERNITY;
si->exp = TICK_ETERNITY;
}
/* default chk_rcv function for scheduled tasks */
static void stream_int_chk_rcv(struct stream_interface *si)
{
- struct channel *ib = si->ib;
+ struct channel *ib = si_ic(si);
DPRINTF(stderr, "%s: si=%p, si->state=%d ib->flags=%08x ob->flags=%08x\n",
__FUNCTION__,
- si, si->state, si->ib->flags, si->ob->flags);
+ si, si->state, si_ic(si)->flags, si_oc(si)->flags);
if (unlikely(si->state != SI_ST_EST || (ib->flags & (CF_SHUTR|CF_DONT_READ))))
return;
/* default chk_snd function for scheduled tasks */
static void stream_int_chk_snd(struct stream_interface *si)
{
- struct channel *ob = si->ob;
+ struct channel *ob = si_oc(si);
DPRINTF(stderr, "%s: si=%p, si->state=%d ib->flags=%08x ob->flags=%08x\n",
__FUNCTION__,
- si, si->state, si->ib->flags, si->ob->flags);
+ si, si->state, si_ic(si)->flags, si_oc(si)->flags);
- if (unlikely(si->state != SI_ST_EST || (si->ob->flags & CF_SHUTW)))
+ if (unlikely(si->state != SI_ST_EST || (si_oc(si)->flags & CF_SHUTW)))
return;
if (!(si->flags & SI_FL_WAIT_DATA) || /* not waiting for data */
*/
if (conn->data == &si_conn_cb) {
struct stream_interface *si = conn->owner;
- struct connection *remote = objt_conn(si->ob->prod->end);
+ struct connection *remote = objt_conn(si_oc(si)->prod->end);
ret = make_proxy_line(trash.str, trash.size, objt_server(conn->target), remote);
}
else {
DPRINTF(stderr, "%s: si=%p, si->state=%d ib->flags=%08x ob->flags=%08x\n",
__FUNCTION__,
- si, si->state, si->ib->flags, si->ob->flags);
+ si, si->state, si_ic(si)->flags, si_oc(si)->flags);
if (conn->flags & CO_FL_ERROR)
si->flags |= SI_FL_ERR;
/* check for recent connection establishment */
if (unlikely(!(conn->flags & (CO_FL_WAIT_L4_CONN | CO_FL_WAIT_L6_CONN | CO_FL_CONNECTED)))) {
si->exp = TICK_ETERNITY;
- si->ob->flags |= CF_WRITE_NULL;
+ si_oc(si)->flags |= CF_WRITE_NULL;
}
/* process consumer side */
- if (channel_is_empty(si->ob)) {
- if (((si->ob->flags & (CF_SHUTW|CF_SHUTW_NOW)) == CF_SHUTW_NOW) &&
+ if (channel_is_empty(si_oc(si))) {
+ if (((si_oc(si)->flags & (CF_SHUTW|CF_SHUTW_NOW)) == CF_SHUTW_NOW) &&
(si->state == SI_ST_EST))
stream_int_shutw_conn(si);
__conn_data_stop_send(conn);
- si->ob->wex = TICK_ETERNITY;
+ si_oc(si)->wex = TICK_ETERNITY;
}
- if ((si->ob->flags & (CF_SHUTW|CF_SHUTW_NOW)) == 0 && channel_may_recv(si->ob))
+ if ((si_oc(si)->flags & (CF_SHUTW|CF_SHUTW_NOW)) == 0 && channel_may_recv(si_oc(si)))
si->flags |= SI_FL_WAIT_DATA;
- if (si->ob->flags & CF_WRITE_ACTIVITY) {
+ if (si_oc(si)->flags & CF_WRITE_ACTIVITY) {
/* update timeouts if we have written something */
- if ((si->ob->flags & (CF_SHUTW|CF_WRITE_PARTIAL)) == CF_WRITE_PARTIAL &&
- !channel_is_empty(si->ob))
- if (tick_isset(si->ob->wex))
- si->ob->wex = tick_add_ifset(now_ms, si->ob->wto);
+ if ((si_oc(si)->flags & (CF_SHUTW|CF_WRITE_PARTIAL)) == CF_WRITE_PARTIAL &&
+ !channel_is_empty(si_oc(si)))
+ if (tick_isset(si_oc(si)->wex))
+ si_oc(si)->wex = tick_add_ifset(now_ms, si_oc(si)->wto);
if (!(si->flags & SI_FL_INDEP_STR))
- if (tick_isset(si->ib->rex))
- si->ib->rex = tick_add_ifset(now_ms, si->ib->rto);
+ if (tick_isset(si_ic(si)->rex))
+ si_ic(si)->rex = tick_add_ifset(now_ms, si_ic(si)->rto);
- if (likely((si->ob->flags & (CF_SHUTW|CF_WRITE_PARTIAL|CF_DONT_READ)) == CF_WRITE_PARTIAL &&
- channel_may_recv(si->ob) &&
- (si->ob->prod->flags & SI_FL_WAIT_ROOM)))
- si_chk_rcv(si->ob->prod);
+ if (likely((si_oc(si)->flags & (CF_SHUTW|CF_WRITE_PARTIAL|CF_DONT_READ)) == CF_WRITE_PARTIAL &&
+ channel_may_recv(si_oc(si)) &&
+ (si_oc(si)->prod->flags & SI_FL_WAIT_ROOM)))
+ si_chk_rcv(si_oc(si)->prod);
}
/* process producer side.
* immediately afterwards once the following data is parsed (eg:
* HTTP chunking).
*/
- if (((si->ib->flags & CF_READ_PARTIAL) && !channel_is_empty(si->ib)) &&
- (si->ib->pipe /* always try to send spliced data */ ||
- (si->ib->buf->i == 0 && (si->ib->cons->flags & SI_FL_WAIT_DATA)))) {
- int last_len = si->ib->pipe ? si->ib->pipe->data : 0;
+ if (((si_ic(si)->flags & CF_READ_PARTIAL) && !channel_is_empty(si_ic(si))) &&
+ (si_ic(si)->pipe /* always try to send spliced data */ ||
+ (si_ic(si)->buf->i == 0 && (si_ic(si)->cons->flags & SI_FL_WAIT_DATA)))) {
+ int last_len = si_ic(si)->pipe ? si_ic(si)->pipe->data : 0;
- si_chk_snd(si->ib->cons);
+ si_chk_snd(si_ic(si)->cons);
/* check if the consumer has freed some space either in the
* buffer or in the pipe.
*/
- if (channel_may_recv(si->ib) &&
- (!last_len || !si->ib->pipe || si->ib->pipe->data < last_len))
+ if (channel_may_recv(si_ic(si)) &&
+ (!last_len || !si_ic(si)->pipe || si_ic(si)->pipe->data < last_len))
si->flags &= ~SI_FL_WAIT_ROOM;
}
if (si->flags & SI_FL_WAIT_ROOM) {
__conn_data_stop_recv(conn);
- si->ib->rex = TICK_ETERNITY;
+ si_ic(si)->rex = TICK_ETERNITY;
}
- else if ((si->ib->flags & (CF_SHUTR|CF_READ_PARTIAL|CF_DONT_READ)) == CF_READ_PARTIAL &&
- channel_may_recv(si->ib)) {
+ else if ((si_ic(si)->flags & (CF_SHUTR|CF_READ_PARTIAL|CF_DONT_READ)) == CF_READ_PARTIAL &&
+ channel_may_recv(si_ic(si))) {
/* we must re-enable reading if si_chk_snd() has freed some space */
__conn_data_want_recv(conn);
- if (!(si->ib->flags & CF_READ_NOEXP) && tick_isset(si->ib->rex))
- si->ib->rex = tick_add_ifset(now_ms, si->ib->rto);
+ if (!(si_ic(si)->flags & CF_READ_NOEXP) && tick_isset(si_ic(si)->rex))
+ si_ic(si)->rex = tick_add_ifset(now_ms, si_ic(si)->rto);
}
/* wake the task up only when needed */
if (/* changes on the production side */
- (si->ib->flags & (CF_READ_NULL|CF_READ_ERROR)) ||
+ (si_ic(si)->flags & (CF_READ_NULL|CF_READ_ERROR)) ||
si->state != SI_ST_EST ||
(si->flags & SI_FL_ERR) ||
- ((si->ib->flags & CF_READ_PARTIAL) &&
- (!si->ib->to_forward || si->ib->cons->state != SI_ST_EST)) ||
+ ((si_ic(si)->flags & CF_READ_PARTIAL) &&
+ (!si_ic(si)->to_forward || si_ic(si)->cons->state != SI_ST_EST)) ||
/* changes on the consumption side */
- (si->ob->flags & (CF_WRITE_NULL|CF_WRITE_ERROR)) ||
- ((si->ob->flags & CF_WRITE_ACTIVITY) &&
- ((si->ob->flags & CF_SHUTW) ||
- ((si->ob->flags & CF_WAKE_WRITE) &&
- (si->ob->prod->state != SI_ST_EST ||
- (channel_is_empty(si->ob) && !si->ob->to_forward)))))) {
+ (si_oc(si)->flags & (CF_WRITE_NULL|CF_WRITE_ERROR)) ||
+ ((si_oc(si)->flags & CF_WRITE_ACTIVITY) &&
+ ((si_oc(si)->flags & CF_SHUTW) ||
+ ((si_oc(si)->flags & CF_WAKE_WRITE) &&
+ (si_oc(si)->prod->state != SI_ST_EST ||
+ (channel_is_empty(si_oc(si)) && !si_oc(si)->to_forward)))))) {
task_wakeup(si->owner, TASK_WOKEN_IO);
}
- if (si->ib->flags & CF_READ_ACTIVITY)
- si->ib->flags &= ~CF_READ_DONTWAIT;
+ if (si_ic(si)->flags & CF_READ_ACTIVITY)
+ si_ic(si)->flags &= ~CF_READ_DONTWAIT;
session_release_buffers(si_sess(si));
return 0;
static void si_conn_send(struct connection *conn)
{
struct stream_interface *si = conn->owner;
- struct channel *chn = si->ob;
+ struct channel *chn = si_oc(si);
int ret;
if (chn->pipe && conn->xprt->snd_pipe) {
*/
void stream_int_update_conn(struct stream_interface *si)
{
- struct channel *ib = si->ib;
- struct channel *ob = si->ob;
+ struct channel *ib = si_ic(si);
+ struct channel *ob = si_oc(si);
struct connection *conn = __objt_conn(si->end);
/* Check if we need to close the read side */
{
struct connection *conn = __objt_conn(si->end);
- si->ib->flags &= ~CF_SHUTR_NOW;
- if (si->ib->flags & CF_SHUTR)
+ si_ic(si)->flags &= ~CF_SHUTR_NOW;
+ if (si_ic(si)->flags & CF_SHUTR)
return;
- si->ib->flags |= CF_SHUTR;
- si->ib->rex = TICK_ETERNITY;
+ si_ic(si)->flags |= CF_SHUTR;
+ si_ic(si)->rex = TICK_ETERNITY;
si->flags &= ~SI_FL_WAIT_ROOM;
if (si->state != SI_ST_EST && si->state != SI_ST_CON)
return;
- if (si->ob->flags & CF_SHUTW) {
+ if (si_oc(si)->flags & CF_SHUTW) {
conn_full_close(conn);
si->state = SI_ST_DIS;
si->exp = TICK_ETERNITY;
{
struct connection *conn = __objt_conn(si->end);
- si->ob->flags &= ~CF_SHUTW_NOW;
- if (si->ob->flags & CF_SHUTW)
+ si_oc(si)->flags &= ~CF_SHUTW_NOW;
+ if (si_oc(si)->flags & CF_SHUTW)
return;
- si->ob->flags |= CF_SHUTW;
- si->ob->wex = TICK_ETERNITY;
+ si_oc(si)->flags |= CF_SHUTW;
+ si_oc(si)->wex = TICK_ETERNITY;
si->flags &= ~SI_FL_WAIT_DATA;
switch (si->state) {
* closed write with pending read (eg: abortonclose while
* waiting for the server).
*/
- if (!(si->flags & SI_FL_NOHALF) || !(si->ib->flags & (CF_SHUTR|CF_DONT_READ))) {
+ if (!(si->flags & SI_FL_NOHALF) || !(si_ic(si)->flags & (CF_SHUTR|CF_DONT_READ))) {
/* We shutdown transport layer */
if (conn_ctrl_ready(conn))
shutdown(conn->t.sock.fd, SHUT_WR);
- if (!(si->ib->flags & (CF_SHUTR|CF_DONT_READ))) {
+ if (!(si_ic(si)->flags & (CF_SHUTR|CF_DONT_READ))) {
/* OK just a shutw, but we want the caller
* to disable polling on this FD if exists.
*/
/* fall through */
default:
si->flags &= ~(SI_FL_WAIT_ROOM | SI_FL_NOLINGER);
- si->ib->flags &= ~CF_SHUTR_NOW;
- si->ib->flags |= CF_SHUTR;
- si->ib->rex = TICK_ETERNITY;
+ si_ic(si)->flags &= ~CF_SHUTR_NOW;
+ si_ic(si)->flags |= CF_SHUTR;
+ si_ic(si)->rex = TICK_ETERNITY;
si->exp = TICK_ETERNITY;
}
}
*/
static void stream_int_chk_rcv_conn(struct stream_interface *si)
{
- struct channel *ib = si->ib;
+ struct channel *ib = si_ic(si);
struct connection *conn = __objt_conn(si->end);
if (unlikely(si->state > SI_ST_EST || (ib->flags & CF_SHUTR)))
*/
static void stream_int_chk_snd_conn(struct stream_interface *si)
{
- struct channel *ob = si->ob;
+ struct channel *ob = si_oc(si);
struct connection *conn = __objt_conn(si->end);
if (unlikely(si->state > SI_ST_EST || (ob->flags & CF_SHUTW)))
!channel_is_empty(ob))
ob->wex = tick_add_ifset(now_ms, ob->wto);
- if (tick_isset(si->ib->rex) && !(si->flags & SI_FL_INDEP_STR)) {
+ if (tick_isset(si_ic(si)->rex) && !(si->flags & SI_FL_INDEP_STR)) {
/* Note: to prevent the client from expiring read timeouts
* during writes, we refresh it. We only do this if the
* interface is not configured for "independent streams",
* of data which can full the socket buffers long before a
* write timeout is detected.
*/
- si->ib->rex = tick_add_ifset(now_ms, si->ib->rto);
+ si_ic(si)->rex = tick_add_ifset(now_ms, si_ic(si)->rto);
}
}
*/
if (likely((ob->flags & (CF_WRITE_NULL|CF_WRITE_ERROR|CF_SHUTW)) ||
((ob->flags & CF_WAKE_WRITE) &&
- ((channel_is_empty(si->ob) && !ob->to_forward) ||
+ ((channel_is_empty(si_oc(si)) && !ob->to_forward) ||
si->state != SI_ST_EST)))) {
out_wakeup:
if (!(si->flags & SI_FL_DONT_WAKE) && si->owner)
static void si_conn_recv_cb(struct connection *conn)
{
struct stream_interface *si = conn->owner;
- struct channel *chn = si->ib;
+ struct channel *chn = si_ic(si);
int ret, max, cur_read;
int read_poll = MAX_READ_POLL_LOOPS;
static void si_conn_send_cb(struct connection *conn)
{
struct stream_interface *si = conn->owner;
- struct channel *chn = si->ob;
+ struct channel *chn = si_oc(si);
if (conn->flags & CO_FL_ERROR)
return;
{
struct connection *conn = __objt_conn(si->end);
- si->ib->flags &= ~CF_SHUTR_NOW;
- if (si->ib->flags & CF_SHUTR)
+ si_ic(si)->flags &= ~CF_SHUTR_NOW;
+ if (si_ic(si)->flags & CF_SHUTR)
return;
- si->ib->flags |= CF_SHUTR;
- si->ib->rex = TICK_ETERNITY;
+ si_ic(si)->flags |= CF_SHUTR;
+ si_ic(si)->rex = TICK_ETERNITY;
si->flags &= ~SI_FL_WAIT_ROOM;
if (si->state != SI_ST_EST && si->state != SI_ST_CON)
return;
- if (si->ob->flags & CF_SHUTW)
+ if (si_oc(si)->flags & CF_SHUTW)
goto do_close;
if (si->flags & SI_FL_NOHALF) {
/* OK we completely close the socket here just as if we went through si_shut[rw]() */
conn_full_close(conn);
- si->ib->flags &= ~CF_SHUTR_NOW;
- si->ib->flags |= CF_SHUTR;
- si->ib->rex = TICK_ETERNITY;
+ si_ic(si)->flags &= ~CF_SHUTR_NOW;
+ si_ic(si)->flags |= CF_SHUTR;
+ si_ic(si)->rex = TICK_ETERNITY;
- si->ob->flags &= ~CF_SHUTW_NOW;
- si->ob->flags |= CF_SHUTW;
- si->ob->wex = TICK_ETERNITY;
+ si_oc(si)->flags &= ~CF_SHUTW_NOW;
+ si_oc(si)->flags |= CF_SHUTW;
+ si_oc(si)->wex = TICK_ETERNITY;
si->flags &= ~(SI_FL_WAIT_DATA | SI_FL_WAIT_ROOM);
projects / thirdparty / haproxy.git / commitdiff
