Merge pull request #6950 from opensourcerouting/bfd-distributed-v3

bfdd: distributed BFD

8 files changed, 1924 insertions, 26 deletions

diff --git a/bfdd/bfd.c b/bfdd/bfd.c
index c77b5cd1a..f7ce0ece3 100644
--- a/bfdd/bfd.c
+++ b/bfdd/bfd.c
@@ -216,6 +216,9 @@ void bfd_session_apply(struct bfd_session *bs)
 	    && (bs->timers.desired_min_tx != min_tx
 		|| bs->timers.required_min_rx != min_rx))
 		bfd_set_polling(bs);
+
+	/* Send updated information to data plane. */
+	bfd_dplane_update_session(bs);
 }
 
 void bfd_profile_remove(struct bfd_session *bs)
@@ -293,6 +296,10 @@ int bfd_session_enable(struct bfd_session *bs)
 	struct vrf *vrf = NULL;
 	int psock;
 
+	/* We are using data plane, we don't need software. */
+	if (bs->bdc)
+		return 0;
+
 	/*
 	 * If the interface or VRF doesn't exist, then we must register
 	 * the session but delay its start.
@@ -332,9 +339,14 @@ int bfd_session_enable(struct bfd_session *bs)
 		bs->vrf = vrf_lookup_by_id(VRF_DEFAULT);
 	assert(bs->vrf);
 
-	if (bs->key.ifname[0]
-	    && CHECK_FLAG(bs->flags, BFD_SESS_FLAG_MH) == 0)
-		bs->ifp = ifp;
+	/* Assign interface pointer (if any). */
+	bs->ifp = ifp;
+
+	/* Attempt to use data plane. */
+	if (bglobal.bg_use_dplane && bfd_dplane_add_session(bs) == 0) {
+		control_notify_config(BCM_NOTIFY_CONFIG_ADD, bs);
+		return 0;
+	}
 
 	/* Sanity check: don't leak open sockets. */
 	if (bs->sock != -1) {
@@ -383,6 +395,10 @@ int bfd_session_enable(struct bfd_session *bs)
  */
 void bfd_session_disable(struct bfd_session *bs)
 {
+	/* We are using data plane, we don't need software. */
+	if (bs->bdc)
+		return;
+
 	/* Free up socket resources. */
 	if (bs->sock != -1) {
 		close(bs->sock);
@@ -393,8 +409,6 @@ void bfd_session_disable(struct bfd_session *bs)
 	bfd_recvtimer_delete(bs);
 	bfd_xmttimer_delete(bs);
 	ptm_bfd_echo_stop(bs);
-	bs->vrf = NULL;
-	bs->ifp = NULL;
 
 	/* Set session down so it doesn't report UP and disabled. */
 	ptm_bfd_sess_dn(bs, BD_PATH_DOWN);
@@ -804,6 +818,9 @@ void bfd_session_free(struct bfd_session *bs)
 
 	bfd_session_disable(bs);
 
+	/* Remove session from data plane if any. */
+	bfd_dplane_delete_session(bs);
+
 	bfd_key_delete(bs->key);
 	bfd_id_delete(bs->discrs.my_discr);
 
@@ -1267,14 +1284,18 @@ void bfd_set_echo(struct bfd_session *bs, bool echo)
 		SET_FLAG(bs->flags, BFD_SESS_FLAG_ECHO);
 
 		/* Activate/update echo receive timeout timer. */
-		bs_echo_timer_handler(bs);
+		if (bs->bdc == NULL)
+			bs_echo_timer_handler(bs);
 	} else {
 		/* Check if echo mode is already disabled. */
 		if (!CHECK_FLAG(bs->flags, BFD_SESS_FLAG_ECHO))
 			return;
 
 		UNSET_FLAG(bs->flags, BFD_SESS_FLAG_ECHO);
-		ptm_bfd_echo_stop(bs);
+
+		/* Deactivate timeout timer. */
+		if (bs->bdc == NULL)
+			ptm_bfd_echo_stop(bs);
 	}
 }
 
@@ -1299,6 +1320,14 @@ void bfd_set_shutdown(struct bfd_session *bs, bool shutdown)
 
 		SET_FLAG(bs->flags, BFD_SESS_FLAG_SHUTDOWN);
 
+		/* Handle data plane shutdown case. */
+		if (bs->bdc) {
+			bs->ses_state = PTM_BFD_ADM_DOWN;
+			bfd_dplane_update_session(bs);
+			control_notify(bs, bs->ses_state);
+			return;
+		}
+
 		/* Disable all events. */
 		bfd_recvtimer_delete(bs);
 		bfd_echo_recvtimer_delete(bs);
@@ -1319,6 +1348,14 @@ void bfd_set_shutdown(struct bfd_session *bs, bool shutdown)
 
 		UNSET_FLAG(bs->flags, BFD_SESS_FLAG_SHUTDOWN);
 
+		/* Handle data plane shutdown case. */
+		if (bs->bdc) {
+			bs->ses_state = PTM_BFD_DOWN;
+			bfd_dplane_update_session(bs);
+			control_notify(bs, bs->ses_state);
+			return;
+		}
+
 		/* Change and notify state change. */
 		bs->ses_state = PTM_BFD_DOWN;
 		control_notify(bs, bs->ses_state);
@@ -2028,6 +2065,16 @@ static int bfd_vrf_enable(struct vrf *vrf)
 		bvrf = XCALLOC(MTYPE_BFDD_VRF, sizeof(struct bfd_vrf_global));
 		bvrf->vrf = vrf;
 		vrf->info = (void *)bvrf;
+
+		/* Disable sockets if using data plane. */
+		if (bglobal.bg_use_dplane) {
+			bvrf->bg_shop = -1;
+			bvrf->bg_mhop = -1;
+			bvrf->bg_shop6 = -1;
+			bvrf->bg_mhop6 = -1;
+			bvrf->bg_echo = -1;
+			bvrf->bg_echov6 = -1;
+		}
 	} else
 		bvrf = vrf->info;
 
@@ -2049,25 +2096,24 @@ static int bfd_vrf_enable(struct vrf *vrf)
 		if (!bvrf->bg_echov6)
 			bvrf->bg_echov6 = bp_echov6_socket(vrf);
 
-		/* Add descriptors to the event loop. */
-		if (!bvrf->bg_ev[0])
-			thread_add_read(master, bfd_recv_cb, bvrf, bvrf->bg_shop,
-					&bvrf->bg_ev[0]);
-		if (!bvrf->bg_ev[1])
-			thread_add_read(master, bfd_recv_cb, bvrf, bvrf->bg_mhop,
-					&bvrf->bg_ev[1]);
+		if (!bvrf->bg_ev[0] && bvrf->bg_shop != -1)
+			thread_add_read(master, bfd_recv_cb, bvrf,
+					bvrf->bg_shop, &bvrf->bg_ev[0]);
+		if (!bvrf->bg_ev[1] && bvrf->bg_mhop != -1)
+			thread_add_read(master, bfd_recv_cb, bvrf,
+					bvrf->bg_mhop, &bvrf->bg_ev[1]);
 		if (!bvrf->bg_ev[2] && bvrf->bg_shop6 != -1)
-			thread_add_read(master, bfd_recv_cb, bvrf, bvrf->bg_shop6,
-					&bvrf->bg_ev[2]);
+			thread_add_read(master, bfd_recv_cb, bvrf,
+					bvrf->bg_shop6, &bvrf->bg_ev[2]);
 		if (!bvrf->bg_ev[3] && bvrf->bg_mhop6 != -1)
-			thread_add_read(master, bfd_recv_cb, bvrf, bvrf->bg_mhop6,
-					&bvrf->bg_ev[3]);
-		if (!bvrf->bg_ev[4])
-			thread_add_read(master, bfd_recv_cb, bvrf, bvrf->bg_echo,
-					&bvrf->bg_ev[4]);
+			thread_add_read(master, bfd_recv_cb, bvrf,
+					bvrf->bg_mhop6, &bvrf->bg_ev[3]);
+		if (!bvrf->bg_ev[4] && bvrf->bg_echo != -1)
+			thread_add_read(master, bfd_recv_cb, bvrf,
+					bvrf->bg_echo, &bvrf->bg_ev[4]);
 		if (!bvrf->bg_ev[5] && bvrf->bg_echov6 != -1)
-			thread_add_read(master, bfd_recv_cb, bvrf, bvrf->bg_echov6,
-					&bvrf->bg_ev[5]);
+			thread_add_read(master, bfd_recv_cb, bvrf,
+					bvrf->bg_echov6, &bvrf->bg_ev[5]);
 	}
 	if (vrf->vrf_id != VRF_DEFAULT) {
 		bfdd_zclient_register(vrf->vrf_id);
diff --git a/bfdd/bfd.h b/bfdd/bfd.h
index af3f92d6a..7c537b40d 100644
--- a/bfdd/bfd.h
+++ b/bfdd/bfd.h
@@ -269,6 +269,7 @@ struct bfd_session {
 	struct bfd_key key;
 	struct peer_label *pl;
 
+	struct bfd_dplane_ctx *bdc;
 	struct sockaddr_any local_address;
 	struct interface *ifp;
 	struct vrf *vrf;
@@ -424,6 +425,10 @@ struct bfd_vrf_global {
 	struct thread *bg_ev[6];
 };
 
+/* Forward declaration of data plane context struct. */
+struct bfd_dplane_ctx;
+TAILQ_HEAD(dplane_queue, bfd_dplane_ctx);
+
 struct bfd_global {
 	int bg_csock;
 	struct thread *bg_csockev;
@@ -441,7 +446,15 @@ struct bfd_global {
 	 */
 	bool bg_shutdown;
 
+	/* Distributed BFD items. */
+	bool bg_use_dplane;
+	int bg_dplane_sock;
+	struct thread *bg_dplane_sockev;
+	struct dplane_queue bg_dplaneq;
+
 	/* Debug options. */
+	/* Show distributed BFD debug messages. */
+	bool debug_dplane;
 	/* Show all peer state changes events. */
 	bool debug_peer_event;
 	/*
@@ -742,4 +755,56 @@ void bfd_session_update_vrf_name(struct bfd_session *bs, struct vrf *vrf);
 
 int ptm_bfd_notify(struct bfd_session *bs, uint8_t notify_state);
 
+/*
+ * dplane.c
+ */
+
+/**
+ * Initialize BFD data plane infrastructure for distributed BFD implementation.
+ *
+ * \param sa socket address.
+ * \param salen socket address structure length.
+ * \param client `true` means connecting socket, `false` listening socket.
+ */
+void bfd_dplane_init(const struct sockaddr *sa, socklen_t salen, bool client);
+
+/**
+ * Attempts to delegate the BFD session liveness detection to hardware.
+ *
+ * \param bs the BFD session data structure.
+ *
+ * \returns
+ * `0` on success and BFD daemon should do nothing or `-1` on failure
+ * and we should fallback to software implementation.
+ */
+int bfd_dplane_add_session(struct bfd_session *bs);
+
+/**
+ * Send new session settings to data plane.
+ *
+ * \param bs the BFD session to update.
+ */
+int bfd_dplane_update_session(const struct bfd_session *bs);
+
+/**
+ * Deletes session from data plane.
+ *
+ * \param bs the BFD session to delete.
+ *
+ * \returns `0` on success otherwise `-1`.
+ */
+int bfd_dplane_delete_session(struct bfd_session *bs);
+
+/**
+ * Asks the data plane for updated counters and update the session data
+ * structure.
+ *
+ * \param bs the BFD session that needs updating.
+ *
+ * \returns `0` on success otherwise `-1` on failure.
+ */
+int bfd_dplane_update_session_counters(struct bfd_session *bs);
+
+void bfd_dplane_show_counters(struct vty *vty);
+
 #endif /* _BFD_H_ */
diff --git a/bfdd/bfdd.c b/bfdd/bfdd.c
index 098e7a289..b8a059708 100644
--- a/bfdd/bfdd.c
+++ b/bfdd/bfdd.c
@@ -20,12 +20,20 @@
 
 #include <zebra.h>
 
+#include <arpa/inet.h>
+#include <netinet/in.h>
+#include <sys/socket.h>
+#include <sys/un.h>
+
+#include <err.h>
+
 #include "filter.h"
 #include "if.h"
 #include "vrf.h"
 
 #include "bfd.h"
 #include "bfdd_nb.h"
+#include "bfddp_packet.h"
 #include "lib/version.h"
 #include "lib/command.h"
 
@@ -129,8 +137,10 @@ FRR_DAEMON_INFO(bfdd, BFD, .vty_port = 2617,
 		.n_yang_modules = array_size(bfdd_yang_modules))
 
 #define OPTION_CTLSOCK 1001
+#define OPTION_DPLANEADDR 2000
 static const struct option longopts[] = {
 	{"bfdctl", required_argument, NULL, OPTION_CTLSOCK},
+	{"dplaneaddr", required_argument, NULL, OPTION_DPLANEADDR},
 	{0}
 };
 
@@ -160,6 +170,143 @@ const struct bfd_state_str_list state_list[] = {
 	{.str = NULL},
 };
 
+static uint16_t
+parse_port(const char *str)
+{
+	char *nulbyte;
+	long rv;
+
+	errno = 0;
+	rv = strtol(str, &nulbyte, 10);
+	/* No conversion performed. */
+	if (rv == 0 && errno == EINVAL) {
+		fprintf(stderr, "invalid BFD data plane address port: %s\n",
+			str);
+		exit(0);
+	}
+	/* Invalid number range. */
+	if ((rv <= 0 || rv >= 65535) || errno == ERANGE) {
+		fprintf(stderr, "invalid BFD data plane port range: %s\n",
+			str);
+		exit(0);
+	}
+	/* There was garbage at the end of the string. */
+	if (*nulbyte != 0) {
+		fprintf(stderr, "invalid BFD data plane port: %s\n",
+			str);
+		exit(0);
+	}
+
+	return (uint16_t)rv;
+}
+
+static void
+distributed_bfd_init(const char *arg)
+{
+	char *sptr, *saux;
+	bool is_client = false;
+	size_t slen;
+	socklen_t salen;
+	char addr[64];
+	char type[64];
+	union {
+		struct sockaddr_in sin;
+		struct sockaddr_in6 sin6;
+		struct sockaddr_un sun;
+	} sa;
+
+	/* Basic parsing: find ':' to figure out type part and address part. */
+	sptr = strchr(arg, ':');
+	if (sptr == NULL) {
+		fprintf(stderr, "invalid BFD data plane socket: %s\n", arg);
+		exit(1);
+	}
+
+	/* Calculate type string length. */
+	slen = (size_t)(sptr - arg);
+
+	/* Copy the address part. */
+	sptr++;
+	strlcpy(addr, sptr, sizeof(addr));
+
+	/* Copy type part. */
+	strlcpy(type, arg, slen + 1);
+
+	/* Reset address data. */
+	memset(&sa, 0, sizeof(sa));
+
+	/* Fill the address information. */
+	if (strcmp(type, "unix") == 0 || strcmp(type, "unixc") == 0) {
+		if (strcmp(type, "unixc") == 0)
+			is_client = true;
+
+		salen = sizeof(sa.sun);
+		sa.sun.sun_family = AF_UNIX;
+		strlcpy(sa.sun.sun_path, addr, sizeof(sa.sun.sun_path));
+	} else if (strcmp(type, "ipv4") == 0 || strcmp(type, "ipv4c") == 0) {
+		if (strcmp(type, "ipv4c") == 0)
+			is_client = true;
+
+		salen = sizeof(sa.sin);
+		sa.sin.sin_family = AF_INET;
+
+		/* Parse port if any. */
+		sptr = strchr(addr, ':');
+		if (sptr == NULL) {
+			sa.sin.sin_port = htons(BFD_DATA_PLANE_DEFAULT_PORT);
+		} else {
+			*sptr = 0;
+			sa.sin.sin_port = htons(parse_port(sptr + 1));
+		}
+
+		if (inet_pton(AF_INET, addr, &sa.sin.sin_addr) != 1)
+			errx(1, "%s: inet_pton: invalid address %s", __func__,
+			     addr);
+	} else if (strcmp(type, "ipv6") == 0 || strcmp(type, "ipv6c") == 0) {
+		if (strcmp(type, "ipv6c") == 0)
+			is_client = true;
+
+		salen = sizeof(sa.sin6);
+		sa.sin6.sin6_family = AF_INET6;
+
+		/* Check for IPv6 enclosures '[]' */
+		sptr = &addr[0];
+		if (*sptr != '[')
+			errx(1, "%s: invalid IPv6 address format: %s", __func__,
+			     addr);
+
+		saux = strrchr(addr, ']');
+		if (saux == NULL)
+			errx(1, "%s: invalid IPv6 address format: %s", __func__,
+			     addr);
+
+		/* Consume the '[]:' part. */
+		slen = saux - sptr;
+		memmove(addr, addr + 1, slen);
+		addr[slen - 1] = 0;
+
+		/* Parse port if any. */
+		saux++;
+		sptr = strrchr(saux, ':');
+		if (sptr == NULL) {
+			sa.sin6.sin6_port = htons(BFD_DATA_PLANE_DEFAULT_PORT);
+		} else {
+			*sptr = 0;
+			sa.sin6.sin6_port = htons(parse_port(sptr + 1));
+		}
+
+		if (inet_pton(AF_INET6, addr, &sa.sin6.sin6_addr) != 1)
+			errx(1, "%s: inet_pton: invalid address %s", __func__,
+			     addr);
+	} else {
+		fprintf(stderr, "invalid BFD data plane socket type: %s\n",
+			type);
+		exit(1);
+	}
+
+	/* Initialize BFD data plane listening socket. */
+	bfd_dplane_init((struct sockaddr *)&sa, salen, is_client);
+}
 
 static void bg_init(void)
 {
@@ -185,7 +332,7 @@ static void bg_init(void)
 
 int main(int argc, char *argv[])
 {
-	char ctl_path[512];
+	char ctl_path[512], dplane_addr[512];
 	bool ctlsockused = false;
 	int opt;
 
@@ -194,7 +341,8 @@ int main(int argc, char *argv[])
 
 	frr_preinit(&bfdd_di, argc, argv);
 	frr_opt_add("", longopts,
-		    "      --bfdctl       Specify bfdd control socket\n");
+		    "      --bfdctl       Specify bfdd control socket\n"
+		    "      --dplaneaddr   Specify BFD data plane address\n");
 
 	snprintf(ctl_path, sizeof(ctl_path), BFDD_CONTROL_SOCKET,
 		 "", "");
@@ -208,6 +356,10 @@ int main(int argc, char *argv[])
 			strlcpy(ctl_path, optarg, sizeof(ctl_path));
 			ctlsockused = true;
 			break;
+		case OPTION_DPLANEADDR:
+			strlcpy(dplane_addr, optarg, sizeof(dplane_addr));
+			bglobal.bg_use_dplane = true;
+			break;
 
 		default:
 			frr_help_exit(1);
@@ -248,6 +400,10 @@ int main(int argc, char *argv[])
 	/* read configuration file and daemonize  */
 	frr_config_fork();
 
+	/* Initialize BFD data plane listening socket. */
+	if (bglobal.bg_use_dplane)
+		distributed_bfd_init(dplane_addr);
+
 	frr_run(master);
 	/* NOTREACHED */
 
diff --git a/bfdd/bfdd_vty.c b/bfdd/bfdd_vty.c
index 837a7b7d7..53e23cf6c 100644
--- a/bfdd/bfdd_vty.c
+++ b/bfdd/bfdd_vty.c
@@ -348,6 +348,11 @@ static void _display_peer_counter(struct vty *vty, struct bfd_session *bs)
 {
 	_display_peer_header(vty, bs);
 
+	/* Ask data plane for updated counters. */
+	if (bfd_dplane_update_session_counters(bs) == -1)
+		zlog_debug("%s: failed to update BFD session counters (%s)",
+			   __func__, bs_to_string(bs));
+
 	vty_out(vty, "\t\tControl packet input: %" PRIu64 " packets\n",
 		bs->stats.rx_ctrl_pkt);
 	vty_out(vty, "\t\tControl packet output: %" PRIu64 " packets\n",
@@ -369,6 +374,11 @@ static struct json_object *__display_peer_counters_json(struct bfd_session *bs)
 {
 	struct json_object *jo = _peer_json_header(bs);
 
+	/* Ask data plane for updated counters. */
+	if (bfd_dplane_update_session_counters(bs) == -1)
+		zlog_debug("%s: failed to update BFD session counters (%s)",
+			   __func__, bs_to_string(bs));
+
 	json_object_int_add(jo, "control-packet-input", bs->stats.rx_ctrl_pkt);
 	json_object_int_add(jo, "control-packet-output", bs->stats.tx_ctrl_pkt);
 	json_object_int_add(jo, "echo-packet-input", bs->stats.rx_echo_pkt);
@@ -748,6 +758,28 @@ DEFPY(bfd_show_peers_brief, bfd_show_peers_brief_cmd,
 	return CMD_SUCCESS;
 }
 
+DEFPY(show_bfd_distributed, show_bfd_distributed_cmd,
+      "show bfd distributed",
+      SHOW_STR
+      "Bidirection Forwarding Detection\n"
+      "Show BFD data plane (distributed BFD) statistics\n")
+{
+	bfd_dplane_show_counters(vty);
+	return CMD_SUCCESS;
+}
+
+DEFPY(
+	bfd_debug_distributed, bfd_debug_distributed_cmd,
+	"[no] debug bfd distributed",
+	NO_STR
+	DEBUG_STR
+	"Bidirection Forwarding Detection\n"
+	"BFD data plane (distributed BFD) debugging\n")
+{
+	bglobal.debug_dplane = !no;
+	return CMD_SUCCESS;
+}
+
 DEFPY(
 	bfd_debug_peer, bfd_debug_peer_cmd,
 	"[no] debug bfd peer",
@@ -888,6 +920,8 @@ DEFUN_NOSH(show_debugging_bfd,
 	   "BFD daemon\n")
 {
 	vty_out(vty, "BFD debugging status:\n");
+	if (bglobal.debug_dplane)
+		vty_out(vty, "  Distributed BFD debugging is on.\n");
 	if (bglobal.debug_peer_event)
 		vty_out(vty, "  Peer events debugging is on.\n");
 	if (bglobal.debug_zebra)
@@ -919,6 +953,11 @@ static int bfdd_write_config(struct vty *vty)
 	struct lyd_node *dnode;
 	int written = 0;
 
+	if (bglobal.debug_dplane) {
+		vty_out(vty, "debug bfd distributed\n");
+		written = 1;
+	}
+
 	if (bglobal.debug_peer_event) {
 		vty_out(vty, "debug bfd peer\n");
 		written = 1;
@@ -951,12 +990,15 @@ void bfdd_vty_init(void)
 	install_element(ENABLE_NODE, &bfd_show_peers_cmd);
 	install_element(ENABLE_NODE, &bfd_show_peer_cmd);
 	install_element(ENABLE_NODE, &bfd_show_peers_brief_cmd);
+	install_element(ENABLE_NODE, &show_bfd_distributed_cmd);
 	install_element(ENABLE_NODE, &show_debugging_bfd_cmd);
 
+	install_element(ENABLE_NODE, &bfd_debug_distributed_cmd);
 	install_element(ENABLE_NODE, &bfd_debug_peer_cmd);
 	install_element(ENABLE_NODE, &bfd_debug_zebra_cmd);
 	install_element(ENABLE_NODE, &bfd_debug_network_cmd);
 
+	install_element(CONFIG_NODE, &bfd_debug_distributed_cmd);
 	install_element(CONFIG_NODE, &bfd_debug_peer_cmd);
 	install_element(CONFIG_NODE, &bfd_debug_zebra_cmd);
 	install_element(CONFIG_NODE, &bfd_debug_network_cmd);
diff --git a/bfdd/bfddp_packet.h b/bfdd/bfddp_packet.h
new file mode 100644
index 000000000..8865baef6
--- /dev/null
+++ b/bfdd/bfddp_packet.h
@@ -0,0 +1,381 @@
+/*
+ * BFD Data Plane protocol messages header.
+ *
+ * Copyright (C) 2020 Network Device Education Foundation, Inc. ("NetDEF")
+ *                    Rafael F. Zalamena
+ *
+ * Permission is hereby granted, free of charge, to any person obtaining a copy
+ * of this software and associated documentation files (the “Software”), to
+ * deal in the Software without restriction, including without limitation the
+ * rights to use, copy, modify, merge, publish, distribute, sublicense, and/or
+ * sell copies of the Software, and to permit persons to whom the Software is
+ * furnished to do so, subject to the following conditions:
+ *
+ * The above copyright notice and this permission notice shall be included in
+ * all copies or substantial portions of the Software.
+ *
+ * THE SOFTWARE IS PROVIDED “AS IS”, WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+ * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+ * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+ * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+ * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING
+ * FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS
+ * IN THE SOFTWARE.
+ */
+
+/**
+ * \file bfddp_packet.h
+ */
+#ifndef BFD_DP_PACKET_H
+#define BFD_DP_PACKET_H
+
+#include <netinet/in.h>
+
+#include <stdint.h>
+
+/*
+ * Protocol definitions.
+ */
+
+/**
+ * BFD protocol version as defined in RFC5880 Section 4.1 Generic BFD Control
+ * Packet Format.
+ */
+#define BFD_PROTOCOL_VERSION 1
+
+/** Default data plane port. */
+#define BFD_DATA_PLANE_DEFAULT_PORT 50700
+
+/** BFD single hop UDP port, as defined in RFC 5881 Section 4. Encapsulation. */
+#define BFD_SINGLE_HOP_PORT 3784
+
+/** BFD multi hop UDP port, as defined in RFC 5883 Section 5. Encapsulation. */
+#define BFD_MULTI_HOP_PORT 4784
+
+/** Default slow start multiplier. */
+#define SLOWSTART_DMULT 3
+/** Default slow start transmission speed. */
+#define SLOWSTART_TX 1000000u
+/** Default slow start receive speed. */
+#define SLOWSTART_RX 1000000u
+/** Default slow start echo receive speed. */
+#define SLOWSTART_ERX 0u
+
+/*
+ * BFD single hop source UDP ports. As defined in RFC 5881 Section 4.
+ * Encapsulation.
+ */
+#define BFD_SOURCE_PORT_BEGIN 49152
+#define BFD_SOURCE_PORT_END 65535
+
+/** BFD data plane protocol version. */
+#define BFD_DP_VERSION 1
+
+/** BFD data plane message types. */
+enum bfddp_message_type {
+	/** Ask for BFD daemon or data plane for echo packet. */
+	ECHO_REQUEST = 0,
+	/** Answer a ECHO_REQUEST packet. */
+	ECHO_REPLY = 1,
+	/** Add or update BFD peer session. */
+	DP_ADD_SESSION = 2,
+	/** Delete BFD peer session. */
+	DP_DELETE_SESSION = 3,
+	/** Tell BFD daemon state changed: timer expired or session down. */
+	BFD_STATE_CHANGE = 4,
+
+	/** Ask for BFD session counters. */
+	DP_REQUEST_SESSION_COUNTERS = 5,
+	/** Tell BFD daemon about counters values. */
+	BFD_SESSION_COUNTERS = 6,
+};
+
+/**
+ * `ECHO_REQUEST`/`ECHO_REPLY` data payload.
+ *
+ * Data plane might use whatever precision it wants for `dp_time`
+ * field, however if you want to be able to tell the delay between
+ * data plane packet send and BFD daemon packet processing you should
+ * use `gettimeofday()` and have the data plane clock synchronized with
+ * BFD daemon (not a problem if data plane runs in the same system).
+ *
+ * Normally data plane will only check the time stamp it sent to determine
+ * the whole packet trip time.
+ */
+struct bfddp_echo {
+	/** Filled by data plane. */
+	uint64_t dp_time;
+	/** Filled by BFD daemon. */
+	uint64_t bfdd_time;
+};
+
+
+/** BFD session flags. */
+enum bfddp_session_flag {
+	/** Set when using multi hop. */
+	SESSION_MULTIHOP = (1 << 0),
+	/** Set when using demand mode. */
+	SESSION_DEMAND = (1 << 1),
+	/** Set when using cbit (Control Plane Independent). */
+	SESSION_CBIT = (1 << 2),
+	/** Set when using echo mode. */
+	SESSION_ECHO = (1 << 3),
+	/** Set when using IPv6. */
+	SESSION_IPV6 = (1 << 4),
+	/** Set when using passive mode. */
+	SESSION_PASSIVE = (1 << 5),
+	/** Set when session is administrative down. */
+	SESSION_SHUTDOWN = (1 << 6),
+};
+
+/**
+ * `DP_ADD_SESSION`/`DP_DELETE_SESSION` data payload.
+ *
+ * `lid` is unique in BFD daemon so it might be used as key for data
+ * structures lookup.
+ */
+struct bfddp_session {
+	/** Important session flags. \see bfddp_session_flag. */
+	uint32_t flags;
+	/**
+	 * Session source address.
+	 *
+	 * Check `flags` field for `SESSION_IPV6` before using as IPv6.
+	 */
+	struct in6_addr src;
+	/**
+	 * Session destination address.
+	 *
+	 * Check `flags` field for `SESSION_IPV6` before using as IPv6.
+	 */
+	struct in6_addr dst;
+
+	/** Local discriminator. */
+	uint32_t lid;
+	/**
+	 * Minimum desired transmission interval (in microseconds) without
+	 * jitter.
+	 */
+	uint32_t min_tx;
+	/**
+	 * Required minimum receive interval rate (in microseconds) without
+	 * jitter.
+	 */
+	uint32_t min_rx;
+	/**
+	 * Required minimum echo receive interval rate (in microseconds)
+	 * without jitter.
+	 */
+	uint32_t min_echo_rx;
+	/** Amount of milliseconds to wait before starting the session */
+	uint32_t hold_time;
+
+	/** Minimum TTL. */
+	uint8_t ttl;
+	/** Detection multiplier. */
+	uint8_t detect_mult;
+	/** Reserved / zeroed. */
+	uint16_t zero;
+
+	/** Interface index (set to `0` when unavailable). */
+	uint32_t ifindex;
+	/** Interface name (empty when unavailable). */
+	char ifname[64];
+
+	/* TODO: missing authentication. */
+};
+
+/** BFD packet state values as defined in RFC 5880, Section 4.1. */
+enum bfd_state_value {
+	/** Session is administratively down. */
+	STATE_ADMINDOWN = 0,
+	/** Session is down or went down. */
+	STATE_DOWN = 1,
+	/** Session is initializing. */
+	STATE_INIT = 2,
+	/** Session is up. */
+	STATE_UP = 3,
+};
+
+/** BFD diagnostic field values as defined in RFC 5880, Section 4.1. */
+enum bfd_diagnostic_value {
+	/** Nothing was diagnosed. */
+	DIAG_NOTHING = 0,
+	/** Control detection time expired. */
+	DIAG_CONTROL_EXPIRED = 1,
+	/** Echo function failed. */
+	DIAG_ECHO_FAILED = 2,
+	/** Neighbor signaled down. */
+	DIAG_DOWN = 3,
+	/** Forwarding plane reset. */
+	DIAG_FP_RESET = 4,
+	/** Path down. */
+	DIAG_PATH_DOWN = 5,
+	/** Concatenated path down. */
+	DIAG_CONCAT_PATH_DOWN = 6,
+	/** Administratively down. */
+	DIAG_ADMIN_DOWN = 7,
+	/** Reverse concatenated path down. */
+	DIAG_REV_CONCAT_PATH_DOWN = 8,
+};
+
+/** BFD remote state flags. */
+enum bfd_remote_flags {
+	/** Control Plane Independent bit. */
+	RBIT_CPI = (1 << 0),
+	/** Demand mode bit. */
+	RBIT_DEMAND = (1 << 1),
+	/** Multipoint bit. */
+	RBIT_MP = (1 << 2),
+};
+
+/**
+ * `BFD_STATE_CHANGE` data payload.
+ */
+struct bfddp_state_change {
+	/** Local discriminator. */
+	uint32_t lid;
+	/** Remote discriminator. */
+	uint32_t rid;
+	/** Remote configurations/bits set. \see bfd_remote_flags. */
+	uint32_t remote_flags;
+	/** Remote minimum desired transmission interval. */
+	uint32_t desired_tx;
+	/** Remote minimum receive interval. */
+	uint32_t required_rx;
+	/** Remote minimum echo receive interval. */
+	uint32_t required_echo_rx;
+	/** Remote state. \see bfd_state_values.*/
+	uint8_t state;
+	/** Remote diagnostics (if any) */
+	uint8_t diagnostics;
+	/** Remote detection multiplier. */
+	uint8_t detection_multiplier;
+};
+
+/**
+ * BFD control packet state bits definition.
+ */
+enum bfddp_control_state_bits {
+	/** Used to request connection establishment signal. */
+	STATE_POLL_BIT = (1 << 5),
+	/** Finalizes the connection establishment signal. */
+	STATE_FINAL_BIT = (1 << 4),
+	/** Signalizes that forward plane doesn't depend on control plane. */
+	STATE_CPI_BIT = (1 << 3),
+	/** Signalizes the use of authentication. */
+	STATE_AUTH_BIT = (1 << 2),
+	/** Signalizes that peer is using demand mode. */
+	STATE_DEMAND_BIT = (1 << 1),
+	/** Used in RFC 8562 implementation. */
+	STATE_MULTI_BIT = (1 << 0),
+};
+
+/**
+ * BFD control packet.
+ *
+ * As defined in 'RFC 5880 Section 4.1 Generic BFD Control Packet Format'.
+ */
+struct bfddp_control_packet {
+	/** (3 bits version << 5) | (5 bits diag). */
+	uint8_t version_diag;
+	/**
+	 * (2 bits state << 6) | (6 bits flags)
+	 *
+	 * \see bfd_state_value, bfddp_control_state_bits.
+	 */
+	uint8_t state_bits;
+	/** Detection multiplier. */
+	uint8_t detection_multiplier;
+	/** Packet length in bytes. */
+	uint8_t length;
+	/** Our discriminator. */
+	uint32_t local_id;
+	/** Remote system discriminator. */
+	uint32_t remote_id;
+	/** Desired minimum send interval in microseconds. */
+	uint32_t desired_tx;
+	/** Desired minimum receive interval in microseconds. */
+	uint32_t required_rx;
+	/** Desired minimum echo receive interval in microseconds. */
+	uint32_t required_echo_rx;
+};
+
+/**
+ * The protocol wire message header structure.
+ */
+struct bfddp_message_header {
+	/** Protocol version format. \see BFD_DP_VERSION. */
+	uint8_t version;
+	/** Reserved / zero field. */
+	uint8_t zero;
+	/** Message contents type. \see bfddp_message_type. */
+	uint16_t type;
+	/**
+	 * Message identification (to pair request/response).
+	 *
+	 * The ID `0` is reserved for asynchronous messages (e.g. unrequested
+	 * messages).
+	 */
+	uint16_t id;
+	/** Message length. */
+	uint16_t length;
+};
+
+/**
+ * Data plane session counters request.
+ *
+ * Message type: `DP_REQUEST_SESSION_COUNTERS`.
+ */
+struct bfddp_request_counters {
+	/** Session local discriminator. */
+	uint32_t lid;
+};
+
+/**
+ * BFD session counters reply.
+ *
+ * Message type: `BFD_SESSION_COUNTERS`.
+ */
+struct bfddp_session_counters {
+	/** Session local discriminator. */
+	uint32_t lid;
+
+	/** Control packet bytes input. */
+	uint64_t control_input_bytes;
+	/** Control packets input. */
+	uint64_t control_input_packets;
+	/** Control packet bytes output. */
+	uint64_t control_output_bytes;
+	/** Control packets output. */
+	uint64_t control_output_packets;
+
+	/** Echo packet bytes input. */
+	uint64_t echo_input_bytes;
+	/** Echo packets input. */
+	uint64_t echo_input_packets;
+	/** Echo packet bytes output. */
+	uint64_t echo_output_bytes;
+	/** Echo packets output. */
+	uint64_t echo_output_packets;
+};
+
+/**
+ * The protocol wire messages structure.
+ */
+struct bfddp_message {
+	/** Message header. \see bfddp_message_header. */
+	struct bfddp_message_header header;
+
+	/** Message payload. \see bfddp_message_type. */
+	union {
+		struct bfddp_echo echo;
+		struct bfddp_session session;
+		struct bfddp_state_change state;
+		struct bfddp_control_packet control;
+		struct bfddp_request_counters counters_req;
+		struct bfddp_session_counters session_counters;
+	} data;
+};
+
+#endif /* BFD_DP_PACKET_H */
diff --git a/bfdd/dplane.c b/bfdd/dplane.c
new file mode 100644
index 000000000..b8f0aadd9
--- /dev/null
+++ b/bfdd/dplane.c
@@ -0,0 +1,1199 @@
+/*
+ * BFD data plane implementation (distributed BFD).
+ *
+ * Copyright (C) 2020 Network Device Education Foundation, Inc. ("NetDEF")
+ *                    Rafael Zalamena
+ *
+ * This program is free software; you can redistribute it and/or modify
+ * it under the terms of the GNU General Public License as published by
+ * the Free Software Foundation; either version 2 of the License, or
+ * (at your option) any later version.
+ *
+ * This program is distributed in the hope that it will be useful, but
+ * WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
+ * General Public License for more details.
+ *
+ * You should have received a copy of the GNU General Public License along
+ * with this program; see the file COPYING; if not, write to the Free Software
+ * Foundation, Inc., 51 Franklin St, Fifth Floor, Boston, MA 02110-1301 USA
+ */
+
+#include <zebra.h>
+
+#include <netinet/in.h>
+#include <netinet/tcp.h>
+#include <sys/socket.h>
+#include <sys/un.h>
+
+#ifdef __FreeBSD__
+#include <sys/endian.h>
+#else
+#include <endian.h>
+#endif /* __FreeBSD__ */
+
+#include <errno.h>
+#include <time.h>
+
+#include "lib/hook.h"
+#include "lib/network.h"
+#include "lib/printfrr.h"
+#include "lib/stream.h"
+#include "lib/thread.h"
+
+#include "bfd.h"
+#include "bfddp_packet.h"
+
+#include "lib/openbsd-queue.h"
+
+DEFINE_MTYPE_STATIC(BFDD, BFDD_DPLANE_CTX, "Data plane client allocated memory")
+
+/** Data plane client socket buffer size. */
+#define BFD_DPLANE_CLIENT_BUF_SIZE 8192
+
+struct bfd_dplane_ctx {
+	/** Client file descriptor. */
+	int sock;
+	/** Is this a connected or accepted? */
+	bool client;
+	/** Is the socket still connecting? */
+	bool connecting;
+	/** Client/server address. */
+	union {
+		struct sockaddr sa;
+		struct sockaddr_in sin;
+		struct sockaddr_in6 sin6;
+		struct sockaddr_un sun;
+	} addr;
+	/** Address length. */
+	socklen_t addrlen;
+	/** Data plane current last used ID. */
+	uint16_t last_id;
+
+	/** Input buffer data. */
+	struct stream *inbuf;
+	/** Output buffer data. */
+	struct stream *outbuf;
+	/** Input event data. */
+	struct thread *inbufev;
+	/** Output event data. */
+	struct thread *outbufev;
+	/** Connection event. */
+	struct thread *connectev;
+
+	/** Amount of bytes read. */
+	uint64_t in_bytes;
+	/** Amount of bytes read peak. */
+	uint64_t in_bytes_peak;
+	/** Amount of bytes written. */
+	uint64_t out_bytes;
+	/** Amount of bytes written peak. */
+	uint64_t out_bytes_peak;
+	/** Amount of output buffer full events (`bfd_dplane_enqueue` failed).
+	 */
+	uint64_t out_fullev;
+
+	/** Amount of messages read (full messages). */
+	uint64_t in_msgs;
+	/** Amount of messages enqueued (maybe written). */
+	uint64_t out_msgs;
+
+	TAILQ_ENTRY(bfd_dplane_ctx) entry;
+};
+
+/**
+ * Callback type for `bfd_dplane_expect`. \see bfd_dplane_expect.
+ */
+typedef void (*bfd_dplane_expect_cb)(struct bfddp_message *msg, void *arg);
+
+static int bfd_dplane_client_connect(struct thread *t);
+static bool bfd_dplane_client_connecting(struct bfd_dplane_ctx *bdc);
+static void bfd_dplane_ctx_free(struct bfd_dplane_ctx *bdc);
+static int _bfd_dplane_add_session(struct bfd_dplane_ctx *bdc,
+				   struct bfd_session *bs);
+
+/*
+ * BFD data plane helper functions.
+ */
+static const char *bfd_dplane_messagetype2str(enum bfddp_message_type bmt)
+{
+	switch (bmt) {
+	case ECHO_REQUEST:
+		return "ECHO_REQUEST";
+	case ECHO_REPLY:
+		return "ECHO_REPLY";
+	case DP_ADD_SESSION:
+		return "DP_ADD_SESSION";
+	case DP_DELETE_SESSION:
+		return "DP_DELETE_SESSION";
+	case BFD_STATE_CHANGE:
+		return "BFD_STATE_CHANGE";
+	case DP_REQUEST_SESSION_COUNTERS:
+		return "DP_REQUEST_SESSION_COUNTERS";
+	case BFD_SESSION_COUNTERS:
+		return "BFD_SESSION_COUNTERS";
+	default:
+		return "UNKNOWN";
+	}
+}
+
+static void bfd_dplane_debug_message(const struct bfddp_message *msg)
+{
+	enum bfddp_message_type bmt;
+	char buf[256], addrs[256];
+	uint32_t flags;
+	int rv;
+
+	if (!bglobal.debug_dplane)
+		return;
+
+	bmt = ntohs(msg->header.type);
+	zlog_debug("dplane-packet: [version=%d length=%d type=%s (%d)]",
+		   msg->header.version, ntohs(msg->header.length),
+		   bfd_dplane_messagetype2str(bmt), bmt);
+
+	switch (bmt) {
+	case ECHO_REPLY:
+	case ECHO_REQUEST:
+		zlog_debug("  [dp_time=%" PRIu64 " bfdd_time=%" PRIu64 "]",
+			   (uint64_t)be64toh(msg->data.echo.dp_time),
+			   (uint64_t)be64toh(msg->data.echo.bfdd_time));
+		break;
+
+	case DP_ADD_SESSION:
+	case DP_DELETE_SESSION:
+		flags = ntohl(msg->data.session.flags);
+		if (flags & SESSION_IPV6)
+			snprintfrr(addrs, sizeof(addrs), "src=%pI6 dst=%pI6",
+				   &msg->data.session.src,
+				   &msg->data.session.dst);
+		else
+			snprintfrr(addrs, sizeof(addrs), "src=%pI4 dst=%pI4",
+				   &msg->data.session.src,
+				   &msg->data.session.dst);
+
+		buf[0] = 0;
+		if (flags & SESSION_CBIT)
+			strlcat(buf, "cpi ", sizeof(buf));
+		if (flags & SESSION_ECHO)
+			strlcat(buf, "echo ", sizeof(buf));
+		if (flags & SESSION_IPV6)
+			strlcat(buf, "ipv6 ", sizeof(buf));
+		if (flags & SESSION_DEMAND)
+			strlcat(buf, "demand ", sizeof(buf));
+		if (flags & SESSION_PASSIVE)
+			strlcat(buf, "passive ", sizeof(buf));
+		if (flags & SESSION_MULTIHOP)
+			strlcat(buf, "multihop ", sizeof(buf));
+		if (flags & SESSION_SHUTDOWN)
+			strlcat(buf, "shutdown ", sizeof(buf));
+
+		/* Remove the last space to make things prettier. */
+		rv = (int)strlen(buf);
+		if (rv > 0)
+			buf[rv - 1] = 0;
+
+		zlog_debug(
+			"  [flags=0x%08x{%s} %s ttl=%d detect_mult=%d "
+			"ifindex=%d ifname=%s]",
+			flags, buf, addrs, msg->data.session.ttl,
+			msg->data.session.detect_mult,
+			ntohl(msg->data.session.ifindex),
+			msg->data.session.ifname);
+		break;
+
+	case BFD_STATE_CHANGE:
+		buf[0] = 0;
+		flags = ntohl(msg->data.state.remote_flags);
+		if (flags & RBIT_CPI)
+			strlcat(buf, "cbit ", sizeof(buf));
+		if (flags & RBIT_DEMAND)
+			strlcat(buf, "demand ", sizeof(buf));
+		if (flags & RBIT_MP)
+			strlcat(buf, "mp ", sizeof(buf));
+
+		/* Remove the last space to make things prettier. */
+		rv = (int)strlen(buf);
+		if (rv > 0)
+			buf[rv - 1] = 0;
+
+		zlog_debug(
+			"  [lid=%u rid=%u flags=0x%02x{%s} state=%s "
+			"diagnostics=%s mult=%d tx=%u rx=%u erx=%u]",
+			ntohl(msg->data.state.lid), ntohl(msg->data.state.rid),
+			flags, buf, state_list[msg->data.state.state].str,
+			diag2str(msg->data.state.diagnostics),
+			msg->data.state.detection_multiplier,
+			ntohl(msg->data.state.desired_tx),
+			ntohl(msg->data.state.required_rx),
+			ntohl(msg->data.state.required_echo_rx));
+		break;
+
+	case DP_REQUEST_SESSION_COUNTERS:
+		zlog_debug("  [lid=%u]", ntohl(msg->data.counters_req.lid));
+		break;
+
+	case BFD_SESSION_COUNTERS:
+		zlog_debug(
+			"  [lid=%u "
+			"control{in %" PRIu64 " bytes (%" PRIu64
+			" packets), "
+			"out %" PRIu64 " bytes (%" PRIu64
+			" packets)} "
+			"echo{in %" PRIu64 " bytes (%" PRIu64
+			" packets), "
+			"out %" PRIu64 " bytes (%" PRIu64 " packets)}]",
+			ntohl(msg->data.session_counters.lid),
+			(uint64_t)be64toh(
+				msg->data.session_counters.control_input_bytes),
+			(uint64_t)be64toh(msg->data.session_counters
+					   .control_input_packets),
+			(uint64_t)be64toh(msg->data.session_counters
+					   .control_output_bytes),
+			(uint64_t)be64toh(msg->data.session_counters
+					   .control_output_packets),
+			(uint64_t)be64toh(msg->data.session_counters.echo_input_bytes),
+			(uint64_t)be64toh(
+				msg->data.session_counters.echo_input_packets),
+			(uint64_t)be64toh(
+				msg->data.session_counters.echo_output_bytes),
+			(uint64_t)be64toh(msg->data.session_counters
+					   .echo_output_packets));
+		break;
+	}
+}
+
+/**
+ * Gets the next unused non zero identification.
+ *
+ * \param bdc the data plane context.
+ *
+ * \returns next usable id.
+ */
+static uint16_t bfd_dplane_next_id(struct bfd_dplane_ctx *bdc)
+{
+	bdc->last_id++;
+
+	/* Don't use reserved id `0`. */
+	if (bdc->last_id == 0)
+		bdc->last_id = 1;
+
+	return bdc->last_id;
+}
+
+static ssize_t bfd_dplane_flush(struct bfd_dplane_ctx *bdc)
+{
+	ssize_t total = 0;
+	int rv;
+
+	while (STREAM_READABLE(bdc->outbuf)) {
+		/* Flush buffer contents to socket. */
+		rv = stream_flush(bdc->outbuf, bdc->sock);
+		if (rv == -1) {
+			/* Interruption: try again. */
+			if (errno == EAGAIN || errno == EWOULDBLOCK
+			    || errno == EINTR)
+				continue;
+
+			zlog_warn("%s: socket failed: %s", __func__,
+				  strerror(errno));
+			bfd_dplane_ctx_free(bdc);
+			return 0;
+		}
+		if (rv == 0) {
+			if (bglobal.debug_dplane)
+				zlog_info("%s: connection closed", __func__);
+
+			bfd_dplane_ctx_free(bdc);
+			return 0;
+		}
+
+		/* Account total written. */
+		total += rv;
+
+		/* Account output bytes. */
+		bdc->out_bytes += (uint64_t)rv;
+
+		/* Forward pointer. */
+		stream_forward_getp(bdc->outbuf, (size_t)rv);
+	}
+
+	/* Make more space for new data. */
+	stream_pulldown(bdc->outbuf);
+
+	/* Disable write ready events. */
+	THREAD_OFF(bdc->outbufev);
+
+	return total;
+}
+
+static int bfd_dplane_write(struct thread *t)
+{
+	struct bfd_dplane_ctx *bdc = THREAD_ARG(t);
+
+	/* Handle connection stage. */
+	if (bdc->connecting && bfd_dplane_client_connecting(bdc))
+		return 0;
+
+	bfd_dplane_flush(bdc);
+
+	return 0;
+}
+
+static void
+bfd_dplane_session_state_change(struct bfd_dplane_ctx *bdc,
+				const struct bfddp_state_change *state)
+{
+	struct bfd_session *bs;
+	uint32_t flags;
+	int old_state;
+
+	/* Look up session. */
+	bs = bfd_id_lookup(ntohl(state->lid));
+	if (bs == NULL) {
+		if (bglobal.debug_dplane)
+			zlog_debug("%s: failed to find session to update",
+				   __func__);
+		return;
+	}
+
+	flags = ntohl(state->remote_flags);
+	old_state = bs->ses_state;
+
+	/* Update session state. */
+	bs->ses_state = state->state;
+	bs->remote_diag = state->diagnostics;
+	bs->discrs.remote_discr = ntohl(state->rid);
+	bs->remote_cbit = !!(flags & RBIT_CPI);
+	bs->remote_detect_mult = state->detection_multiplier;
+	bs->remote_timers.desired_min_tx = ntohl(state->desired_tx);
+	bs->remote_timers.required_min_rx = ntohl(state->required_rx);
+	bs->remote_timers.required_min_echo = ntohl(state->required_echo_rx);
+
+	/* Notify and update counters. */
+	control_notify(bs, bs->ses_state);
+
+	/* No state change. */
+	if (old_state == bs->ses_state)
+		return;
+
+	switch (bs->ses_state) {
+	case PTM_BFD_ADM_DOWN:
+	case PTM_BFD_DOWN:
+		/* Both states mean down. */
+		if (old_state == PTM_BFD_ADM_DOWN || old_state == PTM_BFD_DOWN)
+			break;
+
+		monotime(&bs->downtime);
+		bs->stats.session_down++;
+		break;
+	case PTM_BFD_UP:
+		monotime(&bs->uptime);
+		bs->stats.session_up++;
+		break;
+	case PTM_BFD_INIT:
+		/* NOTHING */
+		break;
+
+	default:
+		zlog_warn("%s: unhandled new state %d", __func__,
+			  bs->ses_state);
+		break;
+	}
+
+	if (bglobal.debug_peer_event)
+		zlog_debug("state-change: [data plane: %s] %s -> %s",
+			   bs_to_string(bs), state_list[old_state].str,
+			   state_list[bs->ses_state].str);
+}
+
+/**
+ * Enqueue message in output buffer.
+ *
+ * \param[in,out] bdc data plane client context.
+ * \param[in] buf the message to buffer.
+ * \param[in] buflen the amount of bytes to buffer.
+ *
+ * \returns `-1` on failure (buffer full) or `0` on success.
+ */
+static int bfd_dplane_enqueue(struct bfd_dplane_ctx *bdc, const void *buf,
+			      size_t buflen)
+{
+	size_t rlen;
+
+	/* Handle not connected yet client. */
+	if (bdc->client && bdc->sock == -1)
+		return -1;
+
+	/* Not enough space. */
+	if (buflen > STREAM_WRITEABLE(bdc->outbuf)) {
+		bdc->out_fullev++;
+		return -1;
+	}
+
+	/* Show debug message if active. */
+	bfd_dplane_debug_message((struct bfddp_message *)buf);
+
+	/* Buffer the message. */
+	stream_write(bdc->outbuf, buf, buflen);
+
+	/* Account message as sent. */
+	bdc->out_msgs++;
+	/* Register peak buffered bytes. */
+	rlen = STREAM_READABLE(bdc->outbuf);
+	if (bdc->out_bytes_peak < rlen)
+		bdc->out_bytes_peak = rlen;
+
+	/* Schedule if it is not yet. */
+	if (bdc->outbufev == NULL)
+		thread_add_write(master, bfd_dplane_write, bdc, bdc->sock,
+				 &bdc->outbufev);
+
+	return 0;
+}
+
+static void bfd_dplane_echo_request_handle(struct bfd_dplane_ctx *bdc,
+					   const struct bfddp_message *bm)
+{
+	struct bfddp_message msg = {};
+	uint16_t msglen = sizeof(msg.header) + sizeof(msg.data.echo);
+	struct timeval tv;
+
+	gettimeofday(&tv, NULL);
+
+	/* Prepare header. */
+	msg.header.version = BFD_DP_VERSION;
+	msg.header.type = htons(ECHO_REPLY);
+	msg.header.length = htons(msglen);
+
+	/* Prepare payload. */
+	msg.data.echo.dp_time = bm->data.echo.dp_time;
+	msg.data.echo.bfdd_time =
+		htobe64((uint64_t)((tv.tv_sec * 1000000) + tv.tv_usec));
+
+	/* Enqueue for output. */
+	bfd_dplane_enqueue(bdc, &msg, msglen);
+}
+
+static void bfd_dplane_handle_message(struct bfddp_message *msg, void *arg)
+{
+	enum bfddp_message_type bmt;
+	struct bfd_dplane_ctx *bdc = arg;
+
+	/* Call the appropriated handler. */
+	bmt = ntohs(msg->header.type);
+	switch (bmt) {
+	case ECHO_REQUEST:
+		bfd_dplane_echo_request_handle(bdc, msg);
+		break;
+	case BFD_STATE_CHANGE:
+		bfd_dplane_session_state_change(bdc, &msg->data.state);
+		break;
+	case ECHO_REPLY:
+		/* NOTHING: we don't do anything with this information. */
+		break;
+	case DP_ADD_SESSION:
+	case DP_DELETE_SESSION:
+	case DP_REQUEST_SESSION_COUNTERS:
+		/* NOTHING: we are not supposed to receive this. */
+		break;
+	case BFD_SESSION_COUNTERS:
+		/*
+		 * NOTHING: caller of DP_REQUEST_SESSION_COUNTERS should
+		 * handle this with `bfd_dplane_expect`.
+		 */
+		break;
+
+	default:
+		zlog_debug("%s: unhandled message type %d", __func__, bmt);
+		break;
+	}
+}
+
+/**
+ * Reads the socket immediately to receive data plane answer to query.
+ *
+ * \param bdc the data plane context.
+ * \param id the message ID waiting response.
+ * \param cb the callback to call when ready.
+ * \param arg the callback argument.
+ *
+ * \return
+ * `-2` on unavailability (try again), `-1` on failure or `0` on success.
+ */
+static int bfd_dplane_expect(struct bfd_dplane_ctx *bdc, uint16_t id,
+			     bfd_dplane_expect_cb cb, void *arg)
+{
+	struct bfddp_message_header *bh;
+	size_t rlen = 0, reads = 0;
+	ssize_t rv;
+
+	/*
+	 * Don't attempt to read if buffer is full, otherwise we'll get a
+	 * bogus 'connection closed' signal (rv == 0).
+	 */
+	if (bdc->inbuf->endp == bdc->inbuf->size)
+		goto skip_read;
+
+read_again:
+	/* Attempt to read message from client. */
+	rv = stream_read_try(bdc->inbuf, bdc->sock,
+			     STREAM_WRITEABLE(bdc->inbuf));
+	if (rv == 0) {
+		if (bglobal.debug_dplane)
+			zlog_info("%s: socket closed", __func__);
+
+		bfd_dplane_ctx_free(bdc);
+		return -1;
+	}
+	if (rv == -1) {
+		zlog_warn("%s: socket failed: %s", __func__, strerror(errno));
+		bfd_dplane_ctx_free(bdc);
+		return -1;
+	}
+
+	/* We got interrupted, reschedule read. */
+	if (rv == -2)
+		return -2;
+
+	/* Account read bytes. */
+	bdc->in_bytes += (uint64_t)rv;
+	/* Register peak buffered bytes. */
+	rlen = STREAM_READABLE(bdc->inbuf);
+	if (bdc->in_bytes_peak < rlen)
+		bdc->in_bytes_peak = rlen;
+
+skip_read:
+	while (rlen > 0) {
+		bh = (struct bfddp_message_header *)stream_pnt(bdc->inbuf);
+		/* Not enough data read. */
+		if (ntohs(bh->length) > rlen)
+			goto read_again;
+
+		/* Account full message read. */
+		bdc->in_msgs++;
+
+		/* Account this message as whole read for buffer reorganize. */
+		reads++;
+
+		/* Check for bad version. */
+		if (bh->version != BFD_DP_VERSION) {
+			zlog_err("%s: bad data plane client version: %d",
+				 __func__, bh->version);
+			return -1;
+		}
+
+		/* Show debug message if active. */
+		bfd_dplane_debug_message((struct bfddp_message *)bh);
+
+		/*
+		 * Handle incoming message with callback if the ID matches,
+		 * otherwise fallback to default handler.
+		 */
+		if (id && ntohs(bh->id) == id)
+			cb((struct bfddp_message *)bh, arg);
+		else
+			bfd_dplane_handle_message((struct bfddp_message *)bh,
+						  bdc);
+
+		/* Advance current read pointer. */
+		stream_forward_getp(bdc->inbuf, ntohs(bh->length));
+
+		/* Reduce the buffer available bytes. */
+		rlen -= ntohs(bh->length);
+
+		/* Reorganize buffer to handle more bytes read. */
+		if (reads >= 3) {
+			stream_pulldown(bdc->inbuf);
+			reads = 0;
+		}
+
+		/* We found the message, return to caller. */
+		if (id && ntohs(bh->id) == id)
+			break;
+	}
+
+	return 0;
+}
+
+static int bfd_dplane_read(struct thread *t)
+{
+	struct bfd_dplane_ctx *bdc = THREAD_ARG(t);
+	int rv;
+
+	rv = bfd_dplane_expect(bdc, 0, bfd_dplane_handle_message, NULL);
+	if (rv == -1)
+		return 0;
+
+	stream_pulldown(bdc->inbuf);
+	thread_add_read(master, bfd_dplane_read, bdc, bdc->sock, &bdc->inbufev);
+	return 0;
+}
+
+static void _bfd_session_register_dplane(struct hash_bucket *hb, void *arg)
+{
+	struct bfd_session *bs = hb->data;
+	struct bfd_dplane_ctx *bdc = arg;
+
+	if (bs->bdc != NULL)
+		return;
+
+	/* Disable software session. */
+	bfd_session_disable(bs);
+
+	/* Move session to data plane. */
+	_bfd_dplane_add_session(bdc, bs);
+}
+
+static struct bfd_dplane_ctx *bfd_dplane_ctx_new(int sock)
+{
+	struct bfd_dplane_ctx *bdc;
+
+	bdc = XCALLOC(MTYPE_BFDD_DPLANE_CTX, sizeof(*bdc));
+	if (bdc == NULL)
+		return NULL;
+
+	bdc->sock = sock;
+	bdc->inbuf = stream_new(BFD_DPLANE_CLIENT_BUF_SIZE);
+	bdc->outbuf = stream_new(BFD_DPLANE_CLIENT_BUF_SIZE);
+
+	/* If not socket ready, skip read and session registration. */
+	if (sock == -1)
+		return bdc;
+
+	thread_add_read(master, bfd_dplane_read, bdc, sock, &bdc->inbufev);
+
+	/* Register all unattached sessions. */
+	bfd_key_iterate(_bfd_session_register_dplane, bdc);
+
+	return bdc;
+}
+
+static void _bfd_session_unregister_dplane(struct hash_bucket *hb, void *arg)
+{
+	struct bfd_session *bs = hb->data;
+	struct bfd_dplane_ctx *bdc = arg;
+
+	if (bs->bdc != bdc)
+		return;
+
+	bs->bdc = NULL;
+
+	/* Fallback to software. */
+	bfd_session_enable(bs);
+}
+
+static void bfd_dplane_ctx_free(struct bfd_dplane_ctx *bdc)
+{
+	if (bglobal.debug_dplane)
+		zlog_debug("%s: terminating data plane client %d", __func__,
+			   bdc->sock);
+
+	/* Client mode has special treatment. */
+	if (bdc->client) {
+		/* Disable connection event if any. */
+		THREAD_OFF(bdc->connectev);
+
+		/* Normal treatment on shutdown. */
+		if (bglobal.bg_shutdown)
+			goto free_resources;
+
+		/* Attempt reconnection. */
+		socket_close(&bdc->sock);
+		THREAD_OFF(bdc->inbufev);
+		THREAD_OFF(bdc->outbufev);
+		thread_add_timer(master, bfd_dplane_client_connect, bdc, 3,
+				 &bdc->connectev);
+		return;
+	}
+
+free_resources:
+	/* Remove from the list of attached data planes. */
+	TAILQ_REMOVE(&bglobal.bg_dplaneq, bdc, entry);
+
+	/* Detach all associated sessions. */
+	if (bglobal.bg_shutdown == false)
+		bfd_key_iterate(_bfd_session_unregister_dplane, bdc);
+
+	/* Free resources. */
+	socket_close(&bdc->sock);
+	stream_free(bdc->inbuf);
+	stream_free(bdc->outbuf);
+	THREAD_OFF(bdc->inbufev);
+	THREAD_OFF(bdc->outbufev);
+	XFREE(MTYPE_BFDD_DPLANE_CTX, bdc);
+}
+
+static void _bfd_dplane_session_fill(const struct bfd_session *bs,
+				     struct bfddp_message *msg)
+{
+	uint16_t msglen = sizeof(msg->header) + sizeof(msg->data.session);
+
+	/* Message header. */
+	msg->header.version = BFD_DP_VERSION;
+	msg->header.length = ntohs(msglen);
+	msg->header.type = ntohs(DP_ADD_SESSION);
+
+	/* Message payload. */
+	msg->data.session.dst = bs->key.peer;
+	msg->data.session.src = bs->key.local;
+	msg->data.session.detect_mult = bs->detect_mult;
+
+	if (bs->ifp) {
+		msg->data.session.ifindex = htonl(bs->ifp->ifindex);
+		strlcpy(msg->data.session.ifname, bs->ifp->name,
+			sizeof(msg->data.session.ifname));
+	}
+	if (bs->flags & BFD_SESS_FLAG_MH) {
+		msg->data.session.flags |= SESSION_MULTIHOP;
+		msg->data.session.ttl = bs->mh_ttl;
+	} else
+		msg->data.session.ttl = BFD_TTL_VAL;
+
+	if (bs->flags & BFD_SESS_FLAG_IPV6)
+		msg->data.session.flags |= SESSION_IPV6;
+	if (bs->flags & BFD_SESS_FLAG_ECHO)
+		msg->data.session.flags |= SESSION_ECHO;
+	if (bs->flags & BFD_SESS_FLAG_CBIT)
+		msg->data.session.flags |= SESSION_CBIT;
+	if (bs->flags & BFD_SESS_FLAG_PASSIVE)
+		msg->data.session.flags |= SESSION_PASSIVE;
+	if (bs->flags & BFD_SESS_FLAG_SHUTDOWN)
+		msg->data.session.flags |= SESSION_SHUTDOWN;
+
+	msg->data.session.flags = htonl(msg->data.session.flags);
+	msg->data.session.lid = htonl(bs->discrs.my_discr);
+	msg->data.session.min_tx = htonl(bs->timers.desired_min_tx);
+	msg->data.session.min_rx = htonl(bs->timers.required_min_rx);
+	msg->data.session.min_echo_rx = htonl(bs->timers.required_min_echo);
+}
+
+static int _bfd_dplane_add_session(struct bfd_dplane_ctx *bdc,
+				   struct bfd_session *bs)
+{
+	int rv;
+
+	/* Associate session. */
+	bs->bdc = bdc;
+
+	/* Reset previous state. */
+	bs->remote_diag = 0;
+	bs->local_diag = 0;
+	bs->ses_state = PTM_BFD_DOWN;
+
+	/* Enqueue message to data plane client. */
+	rv = bfd_dplane_update_session(bs);
+	if (rv != 0)
+		bs->bdc = NULL;
+
+	return rv;
+}
+
+static void _bfd_dplane_update_session_counters(struct bfddp_message *msg,
+						void *arg)
+{
+	struct bfd_session *bs = arg;
+
+	bs->stats.rx_ctrl_pkt =
+		be64toh(msg->data.session_counters.control_input_packets);
+	bs->stats.tx_ctrl_pkt =
+		be64toh(msg->data.session_counters.control_output_packets);
+	bs->stats.rx_echo_pkt =
+		be64toh(msg->data.session_counters.echo_input_packets);
+	bs->stats.tx_echo_pkt =
+		be64toh(msg->data.session_counters.echo_output_bytes);
+}
+
+/**
+ * Send message to data plane requesting the session counters.
+ *
+ * \param bs the BFD session.
+ *
+ * \returns `0` on failure or the request id.
+ */
+static uint16_t bfd_dplane_request_counters(const struct bfd_session *bs)
+{
+	struct bfddp_message msg = {};
+	size_t msglen = sizeof(msg.header) + sizeof(msg.data.counters_req);
+
+	/* Fill header information. */
+	msg.header.version = BFD_DP_VERSION;
+	msg.header.length = htons(msglen);
+	msg.header.type = htons(DP_REQUEST_SESSION_COUNTERS);
+	msg.header.id = htons(bfd_dplane_next_id(bs->bdc));
+
+	/* Session to get counters. */
+	msg.data.counters_req.lid = htonl(bs->discrs.my_discr);
+
+	/* If enqueue failed, let caller know. */
+	if (bfd_dplane_enqueue(bs->bdc, &msg, msglen) == -1)
+		return 0;
+
+	/* Flush socket. */
+	bfd_dplane_flush(bs->bdc);
+
+	return ntohs(msg.header.id);
+}
+
+/*
+ * Data plane listening socket.
+ */
+static int bfd_dplane_accept(struct thread *t)
+{
+	struct bfd_global *bg = THREAD_ARG(t);
+	struct bfd_dplane_ctx *bdc;
+	int sock;
+
+	/* Accept new connection. */
+	sock = accept(bg->bg_dplane_sock, NULL, 0);
+	if (sock == -1) {
+		zlog_warn("%s: accept failed: %s", __func__, strerror(errno));
+		goto reschedule_and_return;
+	}
+
+	/* Create and handle new connection. */
+	bdc = bfd_dplane_ctx_new(sock);
+	TAILQ_INSERT_TAIL(&bglobal.bg_dplaneq, bdc, entry);
+
+	if (bglobal.debug_dplane)
+		zlog_debug("%s: new data plane client connected", __func__);
+
+reschedule_and_return:
+	thread_add_read(master, bfd_dplane_accept, bg, bg->bg_dplane_sock,
+			&bglobal.bg_dplane_sockev);
+	return 0;
+}
+
+/*
+ * Data plane connecting socket.
+ */
+static void _bfd_dplane_client_bootstrap(struct bfd_dplane_ctx *bdc)
+{
+	bdc->connecting = false;
+
+	/* Clean up buffers. */
+	stream_reset(bdc->inbuf);
+	stream_reset(bdc->outbuf);
+
+	/* Ask for read notifications. */
+	thread_add_read(master, bfd_dplane_read, bdc, bdc->sock, &bdc->inbufev);
+
+	/* Remove all sessions then register again to send them all. */
+	bfd_key_iterate(_bfd_session_unregister_dplane, bdc);
+	bfd_key_iterate(_bfd_session_register_dplane, bdc);
+}
+
+static bool bfd_dplane_client_connecting(struct bfd_dplane_ctx *bdc)
+{
+	int rv;
+	socklen_t rvlen = sizeof(rv);
+
+	/* Make sure `errno` is reset, then test `getsockopt` success. */
+	errno = 0;
+	if (getsockopt(bdc->sock, SOL_SOCKET, SO_ERROR, &rv, &rvlen) == -1)
+		rv = -1;
+
+	/* Connection successful. */
+	if (rv == 0) {
+		if (bglobal.debug_dplane)
+			zlog_debug("%s: connected to server: %d", __func__,
+				   bdc->sock);
+
+		_bfd_dplane_client_bootstrap(bdc);
+		return false;
+	}
+
+	switch (rv) {
+	case EINTR:
+	case EAGAIN:
+	case EALREADY:
+	case EINPROGRESS:
+		/* non error, wait more. */
+		return true;
+
+	default:
+		zlog_warn("%s: connection failed: %s", __func__,
+			  strerror(errno));
+		bfd_dplane_ctx_free(bdc);
+		return true;
+	}
+}
+
+static int bfd_dplane_client_connect(struct thread *t)
+{
+	struct bfd_dplane_ctx *bdc = THREAD_ARG(t);
+	int rv, sock;
+	socklen_t rvlen = sizeof(rv);
+
+	/* Allocate new socket. */
+	sock = socket(bdc->addr.sa.sa_family, SOCK_STREAM, 0);
+	if (sock == -1) {
+		zlog_warn("%s: failed to initialize socket: %s", __func__,
+			  strerror(errno));
+		goto reschedule_connect;
+	}
+
+	/* Set non blocking socket. */
+	set_nonblocking(sock);
+
+	/* Set 'no delay' (disables nagle algorithm) for IPv4/IPv6. */
+	rv = 1;
+	if (bdc->addr.sa.sa_family != AF_UNIX
+	    && setsockopt(sock, IPPROTO_TCP, TCP_NODELAY, &rv, rvlen) == -1)
+		zlog_warn("%s: TCP_NODELAY: %s", __func__, strerror(errno));
+
+	/* Attempt to connect. */
+	rv = connect(sock, &bdc->addr.sa, bdc->addrlen);
+	if (rv == -1 && (errno != EINPROGRESS && errno != EAGAIN)) {
+		zlog_warn("%s: data plane connection failed: %s", __func__,
+			  strerror(errno));
+		goto reschedule_connect;
+	}
+
+	bdc->sock = sock;
+	if (rv == -1) {
+		if (bglobal.debug_dplane)
+			zlog_debug("%s: server connection in progress: %d",
+				   __func__, sock);
+
+		/* If we are not connected yet, ask for write notifications. */
+		bdc->connecting = true;
+		thread_add_write(master, bfd_dplane_write, bdc, bdc->sock,
+				 &bdc->outbufev);
+	} else {
+		if (bglobal.debug_dplane)
+			zlog_debug("%s: server connection: %d", __func__, sock);
+
+		/* Otherwise just start accepting data. */
+		_bfd_dplane_client_bootstrap(bdc);
+	}
+
+	return 0;
+
+reschedule_connect:
+	THREAD_OFF(bdc->inbufev);
+	THREAD_OFF(bdc->outbufev);
+	socket_close(&sock);
+	thread_add_timer(master, bfd_dplane_client_connect, bdc, 3,
+			 &bdc->connectev);
+	return 0;
+}
+
+static void bfd_dplane_client_init(const struct sockaddr *sa, socklen_t salen)
+{
+	struct bfd_dplane_ctx *bdc;
+
+	/* Allocate context and copy address for reconnection. */
+	bdc = bfd_dplane_ctx_new(-1);
+	if (salen <= sizeof(bdc->addr)) {
+		memcpy(&bdc->addr, sa, salen);
+		bdc->addrlen = sizeof(bdc->addr);
+	} else {
+		memcpy(&bdc->addr, sa, sizeof(bdc->addr));
+		bdc->addrlen = sizeof(bdc->addr);
+		zlog_warn("%s: server address truncated (from %d to %d)",
+			  __func__, salen, bdc->addrlen);
+	}
+
+	bdc->client = true;
+
+	thread_add_timer(master, bfd_dplane_client_connect, bdc, 0,
+			 &bdc->connectev);
+
+	/* Insert into data plane lists. */
+	TAILQ_INSERT_TAIL(&bglobal.bg_dplaneq, bdc, entry);
+}
+
+/**
+ * Termination phase of the distributed BFD infrastructure: free all allocated
+ * resources.
+ */
+static int bfd_dplane_finish_late(void)
+{
+	struct bfd_dplane_ctx *bdc;
+
+	if (bglobal.debug_dplane)
+		zlog_debug("%s: terminating distributed BFD", __func__);
+
+	/* Free all data plane client contexts. */
+	while ((bdc = TAILQ_FIRST(&bglobal.bg_dplaneq)) != NULL)
+		bfd_dplane_ctx_free(bdc);
+
+	/* Cancel accept thread and close socket. */
+	THREAD_OFF(bglobal.bg_dplane_sockev);
+	close(bglobal.bg_dplane_sock);
+
+	return 0;
+}
+
+/*
+ * Data plane exported functions.
+ */
+void bfd_dplane_init(const struct sockaddr *sa, socklen_t salen, bool client)
+{
+	int sock;
+
+	zlog_info("initializing distributed BFD");
+
+	/* Initialize queue header. */
+	TAILQ_INIT(&bglobal.bg_dplaneq);
+
+	/* Initialize listening socket. */
+	bglobal.bg_dplane_sock = -1;
+
+	/* Observe shutdown events. */
+	hook_register(frr_fini, bfd_dplane_finish_late);
+
+	/* Handle client mode. */
+	if (client) {
+		bfd_dplane_client_init(sa, salen);
+		return;
+	}
+
+	/*
+	 * Data plane socket creation:
+	 * - Set REUSEADDR option for taking over previously open socket.
+	 * - Bind to address requested (maybe IPv4, IPv6, UNIX etc...).
+	 * - Listen on that address for new connections.
+	 * - Ask to be waken up when a new connection comes.
+	 */
+	sock = socket(sa->sa_family, SOCK_STREAM, 0);
+	if (sock == -1) {
+		zlog_warn("%s: failed to initialize socket: %s", __func__,
+			  strerror(errno));
+		return;
+	}
+
+	if (sockopt_reuseaddr(sock) == -1) {
+		zlog_warn("%s: failed to set reuseaddr: %s", __func__,
+			  strerror(errno));
+		close(sock);
+		return;
+	}
+
+	/* Handle UNIX socket: delete previous socket if any. */
+	if (sa->sa_family == AF_UNIX)
+		unlink(((struct sockaddr_un *)sa)->sun_path);
+
+	if (bind(sock, sa, salen) == -1) {
+		zlog_warn("%s: failed to bind socket: %s", __func__,
+			  strerror(errno));
+		close(sock);
+		return;
+	}
+
+	if (listen(sock, SOMAXCONN) == -1) {
+		zlog_warn("%s: failed to put socket on listen: %s", __func__,
+			  strerror(errno));
+		close(sock);
+		return;
+	}
+
+	bglobal.bg_dplane_sock = sock;
+	thread_add_read(master, bfd_dplane_accept, &bglobal, sock,
+			&bglobal.bg_dplane_sockev);
+}
+
+int bfd_dplane_add_session(struct bfd_session *bs)
+{
+	struct bfd_dplane_ctx *bdc;
+
+	/* Select a data plane client to install session. */
+	TAILQ_FOREACH (bdc, &bglobal.bg_dplaneq, entry) {
+		if (_bfd_dplane_add_session(bdc, bs) == 0)
+			return 0;
+	}
+
+	return -1;
+}
+
+int bfd_dplane_update_session(const struct bfd_session *bs)
+{
+	struct bfddp_message msg = {};
+
+	if (bs->bdc == NULL)
+		return 0;
+
+	_bfd_dplane_session_fill(bs, &msg);
+
+	/* Enqueue message to data plane client. */
+	return bfd_dplane_enqueue(bs->bdc, &msg, ntohs(msg.header.length));
+}
+
+int bfd_dplane_delete_session(struct bfd_session *bs)
+{
+	struct bfddp_message msg = {};
+	int rv;
+
+	/* Not using data plane, just return success. */
+	if (bs->bdc == NULL)
+		return 0;
+
+	/* Fill most of the common fields. */
+	_bfd_dplane_session_fill(bs, &msg);
+
+	/* Change the message type. */
+	msg.header.type = ntohs(DP_DELETE_SESSION);
+
+	/* Enqueue message to data plane client. */
+	rv = bfd_dplane_enqueue(bs->bdc, &msg, ntohs(msg.header.length));
+
+	/* Remove association. */
+	bs->bdc = NULL;
+
+	return rv;
+}
+
+/*
+ * Data plane CLI.
+ */
+void bfd_dplane_show_counters(struct vty *vty)
+{
+	struct bfd_dplane_ctx *bdc;
+
+#define SHOW_COUNTER(label, counter, formatter)                                \
+	vty_out(vty, "%28s: %" formatter "\n", (label), (counter))
+
+	vty_out(vty, "%28s\n%28s\n", "Data plane", "==========");
+	TAILQ_FOREACH (bdc, &bglobal.bg_dplaneq, entry) {
+		SHOW_COUNTER("File descriptor", bdc->sock, "d");
+		SHOW_COUNTER("Input bytes", bdc->in_bytes, PRIu64);
+		SHOW_COUNTER("Input bytes peak", bdc->in_bytes_peak, PRIu64);
+		SHOW_COUNTER("Input messages", bdc->in_msgs, PRIu64);
+		SHOW_COUNTER("Input current usage", STREAM_READABLE(bdc->inbuf),
+			     "zu");
+		SHOW_COUNTER("Output bytes", bdc->out_bytes, PRIu64);
+		SHOW_COUNTER("Output bytes peak", bdc->out_bytes_peak, PRIu64);
+		SHOW_COUNTER("Output messages", bdc->out_msgs, PRIu64);
+		SHOW_COUNTER("Output full events", bdc->out_fullev, PRIu64);
+		SHOW_COUNTER("Output current usage",
+			     STREAM_READABLE(bdc->inbuf), "zu");
+		vty_out(vty, "\n");
+	}
+#undef SHOW_COUNTER
+}
+
+int bfd_dplane_update_session_counters(struct bfd_session *bs)
+{
+	uint16_t id;
+	int rv;
+
+	/* If session is not using data plane, then just return success. */
+	if (bs->bdc == NULL)
+		return 0;
+
+	/* Make the request. */
+	id = bfd_dplane_request_counters(bs);
+	if (id == 0) {
+		zlog_debug("%s: counters request failed", __func__);
+		return -1;
+	}
+
+	/* Handle interruptions. */
+	do {
+		rv = bfd_dplane_expect(bs->bdc, id,
+				       _bfd_dplane_update_session_counters, bs);
+	} while (rv == -2);
+
+	return rv;
+}
diff --git a/bfdd/ptm_adapter.c b/bfdd/ptm_adapter.c
index 90e2df236..44519c47b 100644
--- a/bfdd/ptm_adapter.c
+++ b/bfdd/ptm_adapter.c
@@ -703,7 +703,7 @@ static void bfdd_sessions_disable_interface(struct interface *ifp)
 			continue;
 
 		bfd_session_disable(bs);
-
+		bs->ifp = NULL;
 	}
 }
 
@@ -752,6 +752,7 @@ void bfdd_sessions_disable_vrf(struct vrf *vrf)
 			continue;
 
 		bfd_session_disable(bs);
+		bs->vrf = NULL;
 	}
 }
 
diff --git a/bfdd/subdir.am b/bfdd/subdir.am
index d0e3c1e8d..e572d4a3c 100644
--- a/bfdd/subdir.am
+++ b/bfdd/subdir.am
@@ -22,10 +22,18 @@ bfdd_libbfd_a_SOURCES = \
 	bfdd/bfd_packet.c \
 	bfdd/config.c \
 	bfdd/control.c \
+	bfdd/dplane.c \
 	bfdd/event.c \
 	bfdd/ptm_adapter.c \
 	# end
 
+# Install headers so it can be used by external data plane
+# implementations.
+bfdd_headersdir = $(pkgincludedir)/bfdd
+bfdd_headers_HEADERS = \
+	bfdd/bfddp_packet.h \
+	# end
+
 clippy_scan += \
 	bfdd/bfdd_cli.c \
 	bfdd/bfdd_vty.c \