ipoe: implemented new load balancing mechanism

new config options:
[ipoe]
weight=N - global weight
interface=ethX,weight=N - per-interface weight

How it works:
On reception of DHCPDISCOVER accel-ppp sends broadcast DHCP message to port 67 with same xid and add special vendor-specific option
where encodes its current session count multipled by weight.
On reception of such message accel-ppp searches session with same xid and compares weight.
If received weight is less than session's weight then it terminates this session.
per-interface weight=0 has special meaning as backup (fail-over) interface, f.e. it terminates session on any received weight.

By default weight based load balancing is disabled.
To enable need to specify global or/and per-interface weight.

4 files changed, 107 insertions, 31 deletions

diff --git a/accel-pppd/ctrl/ipoe/dhcpv4.c b/accel-pppd/ctrl/ipoe/dhcpv4.c
index ccad318..7bbed3f 100644
--- a/accel-pppd/ctrl/ipoe/dhcpv4.c
+++ b/accel-pppd/ctrl/ipoe/dhcpv4.c
@@ -45,33 +45,21 @@ static mempool_t opt_pool;
 static LIST_HEAD(relay_list);
 static pthread_mutex_t relay_lock = PTHREAD_MUTEX_INITIALIZER;
 
-static pthread_key_t raw_sock_key;
-static __thread int raw_sock = -1;
+static int raw_sock = -1;
 
 static int dhcpv4_read(struct triton_md_handler_t *h);
 int dhcpv4_packet_add_opt(struct dhcpv4_packet *pack, int type, const void *data, int len);
 
-static int open_raw_sock(void)
+static void open_raw_sock(void)
 {
-	if (raw_sock == -1) {
-		raw_sock = socket(AF_PACKET, SOCK_RAW, 0);
-		if (raw_sock < 0) {
-			log_error("dhcpv4: socket(AF_PACKET, SOCK_RAW): %s\n", strerror(errno));
-			return -1;
-		}
-
-		fcntl(raw_sock, F_SETFL, O_NONBLOCK);
-		fcntl(raw_sock, F_SETFD, fcntl(raw_sock, F_GETFD) | FD_CLOEXEC);
-
-		pthread_setspecific(raw_sock_key, (void *)(long)raw_sock);
+	raw_sock = socket(AF_PACKET, SOCK_RAW, 0);
+	if (raw_sock < 0) {
+		log_error("dhcpv4: socket(AF_PACKET, SOCK_RAW): %s\n", strerror(errno));
+		return;
 	}
 
-	return raw_sock;
-}
-
-static void close_raw_sock(void *arg)
-{
-	close((long)arg);
+	fcntl(raw_sock, F_SETFL, O_NONBLOCK);
+	fcntl(raw_sock, F_SETFD, FD_CLOEXEC);
 }
 
 static struct dhcpv4_iprange *parse_range(const char *str)
@@ -586,7 +574,7 @@ uint16_t ip_csum(uint16_t *buf, int len)
 }
 
 
-static int dhcpv4_send_raw(struct dhcpv4_serv *serv, struct dhcpv4_packet *pack, in_addr_t saddr, in_addr_t daddr)
+static int dhcpv4_send_raw(struct dhcpv4_serv *serv, struct dhcpv4_packet *pack, in_addr_t saddr, in_addr_t daddr, int dport)
 {
 	uint8_t hdr[sizeof(struct ether_header) + sizeof(struct iphdr) + sizeof(struct udphdr)];
 	struct ether_header *eth = (struct ether_header *)hdr;
@@ -597,7 +585,6 @@ static int dhcpv4_send_raw(struct dhcpv4_serv *serv, struct dhcpv4_packet *pack,
 	static uint8_t bc_addr[ETH_ALEN] = {0xff, 0xff, 0xff, 0xff, 0xff, 0xff};
 	struct sockaddr_ll ll_addr;
 	struct msghdr msg;
-	int sock = open_raw_sock();
 
 	memset(&ll_addr, 0, sizeof(ll_addr));
 	ll_addr.sll_family = AF_PACKET;
@@ -626,10 +613,11 @@ static int dhcpv4_send_raw(struct dhcpv4_serv *serv, struct dhcpv4_packet *pack,
 	ip->check = 0;
 	ip->saddr = saddr;
 	ip->daddr = (pack->hdr->flags & DHCP_F_BROADCAST) ? INADDR_BROADCAST : daddr;
+	barrier();
 	ip->check = ip_csum((uint16_t *)ip, 20);
 
 	udp->source = ntohs(DHCP_SERV_PORT);
-	udp->dest = ntohs(DHCP_CLIENT_PORT);
+	udp->dest = ntohs(dport);
 	udp->len = htons(sizeof(*udp) + len);
 	udp->check = 0;
 
@@ -638,13 +626,10 @@ static int dhcpv4_send_raw(struct dhcpv4_serv *serv, struct dhcpv4_packet *pack,
 	iov[1].iov_base = pack->data;
 	iov[1].iov_len = len;
 
-	len = sendmsg(sock, &msg, 0);
+	len = sendmsg(raw_sock, &msg, 0);
 
-	if (len < 0) {
-		perror("sendmsg");
-		printf("%i %i\n", errno, serv->ifindex);
+	if (len < 0)
 		return -1;
-	}
 
 	return 0;
 }
@@ -790,7 +775,7 @@ int dhcpv4_send_reply(int msg_type, struct dhcpv4_serv *serv, struct dhcpv4_pack
 	else if (req->hdr->ciaddr && !(pack->hdr->flags & DHCP_F_BROADCAST))
 		r = dhcpv4_send_udp(serv, pack, req->hdr->ciaddr, DHCP_CLIENT_PORT);
 	else
-		r = dhcpv4_send_raw(serv, pack, siaddr, yiaddr);
+		r = dhcpv4_send_raw(serv, pack, siaddr, yiaddr, DHCP_CLIENT_PORT);
 
 	dhcpv4_packet_free(pack);
 
@@ -835,7 +820,7 @@ int dhcpv4_send_nak(struct dhcpv4_serv *serv, struct dhcpv4_packet *req)
 	if (req->hdr->giaddr)
 		r = dhcpv4_send_udp(serv, pack, req->hdr->giaddr, DHCP_SERV_PORT);
 	else
-		r = dhcpv4_send_raw(serv, pack, 0, 0xffffffff);
+		r = dhcpv4_send_raw(serv, pack, 0, 0xffffffff, DHCP_CLIENT_PORT);
 
 	dhcpv4_packet_free(pack);
 
@@ -848,6 +833,34 @@ out_err:
 	return 0;
 }
 
+void dhcpv4_send_notify(struct dhcpv4_serv *serv, struct dhcpv4_packet *req, unsigned int weight)
+{
+	struct dhcpv4_packet *pack = dhcpv4_packet_alloc();
+	uint8_t opt[8];
+
+	if (!pack) {
+		log_emerg("out of memory\n");
+		return;
+	}
+
+	memcpy(pack->hdr, req->hdr, sizeof(*req->hdr));
+	pack->hdr->flags = DHCP_F_BROADCAST;
+	pack->hdr->ciaddr = 0;
+	pack->hdr->yiaddr = 0;
+	pack->hdr->siaddr = 0;
+	pack->hdr->giaddr = 0;
+
+	*(uint32_t *)opt = htonl(ACCEL_PPP_MAGIC);
+	*(uint32_t *)(opt + 4) = htonl(weight);
+
+	dhcpv4_packet_add_opt_u8(pack, 53, DHCPDISCOVER);
+	dhcpv4_packet_add_opt(pack, 43, opt, sizeof(opt));
+
+	dhcpv4_send_raw(serv, pack, 0, INADDR_BROADCAST, DHCP_SERV_PORT);
+
+	dhcpv4_packet_free(pack);
+}
+
 struct dhcpv4_relay *dhcpv4_relay_create(const char *_addr, in_addr_t giaddr, struct triton_context_t *ctx, triton_event_func recv)
 {
 	char str[17], *ptr;
@@ -1190,7 +1203,7 @@ static void init()
 	pack_pool = mempool_create(BUF_SIZE + sizeof(struct dhcpv4_packet));
 	opt_pool = mempool_create(sizeof(struct dhcpv4_option));
 
-	pthread_key_create(&raw_sock_key, close_raw_sock);
+	open_raw_sock();
 
 	load_config();
 
diff --git a/accel-pppd/ctrl/ipoe/dhcpv4.h b/accel-pppd/ctrl/ipoe/dhcpv4.h
index a7fa5a9..ebb67cf 100644
--- a/accel-pppd/ctrl/ipoe/dhcpv4.h
+++ b/accel-pppd/ctrl/ipoe/dhcpv4.h
@@ -32,6 +32,8 @@
 #define DHCPRELEASE  7
 #define DHCPINFORM   8
 
+#define ACCEL_PPP_MAGIC 0xfd56b60a
+
 struct dhcpv4_hdr {
 	uint8_t op;
 	uint8_t htype;
@@ -117,6 +119,8 @@ int dhcpv4_relay_send_release(struct dhcpv4_relay *relay, uint8_t *chaddr, uint3
 int dhcpv4_send_reply(int msg_type, struct dhcpv4_serv *serv, struct dhcpv4_packet *req, uint32_t yiaddr, uint32_t siaddr, uint32_t router, uint32_t mask, int lease_time, int renew_time, struct dhcpv4_packet *relay_reply);
 int dhcpv4_send_nak(struct dhcpv4_serv *serv, struct dhcpv4_packet *req);
 
+void dhcpv4_send_notify(struct dhcpv4_serv *serv, struct dhcpv4_packet *req, unsigned int weight);
+
 void dhcpv4_packet_ref(struct dhcpv4_packet *pack);
 struct dhcpv4_option *dhcpv4_packet_find_opt(struct dhcpv4_packet *pack, int type);
 int dhcpv4_packet_insert_opt82(struct dhcpv4_packet *pack, const char *agent_circuit_id, const char *agent_remote_id);
diff --git a/accel-pppd/ctrl/ipoe/ipoe.c b/accel-pppd/ctrl/ipoe/ipoe.c
index 5db6657..16cc65f 100644
--- a/accel-pppd/ctrl/ipoe/ipoe.c
+++ b/accel-pppd/ctrl/ipoe/ipoe.c
@@ -145,6 +145,7 @@ static int conf_vlan_timeout;
 static int conf_max_request = 3;
 static int conf_session_timeout;
 static int conf_idle_timeout;
+static int conf_weight;
 
 static const char *conf_relay;
 
@@ -212,6 +213,7 @@ static int ipoe_rad_send_acct_request(struct rad_plugin_t *rad, struct rad_packe
 static void ipoe_session_create_auto(struct ipoe_serv *serv);
 static void ipoe_serv_timeout(struct triton_timer_t *t);
 static struct ipoe_session *ipoe_session_create_up(struct ipoe_serv *serv, struct ethhdr *eth, struct iphdr *iph, struct _arphdr *arph);
+static void __terminate(struct ap_session *ses);
 
 static void ipoe_ctx_switch(struct triton_context_t *ctx, void *arg)
 {
@@ -1187,6 +1189,7 @@ static void ipoe_session_finished(struct ap_session *s)
 
 	pthread_mutex_lock(&ses->serv->lock);
 	list_del(&ses->entry);
+	ses->serv->sess_cnt--;
 	if  ((ses->serv->vlan_mon || ses->serv->need_close) && list_empty(&ses->serv->sessions))
 		triton_context_call(&ses->serv->ctx, (triton_event_func)ipoe_serv_release, ses->serv);
 	pthread_mutex_unlock(&ses->serv->lock);
@@ -1323,6 +1326,7 @@ static struct ipoe_session *ipoe_session_create_dhcpv4(struct ipoe_serv *serv, s
 
 	//pthread_mutex_lock(&serv->lock);
 	list_add_tail(&ses->entry, &serv->sessions);
+	serv->sess_cnt++;
 	//pthread_mutex_unlock(&serv->lock);
 
 	if (serv->timer.tpd)
@@ -1672,11 +1676,42 @@ static void mac_change_detected(struct dhcpv4_packet *pack)
 	ap_session_terminate(&ses->ses, TERM_USER_REQUEST, 1);
 }
 
+static int check_notify(struct ipoe_serv *serv, struct dhcpv4_packet *pack)
+{
+	struct dhcpv4_option *opt = dhcpv4_packet_find_opt(pack, 43);
+	struct ipoe_session *ses;
+	unsigned int w;
+
+	if (!opt)
+		return 0;
+
+	if (opt->len != 8)
+		return 0;
+
+	if (*(uint32_t *)opt->data != htonl(ACCEL_PPP_MAGIC))
+		return 0;
+
+	w = htonl(*(uint32_t *)(opt->data + 4));
+
+	list_for_each_entry(ses, &serv->sessions, entry) {
+		if (ses->xid == pack->hdr->xid) {
+			if (w < ses->weight || ses->weight == 0) {
+				log_debug("ipoe: terminate %s by weight %u (%u)\n", ses->ses.ifname, w, ses->weight);
+				triton_context_call(&ses->ctx, (triton_event_func)__terminate, &ses->ses);
+			}
+			break;
+		}
+	}
+
+	return 1;
+}
+
 static void __ipoe_recv_dhcpv4(struct dhcpv4_serv *dhcpv4, struct dhcpv4_packet *pack, int force)
 {
 	struct ipoe_serv *serv = container_of(dhcpv4->ctx, typeof(*serv), ctx);
 	struct ipoe_session *ses, *opt82_ses;
 	int offer_delay;
+	unsigned int weight = 0;
 	//struct dhcpv4_packet *reply;
 
 	if (serv->timer.tpd)
@@ -1687,6 +1722,9 @@ static void __ipoe_recv_dhcpv4(struct dhcpv4_serv *dhcpv4, struct dhcpv4_packet
 
 	pthread_mutex_lock(&serv->lock);
 	if (pack->msg_type == DHCPDISCOVER) {
+		if (check_notify(serv, pack))
+			goto out;
+
 		ses = ipoe_session_lookup(serv, pack, &opt82_ses);
 		if (!ses) {
 			if (serv->opt_shared == 0)
@@ -1709,6 +1747,8 @@ static void __ipoe_recv_dhcpv4(struct dhcpv4_serv *dhcpv4, struct dhcpv4_packet
 			}
 
 			ses = ipoe_session_create_dhcpv4(serv, pack);
+
+			ses->weight = weight = serv->opt_weight >= 0 ? serv->sess_cnt * serv->opt_weight : (stat_active + 1) * conf_weight;
 		}	else {
 			if (ses->terminate) {
 				triton_context_call(ses->ctrl.ctx, (triton_event_func)ipoe_session_terminated, ses);
@@ -1779,6 +1819,9 @@ static void __ipoe_recv_dhcpv4(struct dhcpv4_serv *dhcpv4, struct dhcpv4_packet
 
 out:
 	pthread_mutex_unlock(&serv->lock);
+
+	if (weight)
+		dhcpv4_send_notify(serv->dhcpv4, pack, weight);
 }
 
 static void ipoe_recv_dhcpv4(struct dhcpv4_serv *dhcpv4, struct dhcpv4_packet *pack)
@@ -1946,6 +1989,7 @@ static struct ipoe_session *ipoe_session_create_up(struct ipoe_serv *serv, struc
 	triton_context_register(&ses->ctx, &ses->ses);
 
 	list_add_tail(&ses->entry, &serv->sessions);
+	serv->sess_cnt++;
 
 	if (serv->timer.tpd)
 		triton_timer_del(&serv->timer);
@@ -1990,6 +2034,7 @@ static void ipoe_session_create_auto(struct ipoe_serv *serv)
 	triton_context_register(&ses->ctx, &ses->ses);
 
 	list_add_tail(&ses->entry, &serv->sessions);
+	serv->sess_cnt++;
 
 	triton_context_call(&ses->ctx, (triton_event_func)ipoe_session_start, ses);
 
@@ -2675,6 +2720,7 @@ static void add_interface(const char *ifname, int ifindex, const char *opt, int
 	int opt_ipv6 = conf_ipv6;
 	int opt_auto = conf_auto;
 	int opt_mtu = 0;
+	int opt_weight = -1;
 #ifdef USE_LUA
 	char *opt_lua_username_func = NULL;
 #endif
@@ -2744,6 +2790,8 @@ static void add_interface(const char *ifname, int ifindex, const char *opt, int
 				opt_ipv6 = atoi(ptr1);
 			} else if (strcmp(str, "mtu") == 0) {
 				opt_mtu = atoi(ptr1);
+			} else if (strcmp(str, "weight") == 0) {
+				opt_weight = atoi(ptr1);
 			} else if (strcmp(str, "username") == 0) {
 				if (strcmp(ptr1, "ifname") == 0)
 					opt_username = USERNAME_IFNAME;
@@ -2849,6 +2897,7 @@ static void add_interface(const char *ifname, int ifindex, const char *opt, int
 		serv->opt_arp = opt_arp;
 		serv->opt_username = opt_username;
 		serv->opt_ipv6 = opt_ipv6;
+		serv->opt_weight = opt_weight;
 #ifdef USE_LUA
 		if (serv->opt_lua_username_func && (!opt_lua_username_func || strcmp(serv->opt_lua_username_func, opt_lua_username_func))) {
 			_free(serv->opt_lua_username_func);
@@ -2934,6 +2983,7 @@ static void add_interface(const char *ifname, int ifindex, const char *opt, int
 	serv->opt_username = opt_username;
 	serv->opt_ipv6 = opt_ipv6;
 	serv->opt_mtu = opt_mtu;
+	serv->opt_weight = opt_weight;
 #ifdef USE_LUA
 	serv->opt_lua_username_func = opt_lua_username_func;
 #endif
@@ -3756,6 +3806,12 @@ static void load_config(void)
 	} else
 		conf_calling_sid = SID_MAC;
 
+	opt = conf_get_opt("ipoe", "weight");
+	if (opt)
+		conf_weight = atoi(opt);
+	else
+		conf_weight = 0;
+
 #ifdef RADIUS
 	if (triton_module_loaded("radius"))
 		load_radius_attrs();
diff --git a/accel-pppd/ctrl/ipoe/ipoe.h b/accel-pppd/ctrl/ipoe/ipoe.h
index 2b04ea0..fc774f6 100644
--- a/accel-pppd/ctrl/ipoe/ipoe.h
+++ b/accel-pppd/ctrl/ipoe/ipoe.h
@@ -37,6 +37,7 @@ struct ipoe_serv {
 	int ifindex;
 	uint8_t hwaddr[ETH_ALEN];
 	struct list_head sessions;
+	unsigned int sess_cnt;
 	struct dhcpv4_serv *dhcpv4;
 	struct dhcpv4_relay *dhcpv4_relay;
 	void *arp;
@@ -57,6 +58,7 @@ struct ipoe_serv {
 #ifdef USE_LUA
 	char *opt_lua_username_func;
 #endif
+	int opt_weight;
 	int opt_shared:1;
 	int opt_dhcpv4:1;
 	int opt_up:1;
@@ -102,6 +104,7 @@ struct ipoe_session {
 	int ifindex;
 	char *username;
 	struct ipv4db_item_t ipv4;
+	unsigned int weight;
 #ifdef RADIUS
 	struct rad_plugin_t radius;
 #endif