[PATCH 0/8] bonding: fixes and enhancements

[PATCH 0/8] bonding: fixes and enhancements

[Stephen Hemminger]

These are bug fixes and some small enhancements to allow bonding
to work with external control (teamd). Please consider integrating
these into DPDK 2.2

Eric Kinzie (8):
  bond: use existing enslaved device queues
  bond mode 4: copy entire config structure
  bond mode 4: do not ignore multicast
  bond mode 4: allow external state machine
  bond: active slaves with no primary
  bond: handle slaves with fewer queues than bonding device
  bond: per-slave intermediate rx ring
  bond: do not activate slave twice

 app/test/test_link_bonding_mode4.c                |   7 +-
 drivers/net/bonding/rte_eth_bond_8023ad.c         | 174 +++++++++++++++++
 drivers/net/bonding/rte_eth_bond_8023ad.h         |  44 +++++
 drivers/net/bonding/rte_eth_bond_8023ad_private.h |   2 +
 drivers/net/bonding/rte_eth_bond_api.c            |  48 ++++-
 drivers/net/bonding/rte_eth_bond_pmd.c            | 217 ++++++++++++++++++----
 drivers/net/bonding/rte_eth_bond_private.h        |   9 +-
 drivers/net/bonding/rte_eth_bond_version.map      |   6 +
 8 files changed, 462 insertions(+), 45 deletions(-)

-- 
2.1.4

[PATCH 1/8] bond: use existing enslaved device queues

[Stephen Hemminger]

From: Eric Kinzie <ehkinzie@gmail.com>

This solves issues when an active device is added to a bond.

If a device to be enslaved already has transmit and/or receive queues
allocated, use those and then create any additional queues that are
necessary.

Signed-off-by: Eric Kinzie <ehkinzie@gmail.com>
Signed-off-by: Stephen Hemminger <stephen@networkplumber.org>
---
 drivers/net/bonding/rte_eth_bond_pmd.c | 8 ++++++--
 1 file changed, 6 insertions(+), 2 deletions(-)

diff --git a/drivers/net/bonding/rte_eth_bond_pmd.c b/drivers/net/bonding/rte_eth_bond_pmd.c
index aa985f5..127b976 100644
--- a/drivers/net/bonding/rte_eth_bond_pmd.c
+++ b/drivers/net/bonding/rte_eth_bond_pmd.c
@@ -1340,7 +1340,9 @@ slave_configure(struct rte_eth_dev *bonded_eth_dev,
 	}
 
 	/* Setup Rx Queues */
-	for (q_id = 0; q_id < bonded_eth_dev->data->nb_rx_queues; q_id++) {
+	/* Use existing queues, if any */
+	for (q_id = slave_eth_dev->data->nb_rx_queues;
+	     q_id < bonded_eth_dev->data->nb_rx_queues; q_id++) {
 		bd_rx_q = (struct bond_rx_queue *)bonded_eth_dev->data->rx_queues[q_id];
 
 		errval = rte_eth_rx_queue_setup(slave_eth_dev->data->port_id, q_id,
@@ -1356,7 +1358,9 @@ slave_configure(struct rte_eth_dev *bonded_eth_dev,
 	}
 
 	/* Setup Tx Queues */
-	for (q_id = 0; q_id < bonded_eth_dev->data->nb_tx_queues; q_id++) {
+	/* Use existing queues, if any */
+	for (q_id = slave_eth_dev->data->nb_tx_queues;
+	     q_id < bonded_eth_dev->data->nb_tx_queues; q_id++) {
 		bd_tx_q = (struct bond_tx_queue *)bonded_eth_dev->data->tx_queues[q_id];
 
 		errval = rte_eth_tx_queue_setup(slave_eth_dev->data->port_id, q_id,
-- 
2.1.4

Re: [PATCH 1/8] bond: use existing enslaved device queues

[Declan Doherty]

On 04/12/15 17:14, Stephen Hemminger wrote:
> From: Eric Kinzie <ehkinzie@gmail.com>
>
> This solves issues when an active device is added to a bond.
>
> If a device to be enslaved already has transmit and/or receive queues
> allocated, use those and then create any additional queues that are
> necessary.
>
> Signed-off-by: Eric Kinzie <ehkinzie@gmail.com>
> Signed-off-by: Stephen Hemminger <stephen@networkplumber.org>
> ---
...
>

Acked-by: Declan Doherty <declan.doherty@intel.com>

[PATCH 2/8] bond mode 4: copy entire config structure

[Stephen Hemminger]

From: Eric Kinzie <ekinzie@brocade.com>

Copy all needed fields from the mode8023ad_private structure in
bond_mode_8023ad_conf_get().  This help ensure that a subsequent call
to rte_eth_bond_8023ad_setup() is not passed uninitialized data that
would result in either incorrect behavior or a failed sanity check.

Fixes: 46fb43683679 ("bond: add mode 4")

Signed-off-by: Eric Kinzie <ehkinzie@gmail.com>
Signed-off-by: Stephen Hemminger <stephen@networkplumber.org>
---
 drivers/net/bonding/rte_eth_bond_8023ad.c | 1 +
 1 file changed, 1 insertion(+)

diff --git a/drivers/net/bonding/rte_eth_bond_8023ad.c b/drivers/net/bonding/rte_eth_bond_8023ad.c
index ee2964a..f2620b8 100644
--- a/drivers/net/bonding/rte_eth_bond_8023ad.c
+++ b/drivers/net/bonding/rte_eth_bond_8023ad.c
@@ -1013,6 +1013,7 @@ bond_mode_8023ad_conf_get(struct rte_eth_dev *dev,
 	conf->aggregate_wait_timeout_ms = mode4->aggregate_wait_timeout / ms_ticks;
 	conf->tx_period_ms = mode4->tx_period_timeout / ms_ticks;
 	conf->update_timeout_ms = mode4->update_timeout_us / 1000;
+	conf->rx_marker_period_ms = mode4->rx_marker_timeout / ms_ticks;
 }
 
 void
-- 
2.1.4

Re: [PATCH 2/8] bond mode 4: copy entire config structure

[Declan Doherty]

On 04/12/15 17:14, Stephen Hemminger wrote:
> From: Eric Kinzie <ekinzie@brocade.com>
>
> Copy all needed fields from the mode8023ad_private structure in
> bond_mode_8023ad_conf_get().  This help ensure that a subsequent call
> to rte_eth_bond_8023ad_setup() is not passed uninitialized data that
> would result in either incorrect behavior or a failed sanity check.
>
> Fixes: 46fb43683679 ("bond: add mode 4")
>
> Signed-off-by: Eric Kinzie <ehkinzie@gmail.com>
> Signed-off-by: Stephen Hemminger <stephen@networkplumber.org>
> ---
...
>

Acked-by: Declan Doherty <declan.doherty@intel.com>

[PATCH 3/8] bond mode 4: do not ignore multicast

[Stephen Hemminger]

From: Eric Kinzie <ekinzie@brocade.com>

The bonding PMD in mode 4 puts all enslaved interfaces into promiscuous
mode in order to receive LACPDUs and must filter unwanted packets
after the traffic has been "collected".  Allow broadcast and multicast
through so that ARP and IPv6 neighbor discovery continue to work.

Fixes: 46fb43683679 ("bond: add mode 4")

Signed-off-by: Eric Kinzie <ehkinzie@gmail.com>
Signed-off-by: Stephen Hemminger <stephen@networkplumber.org>
---
 app/test/test_link_bonding_mode4.c     | 7 +++++--
 drivers/net/bonding/rte_eth_bond_pmd.c | 1 +
 2 files changed, 6 insertions(+), 2 deletions(-)

diff --git a/app/test/test_link_bonding_mode4.c b/app/test/test_link_bonding_mode4.c
index 713368d..31640cd 100644
--- a/app/test/test_link_bonding_mode4.c
+++ b/app/test/test_link_bonding_mode4.c
@@ -747,8 +747,11 @@ test_mode4_rx(void)
 	rte_eth_macaddr_get(test_params.bonded_port_id, &bonded_mac);
 	ether_addr_copy(&bonded_mac, &dst_mac);
 
-	/* Assert that dst address is not bonding address */
-	dst_mac.addr_bytes[0]++;
+	/* Assert that dst address is not bonding address.  Do not set the
+	 * least significant bit of the zero byte as this would create a
+	 * multicast address.
+	 */
+	dst_mac.addr_bytes[0] += 2;
 
 	/* First try with promiscuous mode enabled.
 	 * Add 2 packets to each slave. First with bonding MAC address, second with
diff --git a/drivers/net/bonding/rte_eth_bond_pmd.c b/drivers/net/bonding/rte_eth_bond_pmd.c
index 127b976..77582dd 100644
--- a/drivers/net/bonding/rte_eth_bond_pmd.c
+++ b/drivers/net/bonding/rte_eth_bond_pmd.c
@@ -170,6 +170,7 @@ bond_ethdev_rx_burst_8023ad(void *queue, struct rte_mbuf **bufs,
 			 * mode and packet address does not match. */
 			if (unlikely(hdr->ether_type == ether_type_slow_be ||
 				!collecting || (!promisc &&
+					!is_multicast_ether_addr(&hdr->d_addr) &&
 					!is_same_ether_addr(&bond_mac, &hdr->d_addr)))) {
 
 				if (hdr->ether_type == ether_type_slow_be) {
-- 
2.1.4

Re: [PATCH 3/8] bond mode 4: do not ignore multicast

[Declan Doherty]

On 04/12/15 17:14, Stephen Hemminger wrote:
> From: Eric Kinzie <ekinzie@brocade.com>
>
> The bonding PMD in mode 4 puts all enslaved interfaces into promiscuous
> mode in order to receive LACPDUs and must filter unwanted packets
> after the traffic has been "collected".  Allow broadcast and multicast
> through so that ARP and IPv6 neighbor discovery continue to work.
>
> Fixes: 46fb43683679 ("bond: add mode 4")
>
> Signed-off-by: Eric Kinzie <ehkinzie@gmail.com>
> Signed-off-by: Stephen Hemminger <stephen@networkplumber.org>
> ---
...
>


Acked-by: Declan Doherty <declan.doherty@intel.com>

[PATCH 4/8] bond mode 4: allow external state machine

[Stephen Hemminger]

From: Eric Kinzie <ekinzie@brocade.com>

Provide functions to allow an external 802.3ad state machine to transmit
and recieve LACPDUs and to set the collection/distribution flags on
slave interfaces.

Signed-off-by: Eric Kinzie <ehkinzie@gmail.com>
Signed-off-by: Stephen Hemminger <stephen@networkplumber.org>
---
 drivers/net/bonding/rte_eth_bond_8023ad.c         | 173 ++++++++++++++++++++++
 drivers/net/bonding/rte_eth_bond_8023ad.h         |  44 ++++++
 drivers/net/bonding/rte_eth_bond_8023ad_private.h |   2 +
 drivers/net/bonding/rte_eth_bond_version.map      |   6 +
 4 files changed, 225 insertions(+)

diff --git a/drivers/net/bonding/rte_eth_bond_8023ad.c b/drivers/net/bonding/rte_eth_bond_8023ad.c
index f2620b8..811623c 100644
--- a/drivers/net/bonding/rte_eth_bond_8023ad.c
+++ b/drivers/net/bonding/rte_eth_bond_8023ad.c
@@ -42,6 +42,8 @@
 
 #include "rte_eth_bond_private.h"
 
+static void bond_mode_8023ad_ext_periodic_cb(void *arg);
+
 #ifdef RTE_LIBRTE_BOND_DEBUG_8023AD
 #define MODE4_DEBUG(fmt, ...) RTE_LOG(DEBUG, PMD, "%6u [Port %u: %s] " fmt, \
 			bond_dbg_get_time_diff_ms(), slave_id, \
@@ -1014,6 +1016,7 @@ bond_mode_8023ad_conf_get(struct rte_eth_dev *dev,
 	conf->tx_period_ms = mode4->tx_period_timeout / ms_ticks;
 	conf->update_timeout_ms = mode4->update_timeout_us / 1000;
 	conf->rx_marker_period_ms = mode4->rx_marker_timeout / ms_ticks;
+	conf->slowrx_cb = mode4->slowrx_cb;
 }
 
 void
@@ -1035,8 +1038,11 @@ bond_mode_8023ad_setup(struct rte_eth_dev *dev,
 		conf->tx_period_ms = BOND_8023AD_TX_MACHINE_PERIOD_MS;
 		conf->rx_marker_period_ms = BOND_8023AD_RX_MARKER_PERIOD_MS;
 		conf->update_timeout_ms = BOND_MODE_8023AX_UPDATE_TIMEOUT_MS;
+		conf->slowrx_cb = NULL;
 	}
 
+	bond_mode_8023ad_stop(dev);
+
 	mode4->fast_periodic_timeout = conf->fast_periodic_ms * ms_ticks;
 	mode4->slow_periodic_timeout = conf->slow_periodic_ms * ms_ticks;
 	mode4->short_timeout = conf->short_timeout_ms * ms_ticks;
@@ -1045,6 +1051,10 @@ bond_mode_8023ad_setup(struct rte_eth_dev *dev,
 	mode4->tx_period_timeout = conf->tx_period_ms * ms_ticks;
 	mode4->rx_marker_timeout = conf->rx_marker_period_ms * ms_ticks;
 	mode4->update_timeout_us = conf->update_timeout_ms * 1000;
+	mode4->slowrx_cb = conf->slowrx_cb;
+
+	if (dev->data->dev_started)
+		bond_mode_8023ad_start(dev);
 }
 
 int
@@ -1062,6 +1072,13 @@ bond_mode_8023ad_enable(struct rte_eth_dev *bond_dev)
 int
 bond_mode_8023ad_start(struct rte_eth_dev *bond_dev)
 {
+	struct bond_dev_private *internals = bond_dev->data->dev_private;
+	struct mode8023ad_private *mode4 = &internals->mode4;
+
+	if (mode4->slowrx_cb)
+		return rte_eal_alarm_set(BOND_MODE_8023AX_UPDATE_TIMEOUT_MS * 1000,
+			&bond_mode_8023ad_ext_periodic_cb, bond_dev);
+
 	return rte_eal_alarm_set(BOND_MODE_8023AX_UPDATE_TIMEOUT_MS * 1000,
 			&bond_mode_8023ad_periodic_cb, bond_dev);
 }
@@ -1069,6 +1086,13 @@ bond_mode_8023ad_start(struct rte_eth_dev *bond_dev)
 void
 bond_mode_8023ad_stop(struct rte_eth_dev *bond_dev)
 {
+	struct bond_dev_private *internals = bond_dev->data->dev_private;
+	struct mode8023ad_private *mode4 = &internals->mode4;
+
+	if (mode4->slowrx_cb) {
+		rte_eal_alarm_cancel(&bond_mode_8023ad_ext_periodic_cb, bond_dev);
+		return;
+	}
 	rte_eal_alarm_cancel(&bond_mode_8023ad_periodic_cb, bond_dev);
 }
 
@@ -1215,3 +1239,152 @@ rte_eth_bond_8023ad_slave_info(uint8_t port_id, uint8_t slave_id,
 	info->agg_port_id = port->aggregator_port_id;
 	return 0;
 }
+
+int
+rte_eth_bond_8023ad_ext_collect(uint8_t port_id, uint8_t slave_id, int enabled)
+{
+	struct rte_eth_dev *bond_dev;
+	struct bond_dev_private *internals;
+	struct mode8023ad_private *mode4;
+	struct port *port;
+
+	if (rte_eth_bond_mode_get(port_id) != BONDING_MODE_8023AD)
+		return -EINVAL;
+
+	bond_dev = &rte_eth_devices[port_id];
+
+	if (!bond_dev->data->dev_started)
+		return -EINVAL;
+
+	internals = bond_dev->data->dev_private;
+	if (find_slave_by_id(internals->active_slaves,
+			internals->active_slave_count, slave_id) ==
+				internals->active_slave_count)
+		return -EINVAL;
+
+	mode4 = &internals->mode4;
+	if (mode4->slowrx_cb == NULL)
+		return -EINVAL;
+
+	port = &mode_8023ad_ports[slave_id];
+
+	if (enabled)
+		ACTOR_STATE_SET(port, COLLECTING);
+	else
+		ACTOR_STATE_CLR(port, COLLECTING);
+
+	return 0;
+}
+
+int
+rte_eth_bond_8023ad_ext_distrib(uint8_t port_id, uint8_t slave_id, int enabled)
+{
+	struct rte_eth_dev *bond_dev;
+	struct bond_dev_private *internals;
+	struct mode8023ad_private *mode4;
+	struct port *port;
+
+	if (rte_eth_bond_mode_get(port_id) != BONDING_MODE_8023AD)
+		return -EINVAL;
+
+	bond_dev = &rte_eth_devices[port_id];
+
+	if (!bond_dev->data->dev_started)
+		return -EINVAL;
+
+	internals = bond_dev->data->dev_private;
+	if (find_slave_by_id(internals->active_slaves,
+			internals->active_slave_count, slave_id) ==
+				internals->active_slave_count)
+		return -EINVAL;
+
+	mode4 = &internals->mode4;
+	if (mode4->slowrx_cb == NULL)
+		return -EINVAL;
+
+	port = &mode_8023ad_ports[slave_id];
+
+	if (enabled)
+		ACTOR_STATE_SET(port, DISTRIBUTING);
+	else
+		ACTOR_STATE_CLR(port, DISTRIBUTING);
+
+	return 0;
+}
+
+int
+rte_eth_bond_8023ad_ext_slowtx(uint8_t port_id, uint8_t slave_id,
+		struct rte_mbuf *lacp_pkt)
+{
+	struct rte_eth_dev *bond_dev;
+	struct bond_dev_private *internals;
+	struct mode8023ad_private *mode4;
+	struct port *port;
+
+	if (rte_eth_bond_mode_get(port_id) != BONDING_MODE_8023AD)
+		return -EINVAL;
+
+	bond_dev = &rte_eth_devices[port_id];
+
+	if (!bond_dev->data->dev_started)
+		return -EINVAL;
+
+	internals = bond_dev->data->dev_private;
+	if (find_slave_by_id(internals->active_slaves,
+			internals->active_slave_count, slave_id) ==
+				internals->active_slave_count)
+		return -EINVAL;
+
+	mode4 = &internals->mode4;
+	if (mode4->slowrx_cb == NULL)
+		return -EINVAL;
+
+	port = &mode_8023ad_ports[slave_id];
+
+	if (rte_pktmbuf_pkt_len(lacp_pkt) < sizeof(struct lacpdu_header))
+		return -EINVAL;
+
+	struct lacpdu_header *lacp;
+
+	/* only enqueue LACPDUs */
+	lacp = rte_pktmbuf_mtod(lacp_pkt, struct lacpdu_header *);
+	if (lacp->lacpdu.subtype != SLOW_SUBTYPE_LACP)
+		return -EINVAL;
+
+	MODE4_DEBUG("sending LACP frame\n");
+
+	return rte_ring_enqueue(port->tx_ring, lacp_pkt);
+}
+
+static void
+bond_mode_8023ad_ext_periodic_cb(void *arg)
+{
+	struct rte_eth_dev *bond_dev = arg;
+	struct bond_dev_private *internals = bond_dev->data->dev_private;
+	struct mode8023ad_private *mode4 = &internals->mode4;
+	struct port *port;
+	void *pkt = NULL;
+	uint16_t i, slave_id;
+
+	for (i = 0; i < internals->active_slave_count; i++) {
+		slave_id = internals->active_slaves[i];
+		port = &mode_8023ad_ports[slave_id];
+
+		if (rte_ring_dequeue(port->rx_ring, &pkt) == 0) {
+			struct rte_mbuf *lacp_pkt = pkt;
+			struct lacpdu_header *lacp;
+
+			lacp = rte_pktmbuf_mtod(lacp_pkt,
+						struct lacpdu_header *);
+			RTE_VERIFY(lacp->lacpdu.subtype == SLOW_SUBTYPE_LACP);
+
+			/* This is LACP frame so pass it to rx callback.
+			 * Callback is responsible for freeing mbuf.
+			 */
+			mode4->slowrx_cb(slave_id, lacp_pkt);
+		}
+	}
+
+	rte_eal_alarm_set(internals->mode4.update_timeout_us,
+			bond_mode_8023ad_ext_periodic_cb, arg);
+}
diff --git a/drivers/net/bonding/rte_eth_bond_8023ad.h b/drivers/net/bonding/rte_eth_bond_8023ad.h
index ebd0e93..8cfa3d3 100644
--- a/drivers/net/bonding/rte_eth_bond_8023ad.h
+++ b/drivers/net/bonding/rte_eth_bond_8023ad.h
@@ -64,6 +64,8 @@ extern "C" {
 #define MARKER_TLV_TYPE_INFO                0x01
 #define MARKER_TLV_TYPE_RESP                0x02
 
+typedef void (*rte_eth_bond_8023ad_ext_slowrx_fn)(uint8_t slave_id, struct rte_mbuf *lacp_pkt);
+
 enum rte_bond_8023ad_selection {
 	UNSELECTED,
 	STANDBY,
@@ -157,6 +159,7 @@ struct rte_eth_bond_8023ad_conf {
 	uint32_t tx_period_ms;
 	uint32_t rx_marker_period_ms;
 	uint32_t update_timeout_ms;
+	rte_eth_bond_8023ad_ext_slowrx_fn slowrx_cb;
 };
 
 struct rte_eth_bond_8023ad_slave_info {
@@ -219,4 +222,45 @@ rte_eth_bond_8023ad_slave_info(uint8_t port_id, uint8_t slave_id,
 }
 #endif
 
+/**
+ * Configure a slave port to start collecting.
+ *
+ * @param port_id	Bonding device id
+ * @param slave_id	Port id of valid slave.
+ * @param enabled	Non-zero when collection enabled.
+ * @return
+ *   0 - if ok
+ *   -EINVAL if slave is not valid.
+ */
+int
+rte_eth_bond_8023ad_ext_collect(uint8_t port_id, uint8_t slave_id, int enabled);
+
+/**
+ * Configure a slave port to start distributing.
+ *
+ * @param port_id	Bonding device id
+ * @param slave_id	Port id of valid slave.
+ * @param enabled	Non-zero when distribution enabled.
+ * @return
+ *   0 - if ok
+ *   -EINVAL if slave is not valid.
+ */
+int
+rte_eth_bond_8023ad_ext_distrib(uint8_t port_id, uint8_t slave_id, int enabled);
+
+/**
+ * LACPDU transmit path for external 802.3ad state machine.  Caller retains
+ * ownership of the packet on failure.
+ *
+ * @param port_id	Bonding device id
+ * @param slave_id	Port ID of valid slave device.
+ * @param lacp_pkt	mbuf containing LACPDU.
+ *
+ * @return
+ *   0 on success, negative value otherwise.
+ */
+int
+rte_eth_bond_8023ad_ext_slowtx(uint8_t port_id, uint8_t slave_id,
+		struct rte_mbuf *lacp_pkt);
+
 #endif /* RTE_ETH_BOND_8023AD_H_ */
diff --git a/drivers/net/bonding/rte_eth_bond_8023ad_private.h b/drivers/net/bonding/rte_eth_bond_8023ad_private.h
index 8adee70..9e15ece 100644
--- a/drivers/net/bonding/rte_eth_bond_8023ad_private.h
+++ b/drivers/net/bonding/rte_eth_bond_8023ad_private.h
@@ -173,6 +173,8 @@ struct mode8023ad_private {
 	uint64_t tx_period_timeout;
 	uint64_t rx_marker_timeout;
 	uint64_t update_timeout_us;
+	rte_eth_bond_8023ad_ext_slowrx_fn slowrx_cb;
+	uint8_t external_sm;
 };
 
 /**
diff --git a/drivers/net/bonding/rte_eth_bond_version.map b/drivers/net/bonding/rte_eth_bond_version.map
index 22bd920..33d73ff 100644
--- a/drivers/net/bonding/rte_eth_bond_version.map
+++ b/drivers/net/bonding/rte_eth_bond_version.map
@@ -27,3 +27,9 @@ DPDK_2.1 {
 	rte_eth_bond_free;
 
 } DPDK_2.0;
+
+DPDK_2.2 {
+	rte_eth_bond_8023ad_ext_collect;
+	rte_eth_bond_8023ad_ext_distrib;
+	rte_eth_bond_8023ad_ext_slowtx;
+} DPDK_2.1;
-- 
2.1.4

Re: [PATCH 4/8] bond mode 4: allow external state machine

[Declan Doherty]

On 04/12/15 17:14, Stephen Hemminger wrote:
> From: Eric Kinzie <ekinzie@brocade.com>
>
> Provide functions to allow an external 802.3ad state machine to transmit
> and recieve LACPDUs and to set the collection/distribution flags on
> slave interfaces.
>
> Signed-off-by: Eric Kinzie <ehkinzie@gmail.com>
> Signed-off-by: Stephen Hemminger <stephen@networkplumber.org>
> ---
...
>

Acked-by: Declan Doherty <declan.doherty@intel.com>

[PATCH 5/8] bond: active slaves with no primary

[Stephen Hemminger]

From: Eric Kinzie <ekinzie@brocade.com>

If the link state of a slave is "up" when added, it is added to the list
of active slaves but, even if it is the only slave, is not selected as
the primary interface.  Generally, handling of link state interrupts
selects an interface to be primary, but only if the active count is zero.
This change avoids the situation where there are active slaves but
no primary.

Signed-off-by: Eric Kinzie <ekinzie@brocade.com>
Signed-off-by: Stephen Hemminger <stephen@networkplumber.org>
---
 drivers/net/bonding/rte_eth_bond_api.c | 7 ++++++-
 1 file changed, 6 insertions(+), 1 deletion(-)

diff --git a/drivers/net/bonding/rte_eth_bond_api.c b/drivers/net/bonding/rte_eth_bond_api.c
index 8a000c8..630a461 100644
--- a/drivers/net/bonding/rte_eth_bond_api.c
+++ b/drivers/net/bonding/rte_eth_bond_api.c
@@ -427,8 +427,13 @@ __eth_bond_slave_add_lock_free(uint8_t bonded_port_id, uint8_t slave_port_id)
 	if (bonded_eth_dev->data->dev_started) {
 		rte_eth_link_get_nowait(slave_port_id, &link_props);
 
-		 if (link_props.link_status == 1)
+		 if (link_props.link_status == 1) {
+			if (internals->active_slave_count == 0 &&
+			    !internals->user_defined_primary_port)
+				bond_ethdev_primary_set(internals,
+							slave_port_id);
 			activate_slave(bonded_eth_dev, slave_port_id);
+		}
 	}
 	return 0;
 
-- 
2.1.4

Re: [PATCH 5/8] bond: active slaves with no primary

[Declan Doherty]

On 04/12/15 17:14, Stephen Hemminger wrote:
> From: Eric Kinzie <ekinzie@brocade.com>
>
> If the link state of a slave is "up" when added, it is added to the list
> of active slaves but, even if it is the only slave, is not selected as
> the primary interface.  Generally, handling of link state interrupts
> selects an interface to be primary, but only if the active count is zero.
> This change avoids the situation where there are active slaves but
> no primary.
>
> Signed-off-by: Eric Kinzie <ekinzie@brocade.com>
> Signed-off-by: Stephen Hemminger <stephen@networkplumber.org>
> ---
...
>

Acked-by: Declan Doherty <declan.doherty@intel.com>

[PATCH 6/8] bond: handle slaves with fewer queues than bonding device

[Stephen Hemminger]

From: Eric Kinzie <ekinzie@brocade.com>

In the event that the bonding device has a greater number of tx and/or rx
queues than the slave being added, track the queue limits of the slave.
On receive, ignore queue identifiers beyond what the slave interface
can support.  During transmit, pick a different queue id to use if the
intended queue is not available on the slave.

Signed-off-by: Eric Kinzie <ekinzie@brocade.com>
Signed-off-by: Stephen Hemminger <stephen@networkplumber.org>
---
 drivers/net/bonding/rte_eth_bond_api.c     |   6 +-
 drivers/net/bonding/rte_eth_bond_pmd.c     | 141 +++++++++++++++++++++++++----
 drivers/net/bonding/rte_eth_bond_private.h |   5 +-
 3 files changed, 129 insertions(+), 23 deletions(-)

diff --git a/drivers/net/bonding/rte_eth_bond_api.c b/drivers/net/bonding/rte_eth_bond_api.c
index 630a461..64058ff 100644
--- a/drivers/net/bonding/rte_eth_bond_api.c
+++ b/drivers/net/bonding/rte_eth_bond_api.c
@@ -340,11 +340,11 @@ __eth_bond_slave_add_lock_free(uint8_t bonded_port_id, uint8_t slave_port_id)
 
 	slave_eth_dev = &rte_eth_devices[slave_port_id];
 
-	/* Add slave details to bonded device */
-	slave_add(internals, slave_eth_dev);
-
 	rte_eth_dev_info_get(slave_port_id, &dev_info);
 
+	/* Add slave details to bonded device */
+	slave_add(internals, slave_eth_dev, &dev_info);
+
 	/* We need to store slaves reta_size to be able to synchronize RETA for all
 	 * slave devices even if its sizes are different.
 	 */
diff --git a/drivers/net/bonding/rte_eth_bond_pmd.c b/drivers/net/bonding/rte_eth_bond_pmd.c
index 77582dd..868e66b 100644
--- a/drivers/net/bonding/rte_eth_bond_pmd.c
+++ b/drivers/net/bonding/rte_eth_bond_pmd.c
@@ -76,6 +76,47 @@ get_vlan_offset(struct ether_hdr *eth_hdr, uint16_t *proto)
 	return vlan_offset;
 }
 
+static uint8_t
+bond_active_slaves_by_rxqid(struct bond_dev_private *internals, int queue_id,
+		uint8_t slaves[])
+{
+	struct bond_slave_details *slave_details;
+	uint8_t num_of_slaves;
+	uint8_t i = 0;
+
+	num_of_slaves = internals->active_slave_count;
+	memcpy(slaves, internals->active_slaves,
+			sizeof(internals->active_slaves[0]) * num_of_slaves);
+
+	if (num_of_slaves < 1 || internals->kvlist)
+		return num_of_slaves;
+
+	/* remove slaves that don't have a queue numbered "queue_id" */
+	while (i < num_of_slaves) {
+		slave_details = &internals->slaves[i];
+		if (unlikely(queue_id >= slave_details->nb_rx_queues)) {
+			slaves[i] = slaves[num_of_slaves-1];
+			num_of_slaves--;
+		} else
+			i++;
+	}
+
+	return num_of_slaves;
+}
+
+static int
+bond_slave_txqid(struct bond_dev_private *internals, uint8_t slave_id,
+		int queue_id)
+{
+	struct bond_slave_details *slave_details;
+
+	if (internals->kvlist)
+		return queue_id;
+
+	slave_details = &internals->slaves[slave_id];
+	return queue_id % slave_details->nb_tx_queues;
+}
+
 static uint16_t
 bond_ethdev_rx_burst(void *queue, struct rte_mbuf **bufs, uint16_t nb_pkts)
 {
@@ -83,6 +124,8 @@ bond_ethdev_rx_burst(void *queue, struct rte_mbuf **bufs, uint16_t nb_pkts)
 
 	uint16_t num_rx_slave = 0;
 	uint16_t num_rx_total = 0;
+	uint8_t slaves[RTE_MAX_ETHPORTS];
+	uint8_t num_of_slaves;
 
 	int i;
 
@@ -91,11 +134,13 @@ bond_ethdev_rx_burst(void *queue, struct rte_mbuf **bufs, uint16_t nb_pkts)
 
 	internals = bd_rx_q->dev_private;
 
+	num_of_slaves = bond_active_slaves_by_rxqid(internals, bd_rx_q->queue_id,
+						    slaves);
 
-	for (i = 0; i < internals->active_slave_count && nb_pkts; i++) {
+	for (i = 0; i < num_of_slaves && nb_pkts; i++) {
 		/* Offset of pointer to *bufs increases as packets are received
 		 * from other slaves */
-		num_rx_slave = rte_eth_rx_burst(internals->active_slaves[i],
+		num_rx_slave = rte_eth_rx_burst(slaves[i],
 				bd_rx_q->queue_id, bufs + num_rx_total, nb_pkts);
 		if (num_rx_slave) {
 			num_rx_total += num_rx_slave;
@@ -117,8 +162,13 @@ bond_ethdev_rx_burst_active_backup(void *queue, struct rte_mbuf **bufs,
 
 	internals = bd_rx_q->dev_private;
 
-	return rte_eth_rx_burst(internals->current_primary_port,
-			bd_rx_q->queue_id, bufs, nb_pkts);
+	uint8_t active_slave = internals->current_primary_port;
+	struct rte_eth_dev *dev = &rte_eth_devices[active_slave];
+
+	if (bd_rx_q->queue_id >= dev->data->nb_rx_queues)
+		return 0;
+
+	return rte_eth_rx_burst(active_slave, bd_rx_q->queue_id, bufs, nb_pkts);
 }
 
 static uint16_t
@@ -144,9 +194,9 @@ bond_ethdev_rx_burst_8023ad(void *queue, struct rte_mbuf **bufs,
 	rte_eth_macaddr_get(internals->port_id, &bond_mac);
 	/* Copy slave list to protect against slave up/down changes during tx
 	 * bursting */
-	slave_count = internals->active_slave_count;
-	memcpy(slaves, internals->active_slaves,
-			sizeof(internals->active_slaves[0]) * slave_count);
+
+	slave_count = bond_active_slaves_by_rxqid(internals, bd_rx_q->queue_id,
+						  slaves);
 
 	for (i = 0; i < slave_count && num_rx_total < nb_pkts; i++) {
 		j = num_rx_total;
@@ -401,6 +451,7 @@ bond_ethdev_tx_burst_round_robin(void *queue, struct rte_mbuf **bufs,
 
 	static int slave_idx = 0;
 	int i, cslave_idx = 0, tx_fail_total = 0;
+	int queue_id;
 
 	bd_tx_q = (struct bond_tx_queue *)queue;
 	internals = bd_tx_q->dev_private;
@@ -427,7 +478,9 @@ bond_ethdev_tx_burst_round_robin(void *queue, struct rte_mbuf **bufs,
 	/* Send packet burst on each slave device */
 	for (i = 0; i < num_of_slaves; i++) {
 		if (slave_nb_pkts[i] > 0) {
-			num_tx_slave = rte_eth_tx_burst(slaves[i], bd_tx_q->queue_id,
+			queue_id = bond_slave_txqid(internals, i,
+						    bd_tx_q->queue_id);
+			num_tx_slave = rte_eth_tx_burst(slaves[i], queue_id,
 					slave_bufs[i], slave_nb_pkts[i]);
 
 			/* if tx burst fails move packets to end of bufs */
@@ -453,14 +506,27 @@ bond_ethdev_tx_burst_active_backup(void *queue,
 {
 	struct bond_dev_private *internals;
 	struct bond_tx_queue *bd_tx_q;
+	int queue_id;
+	int i;
+	uint8_t num_of_slaves;
+	uint8_t slaves[RTE_MAX_ETHPORTS];
 
 	bd_tx_q = (struct bond_tx_queue *)queue;
 	internals = bd_tx_q->dev_private;
 
-	if (internals->active_slave_count < 1)
+	num_of_slaves = internals->active_slave_count;
+	memcpy(slaves, internals->active_slaves,
+			sizeof(internals->active_slaves[0]) * num_of_slaves);
+
+	if (num_of_slaves < 1)
 		return 0;
 
-	return rte_eth_tx_burst(internals->current_primary_port, bd_tx_q->queue_id,
+	for (i = 0; i < num_of_slaves; i++)
+		if (slaves[i] == internals->current_primary_port)
+			break;
+
+	queue_id = bond_slave_txqid(internals, i, bd_tx_q->queue_id);
+	return rte_eth_tx_burst(internals->current_primary_port, queue_id,
 			bufs, nb_pkts);
 }
 
@@ -696,6 +762,7 @@ bond_ethdev_tx_burst_tlb(void *queue, struct rte_mbuf **bufs, uint16_t nb_pkts)
 	struct ether_hdr *ether_hdr;
 	struct ether_addr primary_slave_addr;
 	struct ether_addr active_slave_addr;
+	int queue_id;
 
 	if (num_of_slaves < 1)
 		return num_tx_total;
@@ -725,7 +792,8 @@ bond_ethdev_tx_burst_tlb(void *queue, struct rte_mbuf **bufs, uint16_t nb_pkts)
 #endif
 		}
 
-		num_tx_total += rte_eth_tx_burst(slaves[i], bd_tx_q->queue_id,
+		queue_id = bond_slave_txqid(internals, i, bd_tx_q->queue_id);
+		num_tx_total += rte_eth_tx_burst(slaves[i], queue_id,
 				bufs + num_tx_total, nb_pkts - num_tx_total);
 
 		if (num_tx_total == nb_pkts)
@@ -903,6 +971,7 @@ bond_ethdev_tx_burst_balance(void *queue, struct rte_mbuf **bufs,
 	uint16_t num_tx_total = 0, num_tx_slave = 0, tx_fail_total = 0;
 
 	int i, op_slave_id;
+	int queue_id;
 
 	struct rte_mbuf *slave_bufs[RTE_MAX_ETHPORTS][nb_pkts];
 	uint16_t slave_nb_pkts[RTE_MAX_ETHPORTS] = { 0 };
@@ -931,7 +1000,9 @@ bond_ethdev_tx_burst_balance(void *queue, struct rte_mbuf **bufs,
 	/* Send packet burst on each slave device */
 	for (i = 0; i < num_of_slaves; i++) {
 		if (slave_nb_pkts[i] > 0) {
-			num_tx_slave = rte_eth_tx_burst(slaves[i], bd_tx_q->queue_id,
+			queue_id = bond_slave_txqid(internals, i,
+						    bd_tx_q->queue_id);
+			num_tx_slave = rte_eth_tx_burst(slaves[i], queue_id,
 					slave_bufs[i], slave_nb_pkts[i]);
 
 			/* if tx burst fails move packets to end of bufs */
@@ -977,6 +1048,8 @@ bond_ethdev_tx_burst_8023ad(void *queue, struct rte_mbuf **bufs,
 	/* Slow packets placed in each slave */
 	uint8_t slave_slow_nb_pkts[RTE_MAX_ETHPORTS] = { 0 };
 
+	int queue_id;
+
 	bd_tx_q = (struct bond_tx_queue *)queue;
 	internals = bd_tx_q->dev_private;
 
@@ -1022,7 +1095,8 @@ bond_ethdev_tx_burst_8023ad(void *queue, struct rte_mbuf **bufs,
 		if (slave_nb_pkts[i] == 0)
 			continue;
 
-		num_tx_slave = rte_eth_tx_burst(slaves[i], bd_tx_q->queue_id,
+		queue_id = bond_slave_txqid(internals, i, bd_tx_q->queue_id);
+		num_tx_slave = rte_eth_tx_burst(slaves[i], queue_id,
 				slave_bufs[i], slave_nb_pkts[i]);
 
 		/* If tx burst fails drop slow packets */
@@ -1057,6 +1131,7 @@ bond_ethdev_tx_burst_broadcast(void *queue, struct rte_mbuf **bufs,
 
 	int slave_tx_total[RTE_MAX_ETHPORTS];
 	int i, most_successful_tx_slave = -1;
+	int queue_id;
 
 	bd_tx_q = (struct bond_tx_queue *)queue;
 	internals = bd_tx_q->dev_private;
@@ -1076,7 +1151,8 @@ bond_ethdev_tx_burst_broadcast(void *queue, struct rte_mbuf **bufs,
 
 	/* Transmit burst on each active slave */
 	for (i = 0; i < num_of_slaves; i++) {
-		slave_tx_total[i] = rte_eth_tx_burst(slaves[i], bd_tx_q->queue_id,
+		queue_id = bond_slave_txqid(internals, i, bd_tx_q->queue_id);
+		slave_tx_total[i] = rte_eth_tx_burst(slaves[i], queue_id,
 					bufs, nb_pkts);
 
 		if (unlikely(slave_tx_total[i] < nb_pkts))
@@ -1298,9 +1374,22 @@ int
 slave_configure(struct rte_eth_dev *bonded_eth_dev,
 		struct rte_eth_dev *slave_eth_dev)
 {
+	struct bond_dev_private *internals;
 	struct bond_rx_queue *bd_rx_q;
 	struct bond_tx_queue *bd_tx_q;
+	int slave_id;
+
+	internals = bonded_eth_dev->data->dev_private;
 
+	for (slave_id = 0; slave_id < internals->slave_count; slave_id++)
+		if (internals->slaves[slave_id].port_id ==
+		    slave_eth_dev->data->port_id)
+			break;
+
+	RTE_VERIFY(slave_id != internals->slave_count);
+
+	uint16_t nb_rx_queues = internals->slaves[slave_id].nb_rx_queues;
+	uint16_t nb_tx_queues = internals->slaves[slave_id].nb_tx_queues;
 	int errval;
 	uint16_t q_id;
 
@@ -1331,8 +1420,7 @@ slave_configure(struct rte_eth_dev *bonded_eth_dev,
 
 	/* Configure device */
 	errval = rte_eth_dev_configure(slave_eth_dev->data->port_id,
-			bonded_eth_dev->data->nb_rx_queues,
-			bonded_eth_dev->data->nb_tx_queues,
+			nb_rx_queues, nb_tx_queues,
 			&(slave_eth_dev->data->dev_conf));
 	if (errval != 0) {
 		RTE_BOND_LOG(ERR, "Cannot configure slave device: port %u , err (%d)",
@@ -1343,7 +1431,7 @@ slave_configure(struct rte_eth_dev *bonded_eth_dev,
 	/* Setup Rx Queues */
 	/* Use existing queues, if any */
 	for (q_id = slave_eth_dev->data->nb_rx_queues;
-	     q_id < bonded_eth_dev->data->nb_rx_queues; q_id++) {
+	     q_id < nb_rx_queues ; q_id++) {
 		bd_rx_q = (struct bond_rx_queue *)bonded_eth_dev->data->rx_queues[q_id];
 
 		errval = rte_eth_rx_queue_setup(slave_eth_dev->data->port_id, q_id,
@@ -1361,7 +1449,7 @@ slave_configure(struct rte_eth_dev *bonded_eth_dev,
 	/* Setup Tx Queues */
 	/* Use existing queues, if any */
 	for (q_id = slave_eth_dev->data->nb_tx_queues;
-	     q_id < bonded_eth_dev->data->nb_tx_queues; q_id++) {
+	     q_id < nb_tx_queues ; q_id++) {
 		bd_tx_q = (struct bond_tx_queue *)bonded_eth_dev->data->tx_queues[q_id];
 
 		errval = rte_eth_tx_queue_setup(slave_eth_dev->data->port_id, q_id,
@@ -1440,7 +1528,8 @@ bond_ethdev_slave_link_status_change_monitor(void *cb_arg);
 
 void
 slave_add(struct bond_dev_private *internals,
-		struct rte_eth_dev *slave_eth_dev)
+		struct rte_eth_dev *slave_eth_dev,
+		const struct rte_eth_dev_info *slave_dev_info)
 {
 	struct bond_slave_details *slave_details =
 			&internals->slaves[internals->slave_count];
@@ -1448,6 +1537,20 @@ slave_add(struct bond_dev_private *internals,
 	slave_details->port_id = slave_eth_dev->data->port_id;
 	slave_details->last_link_status = 0;
 
+	uint16_t bond_nb_rx_queues =
+		rte_eth_devices[internals->port_id].data->nb_rx_queues;
+	uint16_t bond_nb_tx_queues =
+		rte_eth_devices[internals->port_id].data->nb_tx_queues;
+
+	slave_details->nb_rx_queues =
+		bond_nb_rx_queues > slave_dev_info->max_rx_queues
+		? slave_dev_info->max_rx_queues
+		: bond_nb_rx_queues;
+	slave_details->nb_tx_queues =
+		bond_nb_tx_queues > slave_dev_info->max_tx_queues
+		? slave_dev_info->max_tx_queues
+		: bond_nb_tx_queues;
+
 	/* If slave device doesn't support interrupts then we need to enabled
 	 * polling to monitor link status */
 	if (!(slave_eth_dev->data->dev_flags & RTE_PCI_DRV_INTR_LSC)) {
diff --git a/drivers/net/bonding/rte_eth_bond_private.h b/drivers/net/bonding/rte_eth_bond_private.h
index 6c47a29..02f6de1 100644
--- a/drivers/net/bonding/rte_eth_bond_private.h
+++ b/drivers/net/bonding/rte_eth_bond_private.h
@@ -101,6 +101,8 @@ struct bond_slave_details {
 	uint8_t link_status_poll_enabled;
 	uint8_t link_status_wait_to_complete;
 	uint8_t last_link_status;
+	uint16_t nb_rx_queues;
+	uint16_t nb_tx_queues;
 	/**< Port Id of slave eth_dev */
 	struct ether_addr persisted_mac_addr;
 
@@ -240,7 +242,8 @@ slave_remove(struct bond_dev_private *internals,
 
 void
 slave_add(struct bond_dev_private *internals,
-		struct rte_eth_dev *slave_eth_dev);
+		struct rte_eth_dev *slave_eth_dev,
+		const struct rte_eth_dev_info *slave_dev_info);
 
 uint16_t
 xmit_l2_hash(const struct rte_mbuf *buf, uint8_t slave_count);
-- 
2.1.4

Re: [PATCH 6/8] bond: handle slaves with fewer queues than bonding device

[Andriy Berestovskyy]

Hi guys,
I'm not quite sure if we can support less TX queues on a slave that easy:

> queue_id = bond_slave_txqid(internals, i, bd_tx_q->queue_id);
> num_tx_slave = rte_eth_tx_burst(slaves[i], queue_id,
>      slave_bufs[i], slave_nb_pkts[i]);

It seems that two different lcores might end up writing to the same
slave queue at the same time, isn't it?

Regards,
Andriy

On Fri, Dec 4, 2015 at 6:14 PM, Stephen Hemminger
<stephen@networkplumber.org> wrote:
> From: Eric Kinzie <ekinzie@brocade.com>
>
> In the event that the bonding device has a greater number of tx and/or rx
> queues than the slave being added, track the queue limits of the slave.
> On receive, ignore queue identifiers beyond what the slave interface
> can support.  During transmit, pick a different queue id to use if the
> intended queue is not available on the slave.
>
> Signed-off-by: Eric Kinzie <ekinzie@brocade.com>
> Signed-off-by: Stephen Hemminger <stephen@networkplumber.org>
> ---
>  drivers/net/bonding/rte_eth_bond_api.c     |   6 +-
>  drivers/net/bonding/rte_eth_bond_pmd.c     | 141 +++++++++++++++++++++++++----
>  drivers/net/bonding/rte_eth_bond_private.h |   5 +-
>  3 files changed, 129 insertions(+), 23 deletions(-)
>
> diff --git a/drivers/net/bonding/rte_eth_bond_api.c b/drivers/net/bonding/rte_eth_bond_api.c
> index 630a461..64058ff 100644
> --- a/drivers/net/bonding/rte_eth_bond_api.c
> +++ b/drivers/net/bonding/rte_eth_bond_api.c
> @@ -340,11 +340,11 @@ __eth_bond_slave_add_lock_free(uint8_t bonded_port_id, uint8_t slave_port_id)
>
>         slave_eth_dev = &rte_eth_devices[slave_port_id];
>
> -       /* Add slave details to bonded device */
> -       slave_add(internals, slave_eth_dev);
> -
>         rte_eth_dev_info_get(slave_port_id, &dev_info);
>
> +       /* Add slave details to bonded device */
> +       slave_add(internals, slave_eth_dev, &dev_info);
> +
>         /* We need to store slaves reta_size to be able to synchronize RETA for all
>          * slave devices even if its sizes are different.
>          */
> diff --git a/drivers/net/bonding/rte_eth_bond_pmd.c b/drivers/net/bonding/rte_eth_bond_pmd.c
> index 77582dd..868e66b 100644
> --- a/drivers/net/bonding/rte_eth_bond_pmd.c
> +++ b/drivers/net/bonding/rte_eth_bond_pmd.c
> @@ -76,6 +76,47 @@ get_vlan_offset(struct ether_hdr *eth_hdr, uint16_t *proto)
>         return vlan_offset;
>  }
>
> +static uint8_t
> +bond_active_slaves_by_rxqid(struct bond_dev_private *internals, int queue_id,
> +               uint8_t slaves[])
> +{
> +       struct bond_slave_details *slave_details;
> +       uint8_t num_of_slaves;
> +       uint8_t i = 0;
> +
> +       num_of_slaves = internals->active_slave_count;
> +       memcpy(slaves, internals->active_slaves,
> +                       sizeof(internals->active_slaves[0]) * num_of_slaves);
> +
> +       if (num_of_slaves < 1 || internals->kvlist)
> +               return num_of_slaves;
> +
> +       /* remove slaves that don't have a queue numbered "queue_id" */
> +       while (i < num_of_slaves) {
> +               slave_details = &internals->slaves[i];
> +               if (unlikely(queue_id >= slave_details->nb_rx_queues)) {
> +                       slaves[i] = slaves[num_of_slaves-1];
> +                       num_of_slaves--;
> +               } else
> +                       i++;
> +       }
> +
> +       return num_of_slaves;
> +}
> +
> +static int
> +bond_slave_txqid(struct bond_dev_private *internals, uint8_t slave_id,
> +               int queue_id)
> +{
> +       struct bond_slave_details *slave_details;
> +
> +       if (internals->kvlist)
> +               return queue_id;
> +
> +       slave_details = &internals->slaves[slave_id];
> +       return queue_id % slave_details->nb_tx_queues;
> +}
> +
>  static uint16_t
>  bond_ethdev_rx_burst(void *queue, struct rte_mbuf **bufs, uint16_t nb_pkts)
>  {
> @@ -83,6 +124,8 @@ bond_ethdev_rx_burst(void *queue, struct rte_mbuf **bufs, uint16_t nb_pkts)
>
>         uint16_t num_rx_slave = 0;
>         uint16_t num_rx_total = 0;
> +       uint8_t slaves[RTE_MAX_ETHPORTS];
> +       uint8_t num_of_slaves;
>
>         int i;
>
> @@ -91,11 +134,13 @@ bond_ethdev_rx_burst(void *queue, struct rte_mbuf **bufs, uint16_t nb_pkts)
>
>         internals = bd_rx_q->dev_private;
>
> +       num_of_slaves = bond_active_slaves_by_rxqid(internals, bd_rx_q->queue_id,
> +                                                   slaves);
>
> -       for (i = 0; i < internals->active_slave_count && nb_pkts; i++) {
> +       for (i = 0; i < num_of_slaves && nb_pkts; i++) {
>                 /* Offset of pointer to *bufs increases as packets are received
>                  * from other slaves */
> -               num_rx_slave = rte_eth_rx_burst(internals->active_slaves[i],
> +               num_rx_slave = rte_eth_rx_burst(slaves[i],
>                                 bd_rx_q->queue_id, bufs + num_rx_total, nb_pkts);
>                 if (num_rx_slave) {
>                         num_rx_total += num_rx_slave;
> @@ -117,8 +162,13 @@ bond_ethdev_rx_burst_active_backup(void *queue, struct rte_mbuf **bufs,
>
>         internals = bd_rx_q->dev_private;
>
> -       return rte_eth_rx_burst(internals->current_primary_port,
> -                       bd_rx_q->queue_id, bufs, nb_pkts);
> +       uint8_t active_slave = internals->current_primary_port;
> +       struct rte_eth_dev *dev = &rte_eth_devices[active_slave];
> +
> +       if (bd_rx_q->queue_id >= dev->data->nb_rx_queues)
> +               return 0;
> +
> +       return rte_eth_rx_burst(active_slave, bd_rx_q->queue_id, bufs, nb_pkts);
>  }
>
>  static uint16_t
> @@ -144,9 +194,9 @@ bond_ethdev_rx_burst_8023ad(void *queue, struct rte_mbuf **bufs,
>         rte_eth_macaddr_get(internals->port_id, &bond_mac);
>         /* Copy slave list to protect against slave up/down changes during tx
>          * bursting */
> -       slave_count = internals->active_slave_count;
> -       memcpy(slaves, internals->active_slaves,
> -                       sizeof(internals->active_slaves[0]) * slave_count);
> +
> +       slave_count = bond_active_slaves_by_rxqid(internals, bd_rx_q->queue_id,
> +                                                 slaves);
>
>         for (i = 0; i < slave_count && num_rx_total < nb_pkts; i++) {
>                 j = num_rx_total;
> @@ -401,6 +451,7 @@ bond_ethdev_tx_burst_round_robin(void *queue, struct rte_mbuf **bufs,
>
>         static int slave_idx = 0;
>         int i, cslave_idx = 0, tx_fail_total = 0;
> +       int queue_id;
>
>         bd_tx_q = (struct bond_tx_queue *)queue;
>         internals = bd_tx_q->dev_private;
> @@ -427,7 +478,9 @@ bond_ethdev_tx_burst_round_robin(void *queue, struct rte_mbuf **bufs,
>         /* Send packet burst on each slave device */
>         for (i = 0; i < num_of_slaves; i++) {
>                 if (slave_nb_pkts[i] > 0) {
> -                       num_tx_slave = rte_eth_tx_burst(slaves[i], bd_tx_q->queue_id,
> +                       queue_id = bond_slave_txqid(internals, i,
> +                                                   bd_tx_q->queue_id);
> +                       num_tx_slave = rte_eth_tx_burst(slaves[i], queue_id,
>                                         slave_bufs[i], slave_nb_pkts[i]);
>
>                         /* if tx burst fails move packets to end of bufs */
> @@ -453,14 +506,27 @@ bond_ethdev_tx_burst_active_backup(void *queue,
>  {
>         struct bond_dev_private *internals;
>         struct bond_tx_queue *bd_tx_q;
> +       int queue_id;
> +       int i;
> +       uint8_t num_of_slaves;
> +       uint8_t slaves[RTE_MAX_ETHPORTS];
>
>         bd_tx_q = (struct bond_tx_queue *)queue;
>         internals = bd_tx_q->dev_private;
>
> -       if (internals->active_slave_count < 1)
> +       num_of_slaves = internals->active_slave_count;
> +       memcpy(slaves, internals->active_slaves,
> +                       sizeof(internals->active_slaves[0]) * num_of_slaves);
> +
> +       if (num_of_slaves < 1)
>                 return 0;
>
> -       return rte_eth_tx_burst(internals->current_primary_port, bd_tx_q->queue_id,
> +       for (i = 0; i < num_of_slaves; i++)
> +               if (slaves[i] == internals->current_primary_port)
> +                       break;
> +
> +       queue_id = bond_slave_txqid(internals, i, bd_tx_q->queue_id);
> +       return rte_eth_tx_burst(internals->current_primary_port, queue_id,
>                         bufs, nb_pkts);
>  }
>
> @@ -696,6 +762,7 @@ bond_ethdev_tx_burst_tlb(void *queue, struct rte_mbuf **bufs, uint16_t nb_pkts)
>         struct ether_hdr *ether_hdr;
>         struct ether_addr primary_slave_addr;
>         struct ether_addr active_slave_addr;
> +       int queue_id;
>
>         if (num_of_slaves < 1)
>                 return num_tx_total;
> @@ -725,7 +792,8 @@ bond_ethdev_tx_burst_tlb(void *queue, struct rte_mbuf **bufs, uint16_t nb_pkts)
>  #endif
>                 }
>
> -               num_tx_total += rte_eth_tx_burst(slaves[i], bd_tx_q->queue_id,
> +               queue_id = bond_slave_txqid(internals, i, bd_tx_q->queue_id);
> +               num_tx_total += rte_eth_tx_burst(slaves[i], queue_id,
>                                 bufs + num_tx_total, nb_pkts - num_tx_total);
>
>                 if (num_tx_total == nb_pkts)
> @@ -903,6 +971,7 @@ bond_ethdev_tx_burst_balance(void *queue, struct rte_mbuf **bufs,
>         uint16_t num_tx_total = 0, num_tx_slave = 0, tx_fail_total = 0;
>
>         int i, op_slave_id;
> +       int queue_id;
>
>         struct rte_mbuf *slave_bufs[RTE_MAX_ETHPORTS][nb_pkts];
>         uint16_t slave_nb_pkts[RTE_MAX_ETHPORTS] = { 0 };
> @@ -931,7 +1000,9 @@ bond_ethdev_tx_burst_balance(void *queue, struct rte_mbuf **bufs,
>         /* Send packet burst on each slave device */
>         for (i = 0; i < num_of_slaves; i++) {
>                 if (slave_nb_pkts[i] > 0) {
> -                       num_tx_slave = rte_eth_tx_burst(slaves[i], bd_tx_q->queue_id,
> +                       queue_id = bond_slave_txqid(internals, i,
> +                                                   bd_tx_q->queue_id);
> +                       num_tx_slave = rte_eth_tx_burst(slaves[i], queue_id,
>                                         slave_bufs[i], slave_nb_pkts[i]);
>
>                         /* if tx burst fails move packets to end of bufs */
> @@ -977,6 +1048,8 @@ bond_ethdev_tx_burst_8023ad(void *queue, struct rte_mbuf **bufs,
>         /* Slow packets placed in each slave */
>         uint8_t slave_slow_nb_pkts[RTE_MAX_ETHPORTS] = { 0 };
>
> +       int queue_id;
> +
>         bd_tx_q = (struct bond_tx_queue *)queue;
>         internals = bd_tx_q->dev_private;
>
> @@ -1022,7 +1095,8 @@ bond_ethdev_tx_burst_8023ad(void *queue, struct rte_mbuf **bufs,
>                 if (slave_nb_pkts[i] == 0)
>                         continue;
>
> -               num_tx_slave = rte_eth_tx_burst(slaves[i], bd_tx_q->queue_id,
> +               queue_id = bond_slave_txqid(internals, i, bd_tx_q->queue_id);
> +               num_tx_slave = rte_eth_tx_burst(slaves[i], queue_id,
>                                 slave_bufs[i], slave_nb_pkts[i]);
>
>                 /* If tx burst fails drop slow packets */
> @@ -1057,6 +1131,7 @@ bond_ethdev_tx_burst_broadcast(void *queue, struct rte_mbuf **bufs,
>
>         int slave_tx_total[RTE_MAX_ETHPORTS];
>         int i, most_successful_tx_slave = -1;
> +       int queue_id;
>
>         bd_tx_q = (struct bond_tx_queue *)queue;
>         internals = bd_tx_q->dev_private;
> @@ -1076,7 +1151,8 @@ bond_ethdev_tx_burst_broadcast(void *queue, struct rte_mbuf **bufs,
>
>         /* Transmit burst on each active slave */
>         for (i = 0; i < num_of_slaves; i++) {
> -               slave_tx_total[i] = rte_eth_tx_burst(slaves[i], bd_tx_q->queue_id,
> +               queue_id = bond_slave_txqid(internals, i, bd_tx_q->queue_id);
> +               slave_tx_total[i] = rte_eth_tx_burst(slaves[i], queue_id,
>                                         bufs, nb_pkts);
>
>                 if (unlikely(slave_tx_total[i] < nb_pkts))
> @@ -1298,9 +1374,22 @@ int
>  slave_configure(struct rte_eth_dev *bonded_eth_dev,
>                 struct rte_eth_dev *slave_eth_dev)
>  {
> +       struct bond_dev_private *internals;
>         struct bond_rx_queue *bd_rx_q;
>         struct bond_tx_queue *bd_tx_q;
> +       int slave_id;
> +
> +       internals = bonded_eth_dev->data->dev_private;
>
> +       for (slave_id = 0; slave_id < internals->slave_count; slave_id++)
> +               if (internals->slaves[slave_id].port_id ==
> +                   slave_eth_dev->data->port_id)
> +                       break;
> +
> +       RTE_VERIFY(slave_id != internals->slave_count);
> +
> +       uint16_t nb_rx_queues = internals->slaves[slave_id].nb_rx_queues;
> +       uint16_t nb_tx_queues = internals->slaves[slave_id].nb_tx_queues;
>         int errval;
>         uint16_t q_id;
>
> @@ -1331,8 +1420,7 @@ slave_configure(struct rte_eth_dev *bonded_eth_dev,
>
>         /* Configure device */
>         errval = rte_eth_dev_configure(slave_eth_dev->data->port_id,
> -                       bonded_eth_dev->data->nb_rx_queues,
> -                       bonded_eth_dev->data->nb_tx_queues,
> +                       nb_rx_queues, nb_tx_queues,
>                         &(slave_eth_dev->data->dev_conf));
>         if (errval != 0) {
>                 RTE_BOND_LOG(ERR, "Cannot configure slave device: port %u , err (%d)",
> @@ -1343,7 +1431,7 @@ slave_configure(struct rte_eth_dev *bonded_eth_dev,
>         /* Setup Rx Queues */
>         /* Use existing queues, if any */
>         for (q_id = slave_eth_dev->data->nb_rx_queues;
> -            q_id < bonded_eth_dev->data->nb_rx_queues; q_id++) {
> +            q_id < nb_rx_queues ; q_id++) {
>                 bd_rx_q = (struct bond_rx_queue *)bonded_eth_dev->data->rx_queues[q_id];
>
>                 errval = rte_eth_rx_queue_setup(slave_eth_dev->data->port_id, q_id,
> @@ -1361,7 +1449,7 @@ slave_configure(struct rte_eth_dev *bonded_eth_dev,
>         /* Setup Tx Queues */
>         /* Use existing queues, if any */
>         for (q_id = slave_eth_dev->data->nb_tx_queues;
> -            q_id < bonded_eth_dev->data->nb_tx_queues; q_id++) {
> +            q_id < nb_tx_queues ; q_id++) {
>                 bd_tx_q = (struct bond_tx_queue *)bonded_eth_dev->data->tx_queues[q_id];
>
>                 errval = rte_eth_tx_queue_setup(slave_eth_dev->data->port_id, q_id,
> @@ -1440,7 +1528,8 @@ bond_ethdev_slave_link_status_change_monitor(void *cb_arg);
>
>  void
>  slave_add(struct bond_dev_private *internals,
> -               struct rte_eth_dev *slave_eth_dev)
> +               struct rte_eth_dev *slave_eth_dev,
> +               const struct rte_eth_dev_info *slave_dev_info)
>  {
>         struct bond_slave_details *slave_details =
>                         &internals->slaves[internals->slave_count];
> @@ -1448,6 +1537,20 @@ slave_add(struct bond_dev_private *internals,
>         slave_details->port_id = slave_eth_dev->data->port_id;
>         slave_details->last_link_status = 0;
>
> +       uint16_t bond_nb_rx_queues =
> +               rte_eth_devices[internals->port_id].data->nb_rx_queues;
> +       uint16_t bond_nb_tx_queues =
> +               rte_eth_devices[internals->port_id].data->nb_tx_queues;
> +
> +       slave_details->nb_rx_queues =
> +               bond_nb_rx_queues > slave_dev_info->max_rx_queues
> +               ? slave_dev_info->max_rx_queues
> +               : bond_nb_rx_queues;
> +       slave_details->nb_tx_queues =
> +               bond_nb_tx_queues > slave_dev_info->max_tx_queues
> +               ? slave_dev_info->max_tx_queues
> +               : bond_nb_tx_queues;
> +
>         /* If slave device doesn't support interrupts then we need to enabled
>          * polling to monitor link status */
>         if (!(slave_eth_dev->data->dev_flags & RTE_PCI_DRV_INTR_LSC)) {
> diff --git a/drivers/net/bonding/rte_eth_bond_private.h b/drivers/net/bonding/rte_eth_bond_private.h
> index 6c47a29..02f6de1 100644
> --- a/drivers/net/bonding/rte_eth_bond_private.h
> +++ b/drivers/net/bonding/rte_eth_bond_private.h
> @@ -101,6 +101,8 @@ struct bond_slave_details {
>         uint8_t link_status_poll_enabled;
>         uint8_t link_status_wait_to_complete;
>         uint8_t last_link_status;
> +       uint16_t nb_rx_queues;
> +       uint16_t nb_tx_queues;
>         /**< Port Id of slave eth_dev */
>         struct ether_addr persisted_mac_addr;
>
> @@ -240,7 +242,8 @@ slave_remove(struct bond_dev_private *internals,
>
>  void
>  slave_add(struct bond_dev_private *internals,
> -               struct rte_eth_dev *slave_eth_dev);
> +               struct rte_eth_dev *slave_eth_dev,
> +               const struct rte_eth_dev_info *slave_dev_info);
>
>  uint16_t
>  xmit_l2_hash(const struct rte_mbuf *buf, uint8_t slave_count);
> --
> 2.1.4
>



-- 
Andriy Berestovskyy

Re: [PATCH 6/8] bond: handle slaves with fewer queues than bonding device

[Eric Kinzie]

On Fri Dec 04 19:36:09 +0100 2015, Andriy Berestovskyy wrote:
> Hi guys,
> I'm not quite sure if we can support less TX queues on a slave that easy:
> 
> > queue_id = bond_slave_txqid(internals, i, bd_tx_q->queue_id);
> > num_tx_slave = rte_eth_tx_burst(slaves[i], queue_id,
> >      slave_bufs[i], slave_nb_pkts[i]);
> 
> It seems that two different lcores might end up writing to the same
> slave queue at the same time, isn't it?
> 
> Regards,
> Andriy

Andriy, I think you're probably right about this.  Perhaps it should
instead refuse to add or refuse to activate a slave with too few
tx queues.  Could probably fix this with another layer of buffering
so that an lcore with a valid tx queue could pick up the mbufs later,
but this doesn't seem very appealing.

Eric


> On Fri, Dec 4, 2015 at 6:14 PM, Stephen Hemminger
> <stephen@networkplumber.org> wrote:
> > From: Eric Kinzie <ekinzie@brocade.com>
> >
> > In the event that the bonding device has a greater number of tx and/or rx
> > queues than the slave being added, track the queue limits of the slave.
> > On receive, ignore queue identifiers beyond what the slave interface
> > can support.  During transmit, pick a different queue id to use if the
> > intended queue is not available on the slave.
> >
> > Signed-off-by: Eric Kinzie <ekinzie@brocade.com>
> > Signed-off-by: Stephen Hemminger <stephen@networkplumber.org>
> > ---
> >  drivers/net/bonding/rte_eth_bond_api.c     |   6 +-
> >  drivers/net/bonding/rte_eth_bond_pmd.c     | 141 +++++++++++++++++++++++++----
> >  drivers/net/bonding/rte_eth_bond_private.h |   5 +-
> >  3 files changed, 129 insertions(+), 23 deletions(-)
> >
> > diff --git a/drivers/net/bonding/rte_eth_bond_api.c b/drivers/net/bonding/rte_eth_bond_api.c
> > index 630a461..64058ff 100644
> > --- a/drivers/net/bonding/rte_eth_bond_api.c
> > +++ b/drivers/net/bonding/rte_eth_bond_api.c
> > @@ -340,11 +340,11 @@ __eth_bond_slave_add_lock_free(uint8_t bonded_port_id, uint8_t slave_port_id)
> >
> >         slave_eth_dev = &rte_eth_devices[slave_port_id];
> >
> > -       /* Add slave details to bonded device */
> > -       slave_add(internals, slave_eth_dev);
> > -
> >         rte_eth_dev_info_get(slave_port_id, &dev_info);
> >
> > +       /* Add slave details to bonded device */
> > +       slave_add(internals, slave_eth_dev, &dev_info);
> > +
> >         /* We need to store slaves reta_size to be able to synchronize RETA for all
> >          * slave devices even if its sizes are different.
> >          */
> > diff --git a/drivers/net/bonding/rte_eth_bond_pmd.c b/drivers/net/bonding/rte_eth_bond_pmd.c
> > index 77582dd..868e66b 100644
> > --- a/drivers/net/bonding/rte_eth_bond_pmd.c
> > +++ b/drivers/net/bonding/rte_eth_bond_pmd.c
> > @@ -76,6 +76,47 @@ get_vlan_offset(struct ether_hdr *eth_hdr, uint16_t *proto)
> >         return vlan_offset;
> >  }
> >
> > +static uint8_t
> > +bond_active_slaves_by_rxqid(struct bond_dev_private *internals, int queue_id,
> > +               uint8_t slaves[])
> > +{
> > +       struct bond_slave_details *slave_details;
> > +       uint8_t num_of_slaves;
> > +       uint8_t i = 0;
> > +
> > +       num_of_slaves = internals->active_slave_count;
> > +       memcpy(slaves, internals->active_slaves,
> > +                       sizeof(internals->active_slaves[0]) * num_of_slaves);
> > +
> > +       if (num_of_slaves < 1 || internals->kvlist)
> > +               return num_of_slaves;
> > +
> > +       /* remove slaves that don't have a queue numbered "queue_id" */
> > +       while (i < num_of_slaves) {
> > +               slave_details = &internals->slaves[i];
> > +               if (unlikely(queue_id >= slave_details->nb_rx_queues)) {
> > +                       slaves[i] = slaves[num_of_slaves-1];
> > +                       num_of_slaves--;
> > +               } else
> > +                       i++;
> > +       }
> > +
> > +       return num_of_slaves;
> > +}
> > +
> > +static int
> > +bond_slave_txqid(struct bond_dev_private *internals, uint8_t slave_id,
> > +               int queue_id)
> > +{
> > +       struct bond_slave_details *slave_details;
> > +
> > +       if (internals->kvlist)
> > +               return queue_id;
> > +
> > +       slave_details = &internals->slaves[slave_id];
> > +       return queue_id % slave_details->nb_tx_queues;
> > +}
> > +
> >  static uint16_t
> >  bond_ethdev_rx_burst(void *queue, struct rte_mbuf **bufs, uint16_t nb_pkts)
> >  {
> > @@ -83,6 +124,8 @@ bond_ethdev_rx_burst(void *queue, struct rte_mbuf **bufs, uint16_t nb_pkts)
> >
> >         uint16_t num_rx_slave = 0;
> >         uint16_t num_rx_total = 0;
> > +       uint8_t slaves[RTE_MAX_ETHPORTS];
> > +       uint8_t num_of_slaves;
> >
> >         int i;
> >
> > @@ -91,11 +134,13 @@ bond_ethdev_rx_burst(void *queue, struct rte_mbuf **bufs, uint16_t nb_pkts)
> >
> >         internals = bd_rx_q->dev_private;
> >
> > +       num_of_slaves = bond_active_slaves_by_rxqid(internals, bd_rx_q->queue_id,
> > +                                                   slaves);
> >
> > -       for (i = 0; i < internals->active_slave_count && nb_pkts; i++) {
> > +       for (i = 0; i < num_of_slaves && nb_pkts; i++) {
> >                 /* Offset of pointer to *bufs increases as packets are received
> >                  * from other slaves */
> > -               num_rx_slave = rte_eth_rx_burst(internals->active_slaves[i],
> > +               num_rx_slave = rte_eth_rx_burst(slaves[i],
> >                                 bd_rx_q->queue_id, bufs + num_rx_total, nb_pkts);
> >                 if (num_rx_slave) {
> >                         num_rx_total += num_rx_slave;
> > @@ -117,8 +162,13 @@ bond_ethdev_rx_burst_active_backup(void *queue, struct rte_mbuf **bufs,
> >
> >         internals = bd_rx_q->dev_private;
> >
> > -       return rte_eth_rx_burst(internals->current_primary_port,
> > -                       bd_rx_q->queue_id, bufs, nb_pkts);
> > +       uint8_t active_slave = internals->current_primary_port;
> > +       struct rte_eth_dev *dev = &rte_eth_devices[active_slave];
> > +
> > +       if (bd_rx_q->queue_id >= dev->data->nb_rx_queues)
> > +               return 0;
> > +
> > +       return rte_eth_rx_burst(active_slave, bd_rx_q->queue_id, bufs, nb_pkts);
> >  }
> >
> >  static uint16_t
> > @@ -144,9 +194,9 @@ bond_ethdev_rx_burst_8023ad(void *queue, struct rte_mbuf **bufs,
> >         rte_eth_macaddr_get(internals->port_id, &bond_mac);
> >         /* Copy slave list to protect against slave up/down changes during tx
> >          * bursting */
> > -       slave_count = internals->active_slave_count;
> > -       memcpy(slaves, internals->active_slaves,
> > -                       sizeof(internals->active_slaves[0]) * slave_count);
> > +
> > +       slave_count = bond_active_slaves_by_rxqid(internals, bd_rx_q->queue_id,
> > +                                                 slaves);
> >
> >         for (i = 0; i < slave_count && num_rx_total < nb_pkts; i++) {
> >                 j = num_rx_total;
> > @@ -401,6 +451,7 @@ bond_ethdev_tx_burst_round_robin(void *queue, struct rte_mbuf **bufs,
> >
> >         static int slave_idx = 0;
> >         int i, cslave_idx = 0, tx_fail_total = 0;
> > +       int queue_id;
> >
> >         bd_tx_q = (struct bond_tx_queue *)queue;
> >         internals = bd_tx_q->dev_private;
> > @@ -427,7 +478,9 @@ bond_ethdev_tx_burst_round_robin(void *queue, struct rte_mbuf **bufs,
> >         /* Send packet burst on each slave device */
> >         for (i = 0; i < num_of_slaves; i++) {
> >                 if (slave_nb_pkts[i] > 0) {
> > -                       num_tx_slave = rte_eth_tx_burst(slaves[i], bd_tx_q->queue_id,
> > +                       queue_id = bond_slave_txqid(internals, i,
> > +                                                   bd_tx_q->queue_id);
> > +                       num_tx_slave = rte_eth_tx_burst(slaves[i], queue_id,
> >                                         slave_bufs[i], slave_nb_pkts[i]);
> >
> >                         /* if tx burst fails move packets to end of bufs */
> > @@ -453,14 +506,27 @@ bond_ethdev_tx_burst_active_backup(void *queue,
> >  {
> >         struct bond_dev_private *internals;
> >         struct bond_tx_queue *bd_tx_q;
> > +       int queue_id;
> > +       int i;
> > +       uint8_t num_of_slaves;
> > +       uint8_t slaves[RTE_MAX_ETHPORTS];
> >
> >         bd_tx_q = (struct bond_tx_queue *)queue;
> >         internals = bd_tx_q->dev_private;
> >
> > -       if (internals->active_slave_count < 1)
> > +       num_of_slaves = internals->active_slave_count;
> > +       memcpy(slaves, internals->active_slaves,
> > +                       sizeof(internals->active_slaves[0]) * num_of_slaves);
> > +
> > +       if (num_of_slaves < 1)
> >                 return 0;
> >
> > -       return rte_eth_tx_burst(internals->current_primary_port, bd_tx_q->queue_id,
> > +       for (i = 0; i < num_of_slaves; i++)
> > +               if (slaves[i] == internals->current_primary_port)
> > +                       break;
> > +
> > +       queue_id = bond_slave_txqid(internals, i, bd_tx_q->queue_id);
> > +       return rte_eth_tx_burst(internals->current_primary_port, queue_id,
> >                         bufs, nb_pkts);
> >  }
> >
> > @@ -696,6 +762,7 @@ bond_ethdev_tx_burst_tlb(void *queue, struct rte_mbuf **bufs, uint16_t nb_pkts)
> >         struct ether_hdr *ether_hdr;
> >         struct ether_addr primary_slave_addr;
> >         struct ether_addr active_slave_addr;
> > +       int queue_id;
> >
> >         if (num_of_slaves < 1)
> >                 return num_tx_total;
> > @@ -725,7 +792,8 @@ bond_ethdev_tx_burst_tlb(void *queue, struct rte_mbuf **bufs, uint16_t nb_pkts)
> >  #endif
> >                 }
> >
> > -               num_tx_total += rte_eth_tx_burst(slaves[i], bd_tx_q->queue_id,
> > +               queue_id = bond_slave_txqid(internals, i, bd_tx_q->queue_id);
> > +               num_tx_total += rte_eth_tx_burst(slaves[i], queue_id,
> >                                 bufs + num_tx_total, nb_pkts - num_tx_total);
> >
> >                 if (num_tx_total == nb_pkts)
> > @@ -903,6 +971,7 @@ bond_ethdev_tx_burst_balance(void *queue, struct rte_mbuf **bufs,
> >         uint16_t num_tx_total = 0, num_tx_slave = 0, tx_fail_total = 0;
> >
> >         int i, op_slave_id;
> > +       int queue_id;
> >
> >         struct rte_mbuf *slave_bufs[RTE_MAX_ETHPORTS][nb_pkts];
> >         uint16_t slave_nb_pkts[RTE_MAX_ETHPORTS] = { 0 };
> > @@ -931,7 +1000,9 @@ bond_ethdev_tx_burst_balance(void *queue, struct rte_mbuf **bufs,
> >         /* Send packet burst on each slave device */
> >         for (i = 0; i < num_of_slaves; i++) {
> >                 if (slave_nb_pkts[i] > 0) {
> > -                       num_tx_slave = rte_eth_tx_burst(slaves[i], bd_tx_q->queue_id,
> > +                       queue_id = bond_slave_txqid(internals, i,
> > +                                                   bd_tx_q->queue_id);
> > +                       num_tx_slave = rte_eth_tx_burst(slaves[i], queue_id,
> >                                         slave_bufs[i], slave_nb_pkts[i]);
> >
> >                         /* if tx burst fails move packets to end of bufs */
> > @@ -977,6 +1048,8 @@ bond_ethdev_tx_burst_8023ad(void *queue, struct rte_mbuf **bufs,
> >         /* Slow packets placed in each slave */
> >         uint8_t slave_slow_nb_pkts[RTE_MAX_ETHPORTS] = { 0 };
> >
> > +       int queue_id;
> > +
> >         bd_tx_q = (struct bond_tx_queue *)queue;
> >         internals = bd_tx_q->dev_private;
> >
> > @@ -1022,7 +1095,8 @@ bond_ethdev_tx_burst_8023ad(void *queue, struct rte_mbuf **bufs,
> >                 if (slave_nb_pkts[i] == 0)
> >                         continue;
> >
> > -               num_tx_slave = rte_eth_tx_burst(slaves[i], bd_tx_q->queue_id,
> > +               queue_id = bond_slave_txqid(internals, i, bd_tx_q->queue_id);
> > +               num_tx_slave = rte_eth_tx_burst(slaves[i], queue_id,
> >                                 slave_bufs[i], slave_nb_pkts[i]);
> >
> >                 /* If tx burst fails drop slow packets */
> > @@ -1057,6 +1131,7 @@ bond_ethdev_tx_burst_broadcast(void *queue, struct rte_mbuf **bufs,
> >
> >         int slave_tx_total[RTE_MAX_ETHPORTS];
> >         int i, most_successful_tx_slave = -1;
> > +       int queue_id;
> >
> >         bd_tx_q = (struct bond_tx_queue *)queue;
> >         internals = bd_tx_q->dev_private;
> > @@ -1076,7 +1151,8 @@ bond_ethdev_tx_burst_broadcast(void *queue, struct rte_mbuf **bufs,
> >
> >         /* Transmit burst on each active slave */
> >         for (i = 0; i < num_of_slaves; i++) {
> > -               slave_tx_total[i] = rte_eth_tx_burst(slaves[i], bd_tx_q->queue_id,
> > +               queue_id = bond_slave_txqid(internals, i, bd_tx_q->queue_id);
> > +               slave_tx_total[i] = rte_eth_tx_burst(slaves[i], queue_id,
> >                                         bufs, nb_pkts);
> >
> >                 if (unlikely(slave_tx_total[i] < nb_pkts))
> > @@ -1298,9 +1374,22 @@ int
> >  slave_configure(struct rte_eth_dev *bonded_eth_dev,
> >                 struct rte_eth_dev *slave_eth_dev)
> >  {
> > +       struct bond_dev_private *internals;
> >         struct bond_rx_queue *bd_rx_q;
> >         struct bond_tx_queue *bd_tx_q;
> > +       int slave_id;
> > +
> > +       internals = bonded_eth_dev->data->dev_private;
> >
> > +       for (slave_id = 0; slave_id < internals->slave_count; slave_id++)
> > +               if (internals->slaves[slave_id].port_id ==
> > +                   slave_eth_dev->data->port_id)
> > +                       break;
> > +
> > +       RTE_VERIFY(slave_id != internals->slave_count);
> > +
> > +       uint16_t nb_rx_queues = internals->slaves[slave_id].nb_rx_queues;
> > +       uint16_t nb_tx_queues = internals->slaves[slave_id].nb_tx_queues;
> >         int errval;
> >         uint16_t q_id;
> >
> > @@ -1331,8 +1420,7 @@ slave_configure(struct rte_eth_dev *bonded_eth_dev,
> >
> >         /* Configure device */
> >         errval = rte_eth_dev_configure(slave_eth_dev->data->port_id,
> > -                       bonded_eth_dev->data->nb_rx_queues,
> > -                       bonded_eth_dev->data->nb_tx_queues,
> > +                       nb_rx_queues, nb_tx_queues,
> >                         &(slave_eth_dev->data->dev_conf));
> >         if (errval != 0) {
> >                 RTE_BOND_LOG(ERR, "Cannot configure slave device: port %u , err (%d)",
> > @@ -1343,7 +1431,7 @@ slave_configure(struct rte_eth_dev *bonded_eth_dev,
> >         /* Setup Rx Queues */
> >         /* Use existing queues, if any */
> >         for (q_id = slave_eth_dev->data->nb_rx_queues;
> > -            q_id < bonded_eth_dev->data->nb_rx_queues; q_id++) {
> > +            q_id < nb_rx_queues ; q_id++) {
> >                 bd_rx_q = (struct bond_rx_queue *)bonded_eth_dev->data->rx_queues[q_id];
> >
> >                 errval = rte_eth_rx_queue_setup(slave_eth_dev->data->port_id, q_id,
> > @@ -1361,7 +1449,7 @@ slave_configure(struct rte_eth_dev *bonded_eth_dev,
> >         /* Setup Tx Queues */
> >         /* Use existing queues, if any */
> >         for (q_id = slave_eth_dev->data->nb_tx_queues;
> > -            q_id < bonded_eth_dev->data->nb_tx_queues; q_id++) {
> > +            q_id < nb_tx_queues ; q_id++) {
> >                 bd_tx_q = (struct bond_tx_queue *)bonded_eth_dev->data->tx_queues[q_id];
> >
> >                 errval = rte_eth_tx_queue_setup(slave_eth_dev->data->port_id, q_id,
> > @@ -1440,7 +1528,8 @@ bond_ethdev_slave_link_status_change_monitor(void *cb_arg);
> >
> >  void
> >  slave_add(struct bond_dev_private *internals,
> > -               struct rte_eth_dev *slave_eth_dev)
> > +               struct rte_eth_dev *slave_eth_dev,
> > +               const struct rte_eth_dev_info *slave_dev_info)
> >  {
> >         struct bond_slave_details *slave_details =
> >                         &internals->slaves[internals->slave_count];
> > @@ -1448,6 +1537,20 @@ slave_add(struct bond_dev_private *internals,
> >         slave_details->port_id = slave_eth_dev->data->port_id;
> >         slave_details->last_link_status = 0;
> >
> > +       uint16_t bond_nb_rx_queues =
> > +               rte_eth_devices[internals->port_id].data->nb_rx_queues;
> > +       uint16_t bond_nb_tx_queues =
> > +               rte_eth_devices[internals->port_id].data->nb_tx_queues;
> > +
> > +       slave_details->nb_rx_queues =
> > +               bond_nb_rx_queues > slave_dev_info->max_rx_queues
> > +               ? slave_dev_info->max_rx_queues
> > +               : bond_nb_rx_queues;
> > +       slave_details->nb_tx_queues =
> > +               bond_nb_tx_queues > slave_dev_info->max_tx_queues
> > +               ? slave_dev_info->max_tx_queues
> > +               : bond_nb_tx_queues;
> > +
> >         /* If slave device doesn't support interrupts then we need to enabled
> >          * polling to monitor link status */
> >         if (!(slave_eth_dev->data->dev_flags & RTE_PCI_DRV_INTR_LSC)) {
> > diff --git a/drivers/net/bonding/rte_eth_bond_private.h b/drivers/net/bonding/rte_eth_bond_private.h
> > index 6c47a29..02f6de1 100644
> > --- a/drivers/net/bonding/rte_eth_bond_private.h
> > +++ b/drivers/net/bonding/rte_eth_bond_private.h
> > @@ -101,6 +101,8 @@ struct bond_slave_details {
> >         uint8_t link_status_poll_enabled;
> >         uint8_t link_status_wait_to_complete;
> >         uint8_t last_link_status;
> > +       uint16_t nb_rx_queues;
> > +       uint16_t nb_tx_queues;
> >         /**< Port Id of slave eth_dev */
> >         struct ether_addr persisted_mac_addr;
> >
> > @@ -240,7 +242,8 @@ slave_remove(struct bond_dev_private *internals,
> >
> >  void
> >  slave_add(struct bond_dev_private *internals,
> > -               struct rte_eth_dev *slave_eth_dev);
> > +               struct rte_eth_dev *slave_eth_dev,
> > +               const struct rte_eth_dev_info *slave_dev_info);
> >
> >  uint16_t
> >  xmit_l2_hash(const struct rte_mbuf *buf, uint8_t slave_count);
> > --
> > 2.1.4
> >
> 
> 
> 
> -- 
> Andriy Berestovskyy

Re: [PATCH 6/8] bond: handle slaves with fewer queues than bonding device

[Declan Doherty]

On 04/12/15 19:18, Eric Kinzie wrote:
> On Fri Dec 04 19:36:09 +0100 2015, Andriy Berestovskyy wrote:
>> Hi guys,
>> I'm not quite sure if we can support less TX queues on a slave that easy:
>>
>>> queue_id = bond_slave_txqid(internals, i, bd_tx_q->queue_id);
>>> num_tx_slave = rte_eth_tx_burst(slaves[i], queue_id,
>>>       slave_bufs[i], slave_nb_pkts[i]);
>>
>> It seems that two different lcores might end up writing to the same
>> slave queue at the same time, isn't it?
>>
>> Regards,
>> Andriy
>
> Andriy, I think you're probably right about this.  Perhaps it should
> instead refuse to add or refuse to activate a slave with too few
> tx queues.  Could probably fix this with another layer of buffering
> so that an lcore with a valid tx queue could pick up the mbufs later,
> but this doesn't seem very appealing.
>
> Eric
>
>
>> On Fri, Dec 4, 2015 at 6:14 PM, Stephen Hemminger
>> <stephen@networkplumber.org> wrote:
>>> From: Eric Kinzie <ekinzie@brocade.com>
>>>
>>> In the event that the bonding device has a greater number of tx and/or rx
>>> queues than the slave being added, track the queue limits of the slave.
>>> On receive, ignore queue identifiers beyond what the slave interface
>>> can support.  During transmit, pick a different queue id to use if the
>>> intended queue is not available on the slave.
>>>
>>> Signed-off-by: Eric Kinzie <ekinzie@brocade.com>
>>> Signed-off-by: Stephen Hemminger <stephen@networkplumber.org>
>>> ---
...


I don't there is any straight forward way of supporting slaves with 
different numbers of queues, the initial library was written with the 
assumption that the number of tx/rx queues would always be the same on 
each slave. This is why,when a slave is added to a bonded device we 
reconfigure the queues. For features like RSS we have to have the same 
number of rx queues otherwise the flow distribution to an application 
could change in the case of a fail over event. Also by supporting 
different numbers of queues between slaves we would be no longer be 
supporting the standard behavior of ethdevs in DPDK were we expect that 
by using different queues we don't require locking to be thread safe.

Re: [PATCH 6/8] bond: handle slaves with fewer queues than bonding device

[Stephen Hemminger]

A common usage scenario is to bond a vnic like virtio which typically has
only a single rx queue with a VF device that has multiple receive queues.
This is done to do live migration
On Jan 5, 2016 05:47, "Declan Doherty" <declan.doherty@intel.com> wrote:

> On 04/12/15 19:18, Eric Kinzie wrote:
>
>> On Fri Dec 04 19:36:09 +0100 2015, Andriy Berestovskyy wrote:
>>
>>> Hi guys,
>>> I'm not quite sure if we can support less TX queues on a slave that easy:
>>>
>>> queue_id = bond_slave_txqid(internals, i, bd_tx_q->queue_id);
>>>> num_tx_slave = rte_eth_tx_burst(slaves[i], queue_id,
>>>>       slave_bufs[i], slave_nb_pkts[i]);
>>>>
>>>
>>> It seems that two different lcores might end up writing to the same
>>> slave queue at the same time, isn't it?
>>>
>>> Regards,
>>> Andriy
>>>
>>
>> Andriy, I think you're probably right about this.  Perhaps it should
>> instead refuse to add or refuse to activate a slave with too few
>> tx queues.  Could probably fix this with another layer of buffering
>> so that an lcore with a valid tx queue could pick up the mbufs later,
>> but this doesn't seem very appealing.
>>
>> Eric
>>
>>
>> On Fri, Dec 4, 2015 at 6:14 PM, Stephen Hemminger
>>> <stephen@networkplumber.org> wrote:
>>>
>>>> From: Eric Kinzie <ekinzie@brocade.com>
>>>>
>>>> In the event that the bonding device has a greater number of tx and/or
>>>> rx
>>>> queues than the slave being added, track the queue limits of the slave.
>>>> On receive, ignore queue identifiers beyond what the slave interface
>>>> can support.  During transmit, pick a different queue id to use if the
>>>> intended queue is not available on the slave.
>>>>
>>>> Signed-off-by: Eric Kinzie <ekinzie@brocade.com>
>>>> Signed-off-by: Stephen Hemminger <stephen@networkplumber.org>
>>>> ---
>>>>
>>> ...
>
>
> I don't there is any straight forward way of supporting slaves with
> different numbers of queues, the initial library was written with the
> assumption that the number of tx/rx queues would always be the same on each
> slave. This is why,when a slave is added to a bonded device we reconfigure
> the queues. For features like RSS we have to have the same number of rx
> queues otherwise the flow distribution to an application could change in
> the case of a fail over event. Also by supporting different numbers of
> queues between slaves we would be no longer be supporting the standard
> behavior of ethdevs in DPDK were we expect that by using different queues
> we don't require locking to be thread safe.
>
>
>

Re: [PATCH 6/8] bond: handle slaves with fewer queues than bonding device

[Bruce Richardson]

On Fri, Dec 04, 2015 at 02:18:34PM -0500, Eric Kinzie wrote:
> On Fri Dec 04 19:36:09 +0100 2015, Andriy Berestovskyy wrote:
> > Hi guys,
> > I'm not quite sure if we can support less TX queues on a slave that easy:
> > 
> > > queue_id = bond_slave_txqid(internals, i, bd_tx_q->queue_id);
> > > num_tx_slave = rte_eth_tx_burst(slaves[i], queue_id,
> > >      slave_bufs[i], slave_nb_pkts[i]);
> > 
> > It seems that two different lcores might end up writing to the same
> > slave queue at the same time, isn't it?
> > 
> > Regards,
> > Andriy
> 
> Andriy, I think you're probably right about this.  Perhaps it should
> instead refuse to add or refuse to activate a slave with too few
> tx queues.  Could probably fix this with another layer of buffering
> so that an lcore with a valid tx queue could pick up the mbufs later,
> but this doesn't seem very appealing.
> 
> Eric
>
Hi Eric, Stephen, Declan,

all patches of the set apart from this one and the next (nos 6 & 7) have no
comments and have been acked. Is there a resolution on these two patches, so they
can be acked and merged?

Regards,
/Bruce

Re: [PATCH 6/8] bond: handle slaves with fewer queues than bonding device

[Declan Doherty]

On 03/02/16 11:28, Bruce Richardson wrote:
> On Fri, Dec 04, 2015 at 02:18:34PM -0500, Eric Kinzie wrote:
>> On Fri Dec 04 19:36:09 +0100 2015, Andriy Berestovskyy wrote:
>>> Hi guys,
>>> I'm not quite sure if we can support less TX queues on a slave that easy:
>>>
>>>> queue_id = bond_slave_txqid(internals, i, bd_tx_q->queue_id);
>>>> num_tx_slave = rte_eth_tx_burst(slaves[i], queue_id,
>>>>       slave_bufs[i], slave_nb_pkts[i]);
>>>
>>> It seems that two different lcores might end up writing to the same
>>> slave queue at the same time, isn't it?
>>>
>>> Regards,
>>> Andriy
>>
>> Andriy, I think you're probably right about this.  Perhaps it should
>> instead refuse to add or refuse to activate a slave with too few
>> tx queues.  Could probably fix this with another layer of buffering
>> so that an lcore with a valid tx queue could pick up the mbufs later,
>> but this doesn't seem very appealing.
>>
>> Eric
>>
> Hi Eric, Stephen, Declan,
>
> all patches of the set apart from this one and the next (nos 6 & 7) have no
> comments and have been acked. Is there a resolution on these two patches, so they
> can be acked and merged?
>
> Regards,
> /Bruce
>


Hey Bruce, Eric, Stephen, sorry about leaving this patchset hanging 
around. Can you apply patches 1-5 & patch 8 in this patch set. I've 
reviewed and acked all of those patches and I believe they are good tof 
go. I need to give further feedback on patches 6 and 7, as I would like 
to avoid bring further rte_ring buffering into the bonded device if 
possible and I think this should be possible but I haven't had time to 
prototype any alternatives but that shouldn't stop the other patches 
being applied.

Thanks
Declan

Re: [PATCH 6/8] bond: handle slaves with fewer queues than bonding device

[Thomas Monjalon]

2016-02-03 15:17, Declan Doherty:
> On 03/02/16 11:28, Bruce Richardson wrote:
> > Hi Eric, Stephen, Declan,
> >
> > all patches of the set apart from this one and the next (nos 6 & 7) have no
> > comments and have been acked. Is there a resolution on these two patches, so they
> > can be acked and merged?
> 
> Hey Bruce, Eric, Stephen, sorry about leaving this patchset hanging 
> around. Can you apply patches 1-5 & patch 8 in this patch set. I've 
> reviewed and acked all of those patches and I believe they are good tof 
> go. I need to give further feedback on patches 6 and 7, as I would like 
> to avoid bring further rte_ring buffering into the bonded device if 
> possible and I think this should be possible but I haven't had time to 
> prototype any alternatives but that shouldn't stop the other patches 
> being applied.

Picking some patches in a series makes tracking confusing.
The better solution is to re-send the series with only the desired patches.
When re-sending, do not forget to embed the acks from the previous version, thanks.

Re: [PATCH 6/8] bond: handle slaves with fewer queues than bonding device

[Iremonger, Bernard]

Hi Stephen,

> -----Original Message-----
> From: dev [mailto:dev-bounces@dpdk.org] On Behalf Of Thomas Monjalon
> Sent: Wednesday, February 3, 2016 3:21 PM
> To: Doherty, Declan <declan.doherty@intel.com>
> Cc: dev@dpdk.org
> Subject: Re: [dpdk-dev] [PATCH 6/8] bond: handle slaves with fewer queues
> than bonding device
> 
> 2016-02-03 15:17, Declan Doherty:
> > On 03/02/16 11:28, Bruce Richardson wrote:
> > > Hi Eric, Stephen, Declan,
> > >
> > > all patches of the set apart from this one and the next (nos 6 & 7)
> > > have no comments and have been acked. Is there a resolution on these
> > > two patches, so they can be acked and merged?
> >
> > Hey Bruce, Eric, Stephen, sorry about leaving this patchset hanging
> > around. Can you apply patches 1-5 & patch 8 in this patch set. I've
> > reviewed and acked all of those patches and I believe they are good
> > tof go. I need to give further feedback on patches 6 and 7, as I would
> > like to avoid bring further rte_ring buffering into the bonded device
> > if possible and I think this should be possible but I haven't had time
> > to prototype any alternatives but that shouldn't stop the other
> > patches being applied.
> 
> Picking some patches in a series makes tracking confusing.
> The better solution is to re-send the series with only the desired patches.
> When re-sending, do not forget to embed the acks from the previous
> version, thanks.

Could you send a V2 of this patch set without patches 6 and 7.
The other 6 patches have been acked  already.

Regards,

Bernard.

[PATCH v2 0/6] bonding: fixes and enhancements

[Eric Kinzie]

These are bug fixes and some small enhancements to allow bonding
to work with external control (teamd). Please consider integrating
these into DPDK 2.2

Changes in v2:
- remove "bond: handle slaves with fewer queues than bonding device"
- remove "bond: per-slave intermediate rx ring"

Eric Kinzie (6):
  bond: use existing enslaved device queues
  bond mode 4: copy entire config structure
  bond mode 4: do not ignore multicast
  bond mode 4: allow external state machine
  bond: active slaves with no primary
  bond: do not activate slave twice

 app/test/test_link_bonding_mode4.c                |    7 +-
 drivers/net/bonding/rte_eth_bond_8023ad.c         |  174 +++++++++++++++++++++
 drivers/net/bonding/rte_eth_bond_8023ad.h         |   44 ++++++
 drivers/net/bonding/rte_eth_bond_8023ad_private.h |    2 +
 drivers/net/bonding/rte_eth_bond_api.c            |   13 +-
 drivers/net/bonding/rte_eth_bond_pmd.c            |    9 +-
 drivers/net/bonding/rte_eth_bond_version.map      |    6 +
 7 files changed, 249 insertions(+), 6 deletions(-)

-- 
1.7.10.4

[PATCH v2 1/6] bond: use existing enslaved device queues

[Eric Kinzie]

This solves issues when an active device is added to a bond.

If a device to be enslaved already has transmit and/or receive queues
allocated, use those and then create any additional queues that are
necessary.

Signed-off-by: Eric Kinzie <ehkinzie@gmail.com>
Signed-off-by: Stephen Hemminger <stephen@networkplumber.org>
Acked-by: Declan Doherty <declan.doherty@intel.com>
---
 drivers/net/bonding/rte_eth_bond_pmd.c |    8 ++++++--
 1 file changed, 6 insertions(+), 2 deletions(-)

diff --git a/drivers/net/bonding/rte_eth_bond_pmd.c b/drivers/net/bonding/rte_eth_bond_pmd.c
index b63c886..2f193db 100644
--- a/drivers/net/bonding/rte_eth_bond_pmd.c
+++ b/drivers/net/bonding/rte_eth_bond_pmd.c
@@ -1344,7 +1344,9 @@ slave_configure(struct rte_eth_dev *bonded_eth_dev,
 	}
 
 	/* Setup Rx Queues */
-	for (q_id = 0; q_id < bonded_eth_dev->data->nb_rx_queues; q_id++) {
+	/* Use existing queues, if any */
+	for (q_id = slave_eth_dev->data->nb_rx_queues;
+	     q_id < bonded_eth_dev->data->nb_rx_queues; q_id++) {
 		bd_rx_q = (struct bond_rx_queue *)bonded_eth_dev->data->rx_queues[q_id];
 
 		errval = rte_eth_rx_queue_setup(slave_eth_dev->data->port_id, q_id,
@@ -1360,7 +1362,9 @@ slave_configure(struct rte_eth_dev *bonded_eth_dev,
 	}
 
 	/* Setup Tx Queues */
-	for (q_id = 0; q_id < bonded_eth_dev->data->nb_tx_queues; q_id++) {
+	/* Use existing queues, if any */
+	for (q_id = slave_eth_dev->data->nb_tx_queues;
+	     q_id < bonded_eth_dev->data->nb_tx_queues; q_id++) {
 		bd_tx_q = (struct bond_tx_queue *)bonded_eth_dev->data->tx_queues[q_id];
 
 		errval = rte_eth_tx_queue_setup(slave_eth_dev->data->port_id, q_id,
-- 
1.7.10.4

[PATCH v2 2/6] bond mode 4: copy entire config structure

[Eric Kinzie]

From: Eric Kinzie <ekinzie@brocade.com>

Copy all needed fields from the mode8023ad_private structure in
bond_mode_8023ad_conf_get().  This help ensure that a subsequent call
to rte_eth_bond_8023ad_setup() is not passed uninitialized data that
would result in either incorrect behavior or a failed sanity check.

Fixes: 46fb43683679 ("bond: add mode 4")

Signed-off-by: Eric Kinzie <ehkinzie@gmail.com>
Signed-off-by: Stephen Hemminger <stephen@networkplumber.org>
Acked-by: Declan Doherty <declan.doherty@intel.com>
---
 drivers/net/bonding/rte_eth_bond_8023ad.c |    1 +
 1 file changed, 1 insertion(+)

diff --git a/drivers/net/bonding/rte_eth_bond_8023ad.c b/drivers/net/bonding/rte_eth_bond_8023ad.c
index b3b30f6..1b7e93a 100644
--- a/drivers/net/bonding/rte_eth_bond_8023ad.c
+++ b/drivers/net/bonding/rte_eth_bond_8023ad.c
@@ -1019,6 +1019,7 @@ bond_mode_8023ad_conf_get(struct rte_eth_dev *dev,
 	conf->aggregate_wait_timeout_ms = mode4->aggregate_wait_timeout / ms_ticks;
 	conf->tx_period_ms = mode4->tx_period_timeout / ms_ticks;
 	conf->update_timeout_ms = mode4->update_timeout_us / 1000;
+	conf->rx_marker_period_ms = mode4->rx_marker_timeout / ms_ticks;
 }
 
 void
-- 
1.7.10.4

[PATCH v2 3/6] bond mode 4: do not ignore multicast

[Eric Kinzie]

From: Eric Kinzie <ekinzie@brocade.com>

The bonding PMD in mode 4 puts all enslaved interfaces into promiscuous
mode in order to receive LACPDUs and must filter unwanted packets
after the traffic has been "collected".  Allow broadcast and multicast
through so that ARP and IPv6 neighbor discovery continue to work.

Fixes: 46fb43683679 ("bond: add mode 4")

Signed-off-by: Eric Kinzie <ehkinzie@gmail.com>
Signed-off-by: Stephen Hemminger <stephen@networkplumber.org>
Acked-by: Declan Doherty <declan.doherty@intel.com>
---
 app/test/test_link_bonding_mode4.c     |    7 +++++--
 drivers/net/bonding/rte_eth_bond_pmd.c |    1 +
 2 files changed, 6 insertions(+), 2 deletions(-)

diff --git a/app/test/test_link_bonding_mode4.c b/app/test/test_link_bonding_mode4.c
index 713368d..31640cd 100644
--- a/app/test/test_link_bonding_mode4.c
+++ b/app/test/test_link_bonding_mode4.c
@@ -747,8 +747,11 @@ test_mode4_rx(void)
 	rte_eth_macaddr_get(test_params.bonded_port_id, &bonded_mac);
 	ether_addr_copy(&bonded_mac, &dst_mac);
 
-	/* Assert that dst address is not bonding address */
-	dst_mac.addr_bytes[0]++;
+	/* Assert that dst address is not bonding address.  Do not set the
+	 * least significant bit of the zero byte as this would create a
+	 * multicast address.
+	 */
+	dst_mac.addr_bytes[0] += 2;
 
 	/* First try with promiscuous mode enabled.
 	 * Add 2 packets to each slave. First with bonding MAC address, second with
diff --git a/drivers/net/bonding/rte_eth_bond_pmd.c b/drivers/net/bonding/rte_eth_bond_pmd.c
index 2f193db..b938a68 100644
--- a/drivers/net/bonding/rte_eth_bond_pmd.c
+++ b/drivers/net/bonding/rte_eth_bond_pmd.c
@@ -171,6 +171,7 @@ bond_ethdev_rx_burst_8023ad(void *queue, struct rte_mbuf **bufs,
 			 * mode and packet address does not match. */
 			if (unlikely(hdr->ether_type == ether_type_slow_be ||
 				!collecting || (!promisc &&
+					!is_multicast_ether_addr(&hdr->d_addr) &&
 					!is_same_ether_addr(&bond_mac, &hdr->d_addr)))) {
 
 				if (hdr->ether_type == ether_type_slow_be) {
-- 
1.7.10.4

[PATCH v2 4/6] bond mode 4: allow external state machine

[Eric Kinzie]

From: Eric Kinzie <ekinzie@brocade.com>

Provide functions to allow an external 802.3ad state machine to transmit
and recieve LACPDUs and to set the collection/distribution flags on
slave interfaces.

Signed-off-by: Eric Kinzie <ehkinzie@gmail.com>
Signed-off-by: Stephen Hemminger <stephen@networkplumber.org>
Acked-by: Declan Doherty <declan.doherty@intel.com>
---
 drivers/net/bonding/rte_eth_bond_8023ad.c         |  173 +++++++++++++++++++++
 drivers/net/bonding/rte_eth_bond_8023ad.h         |   44 ++++++
 drivers/net/bonding/rte_eth_bond_8023ad_private.h |    2 +
 drivers/net/bonding/rte_eth_bond_version.map      |    6 +
 4 files changed, 225 insertions(+)

diff --git a/drivers/net/bonding/rte_eth_bond_8023ad.c b/drivers/net/bonding/rte_eth_bond_8023ad.c
index 1b7e93a..a260e06 100644
--- a/drivers/net/bonding/rte_eth_bond_8023ad.c
+++ b/drivers/net/bonding/rte_eth_bond_8023ad.c
@@ -42,6 +42,8 @@
 
 #include "rte_eth_bond_private.h"
 
+static void bond_mode_8023ad_ext_periodic_cb(void *arg);
+
 #ifdef RTE_LIBRTE_BOND_DEBUG_8023AD
 #define MODE4_DEBUG(fmt, ...) RTE_LOG(DEBUG, PMD, "%6u [Port %u: %s] " fmt, \
 			bond_dbg_get_time_diff_ms(), slave_id, \
@@ -1020,6 +1022,7 @@ bond_mode_8023ad_conf_get(struct rte_eth_dev *dev,
 	conf->tx_period_ms = mode4->tx_period_timeout / ms_ticks;
 	conf->update_timeout_ms = mode4->update_timeout_us / 1000;
 	conf->rx_marker_period_ms = mode4->rx_marker_timeout / ms_ticks;
+	conf->slowrx_cb = mode4->slowrx_cb;
 }
 
 void
@@ -1041,8 +1044,11 @@ bond_mode_8023ad_setup(struct rte_eth_dev *dev,
 		conf->tx_period_ms = BOND_8023AD_TX_MACHINE_PERIOD_MS;
 		conf->rx_marker_period_ms = BOND_8023AD_RX_MARKER_PERIOD_MS;
 		conf->update_timeout_ms = BOND_MODE_8023AX_UPDATE_TIMEOUT_MS;
+		conf->slowrx_cb = NULL;
 	}
 
+	bond_mode_8023ad_stop(dev);
+
 	mode4->fast_periodic_timeout = conf->fast_periodic_ms * ms_ticks;
 	mode4->slow_periodic_timeout = conf->slow_periodic_ms * ms_ticks;
 	mode4->short_timeout = conf->short_timeout_ms * ms_ticks;
@@ -1051,6 +1057,10 @@ bond_mode_8023ad_setup(struct rte_eth_dev *dev,
 	mode4->tx_period_timeout = conf->tx_period_ms * ms_ticks;
 	mode4->rx_marker_timeout = conf->rx_marker_period_ms * ms_ticks;
 	mode4->update_timeout_us = conf->update_timeout_ms * 1000;
+	mode4->slowrx_cb = conf->slowrx_cb;
+
+	if (dev->data->dev_started)
+		bond_mode_8023ad_start(dev);
 }
 
 int
@@ -1068,6 +1078,13 @@ bond_mode_8023ad_enable(struct rte_eth_dev *bond_dev)
 int
 bond_mode_8023ad_start(struct rte_eth_dev *bond_dev)
 {
+	struct bond_dev_private *internals = bond_dev->data->dev_private;
+	struct mode8023ad_private *mode4 = &internals->mode4;
+
+	if (mode4->slowrx_cb)
+		return rte_eal_alarm_set(BOND_MODE_8023AX_UPDATE_TIMEOUT_MS * 1000,
+			&bond_mode_8023ad_ext_periodic_cb, bond_dev);
+
 	return rte_eal_alarm_set(BOND_MODE_8023AX_UPDATE_TIMEOUT_MS * 1000,
 			&bond_mode_8023ad_periodic_cb, bond_dev);
 }
@@ -1075,6 +1092,13 @@ bond_mode_8023ad_start(struct rte_eth_dev *bond_dev)
 void
 bond_mode_8023ad_stop(struct rte_eth_dev *bond_dev)
 {
+	struct bond_dev_private *internals = bond_dev->data->dev_private;
+	struct mode8023ad_private *mode4 = &internals->mode4;
+
+	if (mode4->slowrx_cb) {
+		rte_eal_alarm_cancel(&bond_mode_8023ad_ext_periodic_cb, bond_dev);
+		return;
+	}
 	rte_eal_alarm_cancel(&bond_mode_8023ad_periodic_cb, bond_dev);
 }
 
@@ -1221,3 +1245,152 @@ rte_eth_bond_8023ad_slave_info(uint8_t port_id, uint8_t slave_id,
 	info->agg_port_id = port->aggregator_port_id;
 	return 0;
 }
+
+int
+rte_eth_bond_8023ad_ext_collect(uint8_t port_id, uint8_t slave_id, int enabled)
+{
+	struct rte_eth_dev *bond_dev;
+	struct bond_dev_private *internals;
+	struct mode8023ad_private *mode4;
+	struct port *port;
+
+	if (rte_eth_bond_mode_get(port_id) != BONDING_MODE_8023AD)
+		return -EINVAL;
+
+	bond_dev = &rte_eth_devices[port_id];
+
+	if (!bond_dev->data->dev_started)
+		return -EINVAL;
+
+	internals = bond_dev->data->dev_private;
+	if (find_slave_by_id(internals->active_slaves,
+			internals->active_slave_count, slave_id) ==
+				internals->active_slave_count)
+		return -EINVAL;
+
+	mode4 = &internals->mode4;
+	if (mode4->slowrx_cb == NULL)
+		return -EINVAL;
+
+	port = &mode_8023ad_ports[slave_id];
+
+	if (enabled)
+		ACTOR_STATE_SET(port, COLLECTING);
+	else
+		ACTOR_STATE_CLR(port, COLLECTING);
+
+	return 0;
+}
+
+int
+rte_eth_bond_8023ad_ext_distrib(uint8_t port_id, uint8_t slave_id, int enabled)
+{
+	struct rte_eth_dev *bond_dev;
+	struct bond_dev_private *internals;
+	struct mode8023ad_private *mode4;
+	struct port *port;
+
+	if (rte_eth_bond_mode_get(port_id) != BONDING_MODE_8023AD)
+		return -EINVAL;
+
+	bond_dev = &rte_eth_devices[port_id];
+
+	if (!bond_dev->data->dev_started)
+		return -EINVAL;
+
+	internals = bond_dev->data->dev_private;
+	if (find_slave_by_id(internals->active_slaves,
+			internals->active_slave_count, slave_id) ==
+				internals->active_slave_count)
+		return -EINVAL;
+
+	mode4 = &internals->mode4;
+	if (mode4->slowrx_cb == NULL)
+		return -EINVAL;
+
+	port = &mode_8023ad_ports[slave_id];
+
+	if (enabled)
+		ACTOR_STATE_SET(port, DISTRIBUTING);
+	else
+		ACTOR_STATE_CLR(port, DISTRIBUTING);
+
+	return 0;
+}
+
+int
+rte_eth_bond_8023ad_ext_slowtx(uint8_t port_id, uint8_t slave_id,
+		struct rte_mbuf *lacp_pkt)
+{
+	struct rte_eth_dev *bond_dev;
+	struct bond_dev_private *internals;
+	struct mode8023ad_private *mode4;
+	struct port *port;
+
+	if (rte_eth_bond_mode_get(port_id) != BONDING_MODE_8023AD)
+		return -EINVAL;
+
+	bond_dev = &rte_eth_devices[port_id];
+
+	if (!bond_dev->data->dev_started)
+		return -EINVAL;
+
+	internals = bond_dev->data->dev_private;
+	if (find_slave_by_id(internals->active_slaves,
+			internals->active_slave_count, slave_id) ==
+				internals->active_slave_count)
+		return -EINVAL;
+
+	mode4 = &internals->mode4;
+	if (mode4->slowrx_cb == NULL)
+		return -EINVAL;
+
+	port = &mode_8023ad_ports[slave_id];
+
+	if (rte_pktmbuf_pkt_len(lacp_pkt) < sizeof(struct lacpdu_header))
+		return -EINVAL;
+
+	struct lacpdu_header *lacp;
+
+	/* only enqueue LACPDUs */
+	lacp = rte_pktmbuf_mtod(lacp_pkt, struct lacpdu_header *);
+	if (lacp->lacpdu.subtype != SLOW_SUBTYPE_LACP)
+		return -EINVAL;
+
+	MODE4_DEBUG("sending LACP frame\n");
+
+	return rte_ring_enqueue(port->tx_ring, lacp_pkt);
+}
+
+static void
+bond_mode_8023ad_ext_periodic_cb(void *arg)
+{
+	struct rte_eth_dev *bond_dev = arg;
+	struct bond_dev_private *internals = bond_dev->data->dev_private;
+	struct mode8023ad_private *mode4 = &internals->mode4;
+	struct port *port;
+	void *pkt = NULL;
+	uint16_t i, slave_id;
+
+	for (i = 0; i < internals->active_slave_count; i++) {
+		slave_id = internals->active_slaves[i];
+		port = &mode_8023ad_ports[slave_id];
+
+		if (rte_ring_dequeue(port->rx_ring, &pkt) == 0) {
+			struct rte_mbuf *lacp_pkt = pkt;
+			struct lacpdu_header *lacp;
+
+			lacp = rte_pktmbuf_mtod(lacp_pkt,
+						struct lacpdu_header *);
+			RTE_VERIFY(lacp->lacpdu.subtype == SLOW_SUBTYPE_LACP);
+
+			/* This is LACP frame so pass it to rx callback.
+			 * Callback is responsible for freeing mbuf.
+			 */
+			mode4->slowrx_cb(slave_id, lacp_pkt);
+		}
+	}
+
+	rte_eal_alarm_set(internals->mode4.update_timeout_us,
+			bond_mode_8023ad_ext_periodic_cb, arg);
+}
diff --git a/drivers/net/bonding/rte_eth_bond_8023ad.h b/drivers/net/bonding/rte_eth_bond_8023ad.h
index ebd0e93..8cfa3d3 100644
--- a/drivers/net/bonding/rte_eth_bond_8023ad.h
+++ b/drivers/net/bonding/rte_eth_bond_8023ad.h
@@ -64,6 +64,8 @@ extern "C" {
 #define MARKER_TLV_TYPE_INFO                0x01
 #define MARKER_TLV_TYPE_RESP                0x02
 
+typedef void (*rte_eth_bond_8023ad_ext_slowrx_fn)(uint8_t slave_id, struct rte_mbuf *lacp_pkt);
+
 enum rte_bond_8023ad_selection {
 	UNSELECTED,
 	STANDBY,
@@ -157,6 +159,7 @@ struct rte_eth_bond_8023ad_conf {
 	uint32_t tx_period_ms;
 	uint32_t rx_marker_period_ms;
 	uint32_t update_timeout_ms;
+	rte_eth_bond_8023ad_ext_slowrx_fn slowrx_cb;
 };
 
 struct rte_eth_bond_8023ad_slave_info {
@@ -219,4 +222,45 @@ rte_eth_bond_8023ad_slave_info(uint8_t port_id, uint8_t slave_id,
 }
 #endif
 
+/**
+ * Configure a slave port to start collecting.
+ *
+ * @param port_id	Bonding device id
+ * @param slave_id	Port id of valid slave.
+ * @param enabled	Non-zero when collection enabled.
+ * @return
+ *   0 - if ok
+ *   -EINVAL if slave is not valid.
+ */
+int
+rte_eth_bond_8023ad_ext_collect(uint8_t port_id, uint8_t slave_id, int enabled);
+
+/**
+ * Configure a slave port to start distributing.
+ *
+ * @param port_id	Bonding device id
+ * @param slave_id	Port id of valid slave.
+ * @param enabled	Non-zero when distribution enabled.
+ * @return
+ *   0 - if ok
+ *   -EINVAL if slave is not valid.
+ */
+int
+rte_eth_bond_8023ad_ext_distrib(uint8_t port_id, uint8_t slave_id, int enabled);
+
+/**
+ * LACPDU transmit path for external 802.3ad state machine.  Caller retains
+ * ownership of the packet on failure.
+ *
+ * @param port_id	Bonding device id
+ * @param slave_id	Port ID of valid slave device.
+ * @param lacp_pkt	mbuf containing LACPDU.
+ *
+ * @return
+ *   0 on success, negative value otherwise.
+ */
+int
+rte_eth_bond_8023ad_ext_slowtx(uint8_t port_id, uint8_t slave_id,
+		struct rte_mbuf *lacp_pkt);
+
 #endif /* RTE_ETH_BOND_8023AD_H_ */
diff --git a/drivers/net/bonding/rte_eth_bond_8023ad_private.h b/drivers/net/bonding/rte_eth_bond_8023ad_private.h
index 8adee70..9e15ece 100644
--- a/drivers/net/bonding/rte_eth_bond_8023ad_private.h
+++ b/drivers/net/bonding/rte_eth_bond_8023ad_private.h
@@ -173,6 +173,8 @@ struct mode8023ad_private {
 	uint64_t tx_period_timeout;
 	uint64_t rx_marker_timeout;
 	uint64_t update_timeout_us;
+	rte_eth_bond_8023ad_ext_slowrx_fn slowrx_cb;
+	uint8_t external_sm;
 };
 
 /**
diff --git a/drivers/net/bonding/rte_eth_bond_version.map b/drivers/net/bonding/rte_eth_bond_version.map
index 22bd920..33d73ff 100644
--- a/drivers/net/bonding/rte_eth_bond_version.map
+++ b/drivers/net/bonding/rte_eth_bond_version.map
@@ -27,3 +27,9 @@ DPDK_2.1 {
 	rte_eth_bond_free;
 
 } DPDK_2.0;
+
+DPDK_2.2 {
+	rte_eth_bond_8023ad_ext_collect;
+	rte_eth_bond_8023ad_ext_distrib;
+	rte_eth_bond_8023ad_ext_slowtx;
+} DPDK_2.1;
-- 
1.7.10.4

Re: [PATCH v2 4/6] bond mode 4: allow external state machine

[Panu Matilainen]

On 02/19/2016 09:17 PM, Eric Kinzie wrote:
> From: Eric Kinzie <ekinzie@brocade.com>
>
> Provide functions to allow an external 802.3ad state machine to transmit
> and recieve LACPDUs and to set the collection/distribution flags on
> slave interfaces.
>
> Signed-off-by: Eric Kinzie <ehkinzie@gmail.com>
> Signed-off-by: Stephen Hemminger <stephen@networkplumber.org>
> Acked-by: Declan Doherty <declan.doherty@intel.com>
[...]
> diff --git a/drivers/net/bonding/rte_eth_bond_8023ad.h b/drivers/net/bonding/rte_eth_bond_8023ad.h
> index ebd0e93..8cfa3d3 100644
> --- a/drivers/net/bonding/rte_eth_bond_8023ad.h
> +++ b/drivers/net/bonding/rte_eth_bond_8023ad.h
> @@ -64,6 +64,8 @@ extern "C" {
>   #define MARKER_TLV_TYPE_INFO                0x01
>   #define MARKER_TLV_TYPE_RESP                0x02
>
> +typedef void (*rte_eth_bond_8023ad_ext_slowrx_fn)(uint8_t slave_id, struct rte_mbuf *lacp_pkt);
> +
>   enum rte_bond_8023ad_selection {
>   	UNSELECTED,
>   	STANDBY,
> @@ -157,6 +159,7 @@ struct rte_eth_bond_8023ad_conf {
>   	uint32_t tx_period_ms;
>   	uint32_t rx_marker_period_ms;
>   	uint32_t update_timeout_ms;
> +	rte_eth_bond_8023ad_ext_slowrx_fn slowrx_cb;
>   };

This still is a likely an ABI break, previously discussed around here:
http://dpdk.org/ml/archives/dev/2015-November/027321.html

It might not be embedded anywhere in DPDK codebase, but there's no 
telling what others might have done with it (have an array of them, 
embed in other structs etc).

Also ultimately ABI compatibility goes both ways: when the library 
soname does not change then an application is free to assume both 
downgrading and upgrading are safe. In this case, upgrading *might* be 
okay, downgrading certainly is not. So by that measure it definitely is 
an ABI break.

[...]
> diff --git a/drivers/net/bonding/rte_eth_bond_version.map b/drivers/net/bonding/rte_eth_bond_version.map
> index 22bd920..33d73ff 100644
> --- a/drivers/net/bonding/rte_eth_bond_version.map
> +++ b/drivers/net/bonding/rte_eth_bond_version.map
> @@ -27,3 +27,9 @@ DPDK_2.1 {
>   	rte_eth_bond_free;
>
>   } DPDK_2.0;
> +
> +DPDK_2.2 {
> +	rte_eth_bond_8023ad_ext_collect;
> +	rte_eth_bond_8023ad_ext_distrib;
> +	rte_eth_bond_8023ad_ext_slowtx;
> +} DPDK_2.1;
>

These symbols are not part of DPDK 2.2, the version here is wrong. 
Technically it would not actually matter much but better not to confuse 
things unnecessarily.

	- Panu -

Re: [PATCH v2 4/6] bond mode 4: allow external state machine

[Iremonger, Bernard]

Hi Eric,
<snip>

> > @@ -157,6 +159,7 @@ struct rte_eth_bond_8023ad_conf {
> >   	uint32_t tx_period_ms;
> >   	uint32_t rx_marker_period_ms;
> >   	uint32_t update_timeout_ms;
> > +	rte_eth_bond_8023ad_ext_slowrx_fn slowrx_cb;
> >   };
> 
> This still is a likely an ABI break, previously discussed around here:
> http://dpdk.org/ml/archives/dev/2015-November/027321.html
> 
> It might not be embedded anywhere in DPDK codebase, but there's no
> telling what others might have done with it (have an array of them, embed in
> other structs etc).
> 
> Also ultimately ABI compatibility goes both ways: when the library soname
> does not change then an application is free to assume both downgrading and
> upgrading are safe. In this case, upgrading *might* be okay, downgrading
> certainly is not. So by that measure it definitely is an ABI break.
> 
> [...]
> > diff --git a/drivers/net/bonding/rte_eth_bond_version.map
> > b/drivers/net/bonding/rte_eth_bond_version.map
> > index 22bd920..33d73ff 100644
> > --- a/drivers/net/bonding/rte_eth_bond_version.map
> > +++ b/drivers/net/bonding/rte_eth_bond_version.map
> > @@ -27,3 +27,9 @@ DPDK_2.1 {
> >   	rte_eth_bond_free;
> >
> >   } DPDK_2.0;
> > +
> > +DPDK_2.2 {
> > +	rte_eth_bond_8023ad_ext_collect;
> > +	rte_eth_bond_8023ad_ext_distrib;
> > +	rte_eth_bond_8023ad_ext_slowtx;
> > +} DPDK_2.1;
> >
> 
> These symbols are not part of DPDK 2.2, the version here is wrong.
> Technically it would not actually matter much but better not to confuse
> things unnecessarily.
> 
> 	- Panu -

It looks like Panu's points are valid, a V3 of this patch set which takes care of these issues will be needed.

Patches 1/6, 5/6 and 6/6 of the patch set are bug fixes, so each patch should contain a fixes line.
Patches 2/6, 3/6 and 4/6 are a new feature, the release notes should be updated for this feature.

Could I suggest splitting the patch set into two patch sets, a bug fix patch set and a new feature patch set.

Regards,

Bernard.

[PATCH V3 0/4] bonding: fixes and enhancements

[Eric Kinzie]

These are bug fixes and some small enhancements to allow bonding
to work with external control (teamd). Please consider integrating
these into DPDK 2.2

Changes in v2:
- remove "bond: handle slaves with fewer queues than bonding device"
- remove "bond: per-slave intermediate rx ring"

Changes in v3:
This version has only fixes.  Patches with new functionality have been
removed and will be submitted separately.
- remove "bond mode 4: allow external state machine"
- remove "bond: use existing enslaved device queues"

Eric Kinzie (4):
  bond mode 4: copy entire config structure
  bond mode 4: do not ignore multicast
  bond: active slaves with no primary
  bond: do not activate slave twice

 app/test/test_link_bonding_mode4.c        |  7 +++++--
 drivers/net/bonding/rte_eth_bond_8023ad.c |  1 +
 drivers/net/bonding/rte_eth_bond_api.c    | 13 +++++++++++--
 drivers/net/bonding/rte_eth_bond_pmd.c    |  1 +
 4 files changed, 18 insertions(+), 4 deletions(-)

-- 
2.1.4

[PATCH V3 1/4] bond mode 4: copy entire config structure

[Eric Kinzie]

From: Eric Kinzie <ekinzie@brocade.com>

Copy all needed fields from the mode8023ad_private structure in
bond_mode_8023ad_conf_get().  This help ensure that a subsequent call
to rte_eth_bond_8023ad_setup() is not passed uninitialized data that
would result in either incorrect behavior or a failed sanity check.

Fixes: 46fb43683679 ("bond: add mode 4")

Signed-off-by: Eric Kinzie <ehkinzie@gmail.com>
Signed-off-by: Stephen Hemminger <stephen@networkplumber.org>
Acked-by: Declan Doherty <declan.doherty@intel.com>
---
 drivers/net/bonding/rte_eth_bond_8023ad.c | 1 +
 1 file changed, 1 insertion(+)

diff --git a/drivers/net/bonding/rte_eth_bond_8023ad.c b/drivers/net/bonding/rte_eth_bond_8023ad.c
index b3b30f6..1b7e93a 100644
--- a/drivers/net/bonding/rte_eth_bond_8023ad.c
+++ b/drivers/net/bonding/rte_eth_bond_8023ad.c
@@ -1019,6 +1019,7 @@ bond_mode_8023ad_conf_get(struct rte_eth_dev *dev,
 	conf->aggregate_wait_timeout_ms = mode4->aggregate_wait_timeout / ms_ticks;
 	conf->tx_period_ms = mode4->tx_period_timeout / ms_ticks;
 	conf->update_timeout_ms = mode4->update_timeout_us / 1000;
+	conf->rx_marker_period_ms = mode4->rx_marker_timeout / ms_ticks;
 }
 
 void
-- 
2.1.4

[PATCH V3 2/4] bond mode 4: do not ignore multicast

[Eric Kinzie]

From: Eric Kinzie <ekinzie@brocade.com>

The bonding PMD in mode 4 puts all enslaved interfaces into promiscuous
mode in order to receive LACPDUs and must filter unwanted packets
after the traffic has been "collected".  Allow broadcast and multicast
through so that ARP and IPv6 neighbor discovery continue to work.

Fixes: 46fb43683679 ("bond: add mode 4")

Signed-off-by: Eric Kinzie <ehkinzie@gmail.com>
Signed-off-by: Stephen Hemminger <stephen@networkplumber.org>
Acked-by: Declan Doherty <declan.doherty@intel.com>
---
 app/test/test_link_bonding_mode4.c     | 7 +++++--
 drivers/net/bonding/rte_eth_bond_pmd.c | 1 +
 2 files changed, 6 insertions(+), 2 deletions(-)

diff --git a/app/test/test_link_bonding_mode4.c b/app/test/test_link_bonding_mode4.c
index 713368d..31640cd 100644
--- a/app/test/test_link_bonding_mode4.c
+++ b/app/test/test_link_bonding_mode4.c
@@ -747,8 +747,11 @@ test_mode4_rx(void)
 	rte_eth_macaddr_get(test_params.bonded_port_id, &bonded_mac);
 	ether_addr_copy(&bonded_mac, &dst_mac);
 
-	/* Assert that dst address is not bonding address */
-	dst_mac.addr_bytes[0]++;
+	/* Assert that dst address is not bonding address.  Do not set the
+	 * least significant bit of the zero byte as this would create a
+	 * multicast address.
+	 */
+	dst_mac.addr_bytes[0] += 2;
 
 	/* First try with promiscuous mode enabled.
 	 * Add 2 packets to each slave. First with bonding MAC address, second with
diff --git a/drivers/net/bonding/rte_eth_bond_pmd.c b/drivers/net/bonding/rte_eth_bond_pmd.c
index b63c886..011150a 100644
--- a/drivers/net/bonding/rte_eth_bond_pmd.c
+++ b/drivers/net/bonding/rte_eth_bond_pmd.c
@@ -171,6 +171,7 @@ bond_ethdev_rx_burst_8023ad(void *queue, struct rte_mbuf **bufs,
 			 * mode and packet address does not match. */
 			if (unlikely(hdr->ether_type == ether_type_slow_be ||
 				!collecting || (!promisc &&
+					!is_multicast_ether_addr(&hdr->d_addr) &&
 					!is_same_ether_addr(&bond_mac, &hdr->d_addr)))) {
 
 				if (hdr->ether_type == ether_type_slow_be) {
-- 
2.1.4

[PATCH V3 3/4] bond: active slaves with no primary

[Eric Kinzie]

From: Eric Kinzie <ekinzie@brocade.com>

If the link state of a slave is "up" when added, it is added to the list
of active slaves but, even if it is the only slave, is not selected as
the primary interface.  Generally, handling of link state interrupts
selects an interface to be primary, but only if the active count is zero.
This change avoids the situation where there are active slaves but
no primary.

Fixes: 2efb58cbab6e ("bond: new link bonding library")

Signed-off-by: Eric Kinzie <ekinzie@brocade.com>
Signed-off-by: Stephen Hemminger <stephen@networkplumber.org>
Acked-by: Declan Doherty <declan.doherty@intel.com>
---
 drivers/net/bonding/rte_eth_bond_api.c | 7 ++++++-
 1 file changed, 6 insertions(+), 1 deletion(-)

diff --git a/drivers/net/bonding/rte_eth_bond_api.c b/drivers/net/bonding/rte_eth_bond_api.c
index 8a000c8..630a461 100644
--- a/drivers/net/bonding/rte_eth_bond_api.c
+++ b/drivers/net/bonding/rte_eth_bond_api.c
@@ -427,8 +427,13 @@ __eth_bond_slave_add_lock_free(uint8_t bonded_port_id, uint8_t slave_port_id)
 	if (bonded_eth_dev->data->dev_started) {
 		rte_eth_link_get_nowait(slave_port_id, &link_props);
 
-		 if (link_props.link_status == 1)
+		 if (link_props.link_status == 1) {
+			if (internals->active_slave_count == 0 &&
+			    !internals->user_defined_primary_port)
+				bond_ethdev_primary_set(internals,
+							slave_port_id);
 			activate_slave(bonded_eth_dev, slave_port_id);
+		}
 	}
 	return 0;
 
-- 
2.1.4

[PATCH V3 4/4] bond: do not activate slave twice

[Eric Kinzie]

From: Eric Kinzie <ekinzie@brocade.com>

The current code for detecting link during slave addition can cause a
slave interface to be activated twice -- once during slave_configure()
and again at the end of __eth_bond_slave_add_lock_free().  This will
either cause the active slave count to be incorrect or will cause the
802.3ad activation function to panic.  Ensure that the interface is not
activated more than once.

Fixes: 46fb43683679 ("bond: add mode 4")

Signed-off-by: Eric Kinzie <ekinzie@brocade.com>
Signed-off-by: Stephen Hemminger <stephen@networkplumber.org>
Acked-by: Declan Doherty <declan.doherty@intel.com>
---
 drivers/net/bonding/rte_eth_bond_api.c | 6 +++++-
 1 file changed, 5 insertions(+), 1 deletion(-)

diff --git a/drivers/net/bonding/rte_eth_bond_api.c b/drivers/net/bonding/rte_eth_bond_api.c
index 630a461..def22d0 100644
--- a/drivers/net/bonding/rte_eth_bond_api.c
+++ b/drivers/net/bonding/rte_eth_bond_api.c
@@ -432,7 +432,11 @@ __eth_bond_slave_add_lock_free(uint8_t bonded_port_id, uint8_t slave_port_id)
 			    !internals->user_defined_primary_port)
 				bond_ethdev_primary_set(internals,
 							slave_port_id);
-			activate_slave(bonded_eth_dev, slave_port_id);
+
+			if (find_slave_by_id(internals->active_slaves,
+					     internals->active_slave_count,
+					     slave_port_id) == internals->active_slave_count)
+				activate_slave(bonded_eth_dev, slave_port_id);
 		}
 	}
 	return 0;
-- 
2.1.4

Re: [PATCH V3 0/4] bonding: fixes and enhancements

[Bruce Richardson]

On Tue, Mar 01, 2016 at 09:31:58AM -0800, Eric Kinzie wrote:
> These are bug fixes and some small enhancements to allow bonding
> to work with external control (teamd). Please consider integrating
> these into DPDK 2.2
> 
> Changes in v2:
> - remove "bond: handle slaves with fewer queues than bonding device"
> - remove "bond: per-slave intermediate rx ring"
> 
> Changes in v3:
> This version has only fixes.  Patches with new functionality have been
> removed and will be submitted separately.
> - remove "bond mode 4: allow external state machine"
> - remove "bond: use existing enslaved device queues"
> 
> Eric Kinzie (4):
>   bond mode 4: copy entire config structure
>   bond mode 4: do not ignore multicast
>   bond: active slaves with no primary
>   bond: do not activate slave twice
> 
>  app/test/test_link_bonding_mode4.c        |  7 +++++--
>  drivers/net/bonding/rte_eth_bond_8023ad.c |  1 +
>  drivers/net/bonding/rte_eth_bond_api.c    | 13 +++++++++++--
>  drivers/net/bonding/rte_eth_bond_pmd.c    |  1 +
>  4 files changed, 18 insertions(+), 4 deletions(-)
> 
> -- 
> 2.1.4
> 
Applied to dpdk-next-net/rel_16_04

/Bruce

Re: [PATCH v2 4/6] bond mode 4: allow external state machine

[Eric Kinzie]

On Thu Feb 25 15:22:35 +0000 2016, Iremonger, Bernard wrote:
> Hi Eric,
> <snip>
> 
> > > @@ -157,6 +159,7 @@ struct rte_eth_bond_8023ad_conf {
> > >   	uint32_t tx_period_ms;
> > >   	uint32_t rx_marker_period_ms;
> > >   	uint32_t update_timeout_ms;
> > > +	rte_eth_bond_8023ad_ext_slowrx_fn slowrx_cb;
> > >   };
> > 
> > This still is a likely an ABI break, previously discussed around here:
> > http://dpdk.org/ml/archives/dev/2015-November/027321.html
> > 
> > It might not be embedded anywhere in DPDK codebase, but there's no
> > telling what others might have done with it (have an array of them, embed in
> > other structs etc).
> > 
> > Also ultimately ABI compatibility goes both ways: when the library soname
> > does not change then an application is free to assume both downgrading and
> > upgrading are safe. In this case, upgrading *might* be okay, downgrading
> > certainly is not. So by that measure it definitely is an ABI break.
> > 
> > [...]
> > > diff --git a/drivers/net/bonding/rte_eth_bond_version.map
> > > b/drivers/net/bonding/rte_eth_bond_version.map
> > > index 22bd920..33d73ff 100644
> > > --- a/drivers/net/bonding/rte_eth_bond_version.map
> > > +++ b/drivers/net/bonding/rte_eth_bond_version.map
> > > @@ -27,3 +27,9 @@ DPDK_2.1 {
> > >   	rte_eth_bond_free;
> > >
> > >   } DPDK_2.0;
> > > +
> > > +DPDK_2.2 {
> > > +	rte_eth_bond_8023ad_ext_collect;
> > > +	rte_eth_bond_8023ad_ext_distrib;
> > > +	rte_eth_bond_8023ad_ext_slowtx;
> > > +} DPDK_2.1;
> > >
> > 
> > These symbols are not part of DPDK 2.2, the version here is wrong.
> > Technically it would not actually matter much but better not to confuse
> > things unnecessarily.
> > 
> > 	- Panu -
> 
> It looks like Panu's points are valid, a V3 of this patch set which takes care of these issues will be needed.
> 
> Patches 1/6, 5/6 and 6/6 of the patch set are bug fixes, so each patch should contain a fixes line.
> Patches 2/6, 3/6 and 4/6 are a new feature, the release notes should be updated for this feature.
> 
> Could I suggest splitting the patch set into two patch sets, a bug fix patch set and a new feature patch set.
> 
> Regards,
> 
> Bernard.

Bernard, a v3 is on the way.  I included only things that are fixes,
but the patch set doesn't quite match the set of patch numbers you listed
above.  1/6 (bond: use existing enslaved device queues) is an improvement,
but doesn't really fix anything that was broken, so I left that out.
2/6 (bond mode 4: copy entire config structure) and 3/6 (bond mode 4:
do not ignore multicast) fix bugs and are included.


Eric

Re: [PATCH v2 4/6] bond mode 4: allow external state machine

[Iremonger, Bernard]

Hi Eric,

<snip>

> > > > @@ -157,6 +159,7 @@ struct rte_eth_bond_8023ad_conf {
> > > >   	uint32_t tx_period_ms;
> > > >   	uint32_t rx_marker_period_ms;
> > > >   	uint32_t update_timeout_ms;
> > > > +	rte_eth_bond_8023ad_ext_slowrx_fn slowrx_cb;
> > > >   };
> > >
> > > This still is a likely an ABI break, previously discussed around here:
> > > http://dpdk.org/ml/archives/dev/2015-November/027321.html
> > >
> > > It might not be embedded anywhere in DPDK codebase, but there's no
> > > telling what others might have done with it (have an array of them,
> > > embed in other structs etc).
> > >
> > > Also ultimately ABI compatibility goes both ways: when the library
> > > soname does not change then an application is free to assume both
> > > downgrading and upgrading are safe. In this case, upgrading *might*
> > > be okay, downgrading certainly is not. So by that measure it definitely is
> an ABI break.
> > >
> > > [...]
> > > > diff --git a/drivers/net/bonding/rte_eth_bond_version.map
> > > > b/drivers/net/bonding/rte_eth_bond_version.map
> > > > index 22bd920..33d73ff 100644
> > > > --- a/drivers/net/bonding/rte_eth_bond_version.map
> > > > +++ b/drivers/net/bonding/rte_eth_bond_version.map
> > > > @@ -27,3 +27,9 @@ DPDK_2.1 {
> > > >   	rte_eth_bond_free;
> > > >
> > > >   } DPDK_2.0;
> > > > +
> > > > +DPDK_2.2 {
> > > > +	rte_eth_bond_8023ad_ext_collect;
> > > > +	rte_eth_bond_8023ad_ext_distrib;
> > > > +	rte_eth_bond_8023ad_ext_slowtx;
> > > > +} DPDK_2.1;
> > > >
> > >
> > > These symbols are not part of DPDK 2.2, the version here is wrong.
> > > Technically it would not actually matter much but better not to
> > > confuse things unnecessarily.
> > >
> > > 	- Panu -
> >
> > It looks like Panu's points are valid, a V3 of this patch set which takes care of
> these issues will be needed.
> >
> > Patches 1/6, 5/6 and 6/6 of the patch set are bug fixes, so each patch
> should contain a fixes line.
> > Patches 2/6, 3/6 and 4/6 are a new feature, the release notes should be
> updated for this feature.
> >
> > Could I suggest splitting the patch set into two patch sets, a bug fix patch
> set and a new feature patch set.
> >
> > Regards,
> >
> > Bernard.
> 
> Bernard, a v3 is on the way.  I included only things that are fixes, but the
> patch set doesn't quite match the set of patch numbers you listed above.
> 1/6 (bond: use existing enslaved device queues) is an improvement, but
> doesn't really fix anything that was broken, so I left that out.
> 2/6 (bond mode 4: copy entire config structure) and 3/6 (bond mode 4:
> do not ignore multicast) fix bugs and are included.
> 
> 
> Eric

Thanks for the V3 patchset and clarifying which patches in the V2 patchset were fixes.

Regards,

Bernard.

  

[PATCH v2 5/6] bond: active slaves with no primary

[Eric Kinzie]

From: Eric Kinzie <ekinzie@brocade.com>

If the link state of a slave is "up" when added, it is added to the list
of active slaves but, even if it is the only slave, is not selected as
the primary interface.  Generally, handling of link state interrupts
selects an interface to be primary, but only if the active count is zero.
This change avoids the situation where there are active slaves but
no primary.

Signed-off-by: Eric Kinzie <ekinzie@brocade.com>
Signed-off-by: Stephen Hemminger <stephen@networkplumber.org>
Acked-by: Declan Doherty <declan.doherty@intel.com>
---
 drivers/net/bonding/rte_eth_bond_api.c |    7 ++++++-
 1 file changed, 6 insertions(+), 1 deletion(-)

diff --git a/drivers/net/bonding/rte_eth_bond_api.c b/drivers/net/bonding/rte_eth_bond_api.c
index 8a000c8..630a461 100644
--- a/drivers/net/bonding/rte_eth_bond_api.c
+++ b/drivers/net/bonding/rte_eth_bond_api.c
@@ -427,8 +427,13 @@ __eth_bond_slave_add_lock_free(uint8_t bonded_port_id, uint8_t slave_port_id)
 	if (bonded_eth_dev->data->dev_started) {
 		rte_eth_link_get_nowait(slave_port_id, &link_props);
 
-		 if (link_props.link_status == 1)
+		 if (link_props.link_status == 1) {
+			if (internals->active_slave_count == 0 &&
+			    !internals->user_defined_primary_port)
+				bond_ethdev_primary_set(internals,
+							slave_port_id);
 			activate_slave(bonded_eth_dev, slave_port_id);
+		}
 	}
 	return 0;
 
-- 
1.7.10.4

[PATCH v2 6/6] bond: do not activate slave twice

[Eric Kinzie]

From: Eric Kinzie <ekinzie@brocade.com>

The current code for detecting link during slave addition can cause a
slave interface to be activated twice -- once during slave_configure()
and again at the end of __eth_bond_slave_add_lock_free().  This will
either cause the active slave count to be incorrect or will cause the
802.3ad activation function to panic.  Ensure that the interface is not
activated more than once.

Signed-off-by: Eric Kinzie <ekinzie@brocade.com>
Signed-off-by: Stephen Hemminger <stephen@networkplumber.org>
Acked-by: Declan Doherty <declan.doherty@intel.com>
---
 drivers/net/bonding/rte_eth_bond_api.c |    6 +++++-
 1 file changed, 5 insertions(+), 1 deletion(-)

diff --git a/drivers/net/bonding/rte_eth_bond_api.c b/drivers/net/bonding/rte_eth_bond_api.c
index 630a461..def22d0 100644
--- a/drivers/net/bonding/rte_eth_bond_api.c
+++ b/drivers/net/bonding/rte_eth_bond_api.c
@@ -432,7 +432,11 @@ __eth_bond_slave_add_lock_free(uint8_t bonded_port_id, uint8_t slave_port_id)
 			    !internals->user_defined_primary_port)
 				bond_ethdev_primary_set(internals,
 							slave_port_id);
-			activate_slave(bonded_eth_dev, slave_port_id);
+
+			if (find_slave_by_id(internals->active_slaves,
+					     internals->active_slave_count,
+					     slave_port_id) == internals->active_slave_count)
+				activate_slave(bonded_eth_dev, slave_port_id);
 		}
 	}
 	return 0;
-- 
1.7.10.4

[PATCH 7/8] bond: per-slave intermediate rx ring

[Stephen Hemminger]

From: Eric Kinzie <ekinzie@brocade.com>

Need to handle the case when bonding two (or more) ixgbe devices
together.  The existing code would break because the ixgbe devices
assumed that if setup in vector mode that the burst size would always
be large.

To solve this, during bond 802.3ad receive, a burst of packets is
fetched from each slave into a local array and appended to per-slave
ring buffer.  Packets are taken from the head of the ring buffer and
returned to the caller.  The number of mbufs provided to each slave is
sufficient to meet the requirements of the ixgbe vector receive.

Signed-off-by: Eric Kinzie <ekinzie@brocade.com>
Signed-off-by: Stephen Hemminger <stephen@networkplumber.org>
---
 drivers/net/bonding/rte_eth_bond_api.c     | 29 ++++++++++++
 drivers/net/bonding/rte_eth_bond_pmd.c     | 71 ++++++++++++++++++++++--------
 drivers/net/bonding/rte_eth_bond_private.h |  4 ++
 3 files changed, 86 insertions(+), 18 deletions(-)

diff --git a/drivers/net/bonding/rte_eth_bond_api.c b/drivers/net/bonding/rte_eth_bond_api.c
index 64058ff..91b3819 100644
--- a/drivers/net/bonding/rte_eth_bond_api.c
+++ b/drivers/net/bonding/rte_eth_bond_api.c
@@ -37,6 +37,8 @@
 #include <rte_malloc.h>
 #include <rte_ethdev.h>
 #include <rte_tcp.h>
+#include <rte_errno.h>
+#include <rte_lcore.h>
 
 #include "rte_eth_bond.h"
 #include "rte_eth_bond_private.h"
@@ -170,6 +172,7 @@ rte_eth_bond_create(const char *name, uint8_t mode, uint8_t socket_id)
 {
 	struct bond_dev_private *internals = NULL;
 	struct rte_eth_dev *eth_dev = NULL;
+	char mem_name[RTE_ETH_NAME_MAX_LEN];
 
 	/* now do all data allocation - for eth_dev structure, dummy pci driver
 	 * and internal (private) data
@@ -254,6 +257,18 @@ rte_eth_bond_create(const char *name, uint8_t mode, uint8_t socket_id)
 	memset(internals->active_slaves, 0, sizeof(internals->active_slaves));
 	memset(internals->slaves, 0, sizeof(internals->slaves));
 
+	snprintf(mem_name, RTE_DIM(mem_name), "%s_rx", name);
+	internals->rx_ring = rte_ring_lookup(mem_name);
+	if (internals->rx_ring == NULL) {
+		internals->rx_ring = rte_ring_create(mem_name,
+				     rte_align32pow2(PMD_BOND_RECV_RING_PKTS *
+						     rte_lcore_count()),
+				     socket_id, 0);
+		if (internals->rx_ring == NULL)
+			rte_panic("%s: Failed to create rx ring '%s': %s\n", name,
+				  mem_name, rte_strerror(rte_errno));
+	}
+
 	/* Set mode 4 default configuration */
 	bond_mode_8023ad_setup(eth_dev, NULL);
 	if (bond_ethdev_mode_set(eth_dev, mode)) {
@@ -532,12 +547,26 @@ __eth_bond_slave_remove_lock_free(uint8_t bonded_port_id, uint8_t slave_port_id)
 			memset(rte_eth_devices[bonded_port_id].data->mac_addrs, 0,
 					sizeof(*(rte_eth_devices[bonded_port_id].data->mac_addrs)));
 	}
+
 	if (internals->slave_count == 0) {
+		/* Remove any remaining packets in the receive ring */
+		struct rte_mbuf *bufs[PMD_BOND_RECV_PKTS_PER_SLAVE];
+		unsigned j, count;
+
 		internals->rx_offload_capa = 0;
 		internals->tx_offload_capa = 0;
 		internals->flow_type_rss_offloads = ETH_RSS_PROTO_MASK;
 		internals->reta_size = 0;
+
+		do {
+			count = rte_ring_dequeue_burst(internals->rx_ring,
+					 (void **)bufs,
+					 PMD_BOND_RECV_PKTS_PER_SLAVE);
+			for  (j = 0; j < count; j++)
+				rte_pktmbuf_free(bufs[j]);
+		} while (count > 0);
 	}
+
 	return 0;
 }
 
diff --git a/drivers/net/bonding/rte_eth_bond_pmd.c b/drivers/net/bonding/rte_eth_bond_pmd.c
index 868e66b..043160e 100644
--- a/drivers/net/bonding/rte_eth_bond_pmd.c
+++ b/drivers/net/bonding/rte_eth_bond_pmd.c
@@ -180,10 +180,15 @@ bond_ethdev_rx_burst_8023ad(void *queue, struct rte_mbuf **bufs,
 	struct bond_dev_private *internals = bd_rx_q->dev_private;
 	struct ether_addr bond_mac;
 
+	unsigned rx_ring_avail = rte_ring_free_count(internals->rx_ring);
+	struct rte_mbuf *mbuf_bounce[PMD_BOND_RECV_PKTS_PER_SLAVE];
+
 	struct ether_hdr *hdr;
 
 	const uint16_t ether_type_slow_be = rte_be_to_cpu_16(ETHER_TYPE_SLOW);
 	uint16_t num_rx_total = 0;	/* Total number of received packets */
+	uint16_t num_rx_slave;
+	uint16_t num_enq_slave;
 	uint8_t slaves[RTE_MAX_ETHPORTS];
 	uint8_t slave_count;
 
@@ -191,6 +196,9 @@ bond_ethdev_rx_burst_8023ad(void *queue, struct rte_mbuf **bufs,
 	const uint8_t promisc = internals->promiscuous_en;
 	uint8_t i, j, k;
 
+	if (rx_ring_avail < PMD_BOND_RECV_PKTS_PER_SLAVE)
+		goto dequeue;
+
 	rte_eth_macaddr_get(internals->port_id, &bond_mac);
 	/* Copy slave list to protect against slave up/down changes during tx
 	 * bursting */
@@ -198,23 +206,27 @@ bond_ethdev_rx_burst_8023ad(void *queue, struct rte_mbuf **bufs,
 	slave_count = bond_active_slaves_by_rxqid(internals, bd_rx_q->queue_id,
 						  slaves);
 
-	for (i = 0; i < slave_count && num_rx_total < nb_pkts; i++) {
-		j = num_rx_total;
+	for (i = 0; i < slave_count && num_rx_total < rx_ring_avail; i++) {
+		j = 0;
 		collecting = ACTOR_STATE(&mode_8023ad_ports[slaves[i]], COLLECTING);
 
 		/* Read packets from this slave */
-		num_rx_total += rte_eth_rx_burst(slaves[i], bd_rx_q->queue_id,
-				&bufs[num_rx_total], nb_pkts - num_rx_total);
-
-		for (k = j; k < 2 && k < num_rx_total; k++)
-			rte_prefetch0(rte_pktmbuf_mtod(bufs[k], void *));
+		if (unlikely(rx_ring_avail - num_rx_total < PMD_BOND_RECV_PKTS_PER_SLAVE))
+			continue;
+		num_rx_slave = rte_eth_rx_burst(slaves[i], bd_rx_q->queue_id,
+						 mbuf_bounce,
+						 PMD_BOND_RECV_PKTS_PER_SLAVE);
+		for (k = j; k < 2 && k < num_rx_slave; k++)
+			rte_prefetch0(rte_pktmbuf_mtod(mbuf_bounce[k], void *));
 
 		/* Handle slow protocol packets. */
-		while (j < num_rx_total) {
-			if (j + 3 < num_rx_total)
-				rte_prefetch0(rte_pktmbuf_mtod(bufs[j + 3], void *));
+		while (j < num_rx_slave) {
+			if (j + 3 < num_rx_slave)
+				rte_prefetch0(rte_pktmbuf_mtod(
+					      mbuf_bounce[j + 3],
+					      void *));
 
-			hdr = rte_pktmbuf_mtod(bufs[j], struct ether_hdr *);
+			hdr = rte_pktmbuf_mtod(mbuf_bounce[j], struct ether_hdr *);
 			/* Remove packet from array if it is slow packet or slave is not
 			 * in collecting state or bondign interface is not in promiscus
 			 * mode and packet address does not match. */
@@ -225,22 +237,45 @@ bond_ethdev_rx_burst_8023ad(void *queue, struct rte_mbuf **bufs,
 
 				if (hdr->ether_type == ether_type_slow_be) {
 					bond_mode_8023ad_handle_slow_pkt(internals, slaves[i],
-						bufs[j]);
+						mbuf_bounce[j]);
 				} else
-					rte_pktmbuf_free(bufs[j]);
+					rte_pktmbuf_free(mbuf_bounce[j]);
 
 				/* Packet is managed by mode 4 or dropped, shift the array */
-				num_rx_total--;
-				if (j < num_rx_total) {
-					memmove(&bufs[j], &bufs[j + 1], sizeof(bufs[0]) *
-						(num_rx_total - j));
+				num_rx_slave--;
+				if (j < num_rx_slave) {
+					memmove(&mbuf_bounce[j],
+						&mbuf_bounce[j + 1],
+						sizeof(mbuf_bounce[0]) *
+						(num_rx_slave - j));
 				}
 			} else
 				j++;
 		}
+
+		if (num_rx_slave > 0) {
+			if (mbuf_bounce[0] == NULL)
+				RTE_LOG(ERR, PMD, "%s: Enqueue a NULL??\n",
+					__func__);
+
+			num_enq_slave = rte_ring_enqueue_burst(internals->rx_ring,
+							       (void **)mbuf_bounce,
+							       num_rx_slave);
+
+			if (num_enq_slave < num_rx_slave) {
+				RTE_LOG(ERR, PMD,
+					"%s: failed to enqueue %u packets",
+					__func__,
+					(num_rx_slave - num_enq_slave));
+				for (j = num_enq_slave; j < num_rx_slave; j++)
+					rte_pktmbuf_free(mbuf_bounce[j]);
+			}
+			num_rx_total += num_enq_slave;
+		}
 	}
 
-	return num_rx_total;
+dequeue:
+	return rte_ring_dequeue_burst(internals->rx_ring, (void **)bufs, nb_pkts);
 }
 
 #if defined(RTE_LIBRTE_BOND_DEBUG_ALB) || defined(RTE_LIBRTE_BOND_DEBUG_ALB_L1)
diff --git a/drivers/net/bonding/rte_eth_bond_private.h b/drivers/net/bonding/rte_eth_bond_private.h
index 02f6de1..2c43bc3 100644
--- a/drivers/net/bonding/rte_eth_bond_private.h
+++ b/drivers/net/bonding/rte_eth_bond_private.h
@@ -50,6 +50,8 @@
 #define PMD_BOND_LSC_POLL_PERIOD_KVARG		("lsc_poll_period_ms")
 #define PMD_BOND_LINK_UP_PROP_DELAY_KVARG	("up_delay")
 #define PMD_BOND_LINK_DOWN_PROP_DELAY_KVARG	("down_delay")
+#define PMD_BOND_RECV_RING_PKTS			512
+#define PMD_BOND_RECV_PKTS_PER_SLAVE		32
 
 #define PMD_BOND_XMIT_POLICY_LAYER2_KVARG	("l2")
 #define PMD_BOND_XMIT_POLICY_LAYER23_KVARG	("l23")
@@ -171,6 +173,8 @@ struct bond_dev_private {
 
 	struct rte_kvargs *kvlist;
 	uint8_t slave_update_idx;
+
+	struct rte_ring *rx_ring;
 };
 
 extern struct eth_dev_ops default_dev_ops;
-- 
2.1.4

[PATCH 8/8] bond: do not activate slave twice

[Stephen Hemminger]

From: Eric Kinzie <ekinzie@brocade.com>

The current code for detecting link during slave addition can cause a
slave interface to be activated twice -- once during slave_configure()
and again at the end of __eth_bond_slave_add_lock_free().  This will
either cause the active slave count to be incorrect or will cause the
802.3ad activation function to panic.  Ensure that the interface is not
activated more than once.

Signed-off-by: Eric Kinzie <ekinzie@brocade.com>
Signed-off-by: Stephen Hemminger <stephen@networkplumber.org>
---
 drivers/net/bonding/rte_eth_bond_api.c | 6 +++++-
 1 file changed, 5 insertions(+), 1 deletion(-)

diff --git a/drivers/net/bonding/rte_eth_bond_api.c b/drivers/net/bonding/rte_eth_bond_api.c
index 91b3819..956a87d 100644
--- a/drivers/net/bonding/rte_eth_bond_api.c
+++ b/drivers/net/bonding/rte_eth_bond_api.c
@@ -447,7 +447,11 @@ __eth_bond_slave_add_lock_free(uint8_t bonded_port_id, uint8_t slave_port_id)
 			    !internals->user_defined_primary_port)
 				bond_ethdev_primary_set(internals,
 							slave_port_id);
-			activate_slave(bonded_eth_dev, slave_port_id);
+
+			if (find_slave_by_id(internals->active_slaves,
+					     internals->active_slave_count,
+					     slave_port_id) == internals->active_slave_count)
+				activate_slave(bonded_eth_dev, slave_port_id);
 		}
 	}
 	return 0;
-- 
2.1.4

Re: [PATCH 8/8] bond: do not activate slave twice

[Declan Doherty]

On 04/12/15 17:14, Stephen Hemminger wrote:
> From: Eric Kinzie <ekinzie@brocade.com>
>
> The current code for detecting link during slave addition can cause a
> slave interface to be activated twice -- once during slave_configure()
> and again at the end of __eth_bond_slave_add_lock_free().  This will
> either cause the active slave count to be incorrect or will cause the
> 802.3ad activation function to panic.  Ensure that the interface is not
> activated more than once.
>
> Signed-off-by: Eric Kinzie <ekinzie@brocade.com>
> Signed-off-by: Stephen Hemminger <stephen@networkplumber.org>
> ---
...
 >

Acked-by: Declan Doherty <declan.doherty@intel.com>

Re: [PATCH 0/8] bonding: fixes and enhancements

[Iremonger, Bernard]

Hi Stephen,

> -----Original Message-----
> From: dev [mailto:dev-bounces@dpdk.org] On Behalf Of Stephen
> Hemminger
> Sent: Friday, December 4, 2015 5:14 PM
> To: Doherty, Declan <declan.doherty@intel.com>
> Cc: dev@dpdk.org
> Subject: [dpdk-dev] [PATCH 0/8] bonding: fixes and enhancements
> 
> These are bug fixes and some small enhancements to allow bonding to work
> with external control (teamd). Please consider integrating these into DPDK
> 2.2
> 
> Eric Kinzie (8):
>   bond: use existing enslaved device queues
>   bond mode 4: copy entire config structure
>   bond mode 4: do not ignore multicast
>   bond mode 4: allow external state machine
>   bond: active slaves with no primary
>   bond: handle slaves with fewer queues than bonding device
>   bond: per-slave intermediate rx ring
>   bond: do not activate slave twice
> 
>  app/test/test_link_bonding_mode4.c                |   7 +-
>  drivers/net/bonding/rte_eth_bond_8023ad.c         | 174
> +++++++++++++++++
>  drivers/net/bonding/rte_eth_bond_8023ad.h         |  44 +++++
>  drivers/net/bonding/rte_eth_bond_8023ad_private.h |   2 +
>  drivers/net/bonding/rte_eth_bond_api.c            |  48 ++++-
>  drivers/net/bonding/rte_eth_bond_pmd.c            | 217
> ++++++++++++++++++----
>  drivers/net/bonding/rte_eth_bond_private.h        |   9 +-
>  drivers/net/bonding/rte_eth_bond_version.map      |   6 +
>  8 files changed, 462 insertions(+), 45 deletions(-)
> 
> --
> 2.1.4

Patches 6 and 7 of this patchset do not apply successfully to DPDK 2.2, a rebase is probably needed.
It might be better to split this patchset into a fixes patchset and a new feature patchset.

Regards,

Bernard.