#include <errno.h>

#include <fcntl.h>

#include <stdio.h>

#include <stdlib.h>

#include <string.h>

#include <sys/socket.h>

#include <sys/stat.h>

#include <sys/types.h>

#include <haproxy/api.h>

#include <haproxy/applet.h>

#include <haproxy/channel.h>

#include <haproxy/cli.h>

#include <haproxy/dict.h>

#include <haproxy/errors.h>

#include <haproxy/fd.h>

#include <haproxy/frontend.h>

#include <haproxy/net_helper.h>

#include <haproxy/obj_type-t.h>

#include <haproxy/peers.h>

#include <haproxy/proxy.h>

#include <haproxy/session-t.h>

#include <haproxy/signal.h>

#include <haproxy/stats-t.h>

#include <haproxy/stick_table.h>

#include <haproxy/stream.h>

#include <haproxy/stream_interface.h>

#include <haproxy/task.h>

#include <haproxy/thread.h>

#include <haproxy/time.h>

#include <haproxy/tools.h>

#define PEERS_F_RESYNC_LOCAL 0x00000001 /* Learn from local finished or no more needed */

#define PEERS_F_RESYNC_REMOTE 0x00000002 /* Learn from remote finished or no more needed */

#define PEERS_F_RESYNC_ASSIGN 0x00000004 /* A peer was assigned to learn our lesson */

#define PEERS_F_RESYNC_PROCESS 0x00000008 /* The assigned peer was requested for resync */

#define PEERS_F_DONOTSTOP 0x00010000 /* Main table sync task block process during soft stop

#define PEERS_RESYNC_STATEMASK (PEERS_F_RESYNC_LOCAL|PEERS_F_RESYNC_REMOTE)

#define PEERS_RESYNC_FROMLOCAL 0x00000000

#define PEERS_RESYNC_FROMREMOTE PEERS_F_RESYNC_LOCAL

#define PEERS_RESYNC_FINISHED (PEERS_F_RESYNC_LOCAL|PEERS_F_RESYNC_REMOTE)

#define SHTABLE_F_TEACH_STAGE1 0x00000001 /* Teach state 1 complete */

#define SHTABLE_F_TEACH_STAGE2 0x00000002 /* Teach state 2 complete */

#define PEER_F_TEACH_PROCESS 0x00000001 /* Teach a lesson to current peer */

#define PEER_F_TEACH_FINISHED 0x00000008 /* Teach conclude, (wait for confirm) */

#define PEER_F_TEACH_COMPLETE 0x00000010 /* All that we know already taught to current peer, used only for a local peer */

#define PEER_F_LEARN_ASSIGN 0x00000100 /* Current peer was assigned for a lesson */

#define PEER_F_LEARN_NOTUP2DATE 0x00000200 /* Learn from peer finished but peer is not up to date */

#define PEER_F_ALIVE 0x20000000 /* Used to flag a peer a alive. */

#define PEER_F_HEARTBEAT 0x40000000 /* Heartbeat message to send. */

#define PEER_F_DWNGRD 0x80000000 /* When this flag is enabled, we must downgrade the supported version announced during peer sessions. */

#define PEER_TEACH_RESET ~(PEER_F_TEACH_PROCESS|PEER_F_TEACH_FINISHED) /* PEER_F_TEACH_COMPLETE should never be reset */

#define PEER_LEARN_RESET ~(PEER_F_LEARN_ASSIGN|PEER_F_LEARN_NOTUP2DATE)

#define PEER_RESYNC_TIMEOUT 5000 /* 5 seconds */

#define PEER_RECONNECT_TIMEOUT 5000 /* 5 seconds */

#define PEER_HEARTBEAT_TIMEOUT 3000 /* 3 seconds */

enum {

PEER_MSG_CLASS_CONTROL = 0,

PEER_MSG_CLASS_ERROR,

PEER_MSG_CLASS_STICKTABLE = 10,

PEER_MSG_CLASS_RESERVED = 255,

};

enum {

PEER_MSG_CTRL_RESYNCREQ = 0,

PEER_MSG_CTRL_RESYNCFINISHED,

PEER_MSG_CTRL_RESYNCPARTIAL,

PEER_MSG_CTRL_RESYNCCONFIRM,

PEER_MSG_CTRL_HEARTBEAT,

};

enum {

PEER_MSG_ERR_PROTOCOL = 0,

PEER_MSG_ERR_SIZELIMIT,

};

enum {

PEER_KT_ANY = 0, /* any type */

PEER_KT_RESV1, /* UNUSED */

PEER_KT_SINT, /* signed 64bits integer type */

PEER_KT_RESV3, /* UNUSED */

PEER_KT_IPV4, /* ipv4 type */

PEER_KT_IPV6, /* ipv6 type */

PEER_KT_STR, /* char string type */

PEER_KT_BIN, /* buffer type */

PEER_KT_TYPES /* number of types, must always be last */

};

static int peer_net_key_type[SMP_TYPES] = {

[SMP_T_SINT] = PEER_KT_SINT,

[SMP_T_IPV4] = PEER_KT_IPV4,

[SMP_T_IPV6] = PEER_KT_IPV6,

[SMP_T_STR] = PEER_KT_STR,

[SMP_T_BIN] = PEER_KT_BIN,

};

static int peer_int_key_type[PEER_KT_TYPES] = {

[PEER_KT_SINT] = SMP_T_SINT,

[PEER_KT_IPV4] = SMP_T_IPV4,

[PEER_KT_IPV6] = SMP_T_IPV6,

[PEER_KT_STR] = SMP_T_STR,

[PEER_KT_BIN] = SMP_T_BIN,

};

struct peer_prep_params {

};

#define PEER_MSG_STKT_UPDATE 0x80

#define PEER_MSG_STKT_INCUPDATE 0x81

#define PEER_MSG_STKT_DEFINE 0x82

#define PEER_MSG_STKT_SWITCH 0x83

#define PEER_MSG_STKT_ACK 0x84

#define PEER_MSG_STKT_UPDATE_TIMED 0x85

#define PEER_MSG_STKT_INCUPDATE_TIMED 0x86

#define PEER_MSG_STKT_BIT 7

#define PEER_MSG_STKT_BIT_MASK (1 << PEER_MSG_STKT_BIT)

#define PEER_MSG_ENC_LENGTH_MAXLEN 5

#define PEER_ENC_2BYTES_MIN 0xf0 /* 0xf0 (or 240) */

#define PEER_ENC_3BYTES_MIN ((1ULL << 11) | PEER_ENC_2BYTES_MIN) /* 0x8f0 (or 2288) */

#define PEER_ENC_4BYTES_MIN ((1ULL << 18) | PEER_ENC_3BYTES_MIN) /* 0x408f0 (or 264432) */

#define PEER_ENC_5BYTES_MIN ((1ULL << 25) | PEER_ENC_4BYTES_MIN) /* 0x20408f0 (or 33818864) */

#define PEER_ENC_6BYTES_MIN ((1ULL << 32) | PEER_ENC_5BYTES_MIN) /* 0x1020408f0 (or 4328786160) */

#define PEER_ENC_7BYTES_MIN ((1ULL << 39) | PEER_ENC_6BYTES_MIN) /* 0x81020408f0 (or 554084600048) */

#define PEER_ENC_8BYTES_MIN ((1ULL << 46) | PEER_ENC_7BYTES_MIN) /* 0x4081020408f0 (or 70922828777712) */

#define PEER_ENC_9BYTES_MIN ((1ULL << 53) | PEER_ENC_8BYTES_MIN) /* 0x204081020408f0 (or 9078122083518704) */

#define PEER_ENC_10BYTES_MIN ((1ULL << 60) | PEER_ENC_9BYTES_MIN) /* 0x10204081020408f0 (or 1161999626690365680) */

#define PEER_ENC_STOP_BIT 7

#define PEER_ENC_STOP_BYTE (1 << PEER_ENC_STOP_BIT)

#define PEER_ENC_2BYTES_MIN_BITS 4

#define PEER_MSG_HEADER_LEN 2

#define PEER_STKT_CACHE_MAX_ENTRIES 128

enum {

PEER_SESS_ST_ACCEPT = 0, /* Initial state for session create by an accept, must be zero! */

PEER_SESS_ST_GETVERSION, /* Validate supported protocol version */

PEER_SESS_ST_GETHOST, /* Validate host ID correspond to local host id */

PEER_SESS_ST_GETPEER, /* Validate peer ID correspond to a known remote peer id */

/* after this point, data were possibly exchanged */

PEER_SESS_ST_SENDSUCCESS, /* Send ret code 200 (success) and wait for message */

PEER_SESS_ST_CONNECT, /* Initial state for session create on a connect, push presentation into buffer */

PEER_SESS_ST_GETSTATUS, /* Wait for the welcome message */

PEER_SESS_ST_WAITMSG, /* Wait for data messages */

PEER_SESS_ST_EXIT, /* Exit with status code */

PEER_SESS_ST_ERRPROTO, /* Send error proto message before exit */

PEER_SESS_ST_ERRSIZE, /* Send error size message before exit */

PEER_SESS_ST_END, /* Killed session */

};

#define PEER_SESS_SC_CONNECTCODE 100 /* connect in progress */

#define PEER_SESS_SC_CONNECTEDCODE 110 /* tcp connect success */

#define PEER_SESS_SC_SUCCESSCODE 200 /* accept or connect successful */

#define PEER_SESS_SC_TRYAGAIN 300 /* try again later */

#define PEER_SESS_SC_ERRPROTO 501 /* error protocol */

#define PEER_SESS_SC_ERRVERSION 502 /* unknown protocol version */

#define PEER_SESS_SC_ERRHOST 503 /* bad host name */

#define PEER_SESS_SC_ERRPEER 504 /* unknown peer */

#define PEER_SESSION_PROTO_NAME "HAProxyS"

#define PEER_MAJOR_VER 2

#define PEER_MINOR_VER 1

#define PEER_DWNGRD_MINOR_VER 0

static size_t proto_len = sizeof(PEER_SESSION_PROTO_NAME) - 1;

struct peers *cfg_peers = NULL;

static void peer_session_forceshutdown(struct peer *peer);

static struct ebpt_node *dcache_tx_insert(struct dcache *dc,

struct dcache_tx_entry *i);

static inline void flush_dcache(struct peer *peer);

static const char *statuscode_str(int statuscode)

{

switch (statuscode) {

case PEER_SESS_SC_CONNECTCODE:

return "CONN";

case PEER_SESS_SC_CONNECTEDCODE:

return "HSHK";

case PEER_SESS_SC_SUCCESSCODE:

return "ESTA";

case PEER_SESS_SC_TRYAGAIN:

return "RETR";

case PEER_SESS_SC_ERRPROTO:

return "PROT";

case PEER_SESS_SC_ERRVERSION:

return "VERS";

case PEER_SESS_SC_ERRHOST:

return "NAME";

case PEER_SESS_SC_ERRPEER:

return "UNKN";

default:

return "NONE";

}

}

int intencode(uint64_t i, char **str) {

int idx = 0;

unsigned char *msg;

msg = (unsigned char *)*str;

if (i < PEER_ENC_2BYTES_MIN) {

msg[0] = (unsigned char)i;

*str = (char *)&msg[idx+1];

return (idx+1);

}

msg[idx] =(unsigned char)i | PEER_ENC_2BYTES_MIN;

i = (i - PEER_ENC_2BYTES_MIN) >> PEER_ENC_2BYTES_MIN_BITS;

while (i >= PEER_ENC_STOP_BYTE) {

msg[++idx] = (unsigned char)i | PEER_ENC_STOP_BYTE;

i = (i - PEER_ENC_STOP_BYTE) >> PEER_ENC_STOP_BIT;

}

msg[++idx] = (unsigned char)i;

*str = (char *)&msg[idx+1];

return (idx+1);

}

uint64_t intdecode(char **str, char *end)

{

unsigned char *msg;

uint64_t i;

int shift;

if (!*str)

return 0;

msg = (unsigned char *)*str;

if (msg >= (unsigned char *)end)

goto fail;

i = *(msg++);

if (i >= PEER_ENC_2BYTES_MIN) {

shift = PEER_ENC_2BYTES_MIN_BITS;

do {

if (msg >= (unsigned char *)end)

goto fail;

i += (uint64_t)*msg << shift;

shift += PEER_ENC_STOP_BIT;

} while (*(msg++) >= PEER_ENC_STOP_BYTE);

}

*str = (char *)msg;

return i;

fail:

*str = NULL;

return 0;

}

static int peer_prepare_hellomsg(char *msg, size_t size, struct peer_prep_params *p)

{

int min_ver, ret;

struct peer *peer;

peer = p->hello.peer;

min_ver = (peer->flags & PEER_F_DWNGRD) ? PEER_DWNGRD_MINOR_VER : PEER_MINOR_VER;

/* Prepare headers */

ret = snprintf(msg, size, PEER_SESSION_PROTO_NAME " %u.%u\n%s\n%s %d %d\n",

PEER_MAJOR_VER, min_ver, peer->id, localpeer, (int)getpid(), relative_pid);

if (ret >= size)

return 0;

return ret;

}

static int peer_prepare_status_successmsg(char *msg, size_t size, struct peer_prep_params *p)

{

int ret;

ret = snprintf(msg, size, "%d\n", PEER_SESS_SC_SUCCESSCODE);

if (ret >= size)

return 0;

return ret;

}

static int peer_prepare_status_errormsg(char *msg, size_t size, struct peer_prep_params *p)

{

int ret;

unsigned int st1;

st1 = p->error_status.st1;

ret = snprintf(msg, size, "%d\n", st1);

if (ret >= size)

return 0;

return ret;

}

static inline void peer_set_update_msg_type(char *msg_type, int use_identifier, int use_timed)

{

if (use_timed) {

if (use_identifier)

*msg_type = PEER_MSG_STKT_UPDATE_TIMED;

else

*msg_type = PEER_MSG_STKT_INCUPDATE_TIMED;

}

else {

if (use_identifier)

*msg_type = PEER_MSG_STKT_UPDATE;

else

*msg_type = PEER_MSG_STKT_INCUPDATE;

}

}

static int peer_prepare_updatemsg(char *msg, size_t size, struct peer_prep_params *p)

{

uint32_t netinteger;

unsigned short datalen;

char *cursor, *datamsg;

unsigned int data_type;

void *data_ptr;

struct stksess *ts;

struct shared_table *st;

unsigned int updateid;

int use_identifier;

int use_timed;

struct peer *peer;

ts = p->updt.stksess;

st = p->updt.shared_table;

updateid = p->updt.updateid;

use_identifier = p->updt.use_identifier;

use_timed = p->updt.use_timed;

peer = p->updt.peer;

cursor = datamsg = msg + PEER_MSG_HEADER_LEN + PEER_MSG_ENC_LENGTH_MAXLEN;

/* construct message */

/* check if we need to send the update identifier */

if (!st->last_pushed || updateid < st->last_pushed || ((updateid - st->last_pushed) != 1)) {

use_identifier = 1;

}

/* encode update identifier if needed */

if (use_identifier) {

netinteger = htonl(updateid);

memcpy(cursor, &netinteger, sizeof(netinteger));

cursor += sizeof(netinteger);

}

if (use_timed) {

netinteger = htonl(tick_remain(now_ms, ts->expire));

memcpy(cursor, &netinteger, sizeof(netinteger));

cursor += sizeof(netinteger);

}

/* encode the key */

if (st->table->type == SMP_T_STR) {

int stlen = strlen((char *)ts->key.key);

intencode(stlen, &cursor);

memcpy(cursor, ts->key.key, stlen);

cursor += stlen;

}

else if (st->table->type == SMP_T_SINT) {

netinteger = htonl(read_u32(ts->key.key));

memcpy(cursor, &netinteger, sizeof(netinteger));

cursor += sizeof(netinteger);

}

else {

memcpy(cursor, ts->key.key, st->table->key_size);

cursor += st->table->key_size;

}

HA_RWLOCK_RDLOCK(STK_SESS_LOCK, &ts->lock);

/* encode values */

for (data_type = 0 ; data_type < STKTABLE_DATA_TYPES ; data_type++) {

data_ptr = stktable_data_ptr(st->table, ts, data_type);

if (data_ptr) {

switch (stktable_data_types[data_type].std_type) {

case STD_T_SINT: {

int data;

data = stktable_data_cast(data_ptr, std_t_sint);

intencode(data, &cursor);

break;

}

case STD_T_UINT: {

unsigned int data;

data = stktable_data_cast(data_ptr, std_t_uint);

intencode(data, &cursor);

break;

}

case STD_T_ULL: {

unsigned long long data;

data = stktable_data_cast(data_ptr, std_t_ull);

intencode(data, &cursor);

break;

}

case STD_T_FRQP: {

struct freq_ctr_period *frqp;

frqp = &stktable_data_cast(data_ptr, std_t_frqp);

intencode((unsigned int)(now_ms - frqp->curr_tick), &cursor);

intencode(frqp->curr_ctr, &cursor);

intencode(frqp->prev_ctr, &cursor);

break;

}

case STD_T_DICT: {

struct dict_entry *de;

struct ebpt_node *cached_de;

struct dcache_tx_entry cde = { };

char *beg, *end;

size_t value_len, data_len;

struct dcache *dc;

de = stktable_data_cast(data_ptr, std_t_dict);

if (!de) {

/* No entry */

intencode(0, &cursor);

break;

}

dc = peer->dcache;

cde.entry.key = de;

cached_de = dcache_tx_insert(dc, &cde);

if (cached_de == &cde.entry) {

if (cde.id + 1 >= PEER_ENC_2BYTES_MIN)

break;

/* Encode the length of the remaining data -> 1 */

intencode(1, &cursor);

/* Encode the cache entry ID */

intencode(cde.id + 1, &cursor);

}

else {

/* Leave enough room to encode the remaining data length. */

end = beg = cursor + PEER_MSG_ENC_LENGTH_MAXLEN;

/* Encode the dictionary entry key */

intencode(cde.id + 1, &end);

/* Encode the length of the dictionary entry data */

value_len = de->len;

intencode(value_len, &end);

/* Copy the data */

memcpy(end, de->value.key, value_len);

end += value_len;

/* Encode the length of the data */

data_len = end - beg;

intencode(data_len, &cursor);

memmove(cursor, beg, data_len);

cursor += data_len;

}

break;

}

}

}

}

HA_RWLOCK_RDUNLOCK(STK_SESS_LOCK, &ts->lock);

/* Compute datalen */

datalen = (cursor - datamsg);

/* prepare message header */

msg[0] = PEER_MSG_CLASS_STICKTABLE;

peer_set_update_msg_type(&msg[1], use_identifier, use_timed);

cursor = &msg[2];

intencode(datalen, &cursor);

/* move data after header */

memmove(cursor, datamsg, datalen);

/* return header size + data_len */

return (cursor - msg) + datalen;

}

static int peer_prepare_switchmsg(char *msg, size_t size, struct peer_prep_params *params)

{

int len;

unsigned short datalen;

struct buffer *chunk;

char *cursor, *datamsg, *chunkp, *chunkq;

uint64_t data = 0;

unsigned int data_type;

struct shared_table *st;

st = params->swtch.shared_table;

cursor = datamsg = msg + PEER_MSG_HEADER_LEN + PEER_MSG_ENC_LENGTH_MAXLEN;

/* Encode data */

/* encode local id */

intencode(st->local_id, &cursor);

/* encode table name */

len = strlen(st->table->nid);

intencode(len, &cursor);

memcpy(cursor, st->table->nid, len);

cursor += len;

/* encode table type */

intencode(peer_net_key_type[st->table->type], &cursor);

/* encode table key size */

intencode(st->table->key_size, &cursor);

chunk = get_trash_chunk();

chunkp = chunkq = chunk->area;

/* encode available known data types in table */

for (data_type = 0 ; data_type < STKTABLE_DATA_TYPES ; data_type++) {

if (st->table->data_ofs[data_type]) {

switch (stktable_data_types[data_type].std_type) {

case STD_T_SINT:

case STD_T_UINT:

case STD_T_ULL:

case STD_T_DICT:

data |= 1 << data_type;

break;

case STD_T_FRQP:

data |= 1 << data_type;

intencode(data_type, &chunkq);

intencode(st->table->data_arg[data_type].u, &chunkq);

break;

}

}

}

intencode(data, &cursor);

/* Encode stick-table entries duration. */

intencode(st->table->expire, &cursor);

if (chunkq > chunkp) {

chunk->data = chunkq - chunkp;

memcpy(cursor, chunk->area, chunk->data);

cursor += chunk->data;

}

/* Compute datalen */

datalen = (cursor - datamsg);

/* prepare message header */

msg[0] = PEER_MSG_CLASS_STICKTABLE;

msg[1] = PEER_MSG_STKT_DEFINE;

cursor = &msg[2];

intencode(datalen, &cursor);

/* move data after header */

memmove(cursor, datamsg, datalen);

/* return header size + data_len */

return (cursor - msg) + datalen;

}

static int peer_prepare_ackmsg(char *msg, size_t size, struct peer_prep_params *p)

{

unsigned short datalen;

char *cursor, *datamsg;

uint32_t netinteger;

struct shared_table *st;

cursor = datamsg = msg + PEER_MSG_HEADER_LEN + PEER_MSG_ENC_LENGTH_MAXLEN;

st = p->ack.shared_table;

intencode(st->remote_id, &cursor);

netinteger = htonl(st->last_get);

memcpy(cursor, &netinteger, sizeof(netinteger));

cursor += sizeof(netinteger);

/* Compute datalen */

datalen = (cursor - datamsg);

/* prepare message header */

msg[0] = PEER_MSG_CLASS_STICKTABLE;

msg[1] = PEER_MSG_STKT_ACK;

cursor = &msg[2];

intencode(datalen, &cursor);

/* move data after header */

memmove(cursor, datamsg, datalen);

/* return header size + data_len */

return (cursor - msg) + datalen;

}

void __peer_session_deinit(struct peer *peer)

{

struct stream_interface *si;

struct stream *s;

struct peers *peers;

if (!peer->appctx)

return;

si = peer->appctx->owner;

if (!si)

return;

s = si_strm(si);

if (!s)

return;

peers = strm_fe(s)->parent;

if (!peers)

return;

if (peer->appctx->st0 == PEER_SESS_ST_WAITMSG)

HA_ATOMIC_SUB(&connected_peers, 1);

HA_ATOMIC_SUB(&active_peers, 1);

flush_dcache(peer);

/* Re-init current table pointers to force announcement on re-connect */

peer->remote_table = peer->last_local_table = NULL;

peer->appctx = NULL;

if (peer->flags & PEER_F_LEARN_ASSIGN) {

/* unassign current peer for learning */

peer->flags &= ~(PEER_F_LEARN_ASSIGN);

peers->flags &= ~(PEERS_F_RESYNC_ASSIGN|PEERS_F_RESYNC_PROCESS);

/* reschedule a resync */

peers->resync_timeout = tick_add(now_ms, MS_TO_TICKS(5000));

}

/* reset teaching and learning flags to 0 */

peer->flags &= PEER_TEACH_RESET;

peer->flags &= PEER_LEARN_RESET;

/* set this peer as dead from heartbeat point of view */

peer->flags &= ~PEER_F_ALIVE;

task_wakeup(peers->sync_task, TASK_WOKEN_MSG);

}

static void peer_session_release(struct appctx *appctx)

{

struct peer *peer = appctx->ctx.peers.ptr;

/* appctx->ctx.peers.ptr is not a peer session */

if (appctx->st0 < PEER_SESS_ST_SENDSUCCESS)

return;

/* peer session identified */

if (peer) {

HA_SPIN_LOCK(PEER_LOCK, &peer->lock);

if (peer->appctx == appctx)

__peer_session_deinit(peer);

HA_SPIN_UNLOCK(PEER_LOCK, &peer->lock);

}

}

static int peer_get_version(const char *str,

unsigned int *maj_ver, unsigned int *min_ver)

{

unsigned int majv, minv;

const char *pos, *saved;

const char *end;

saved = pos = str;

end = str + strlen(str);

majv = read_uint(&pos, end);

if (saved == pos || *pos++ != '.')

return -1;

saved = pos;

minv = read_uint(&pos, end);

if (saved == pos || pos != end)

return -1;

*maj_ver = majv;

*min_ver = minv;

return 0;

}

static inline int peer_getline(struct appctx *appctx)

{

int n;

struct stream_interface *si = appctx->owner;

n = co_getline(si_oc(si), trash.area, trash.size);

if (!n)

return 0;

if (n < 0 || trash.area[n - 1] != '\n') {

appctx->st0 = PEER_SESS_ST_END;

return -1;

}

if (n > 1 && (trash.area[n - 2] == '\r'))

trash.area[n - 2] = 0;

else

trash.area[n - 1] = 0;

co_skip(si_oc(si), n);

return n;

}

static inline int peer_send_msg(struct appctx *appctx,

int (*peer_prepare_msg)(char *, size_t, struct peer_prep_params *),

struct peer_prep_params *params)

{

int ret, msglen;

struct stream_interface *si = appctx->owner;

msglen = peer_prepare_msg(trash.area, trash.size, params);

if (!msglen) {

/* internal error: message does not fit in trash */

appctx->st0 = PEER_SESS_ST_END;

return 0;

}

/* message to buffer */

ret = ci_putblk(si_ic(si), trash.area, msglen);

if (ret <= 0) {

if (ret == -1) {

/* No more write possible */

si_rx_room_blk(si);

return -1;

}

appctx->st0 = PEER_SESS_ST_END;

}

return ret;

}

static inline int peer_send_hellomsg(struct appctx *appctx, struct peer *peer)

{

struct peer_prep_params p = {

.hello.peer = peer,

};

return peer_send_msg(appctx, peer_prepare_hellomsg, &p);

}

static inline int peer_send_status_successmsg(struct appctx *appctx)

{

return peer_send_msg(appctx, peer_prepare_status_successmsg, NULL);

}

static inline int peer_send_status_errormsg(struct appctx *appctx)

{

struct peer_prep_params p = {

.error_status.st1 = appctx->st1,

};

return peer_send_msg(appctx, peer_prepare_status_errormsg, &p);

}

static inline int peer_send_switchmsg(struct shared_table *st, struct appctx *appctx)

{

struct peer_prep_params p = {

.swtch.shared_table = st,

};

return peer_send_msg(appctx, peer_prepare_switchmsg, &p);

}

static inline int peer_send_ackmsg(struct shared_table *st, struct appctx *appctx)

{

struct peer_prep_params p = {

.ack.shared_table = st,

};

return peer_send_msg(appctx, peer_prepare_ackmsg, &p);

}

static inline int peer_send_updatemsg(struct shared_table *st, struct appctx *appctx, struct stksess *ts,

unsigned int updateid, int use_identifier, int use_timed)

{

struct peer_prep_params p = {

.updt.stksess = ts,

.updt.shared_table = st,

.updt.updateid = updateid,