From 96084e1a1f2e0a49c961bbddb9fffd2e03bfae3f Mon Sep 17 00:00:00 2001 From: "/C=EU/ST=EU/CN=Pablo Neira Ayuso/emailAddress=pablo@netfilter.org" Date: Mon, 9 Jul 2007 19:11:53 +0000 Subject: - conntrack-tools requires libnetfilter_conntrack >= 0.0.81 - add len field to nethdr - implement buffered send/recv to batch messages - stop using netlink format for network messages: use similar TLV-based format - reduce synchronization messages size up to 60% - introduce periodic alive messages for sync-nack protocol - timeslice alarm implementation: remove alarm pthread, remove locking - simplify debugging functions: use nfct_snprintf instead - remove major use of libnfnetlink functions: use libnetfilter_conntrack API - deprecate conntrackd -F, use conntrack -F instead - major rework of the network infrastructure: much simple, less messy --- src/sync-mode.c | 152 +++++++++++++++++++++++++++++++++----------------------- 1 file changed, 89 insertions(+), 63 deletions(-) (limited to 'src/sync-mode.c') diff --git a/src/sync-mode.c b/src/sync-mode.c index 38ab016..f30cb95 100644 --- a/src/sync-mode.c +++ b/src/sync-mode.c @@ -27,43 +27,27 @@ #include #include "sync.h" #include "network.h" +#include "buffer.h" +#include "debug.h" -/* handler for multicast messages received */ -static void mcast_handler() +static void do_mcast_handler_step(struct nethdr *net) { - int ret; - unsigned int type; - char __net[4096]; - struct nethdr *net = (struct nethdr *) __net; - struct nlmsghdr *nlh = (struct nlmsghdr *) (__net + NETHDR_SIZ); + unsigned int query; + struct netpld *pld = NETHDR_DATA(net); char __ct[nfct_maxsize()]; struct nf_conntrack *ct = (struct nf_conntrack *) __ct; struct us_conntrack *u = NULL; - ret = mcast_recv_netmsg(STATE_SYNC(mcast_server), net, sizeof(__net)); - if (ret <= 0) { - STATE(malformed)++; - return; - } - if (STATE_SYNC(sync)->recv(net)) return; memset(ct, 0, sizeof(__ct)); - if ((type = parse_network_msg(ct, nlh)) == NFCT_T_ERROR) { - STATE(malformed)++; - return; - } + /* XXX: check for malformed */ + parse_netpld(ct, pld, &query); - nfct_attr_unset(ct, ATTR_TIMEOUT); - nfct_attr_unset(ct, ATTR_ORIG_COUNTER_BYTES); - nfct_attr_unset(ct, ATTR_ORIG_COUNTER_PACKETS); - nfct_attr_unset(ct, ATTR_REPL_COUNTER_BYTES); - nfct_attr_unset(ct, ATTR_REPL_COUNTER_PACKETS); - - switch(type) { - case NFCT_T_NEW: + switch(query) { + case NFCT_Q_CREATE: retry: if ((u = cache_add(STATE_SYNC(external), ct))) { debug_ct(u->ct, "external new"); @@ -80,24 +64,57 @@ retry: debug_ct(ct, "can't add"); } break; - case NFCT_T_UPDATE: + case NFCT_Q_UPDATE: if ((u = cache_update_force(STATE_SYNC(external), ct))) { debug_ct(u->ct, "external update"); } else debug_ct(ct, "can't update"); break; - case NFCT_T_DESTROY: + case NFCT_Q_DESTROY: if (cache_del(STATE_SYNC(external), ct)) debug_ct(ct, "external destroy"); else debug_ct(ct, "can't destroy"); break; default: - dlog(STATE(log), "mcast received unknown msg type %d\n", type); + dlog(STATE(log), "mcast received unknown query %d\n", query); break; } } +/* handler for multicast messages received */ +static void mcast_handler() +{ + int numbytes, remain; + char __net[4096], *ptr = __net; + + numbytes = mcast_recv(STATE_SYNC(mcast_server), __net, sizeof(__net)); + if (numbytes <= 0) + return; + + remain = numbytes; + while (remain > 0) { + struct nethdr *net = (struct nethdr *) ptr; + + if (ntohs(net->len) > remain) { + dlog(STATE(log), "fragmented messages"); + break; + } + + debug("recv sq: %u fl:%u len:%u (rem:%d)\n", + ntohl(net->seq), ntohs(net->flags), + ntohs(net->len), remain); + + if (handle_netmsg(net) == -1) { + STATE(malformed)++; + return; + } + do_mcast_handler_step(net); + ptr += net->len; + remain -= net->len; + } +} + static int init_sync(void) { int ret; @@ -159,11 +176,6 @@ static int init_sync(void) /* initialization of multicast sequence generation */ STATE_SYNC(last_seq_sent) = time(NULL); - if (create_alarm_thread() == -1) { - dlog(STATE(log), "[FAIL] can't initialize alarm thread"); - return -1; - } - return 0; } @@ -174,11 +186,17 @@ static int add_fds_to_set_sync(fd_set *readfds) return STATE_SYNC(mcast_server->fd); } -static void step_sync(fd_set *readfds) +static void run_sync(fd_set *readfds, int step) { /* multicast packet has been received */ if (FD_ISSET(STATE_SYNC(mcast_server->fd), readfds)) mcast_handler(); + + if (STATE_SYNC(sync)->run) + STATE_SYNC(sync)->run(step); + + /* flush pending messages */ + mcast_buffered_pending_netmsg(STATE_SYNC(mcast_client)); } static void kill_sync() @@ -189,8 +207,6 @@ static void kill_sync() mcast_server_destroy(STATE_SYNC(mcast_server)); mcast_client_destroy(STATE_SYNC(mcast_client)); - destroy_alarm_thread(); - if (STATE_SYNC(sync)->kill) STATE_SYNC(sync)->kill(); } @@ -267,10 +283,6 @@ static int local_handler_sync(int fd, int type, void *data) STATE_SYNC(mcast_server)); dump_stats_sync(fd); break; - case SEND_BULK: - dlog(STATE(log), "[REQ] sending bulk update"); - cache_bulk(STATE_SYNC(internal)); - break; default: if (STATE_SYNC(sync)->local) ret = STATE_SYNC(sync)->local(fd, type, data); @@ -280,7 +292,7 @@ static int local_handler_sync(int fd, int type, void *data) return ret; } -static void dump_sync(struct nf_conntrack *ct, struct nlmsghdr *nlh) +static void dump_sync(struct nf_conntrack *ct) { /* This is required by kernels < 2.6.20 */ nfct_attr_unset(ct, ATTR_TIMEOUT); @@ -294,23 +306,21 @@ static void dump_sync(struct nf_conntrack *ct, struct nlmsghdr *nlh) debug_ct(ct, "resync"); } -static void mcast_send_sync(struct nlmsghdr *nlh, - struct us_conntrack *u, +static void mcast_send_sync(struct us_conntrack *u, struct nf_conntrack *ct, - int type) + int query) { - char __net[4096]; - struct nethdr *net = (struct nethdr *) __net; - - memset(__net, 0, sizeof(__net)); + int len; + struct nethdr *net; - if (!state_helper_verdict(type, ct)) + if (!state_helper_verdict(query, ct)) return; - memcpy(__net + NETHDR_SIZ, nlh, nlh->nlmsg_len); - mcast_send_netmsg(STATE_SYNC(mcast_client), net); + net = BUILD_NETMSG(ct, query); + len = prepare_send_netmsg(STATE_SYNC(mcast_client), net); + mcast_buffered_send_netmsg(STATE_SYNC(mcast_client), net, len); if (STATE_SYNC(sync)->send) - STATE_SYNC(sync)->send(type, net, u); + STATE_SYNC(sync)->send(net, u); } static int overrun_cb(enum nf_conntrack_msg_type type, @@ -332,8 +342,16 @@ static int overrun_cb(enum nf_conntrack_msg_type type, if (!cache_test(STATE_SYNC(internal), ct)) { if ((u = cache_update_force(STATE_SYNC(internal), ct))) { + int len; + debug_ct(u->ct, "overrun resync"); - mcast_build_send_update(u); + + struct nethdr *net = BUILD_NETMSG(u->ct, NFCT_Q_UPDATE); + len = prepare_send_netmsg(STATE_SYNC(mcast_client),net); + mcast_buffered_send_netmsg(STATE_SYNC(mcast_client), + net, len); + if (STATE_SYNC(sync)->send) + STATE_SYNC(sync)->send(net, u); } } @@ -348,9 +366,17 @@ static int overrun_purge_step(void *data1, void *data2) ret = nfct_query(h, NFCT_Q_GET, u->ct); if (ret == -1 && errno == ENOENT) { + int len; + struct nethdr *net = BUILD_NETMSG(u->ct, NFCT_Q_DESTROY); + debug_ct(u->ct, "overrun purge resync"); - mcast_build_send_destroy(u); - __cache_del(STATE_SYNC(internal), u->ct); + + len = prepare_send_netmsg(STATE_SYNC(mcast_client), net); + mcast_buffered_send_netmsg(STATE_SYNC(mcast_client), net, len); + if (STATE_SYNC(sync)->send) + STATE_SYNC(sync)->send(net, u); + + cache_del(STATE_SYNC(internal), u->ct); } return 0; @@ -382,7 +408,7 @@ static void overrun_sync() nfct_close(h); } -static void event_new_sync(struct nf_conntrack *ct, struct nlmsghdr *nlh) +static void event_new_sync(struct nf_conntrack *ct) { struct us_conntrack *u; @@ -394,12 +420,12 @@ static void event_new_sync(struct nf_conntrack *ct, struct nlmsghdr *nlh) nfct_attr_unset(ct, ATTR_TIMEOUT); retry: if ((u = cache_add(STATE_SYNC(internal), ct))) { - mcast_send_sync(nlh, u, ct, NFCT_T_NEW); + mcast_send_sync(u, ct, NFCT_Q_CREATE); debug_ct(u->ct, "internal new"); } else { if (errno == EEXIST) { cache_del(STATE_SYNC(internal), ct); - mcast_send_sync(nlh, NULL, ct, NFCT_T_DESTROY); + mcast_send_sync(NULL, ct, NFCT_Q_DESTROY); goto retry; } @@ -409,7 +435,7 @@ retry: } } -static void event_update_sync(struct nf_conntrack *ct, struct nlmsghdr *nlh) +static void event_update_sync(struct nf_conntrack *ct) { struct us_conntrack *u; @@ -420,15 +446,15 @@ static void event_update_sync(struct nf_conntrack *ct, struct nlmsghdr *nlh) return; } debug_ct(u->ct, "internal update"); - mcast_send_sync(nlh, u, ct, NFCT_T_UPDATE); + mcast_send_sync(u, ct, NFCT_Q_UPDATE); } -static int event_destroy_sync(struct nf_conntrack *ct, struct nlmsghdr *nlh) +static int event_destroy_sync(struct nf_conntrack *ct) { nfct_attr_unset(ct, ATTR_TIMEOUT); if (cache_del(STATE_SYNC(internal), ct)) { - mcast_send_sync(nlh, NULL, ct, NFCT_T_DESTROY); + mcast_send_sync(NULL, ct, NFCT_Q_DESTROY); debug_ct(ct, "internal destroy"); } else debug_ct(ct, "can't destroy"); @@ -437,7 +463,7 @@ static int event_destroy_sync(struct nf_conntrack *ct, struct nlmsghdr *nlh) struct ct_mode sync_mode = { .init = init_sync, .add_fds_to_set = add_fds_to_set_sync, - .step = step_sync, + .run = run_sync, .local = local_handler_sync, .kill = kill_sync, .dump = dump_sync, -- cgit v1.2.3