[PATCH net 0/8] amt: fix validation and synchronization bugs

[PATCH net 0/8] amt: fix validation and synchronization bugs

[Taehee Yoo]

There are some synchronization issues in the amt module.
Especially, an amt gateway doesn't well synchronize its own variables
and status(amt->status).
It tries to use a workqueue for handles in a single thread.
A global lock is also good, but it would occur complex locking complex.

In this patchset, only the gateway uses workqueue.
The reason why only gateway interface uses workqueue is that gateway
should manage its own states and variables a little bit statefully.
But relay doesn't need to manage tunnels statefully, stateless is okay.
So, relay side message handlers are okay to be called concurrently.
But it doesn't mean that no lock is needed.

Only amt multicast data message type will not be processed by the work
queue because It contains actual multicast data.
So, it should be processed immediately.

When any amt gateway events are triggered(sending discovery message by
delayed_work, sending request message by delayed_work and receiving
messages), it stores event and skb into the event queue(amt->events[16]).
Then, workqueue processes these events one by one.

The first patch is to use the work queue.

The second patch is to remove unnecessary lock due to a previous patch.

The third patch is to use READ_ONCE() in the amt module.
Even if the amt module uses a single thread, some variables (ready4,
ready6, amt->status) can be accessed concurrently.

The fourth patch is to add missing nonce generation logic when it sends a
new request message.

The fifth patch is to drop unexpected advertisement messages.
advertisement message should be received only after the gateway sends
a discovery message first.
So, the gateway should drop advertisement messages if it has never
sent a discovery message and it also should drop duplicate advertisement
messages.
Using nonce is good to distinguish whether a received message is an
expected message or not.

The sixth patch is to drop unexpected query messages.
This is the same behavior as the fourth patch.
Query messages should be received only after the gateway sends a request
message first.
The nonce variable is used to distinguish whether it is a reply to a
previous request message or not.
amt->ready4 and amt->ready6 are used to distinguish duplicate messages.

The seventh patch is to drop unexpected multicast data.
AMT gateway should not receive multicast data message type before
establish between gateway and relay.
In order to drop unexpected multicast data messages, it checks amt->status.

The last patch is to fix a locking problem on the relay side.
amt->nr_tunnels variable is protected by amt->lock.
But amt_request_handler() doesn't protect this variable.

Taehee Yoo (8):
  amt: use workqueue for gateway side message handling
  amt: remove unnecessary locks
  amt: use READ_ONCE() in amt module
  amt: add missing regeneration nonce logic in request logic
  amt: drop unexpected advertisement message
  amt: drop unexpected query message
  amt: drop unexpected multicast data
  amt: do not use amt->nr_tunnels outside of lock

 drivers/net/amt.c | 238 ++++++++++++++++++++++++++++++++++++----------
 include/net/amt.h |  20 ++++
 2 files changed, 206 insertions(+), 52 deletions(-)

-- 
2.17.1

[PATCH net 1/8] amt: use workqueue for gateway side message handling

[Taehee Yoo]

There are some synchronization issues(amt->status, amt->req_cnt, etc)
if the interface is in gateway mode because gateway message handlers
are processed concurrently.
This applies a work queue for processing these messages instead of
expanding the locking context.

So, the purposes of this patch are to fix exist race conditions and to make
gateway to be able to validate a gateway status more correctly.

When the AMT gateway interface is created, it tries to establish to relay.
The establishment step looks stateless, but it should be managed well.
In order to handle messages in the gateway, it saves the current
status(i.e. AMT_STATUS_XXX).
This patch makes gateway code to be worked with a single thread.

Now, all messages except the multicast are triggered(received or
delay expired), and these messages will be stored in the event
queue(amt->events).
Then, the single worker processes stored messages asynchronously one
by one.
The multicast data message type will be still processed immediately.

Now, amt->lock is only needed to access the event queue(amt->events)
if an interface is the gateway mode.

Fixes: cbc21dc1cfe9 ("amt: add data plane of amt interface")
Signed-off-by: Taehee Yoo <ap420073@gmail.com>
---
 drivers/net/amt.c | 158 +++++++++++++++++++++++++++++++++++++++++-----
 include/net/amt.h |  20 ++++++
 2 files changed, 163 insertions(+), 15 deletions(-)

diff --git a/drivers/net/amt.c b/drivers/net/amt.c
index be2719a3ba70..032c2934e466 100644
--- a/drivers/net/amt.c
+++ b/drivers/net/amt.c
@@ -900,6 +900,28 @@ static void amt_send_mld_gq(struct amt_dev *amt, struct amt_tunnel_list *tunnel)
 }
 #endif
 
+static bool amt_queue_events(struct amt_dev *amt, enum amt_event event,
+			     struct sk_buff *skb)
+{
+	int index;
+
+	spin_lock_bh(&amt->lock);
+	if (amt->nr_events >= AMT_MAX_EVENTS) {
+		spin_unlock_bh(&amt->lock);
+		return 1;
+	}
+
+	index = (amt->event_idx + amt->nr_events) % AMT_MAX_EVENTS;
+	amt->events[index].event = event;
+	amt->events[index].skb = skb;
+	amt->nr_events++;
+	amt->event_idx %= AMT_MAX_EVENTS;
+	queue_work(amt_wq, &amt->event_wq);
+	spin_unlock_bh(&amt->lock);
+
+	return 0;
+}
+
 static void amt_secret_work(struct work_struct *work)
 {
 	struct amt_dev *amt = container_of(to_delayed_work(work),
@@ -913,12 +935,8 @@ static void amt_secret_work(struct work_struct *work)
 			 msecs_to_jiffies(AMT_SECRET_TIMEOUT));
 }
 
-static void amt_discovery_work(struct work_struct *work)
+static void amt_event_send_discovery(struct amt_dev *amt)
 {
-	struct amt_dev *amt = container_of(to_delayed_work(work),
-					   struct amt_dev,
-					   discovery_wq);
-
 	spin_lock_bh(&amt->lock);
 	if (amt->status > AMT_STATUS_SENT_DISCOVERY)
 		goto out;
@@ -933,11 +951,19 @@ static void amt_discovery_work(struct work_struct *work)
 	spin_unlock_bh(&amt->lock);
 }
 
-static void amt_req_work(struct work_struct *work)
+static void amt_discovery_work(struct work_struct *work)
 {
 	struct amt_dev *amt = container_of(to_delayed_work(work),
 					   struct amt_dev,
-					   req_wq);
+					   discovery_wq);
+
+	if (amt_queue_events(amt, AMT_EVENT_SEND_DISCOVERY, NULL))
+		mod_delayed_work(amt_wq, &amt->discovery_wq,
+				 msecs_to_jiffies(AMT_DISCOVERY_TIMEOUT));
+}
+
+static void amt_event_send_request(struct amt_dev *amt)
+{
 	u32 exp;
 
 	spin_lock_bh(&amt->lock);
@@ -967,6 +993,17 @@ static void amt_req_work(struct work_struct *work)
 	spin_unlock_bh(&amt->lock);
 }
 
+static void amt_req_work(struct work_struct *work)
+{
+	struct amt_dev *amt = container_of(to_delayed_work(work),
+					   struct amt_dev,
+					   req_wq);
+
+	if (amt_queue_events(amt, AMT_EVENT_SEND_REQUEST, NULL))
+		mod_delayed_work(amt_wq, &amt->req_wq,
+				 msecs_to_jiffies(100));
+}
+
 static bool amt_send_membership_update(struct amt_dev *amt,
 				       struct sk_buff *skb,
 				       bool v6)
@@ -2392,12 +2429,14 @@ static bool amt_membership_query_handler(struct amt_dev *amt,
 	skb->pkt_type = PACKET_MULTICAST;
 	skb->ip_summed = CHECKSUM_NONE;
 	len = skb->len;
+	rcu_read_lock_bh();
 	if (__netif_rx(skb) == NET_RX_SUCCESS) {
 		amt_update_gw_status(amt, AMT_STATUS_RECEIVED_QUERY, true);
 		dev_sw_netstats_rx_add(amt->dev, len);
 	} else {
 		amt->dev->stats.rx_dropped++;
 	}
+	rcu_read_unlock_bh();
 
 	return false;
 }
@@ -2688,6 +2727,38 @@ static bool amt_request_handler(struct amt_dev *amt, struct sk_buff *skb)
 	return false;
 }
 
+static void amt_gw_rcv(struct amt_dev *amt, struct sk_buff *skb)
+{
+	int type = amt_parse_type(skb);
+	int err = 1;
+
+	if (type == -1)
+		goto drop;
+
+	if (amt->mode == AMT_MODE_GATEWAY) {
+		switch (type) {
+		case AMT_MSG_ADVERTISEMENT:
+			err = amt_advertisement_handler(amt, skb);
+			break;
+		case AMT_MSG_MEMBERSHIP_QUERY:
+			err = amt_membership_query_handler(amt, skb);
+			if (!err)
+				return;
+			break;
+		default:
+			netdev_dbg(amt->dev, "Invalid type of Gateway\n");
+			break;
+		}
+	}
+drop:
+	if (err) {
+		amt->dev->stats.rx_dropped++;
+		kfree_skb(skb);
+	} else {
+		consume_skb(skb);
+	}
+}
+
 static int amt_rcv(struct sock *sk, struct sk_buff *skb)
 {
 	struct amt_dev *amt;
@@ -2719,8 +2790,12 @@ static int amt_rcv(struct sock *sk, struct sk_buff *skb)
 				err = true;
 				goto drop;
 			}
-			err = amt_advertisement_handler(amt, skb);
-			break;
+			if (amt_queue_events(amt, AMT_EVENT_RECEIVE, skb)) {
+				netdev_dbg(amt->dev, "AMT Event queue full\n");
+				err = true;
+				goto drop;
+			}
+			goto out;
 		case AMT_MSG_MULTICAST_DATA:
 			if (iph->saddr != amt->remote_ip) {
 				netdev_dbg(amt->dev, "Invalid Relay IP\n");
@@ -2738,11 +2813,12 @@ static int amt_rcv(struct sock *sk, struct sk_buff *skb)
 				err = true;
 				goto drop;
 			}
-			err = amt_membership_query_handler(amt, skb);
-			if (err)
+			if (amt_queue_events(amt, AMT_EVENT_RECEIVE, skb)) {
+				netdev_dbg(amt->dev, "AMT Event queue full\n");
+				err = true;
 				goto drop;
-			else
-				goto out;
+			}
+			goto out;
 		default:
 			err = true;
 			netdev_dbg(amt->dev, "Invalid type of Gateway\n");
@@ -2780,6 +2856,45 @@ static int amt_rcv(struct sock *sk, struct sk_buff *skb)
 	return 0;
 }
 
+static void amt_event_work(struct work_struct *work)
+{
+	struct amt_dev *amt = container_of(work, struct amt_dev, event_wq);
+	struct sk_buff *skb;
+	u8 event;
+
+	while (1) {
+		spin_lock(&amt->lock);
+		if (amt->nr_events == 0) {
+			spin_unlock(&amt->lock);
+			return;
+		}
+		event = amt->events[amt->event_idx].event;
+		skb = amt->events[amt->event_idx].skb;
+		amt->events[amt->event_idx].event = AMT_EVENT_NONE;
+		amt->events[amt->event_idx].skb = NULL;
+		amt->nr_events--;
+		amt->event_idx++;
+		amt->event_idx %= AMT_MAX_EVENTS;
+		spin_unlock(&amt->lock);
+
+		switch (event) {
+		case AMT_EVENT_RECEIVE:
+			amt_gw_rcv(amt, skb);
+			break;
+		case AMT_EVENT_SEND_DISCOVERY:
+			amt_event_send_discovery(amt);
+			break;
+		case AMT_EVENT_SEND_REQUEST:
+			amt_event_send_request(amt);
+			break;
+		default:
+			if (skb)
+				kfree_skb(skb);
+			break;
+		}
+	}
+}
+
 static int amt_err_lookup(struct sock *sk, struct sk_buff *skb)
 {
 	struct amt_dev *amt;
@@ -2892,10 +3007,21 @@ static int amt_dev_stop(struct net_device *dev)
 	struct amt_dev *amt = netdev_priv(dev);
 	struct amt_tunnel_list *tunnel, *tmp;
 	struct socket *sock;
+	struct sk_buff *skb;
+	int i;
 
 	cancel_delayed_work_sync(&amt->req_wq);
 	cancel_delayed_work_sync(&amt->discovery_wq);
 	cancel_delayed_work_sync(&amt->secret_wq);
+	cancel_work_sync(&amt->event_wq);
+
+	for (i = 0; i < AMT_MAX_EVENTS; i++) {
+		skb = amt->events[i].skb;
+		if (skb)
+			kfree_skb(skb);
+		amt->events[i].event = AMT_EVENT_NONE;
+		amt->events[i].skb = NULL;
+	}
 
 	/* shutdown */
 	sock = rtnl_dereference(amt->sock);
@@ -3051,6 +3177,8 @@ static int amt_newlink(struct net *net, struct net_device *dev,
 		amt->max_tunnels = AMT_MAX_TUNNELS;
 
 	spin_lock_init(&amt->lock);
+	amt->event_idx = 0;
+	amt->nr_events = 0;
 	amt->max_groups = AMT_MAX_GROUP;
 	amt->max_sources = AMT_MAX_SOURCE;
 	amt->hash_buckets = AMT_HSIZE;
@@ -3146,8 +3274,8 @@ static int amt_newlink(struct net *net, struct net_device *dev,
 	INIT_DELAYED_WORK(&amt->discovery_wq, amt_discovery_work);
 	INIT_DELAYED_WORK(&amt->req_wq, amt_req_work);
 	INIT_DELAYED_WORK(&amt->secret_wq, amt_secret_work);
+	INIT_WORK(&amt->event_wq, amt_event_work);
 	INIT_LIST_HEAD(&amt->tunnel_list);
-
 	return 0;
 err:
 	dev_put(amt->stream_dev);
@@ -3280,7 +3408,7 @@ static int __init amt_init(void)
 	if (err < 0)
 		goto unregister_notifier;
 
-	amt_wq = alloc_workqueue("amt", WQ_UNBOUND, 1);
+	amt_wq = alloc_workqueue("amt", WQ_UNBOUND, 0);
 	if (!amt_wq) {
 		err = -ENOMEM;
 		goto rtnl_unregister;
diff --git a/include/net/amt.h b/include/net/amt.h
index 0e40c3d64fcf..08fc30cf2f34 100644
--- a/include/net/amt.h
+++ b/include/net/amt.h
@@ -78,6 +78,15 @@ enum amt_status {
 
 #define AMT_STATUS_MAX (__AMT_STATUS_MAX - 1)
 
+/* Gateway events only */
+enum amt_event {
+	AMT_EVENT_NONE,
+	AMT_EVENT_RECEIVE,
+	AMT_EVENT_SEND_DISCOVERY,
+	AMT_EVENT_SEND_REQUEST,
+	__AMT_EVENT_MAX,
+};
+
 struct amt_header {
 #if defined(__LITTLE_ENDIAN_BITFIELD)
 	u8 type:4,
@@ -292,6 +301,12 @@ struct amt_group_node {
 	struct hlist_head	sources[];
 };
 
+#define AMT_MAX_EVENTS	16
+struct amt_events {
+	enum amt_event event;
+	struct sk_buff *skb;
+};
+
 struct amt_dev {
 	struct net_device       *dev;
 	struct net_device       *stream_dev;
@@ -308,6 +323,7 @@ struct amt_dev {
 	struct delayed_work     req_wq;
 	/* Protected by RTNL */
 	struct delayed_work     secret_wq;
+	struct work_struct	event_wq;
 	/* AMT status */
 	enum amt_status		status;
 	/* Generated key */
@@ -345,6 +361,10 @@ struct amt_dev {
 	/* Used only in gateway mode */
 	u64			mac:48,
 				reserved:16;
+	/* AMT gateway side message handler queue */
+	struct amt_events	events[AMT_MAX_EVENTS];
+	u8			event_idx;
+	u8			nr_events;
 };
 
 #define AMT_TOS			0xc0
-- 
2.17.1

Re: [PATCH net 1/8] amt: use workqueue for gateway side message handling

[Jakub Kicinski]

On Tue, 12 Jul 2022 10:57:07 +0000 Taehee Yoo wrote:
> @@ -2392,12 +2429,14 @@ static bool amt_membership_query_handler(struct amt_dev *amt,
>  	skb->pkt_type = PACKET_MULTICAST;
>  	skb->ip_summed = CHECKSUM_NONE;
>  	len = skb->len;
> +	rcu_read_lock_bh();
>  	if (__netif_rx(skb) == NET_RX_SUCCESS) {
>  		amt_update_gw_status(amt, AMT_STATUS_RECEIVED_QUERY, true);
>  		dev_sw_netstats_rx_add(amt->dev, len);
>  	} else {
>  		amt->dev->stats.rx_dropped++;
>  	}
> +	rcu_read_unlock_bh();
>  
>  	return false;
>  }

The RCU lock addition looks potentially unrelated?

> @@ -2892,10 +3007,21 @@ static int amt_dev_stop(struct net_device *dev)
>  	struct amt_dev *amt = netdev_priv(dev);
>  	struct amt_tunnel_list *tunnel, *tmp;
>  	struct socket *sock;
> +	struct sk_buff *skb;
> +	int i;
>  
>  	cancel_delayed_work_sync(&amt->req_wq);
>  	cancel_delayed_work_sync(&amt->discovery_wq);
>  	cancel_delayed_work_sync(&amt->secret_wq);
> +	cancel_work_sync(&amt->event_wq);

Are you sure the work will not get scheduled again?
What has stopped packet Rx at this point?

> +	for (i = 0; i < AMT_MAX_EVENTS; i++) {
> +		skb = amt->events[i].skb;
> +		if (skb)
> +			kfree_skb(skb);
> +		amt->events[i].event = AMT_EVENT_NONE;
> +		amt->events[i].skb = NULL;
> +	}
> 
>  	/* shutdown */
>  	sock = rtnl_dereference(amt->sock);
> @@ -3051,6 +3177,8 @@ static int amt_newlink(struct net *net, struct net_device *dev,
>  		amt->max_tunnels = AMT_MAX_TUNNELS;
>  
>  	spin_lock_init(&amt->lock);
> +	amt->event_idx = 0;
> +	amt->nr_events = 0;

no need to init member of netdev_priv() to 0, it's zalloc'ed

>  	amt->max_groups = AMT_MAX_GROUP;
>  	amt->max_sources = AMT_MAX_SOURCE;
>  	amt->hash_buckets = AMT_HSIZE;

Re: [PATCH net 1/8] amt: use workqueue for gateway side message handling

[Taehee Yoo]

Hi Jakub,
Thank you so much for your review!

On 7/14/22 12:55, Jakub Kicinski wrote:
 > On Tue, 12 Jul 2022 10:57:07 +0000 Taehee Yoo wrote:
 >> @@ -2392,12 +2429,14 @@ static bool 
amt_membership_query_handler(struct amt_dev *amt,
 >>   	skb->pkt_type = PACKET_MULTICAST;
 >>   	skb->ip_summed = CHECKSUM_NONE;
 >>   	len = skb->len;
 >> +	rcu_read_lock_bh();
 >>   	if (__netif_rx(skb) == NET_RX_SUCCESS) {
 >>   		amt_update_gw_status(amt, AMT_STATUS_RECEIVED_QUERY, true);
 >>   		dev_sw_netstats_rx_add(amt->dev, len);
 >>   	} else {
 >>   		amt->dev->stats.rx_dropped++;
 >>   	}
 >> +	rcu_read_unlock_bh();
 >>
 >>   	return false;
 >>   }
 >
 > The RCU lock addition looks potentially unrelated?
 >

I checked that RCU is not necessary here.
I tested local_bh_disable(), and it works well. no splats appeared.
So, as Paolo suggested, I will use local_bh_disable() instead of 
rcu_read_lock_bh() in the v2.

 >> @@ -2892,10 +3007,21 @@ static int amt_dev_stop(struct net_device *dev)
 >>   	struct amt_dev *amt = netdev_priv(dev);
 >>   	struct amt_tunnel_list *tunnel, *tmp;
 >>   	struct socket *sock;
 >> +	struct sk_buff *skb;
 >> +	int i;
 >>
 >>   	cancel_delayed_work_sync(&amt->req_wq);
 >>   	cancel_delayed_work_sync(&amt->discovery_wq);
 >>   	cancel_delayed_work_sync(&amt->secret_wq);
 >> +	cancel_work_sync(&amt->event_wq);
 >
 > Are you sure the work will not get scheduled again?
 > What has stopped packet Rx at this point?

I expected that RX was already stopped then ->ndo_stop() can be called.
But as you pointed out, it isn't.
In order to ensure that no more RX concurrently, amt_rcv() should not be 
called.
So, I think cancel_delayed_work() should be called after setting null to 
the amt socket in amt_dev_stop().
code looks like:
    RCU_INIT_POINTER(amt->sock, NULL);
    synchronize_net();
    cancel_delayed_work(&amt->event_wq).

 >
 >> +	for (i = 0; i < AMT_MAX_EVENTS; i++) {
 >> +		skb = amt->events[i].skb;
 >> +		if (skb)
 >> +			kfree_skb(skb);
 >> +		amt->events[i].event = AMT_EVENT_NONE;
 >> +		amt->events[i].skb = NULL;
 >> +	}
 >>
 >>   	/* shutdown */
 >>   	sock = rtnl_dereference(amt->sock);
 >> @@ -3051,6 +3177,8 @@ static int amt_newlink(struct net *net, struct 
net_device *dev,
 >>   		amt->max_tunnels = AMT_MAX_TUNNELS;
 >>
 >>   	spin_lock_init(&amt->lock);
 >> +	amt->event_idx = 0;
 >> +	amt->nr_events = 0;
 >
 > no need to init member of netdev_priv() to 0, it's zalloc'ed
 >

Thanks a lot for pointing it out.
This is my mistake.
If an amt interface is down and then up again, it would have incorrect 
values of event_idx and nr_events.
Because there is no init for these values.
So, init for these variable in ->ndo_open() or ->ndo_stop() is needed.
Initilization in the ->ndo_newlink() can't fix anything.

As a test, there is a bug certainly.
So, I will move it to ->ndo_open().

 >>   	amt->max_groups = AMT_MAX_GROUP;
 >>   	amt->max_sources = AMT_MAX_SOURCE;
 >>   	amt->hash_buckets = AMT_HSIZE;

I will send the v2 patch after some tests!

Thanks a lot for your review!
Taehee Yoo

Re: [PATCH net 1/8] amt: use workqueue for gateway side message handling

[Paolo Abeni]

On Tue, 2022-07-12 at 10:57 +0000, Taehee Yoo wrote:
> There are some synchronization issues(amt->status, amt->req_cnt, etc)
> if the interface is in gateway mode because gateway message handlers
> are processed concurrently.
> This applies a work queue for processing these messages instead of
> expanding the locking context.
> 
> So, the purposes of this patch are to fix exist race conditions and to make
> gateway to be able to validate a gateway status more correctly.
> 
> When the AMT gateway interface is created, it tries to establish to relay.
> The establishment step looks stateless, but it should be managed well.
> In order to handle messages in the gateway, it saves the current
> status(i.e. AMT_STATUS_XXX).
> This patch makes gateway code to be worked with a single thread.
> 
> Now, all messages except the multicast are triggered(received or
> delay expired), and these messages will be stored in the event
> queue(amt->events).
> Then, the single worker processes stored messages asynchronously one
> by one.
> The multicast data message type will be still processed immediately.
> 
> Now, amt->lock is only needed to access the event queue(amt->events)
> if an interface is the gateway mode.
> 
> Fixes: cbc21dc1cfe9 ("amt: add data plane of amt interface")
> Signed-off-by: Taehee Yoo <ap420073@gmail.com>
> ---
>  drivers/net/amt.c | 158 +++++++++++++++++++++++++++++++++++++++++-----
>  include/net/amt.h |  20 ++++++
>  2 files changed, 163 insertions(+), 15 deletions(-)
> 
> diff --git a/drivers/net/amt.c b/drivers/net/amt.c
> index be2719a3ba70..032c2934e466 100644
> --- a/drivers/net/amt.c
> +++ b/drivers/net/amt.c
> @@ -900,6 +900,28 @@ static void amt_send_mld_gq(struct amt_dev *amt, struct amt_tunnel_list *tunnel)
>  }
>  #endif
>  
> +static bool amt_queue_events(struct amt_dev *amt, enum amt_event event,
> +			     struct sk_buff *skb)
> +{
> +	int index;
> +
> +	spin_lock_bh(&amt->lock);
> +	if (amt->nr_events >= AMT_MAX_EVENTS) {
> +		spin_unlock_bh(&amt->lock);
> +		return 1;
> +	}
> +
> +	index = (amt->event_idx + amt->nr_events) % AMT_MAX_EVENTS;
> +	amt->events[index].event = event;
> +	amt->events[index].skb = skb;
> +	amt->nr_events++;
> +	amt->event_idx %= AMT_MAX_EVENTS;
> +	queue_work(amt_wq, &amt->event_wq);
> +	spin_unlock_bh(&amt->lock);
> +
> +	return 0;
> +}
> +
>  static void amt_secret_work(struct work_struct *work)
>  {
>  	struct amt_dev *amt = container_of(to_delayed_work(work),
> @@ -913,12 +935,8 @@ static void amt_secret_work(struct work_struct *work)
>  			 msecs_to_jiffies(AMT_SECRET_TIMEOUT));
>  }
>  
> -static void amt_discovery_work(struct work_struct *work)
> +static void amt_event_send_discovery(struct amt_dev *amt)
>  {
> -	struct amt_dev *amt = container_of(to_delayed_work(work),
> -					   struct amt_dev,
> -					   discovery_wq);
> -
>  	spin_lock_bh(&amt->lock);
>  	if (amt->status > AMT_STATUS_SENT_DISCOVERY)
>  		goto out;
> @@ -933,11 +951,19 @@ static void amt_discovery_work(struct work_struct *work)
>  	spin_unlock_bh(&amt->lock);
>  }
>  
> -static void amt_req_work(struct work_struct *work)
> +static void amt_discovery_work(struct work_struct *work)
>  {
>  	struct amt_dev *amt = container_of(to_delayed_work(work),
>  					   struct amt_dev,
> -					   req_wq);
> +					   discovery_wq);
> +
> +	if (amt_queue_events(amt, AMT_EVENT_SEND_DISCOVERY, NULL))
> +		mod_delayed_work(amt_wq, &amt->discovery_wq,
> +				 msecs_to_jiffies(AMT_DISCOVERY_TIMEOUT));
> +}
> +
> +static void amt_event_send_request(struct amt_dev *amt)
> +{
>  	u32 exp;
>  
>  	spin_lock_bh(&amt->lock);
> @@ -967,6 +993,17 @@ static void amt_req_work(struct work_struct *work)
>  	spin_unlock_bh(&amt->lock);
>  }
>  
> +static void amt_req_work(struct work_struct *work)
> +{
> +	struct amt_dev *amt = container_of(to_delayed_work(work),
> +					   struct amt_dev,
> +					   req_wq);
> +
> +	if (amt_queue_events(amt, AMT_EVENT_SEND_REQUEST, NULL))
> +		mod_delayed_work(amt_wq, &amt->req_wq,
> +				 msecs_to_jiffies(100));
> +}
> +
>  static bool amt_send_membership_update(struct amt_dev *amt,
>  				       struct sk_buff *skb,
>  				       bool v6)
> @@ -2392,12 +2429,14 @@ static bool amt_membership_query_handler(struct amt_dev *amt,
>  	skb->pkt_type = PACKET_MULTICAST;
>  	skb->ip_summed = CHECKSUM_NONE;
>  	len = skb->len;
> +	rcu_read_lock_bh();

Here you only need local_bh_disable(), the RCU part is confusing as
Jakub noted, and not needed.

>  	if (__netif_rx(skb) == NET_RX_SUCCESS) {
>  		amt_update_gw_status(amt, AMT_STATUS_RECEIVED_QUERY, true);
>  		dev_sw_netstats_rx_add(amt->dev, len);
>  	} else {
>  		amt->dev->stats.rx_dropped++;
>  	}
> +	rcu_read_unlock_bh();
>  
>  	return false;
>  }
> @@ -2688,6 +2727,38 @@ static bool amt_request_handler(struct amt_dev *amt, struct sk_buff *skb)
>  	return false;
>  }
>  
> +static void amt_gw_rcv(struct amt_dev *amt, struct sk_buff *skb)
> +{
> +	int type = amt_parse_type(skb);
> +	int err = 1;
> +
> +	if (type == -1)
> +		goto drop;
> +
> +	if (amt->mode == AMT_MODE_GATEWAY) {
> +		switch (type) {
> +		case AMT_MSG_ADVERTISEMENT:
> +			err = amt_advertisement_handler(amt, skb);
> +			break;
> +		case AMT_MSG_MEMBERSHIP_QUERY:
> +			err = amt_membership_query_handler(amt, skb);
> +			if (!err)
> +				return;
> +			break;
> +		default:
> +			netdev_dbg(amt->dev, "Invalid type of Gateway\n");
> +			break;
> +		}
> +	}
> +drop:
> +	if (err) {
> +		amt->dev->stats.rx_dropped++;
> +		kfree_skb(skb);
> +	} else {
> +		consume_skb(skb);
> +	}
> +}
> +
>  static int amt_rcv(struct sock *sk, struct sk_buff *skb)
>  {
>  	struct amt_dev *amt;
> @@ -2719,8 +2790,12 @@ static int amt_rcv(struct sock *sk, struct sk_buff *skb)
>  				err = true;
>  				goto drop;
>  			}
> -			err = amt_advertisement_handler(amt, skb);
> -			break;
> +			if (amt_queue_events(amt, AMT_EVENT_RECEIVE, skb)) {
> +				netdev_dbg(amt->dev, "AMT Event queue full\n");
> +				err = true;
> +				goto drop;
> +			}
> +			goto out;
>  		case AMT_MSG_MULTICAST_DATA:
>  			if (iph->saddr != amt->remote_ip) {
>  				netdev_dbg(amt->dev, "Invalid Relay IP\n");
> @@ -2738,11 +2813,12 @@ static int amt_rcv(struct sock *sk, struct sk_buff *skb)
>  				err = true;
>  				goto drop;
>  			}
> -			err = amt_membership_query_handler(amt, skb);
> -			if (err)
> +			if (amt_queue_events(amt, AMT_EVENT_RECEIVE, skb)) {
> +				netdev_dbg(amt->dev, "AMT Event queue full\n");
> +				err = true;
>  				goto drop;
> -			else
> -				goto out;
> +			}
> +			goto out;
>  		default:
>  			err = true;
>  			netdev_dbg(amt->dev, "Invalid type of Gateway\n");
> @@ -2780,6 +2856,45 @@ static int amt_rcv(struct sock *sk, struct sk_buff *skb)
>  	return 0;
>  }
>  
> +static void amt_event_work(struct work_struct *work)
> +{
> +	struct amt_dev *amt = container_of(work, struct amt_dev, event_wq);
> +	struct sk_buff *skb;
> +	u8 event;
> +
> +	while (1) {
> +		spin_lock(&amt->lock);

This is called in process context, amd amt->lock can be acquired from
BH context, you need spin_lock_bh() here.

Lockdep should help finding this kind of issue.

> +		if (amt->nr_events == 0) {
> +			spin_unlock(&amt->lock);
> +			return;
> +		}
> +		event = amt->events[amt->event_idx].event;
> +		skb = amt->events[amt->event_idx].skb;
> +		amt->events[amt->event_idx].event = AMT_EVENT_NONE;
> +		amt->events[amt->event_idx].skb = NULL;
> +		amt->nr_events--;
> +		amt->event_idx++;
> +		amt->event_idx %= AMT_MAX_EVENTS;
> +		spin_unlock(&amt->lock);
> +
> +		switch (event) {
> +		case AMT_EVENT_RECEIVE:
> +			amt_gw_rcv(amt, skb);
> +			break;
> +		case AMT_EVENT_SEND_DISCOVERY:
> +			amt_event_send_discovery(amt);
> +			break;
> +		case AMT_EVENT_SEND_REQUEST:
> +			amt_event_send_request(amt);
> +			break;
> +		default:
> +			if (skb)
> +				kfree_skb(skb);
> +			break;
> +		}

This loops is unbound. If the socket keep adding events, it can keep
running forever. You need either to add cond_schedule() or even better
break it after a low max number of iterations - pending event will be
served when the work struct is dequeued next

> +	}
> +}
> +
>  static int amt_err_lookup(struct sock *sk, struct sk_buff *skb)
>  {
>  	struct amt_dev *amt;
> @@ -2892,10 +3007,21 @@ static int amt_dev_stop(struct net_device *dev)
>  	struct amt_dev *amt = netdev_priv(dev);
>  	struct amt_tunnel_list *tunnel, *tmp;
>  	struct socket *sock;
> +	struct sk_buff *skb;
> +	int i;
>  
>  	cancel_delayed_work_sync(&amt->req_wq);
>  	cancel_delayed_work_sync(&amt->discovery_wq);
>  	cancel_delayed_work_sync(&amt->secret_wq);
> +	cancel_work_sync(&amt->event_wq);
> +
> +	for (i = 0; i < AMT_MAX_EVENTS; i++) {
> +		skb = amt->events[i].skb;
> +		if (skb)
> +			kfree_skb(skb);
> +		amt->events[i].event = AMT_EVENT_NONE;
> +		amt->events[i].skb = NULL;
> +	}
>  
>  	/* shutdown */
>  	sock = rtnl_dereference(amt->sock);
> @@ -3051,6 +3177,8 @@ static int amt_newlink(struct net *net, struct net_device *dev,
>  		amt->max_tunnels = AMT_MAX_TUNNELS;
>  
>  	spin_lock_init(&amt->lock);
> +	amt->event_idx = 0;
> +	amt->nr_events = 0;
>  	amt->max_groups = AMT_MAX_GROUP;
>  	amt->max_sources = AMT_MAX_SOURCE;
>  	amt->hash_buckets = AMT_HSIZE;
> @@ -3146,8 +3274,8 @@ static int amt_newlink(struct net *net, struct net_device *dev,
>  	INIT_DELAYED_WORK(&amt->discovery_wq, amt_discovery_work);
>  	INIT_DELAYED_WORK(&amt->req_wq, amt_req_work);
>  	INIT_DELAYED_WORK(&amt->secret_wq, amt_secret_work);
> +	INIT_WORK(&amt->event_wq, amt_event_work);
>  	INIT_LIST_HEAD(&amt->tunnel_list);
> -
>  	return 0;
>  err:
>  	dev_put(amt->stream_dev);
> @@ -3280,7 +3408,7 @@ static int __init amt_init(void)
>  	if (err < 0)
>  		goto unregister_notifier;
>  
> -	amt_wq = alloc_workqueue("amt", WQ_UNBOUND, 1);
> +	amt_wq = alloc_workqueue("amt", WQ_UNBOUND, 0);
>  	if (!amt_wq) {
>  		err = -ENOMEM;
>  		goto rtnl_unregister;
> diff --git a/include/net/amt.h b/include/net/amt.h
> index 0e40c3d64fcf..08fc30cf2f34 100644
> --- a/include/net/amt.h
> +++ b/include/net/amt.h
> @@ -78,6 +78,15 @@ enum amt_status {
>  
>  #define AMT_STATUS_MAX (__AMT_STATUS_MAX - 1)
>  
> +/* Gateway events only */
> +enum amt_event {
> +	AMT_EVENT_NONE,
> +	AMT_EVENT_RECEIVE,
> +	AMT_EVENT_SEND_DISCOVERY,
> +	AMT_EVENT_SEND_REQUEST,
> +	__AMT_EVENT_MAX,
> +};
> +
>  struct amt_header {
>  #if defined(__LITTLE_ENDIAN_BITFIELD)
>  	u8 type:4,
> @@ -292,6 +301,12 @@ struct amt_group_node {
>  	struct hlist_head	sources[];
>  };
>  
> +#define AMT_MAX_EVENTS	16
> +struct amt_events {
> +	enum amt_event event;
> +	struct sk_buff *skb;
> +};
> +
>  struct amt_dev {
>  	struct net_device       *dev;
>  	struct net_device       *stream_dev;
> @@ -308,6 +323,7 @@ struct amt_dev {
>  	struct delayed_work     req_wq;
>  	/* Protected by RTNL */
>  	struct delayed_work     secret_wq;
> +	struct work_struct	event_wq;
>  	/* AMT status */
>  	enum amt_status		status;
>  	/* Generated key */
> @@ -345,6 +361,10 @@ struct amt_dev {
>  	/* Used only in gateway mode */
>  	u64			mac:48,
>  				reserved:16;
> +	/* AMT gateway side message handler queue */
> +	struct amt_events	events[AMT_MAX_EVENTS];
> +	u8			event_idx;
> +	u8			nr_events;
>  };
>  
>  #define AMT_TOS			0xc0

Re: [PATCH net 1/8] amt: use workqueue for gateway side message handling

[Taehee Yoo]

Hi Paolo,
Thank you so much for your review!

On 7/14/22 17:09, Paolo Abeni wrote:
 > On Tue, 2022-07-12 at 10:57 +0000, Taehee Yoo wrote:
 >> There are some synchronization issues(amt->status, amt->req_cnt, etc)
 >> if the interface is in gateway mode because gateway message handlers
 >> are processed concurrently.
 >> This applies a work queue for processing these messages instead of
 >> expanding the locking context.
 >>
 >> So, the purposes of this patch are to fix exist race conditions and 
to make
 >> gateway to be able to validate a gateway status more correctly.
 >>
 >> When the AMT gateway interface is created, it tries to establish to 
relay.
 >> The establishment step looks stateless, but it should be managed well.
 >> In order to handle messages in the gateway, it saves the current
 >> status(i.e. AMT_STATUS_XXX).
 >> This patch makes gateway code to be worked with a single thread.
 >>
 >> Now, all messages except the multicast are triggered(received or
 >> delay expired), and these messages will be stored in the event
 >> queue(amt->events).
 >> Then, the single worker processes stored messages asynchronously one
 >> by one.
 >> The multicast data message type will be still processed immediately.
 >>
 >> Now, amt->lock is only needed to access the event queue(amt->events)
 >> if an interface is the gateway mode.
 >>
 >> Fixes: cbc21dc1cfe9 ("amt: add data plane of amt interface")
 >> Signed-off-by: Taehee Yoo <ap420073@gmail.com>
 >> ---
 >>   drivers/net/amt.c | 158 +++++++++++++++++++++++++++++++++++++++++-----
 >>   include/net/amt.h |  20 ++++++
 >>   2 files changed, 163 insertions(+), 15 deletions(-)
 >>
 >> diff --git a/drivers/net/amt.c b/drivers/net/amt.c
 >> index be2719a3ba70..032c2934e466 100644
 >> --- a/drivers/net/amt.c
 >> +++ b/drivers/net/amt.c
 >> @@ -900,6 +900,28 @@ static void amt_send_mld_gq(struct amt_dev 
*amt, struct amt_tunnel_list *tunnel)
 >>   }
 >>   #endif
 >>
 >> +static bool amt_queue_events(struct amt_dev *amt, enum amt_event event,
 >> +			     struct sk_buff *skb)
 >> +{
 >> +	int index;
 >> +
 >> +	spin_lock_bh(&amt->lock);
 >> +	if (amt->nr_events >= AMT_MAX_EVENTS) {
 >> +		spin_unlock_bh(&amt->lock);
 >> +		return 1;
 >> +	}
 >> +
 >> +	index = (amt->event_idx + amt->nr_events) % AMT_MAX_EVENTS;
 >> +	amt->events[index].event = event;
 >> +	amt->events[index].skb = skb;
 >> +	amt->nr_events++;
 >> +	amt->event_idx %= AMT_MAX_EVENTS;
 >> +	queue_work(amt_wq, &amt->event_wq);
 >> +	spin_unlock_bh(&amt->lock);
 >> +
 >> +	return 0;
 >> +}
 >> +
 >>   static void amt_secret_work(struct work_struct *work)
 >>   {
 >>   	struct amt_dev *amt = container_of(to_delayed_work(work),
 >> @@ -913,12 +935,8 @@ static void amt_secret_work(struct work_struct 
*work)
 >>   			 msecs_to_jiffies(AMT_SECRET_TIMEOUT));
 >>   }
 >>
 >> -static void amt_discovery_work(struct work_struct *work)
 >> +static void amt_event_send_discovery(struct amt_dev *amt)
 >>   {
 >> -	struct amt_dev *amt = container_of(to_delayed_work(work),
 >> -					   struct amt_dev,
 >> -					   discovery_wq);
 >> -
 >>   	spin_lock_bh(&amt->lock);
 >>   	if (amt->status > AMT_STATUS_SENT_DISCOVERY)
 >>   		goto out;
 >> @@ -933,11 +951,19 @@ static void amt_discovery_work(struct 
work_struct *work)
 >>   	spin_unlock_bh(&amt->lock);
 >>   }
 >>
 >> -static void amt_req_work(struct work_struct *work)
 >> +static void amt_discovery_work(struct work_struct *work)
 >>   {
 >>   	struct amt_dev *amt = container_of(to_delayed_work(work),
 >>   					   struct amt_dev,
 >> -					   req_wq);
 >> +					   discovery_wq);
 >> +
 >> +	if (amt_queue_events(amt, AMT_EVENT_SEND_DISCOVERY, NULL))
 >> +		mod_delayed_work(amt_wq, &amt->discovery_wq,
 >> +				 msecs_to_jiffies(AMT_DISCOVERY_TIMEOUT));
 >> +}
 >> +
 >> +static void amt_event_send_request(struct amt_dev *amt)
 >> +{
 >>   	u32 exp;
 >>
 >>   	spin_lock_bh(&amt->lock);
 >> @@ -967,6 +993,17 @@ static void amt_req_work(struct work_struct *work)
 >>   	spin_unlock_bh(&amt->lock);
 >>   }
 >>
 >> +static void amt_req_work(struct work_struct *work)
 >> +{
 >> +	struct amt_dev *amt = container_of(to_delayed_work(work),
 >> +					   struct amt_dev,
 >> +					   req_wq);
 >> +
 >> +	if (amt_queue_events(amt, AMT_EVENT_SEND_REQUEST, NULL))
 >> +		mod_delayed_work(amt_wq, &amt->req_wq,
 >> +				 msecs_to_jiffies(100));
 >> +}
 >> +
 >>   static bool amt_send_membership_update(struct amt_dev *amt,
 >>   				       struct sk_buff *skb,
 >>   				       bool v6)
 >> @@ -2392,12 +2429,14 @@ static bool 
amt_membership_query_handler(struct amt_dev *amt,
 >>   	skb->pkt_type = PACKET_MULTICAST;
 >>   	skb->ip_summed = CHECKSUM_NONE;
 >>   	len = skb->len;
 >> +	rcu_read_lock_bh();
 >
 > Here you only need local_bh_disable(), the RCU part is confusing as
 > Jakub noted, and not needed.

Thanks for suggesting, I will use local_bh_disabled() instead of 
rcu_read_lock_bh().

 >
 >>   	if (__netif_rx(skb) == NET_RX_SUCCESS) {
 >>   		amt_update_gw_status(amt, AMT_STATUS_RECEIVED_QUERY, true);
 >>   		dev_sw_netstats_rx_add(amt->dev, len);
 >>   	} else {
 >>   		amt->dev->stats.rx_dropped++;
 >>   	}
 >> +	rcu_read_unlock_bh();
 >>
 >>   	return false;
 >>   }
 >> @@ -2688,6 +2727,38 @@ static bool amt_request_handler(struct 
amt_dev *amt, struct sk_buff *skb)
 >>   	return false;
 >>   }
 >>
 >> +static void amt_gw_rcv(struct amt_dev *amt, struct sk_buff *skb)
 >> +{
 >> +	int type = amt_parse_type(skb);
 >> +	int err = 1;
 >> +
 >> +	if (type == -1)
 >> +		goto drop;
 >> +
 >> +	if (amt->mode == AMT_MODE_GATEWAY) {
 >> +		switch (type) {
 >> +		case AMT_MSG_ADVERTISEMENT:
 >> +			err = amt_advertisement_handler(amt, skb);
 >> +			break;
 >> +		case AMT_MSG_MEMBERSHIP_QUERY:
 >> +			err = amt_membership_query_handler(amt, skb);
 >> +			if (!err)
 >> +				return;
 >> +			break;
 >> +		default:
 >> +			netdev_dbg(amt->dev, "Invalid type of Gateway\n");
 >> +			break;
 >> +		}
 >> +	}
 >> +drop:
 >> +	if (err) {
 >> +		amt->dev->stats.rx_dropped++;
 >> +		kfree_skb(skb);
 >> +	} else {
 >> +		consume_skb(skb);
 >> +	}
 >> +}
 >> +
 >>   static int amt_rcv(struct sock *sk, struct sk_buff *skb)
 >>   {
 >>   	struct amt_dev *amt;
 >> @@ -2719,8 +2790,12 @@ static int amt_rcv(struct sock *sk, struct 
sk_buff *skb)
 >>   				err = true;
 >>   				goto drop;
 >>   			}
 >> -			err = amt_advertisement_handler(amt, skb);
 >> -			break;
 >> +			if (amt_queue_events(amt, AMT_EVENT_RECEIVE, skb)) {
 >> +				netdev_dbg(amt->dev, "AMT Event queue full\n");
 >> +				err = true;
 >> +				goto drop;
 >> +			}
 >> +			goto out;
 >>   		case AMT_MSG_MULTICAST_DATA:
 >>   			if (iph->saddr != amt->remote_ip) {
 >>   				netdev_dbg(amt->dev, "Invalid Relay IP\n");
 >> @@ -2738,11 +2813,12 @@ static int amt_rcv(struct sock *sk, struct 
sk_buff *skb)
 >>   				err = true;
 >>   				goto drop;
 >>   			}
 >> -			err = amt_membership_query_handler(amt, skb);
 >> -			if (err)
 >> +			if (amt_queue_events(amt, AMT_EVENT_RECEIVE, skb)) {
 >> +				netdev_dbg(amt->dev, "AMT Event queue full\n");
 >> +				err = true;
 >>   				goto drop;
 >> -			else
 >> -				goto out;
 >> +			}
 >> +			goto out;
 >>   		default:
 >>   			err = true;
 >>   			netdev_dbg(amt->dev, "Invalid type of Gateway\n");
 >> @@ -2780,6 +2856,45 @@ static int amt_rcv(struct sock *sk, struct 
sk_buff *skb)
 >>   	return 0;
 >>   }
 >>
 >> +static void amt_event_work(struct work_struct *work)
 >> +{
 >> +	struct amt_dev *amt = container_of(work, struct amt_dev, event_wq);
 >> +	struct sk_buff *skb;
 >> +	u8 event;
 >> +
 >> +	while (1) {
 >> +		spin_lock(&amt->lock);
 >
 > This is called in process context, amd amt->lock can be acquired from
 > BH context, you need spin_lock_bh() here.
 >
 > Lockdep should help finding this kind of issue.
 >

Ah, I found that lockdep is disabled in boot time due to other bugs.
I will use spin_lock_bh() and test with the lockdep!

 >> +		if (amt->nr_events == 0) {
 >> +			spin_unlock(&amt->lock);
 >> +			return;
 >> +		}
 >> +		event = amt->events[amt->event_idx].event;
 >> +		skb = amt->events[amt->event_idx].skb;
 >> +		amt->events[amt->event_idx].event = AMT_EVENT_NONE;
 >> +		amt->events[amt->event_idx].skb = NULL;
 >> +		amt->nr_events--;
 >> +		amt->event_idx++;
 >> +		amt->event_idx %= AMT_MAX_EVENTS;
 >> +		spin_unlock(&amt->lock);
 >> +
 >> +		switch (event) {
 >> +		case AMT_EVENT_RECEIVE:
 >> +			amt_gw_rcv(amt, skb);
 >> +			break;
 >> +		case AMT_EVENT_SEND_DISCOVERY:
 >> +			amt_event_send_discovery(amt);
 >> +			break;
 >> +		case AMT_EVENT_SEND_REQUEST:
 >> +			amt_event_send_request(amt);
 >> +			break;
 >> +		default:
 >> +			if (skb)
 >> +				kfree_skb(skb);
 >> +			break;
 >> +		}
 >
 > This loops is unbound. If the socket keep adding events, it can keep
 > running forever. You need either to add cond_schedule() or even better
 > break it after a low max number of iterations - pending event will be
 > served when the work struct is dequeued next
 >

Thanks, I will use a limit variable.


 >> +	}
 >> +}
 >> +
 >>   static int amt_err_lookup(struct sock *sk, struct sk_buff *skb)
 >>   {
 >>   	struct amt_dev *amt;
 >> @@ -2892,10 +3007,21 @@ static int amt_dev_stop(struct net_device *dev)
 >>   	struct amt_dev *amt = netdev_priv(dev);
 >>   	struct amt_tunnel_list *tunnel, *tmp;
 >>   	struct socket *sock;
 >> +	struct sk_buff *skb;
 >> +	int i;
 >>
 >>   	cancel_delayed_work_sync(&amt->req_wq);
 >>   	cancel_delayed_work_sync(&amt->discovery_wq);
 >>   	cancel_delayed_work_sync(&amt->secret_wq);
 >> +	cancel_work_sync(&amt->event_wq);
 >> +
 >> +	for (i = 0; i < AMT_MAX_EVENTS; i++) {
 >> +		skb = amt->events[i].skb;
 >> +		if (skb)
 >> +			kfree_skb(skb);
 >> +		amt->events[i].event = AMT_EVENT_NONE;
 >> +		amt->events[i].skb = NULL;
 >> +	}
 >>
 >>   	/* shutdown */
 >>   	sock = rtnl_dereference(amt->sock);
 >> @@ -3051,6 +3177,8 @@ static int amt_newlink(struct net *net, struct 
net_device *dev,
 >>   		amt->max_tunnels = AMT_MAX_TUNNELS;
 >>
 >>   	spin_lock_init(&amt->lock);
 >> +	amt->event_idx = 0;
 >> +	amt->nr_events = 0;
 >>   	amt->max_groups = AMT_MAX_GROUP;
 >>   	amt->max_sources = AMT_MAX_SOURCE;
 >>   	amt->hash_buckets = AMT_HSIZE;
 >> @@ -3146,8 +3274,8 @@ static int amt_newlink(struct net *net, struct 
net_device *dev,
 >>   	INIT_DELAYED_WORK(&amt->discovery_wq, amt_discovery_work);
 >>   	INIT_DELAYED_WORK(&amt->req_wq, amt_req_work);
 >>   	INIT_DELAYED_WORK(&amt->secret_wq, amt_secret_work);
 >> +	INIT_WORK(&amt->event_wq, amt_event_work);
 >>   	INIT_LIST_HEAD(&amt->tunnel_list);
 >> -
 >>   	return 0;
 >>   err:
 >>   	dev_put(amt->stream_dev);
 >> @@ -3280,7 +3408,7 @@ static int __init amt_init(void)
 >>   	if (err < 0)
 >>   		goto unregister_notifier;
 >>
 >> -	amt_wq = alloc_workqueue("amt", WQ_UNBOUND, 1);
 >> +	amt_wq = alloc_workqueue("amt", WQ_UNBOUND, 0);
 >>   	if (!amt_wq) {
 >>   		err = -ENOMEM;
 >>   		goto rtnl_unregister;
 >> diff --git a/include/net/amt.h b/include/net/amt.h
 >> index 0e40c3d64fcf..08fc30cf2f34 100644
 >> --- a/include/net/amt.h
 >> +++ b/include/net/amt.h
 >> @@ -78,6 +78,15 @@ enum amt_status {
 >>
 >>   #define AMT_STATUS_MAX (__AMT_STATUS_MAX - 1)
 >>
 >> +/* Gateway events only */
 >> +enum amt_event {
 >> +	AMT_EVENT_NONE,
 >> +	AMT_EVENT_RECEIVE,
 >> +	AMT_EVENT_SEND_DISCOVERY,
 >> +	AMT_EVENT_SEND_REQUEST,
 >> +	__AMT_EVENT_MAX,
 >> +};
 >> +
 >>   struct amt_header {
 >>   #if defined(__LITTLE_ENDIAN_BITFIELD)
 >>   	u8 type:4,
 >> @@ -292,6 +301,12 @@ struct amt_group_node {
 >>   	struct hlist_head	sources[];
 >>   };
 >>
 >> +#define AMT_MAX_EVENTS	16
 >> +struct amt_events {
 >> +	enum amt_event event;
 >> +	struct sk_buff *skb;
 >> +};
 >> +
 >>   struct amt_dev {
 >>   	struct net_device       *dev;
 >>   	struct net_device       *stream_dev;
 >> @@ -308,6 +323,7 @@ struct amt_dev {
 >>   	struct delayed_work     req_wq;
 >>   	/* Protected by RTNL */
 >>   	struct delayed_work     secret_wq;
 >> +	struct work_struct	event_wq;
 >>   	/* AMT status */
 >>   	enum amt_status		status;
 >>   	/* Generated key */
 >> @@ -345,6 +361,10 @@ struct amt_dev {
 >>   	/* Used only in gateway mode */
 >>   	u64			mac:48,
 >>   				reserved:16;
 >> +	/* AMT gateway side message handler queue */
 >> +	struct amt_events	events[AMT_MAX_EVENTS];
 >> +	u8			event_idx;
 >> +	u8			nr_events;
 >>   };
 >>
 >>   #define AMT_TOS			0xc0
 >

I will send the v2 patch after some tests with lockdep.

Thanks a lot!
Taehee Yoo

[PATCH net 2/8] amt: remove unnecessary locks

[Taehee Yoo]

By the previous patch, amt gateway handlers are changed to worked by
a single thread.
So, most locks for gateway are not needed.
So, it removes.

Fixes: cbc21dc1cfe9 ("amt: add data plane of amt interface")
Signed-off-by: Taehee Yoo <ap420073@gmail.com>
---
 drivers/net/amt.c | 32 +++++---------------------------
 1 file changed, 5 insertions(+), 27 deletions(-)

diff --git a/drivers/net/amt.c b/drivers/net/amt.c
index 032c2934e466..3ff8e522b92a 100644
--- a/drivers/net/amt.c
+++ b/drivers/net/amt.c
@@ -577,8 +577,8 @@ static struct sk_buff *amt_build_igmp_gq(struct amt_dev *amt)
 	return skb;
 }
 
-static void __amt_update_gw_status(struct amt_dev *amt, enum amt_status status,
-				   bool validate)
+static void amt_update_gw_status(struct amt_dev *amt, enum amt_status status,
+				 bool validate)
 {
 	if (validate && amt->status >= status)
 		return;
@@ -600,14 +600,6 @@ static void __amt_update_relay_status(struct amt_tunnel_list *tunnel,
 	tunnel->status = status;
 }
 
-static void amt_update_gw_status(struct amt_dev *amt, enum amt_status status,
-				 bool validate)
-{
-	spin_lock_bh(&amt->lock);
-	__amt_update_gw_status(amt, status, validate);
-	spin_unlock_bh(&amt->lock);
-}
-
 static void amt_update_relay_status(struct amt_tunnel_list *tunnel,
 				    enum amt_status status, bool validate)
 {
@@ -700,9 +692,7 @@ static void amt_send_discovery(struct amt_dev *amt)
 	if (unlikely(net_xmit_eval(err)))
 		amt->dev->stats.tx_errors++;
 
-	spin_lock_bh(&amt->lock);
-	__amt_update_gw_status(amt, AMT_STATUS_SENT_DISCOVERY, true);
-	spin_unlock_bh(&amt->lock);
+	amt_update_gw_status(amt, AMT_STATUS_SENT_DISCOVERY, true);
 out:
 	rcu_read_unlock();
 }
@@ -937,18 +927,14 @@ static void amt_secret_work(struct work_struct *work)
 
 static void amt_event_send_discovery(struct amt_dev *amt)
 {
-	spin_lock_bh(&amt->lock);
 	if (amt->status > AMT_STATUS_SENT_DISCOVERY)
 		goto out;
 	get_random_bytes(&amt->nonce, sizeof(__be32));
-	spin_unlock_bh(&amt->lock);
 
 	amt_send_discovery(amt);
-	spin_lock_bh(&amt->lock);
 out:
 	mod_delayed_work(amt_wq, &amt->discovery_wq,
 			 msecs_to_jiffies(AMT_DISCOVERY_TIMEOUT));
-	spin_unlock_bh(&amt->lock);
 }
 
 static void amt_discovery_work(struct work_struct *work)
@@ -966,7 +952,6 @@ static void amt_event_send_request(struct amt_dev *amt)
 {
 	u32 exp;
 
-	spin_lock_bh(&amt->lock);
 	if (amt->status < AMT_STATUS_RECEIVED_ADVERTISEMENT)
 		goto out;
 
@@ -976,21 +961,18 @@ static void amt_event_send_request(struct amt_dev *amt)
 		amt->ready4 = false;
 		amt->ready6 = false;
 		amt->remote_ip = 0;
-		__amt_update_gw_status(amt, AMT_STATUS_INIT, false);
+		amt_update_gw_status(amt, AMT_STATUS_INIT, false);
 		amt->req_cnt = 0;
 		goto out;
 	}
-	spin_unlock_bh(&amt->lock);
 
 	amt_send_request(amt, false);
 	amt_send_request(amt, true);
-	spin_lock_bh(&amt->lock);
-	__amt_update_gw_status(amt, AMT_STATUS_SENT_REQUEST, true);
+	amt_update_gw_status(amt, AMT_STATUS_SENT_REQUEST, true);
 	amt->req_cnt++;
 out:
 	exp = min_t(u32, (1 * (1 << amt->req_cnt)), AMT_MAX_REQ_TIMEOUT);
 	mod_delayed_work(amt_wq, &amt->req_wq, msecs_to_jiffies(exp * 1000));
-	spin_unlock_bh(&amt->lock);
 }
 
 static void amt_req_work(struct work_struct *work)
@@ -2386,12 +2368,10 @@ static bool amt_membership_query_handler(struct amt_dev *amt,
 		ihv3 = skb_pull(skb, sizeof(*iph) + AMT_IPHDR_OPTS);
 		skb_reset_transport_header(skb);
 		skb_push(skb, sizeof(*iph) + AMT_IPHDR_OPTS);
-		spin_lock_bh(&amt->lock);
 		amt->ready4 = true;
 		amt->mac = amtmq->response_mac;
 		amt->req_cnt = 0;
 		amt->qi = ihv3->qqic;
-		spin_unlock_bh(&amt->lock);
 		skb->protocol = htons(ETH_P_IP);
 		eth->h_proto = htons(ETH_P_IP);
 		ip_eth_mc_map(iph->daddr, eth->h_dest);
@@ -2411,12 +2391,10 @@ static bool amt_membership_query_handler(struct amt_dev *amt,
 		mld2q = skb_pull(skb, sizeof(*ip6h) + AMT_IP6HDR_OPTS);
 		skb_reset_transport_header(skb);
 		skb_push(skb, sizeof(*ip6h) + AMT_IP6HDR_OPTS);
-		spin_lock_bh(&amt->lock);
 		amt->ready6 = true;
 		amt->mac = amtmq->response_mac;
 		amt->req_cnt = 0;
 		amt->qi = mld2q->mld2q_qqic;
-		spin_unlock_bh(&amt->lock);
 		skb->protocol = htons(ETH_P_IPV6);
 		eth->h_proto = htons(ETH_P_IPV6);
 		ipv6_eth_mc_map(&ip6h->daddr, eth->h_dest);
-- 
2.17.1

[PATCH net 3/8] amt: use READ_ONCE() in amt module

[Taehee Yoo]

There are some data races in the amt module.
amt->ready4, amt->ready6, and amt->status can be accessed concurrently
without locks.
So, it uses READ_ONCE() and WRITE_ONCE().

Fixes: cbc21dc1cfe9 ("amt: add data plane of amt interface")
Signed-off-by: Taehee Yoo <ap420073@gmail.com>
---
 drivers/net/amt.c | 15 ++++++++-------
 1 file changed, 8 insertions(+), 7 deletions(-)

diff --git a/drivers/net/amt.c b/drivers/net/amt.c
index 3ff8e522b92a..9977ce9e5ae0 100644
--- a/drivers/net/amt.c
+++ b/drivers/net/amt.c
@@ -584,7 +584,7 @@ static void amt_update_gw_status(struct amt_dev *amt, enum amt_status status,
 		return;
 	netdev_dbg(amt->dev, "Update GW status %s -> %s",
 		   status_str[amt->status], status_str[status]);
-	amt->status = status;
+	WRITE_ONCE(amt->status, status);
 }
 
 static void __amt_update_relay_status(struct amt_tunnel_list *tunnel,
@@ -958,8 +958,8 @@ static void amt_event_send_request(struct amt_dev *amt)
 	if (amt->req_cnt > AMT_MAX_REQ_COUNT) {
 		netdev_dbg(amt->dev, "Gateway is not ready");
 		amt->qi = AMT_INIT_REQ_TIMEOUT;
-		amt->ready4 = false;
-		amt->ready6 = false;
+		WRITE_ONCE(amt->ready4, false);
+		WRITE_ONCE(amt->ready6, false);
 		amt->remote_ip = 0;
 		amt_update_gw_status(amt, AMT_STATUS_INIT, false);
 		amt->req_cnt = 0;
@@ -1239,7 +1239,8 @@ static netdev_tx_t amt_dev_xmit(struct sk_buff *skb, struct net_device *dev)
 		/* Gateway only passes IGMP/MLD packets */
 		if (!report)
 			goto free;
-		if ((!v6 && !amt->ready4) || (v6 && !amt->ready6))
+		if ((!v6 && !READ_ONCE(amt->ready4)) ||
+		    (v6 && !READ_ONCE(amt->ready6)))
 			goto free;
 		if (amt_send_membership_update(amt, skb,  v6))
 			goto free;
@@ -2368,7 +2369,7 @@ static bool amt_membership_query_handler(struct amt_dev *amt,
 		ihv3 = skb_pull(skb, sizeof(*iph) + AMT_IPHDR_OPTS);
 		skb_reset_transport_header(skb);
 		skb_push(skb, sizeof(*iph) + AMT_IPHDR_OPTS);
-		amt->ready4 = true;
+		WRITE_ONCE(amt->ready4, true);
 		amt->mac = amtmq->response_mac;
 		amt->req_cnt = 0;
 		amt->qi = ihv3->qqic;
@@ -2391,7 +2392,7 @@ static bool amt_membership_query_handler(struct amt_dev *amt,
 		mld2q = skb_pull(skb, sizeof(*ip6h) + AMT_IP6HDR_OPTS);
 		skb_reset_transport_header(skb);
 		skb_push(skb, sizeof(*ip6h) + AMT_IP6HDR_OPTS);
-		amt->ready6 = true;
+		WRITE_ONCE(amt->ready6, true);
 		amt->mac = amtmq->response_mac;
 		amt->req_cnt = 0;
 		amt->qi = mld2q->mld2q_qqic;
@@ -2897,7 +2898,7 @@ static int amt_err_lookup(struct sock *sk, struct sk_buff *skb)
 		break;
 	case AMT_MSG_REQUEST:
 	case AMT_MSG_MEMBERSHIP_UPDATE:
-		if (amt->status >= AMT_STATUS_RECEIVED_ADVERTISEMENT)
+		if (READ_ONCE(amt->status) >= AMT_STATUS_RECEIVED_ADVERTISEMENT)
 			mod_delayed_work(amt_wq, &amt->req_wq, 0);
 		break;
 	default:
-- 
2.17.1

[PATCH net 4/8] amt: add missing regeneration nonce logic in request logic

[Taehee Yoo]

When AMT gateway starts sending a new request message, it should
regenerate the nonce variable.

Fixes: cbc21dc1cfe9 ("amt: add data plane of amt interface")
Signed-off-by: Taehee Yoo <ap420073@gmail.com>
---
 drivers/net/amt.c | 4 ++++
 1 file changed, 4 insertions(+)

diff --git a/drivers/net/amt.c b/drivers/net/amt.c
index 9977ce9e5ae0..da2023d44da4 100644
--- a/drivers/net/amt.c
+++ b/drivers/net/amt.c
@@ -963,9 +963,13 @@ static void amt_event_send_request(struct amt_dev *amt)
 		amt->remote_ip = 0;
 		amt_update_gw_status(amt, AMT_STATUS_INIT, false);
 		amt->req_cnt = 0;
+		amt->nonce = 0;
 		goto out;
 	}
 
+	if (!amt->req_cnt)
+		get_random_bytes(&amt->nonce, sizeof(__be32));
+
 	amt_send_request(amt, false);
 	amt_send_request(amt, true);
 	amt_update_gw_status(amt, AMT_STATUS_SENT_REQUEST, true);
-- 
2.17.1

[PATCH net 5/8] amt: drop unexpected advertisement message

[Taehee Yoo]

AMT gateway interface should not receive unexpected advertisement messages.
In order to drop these packets, it should check nonce and amt->status.

Fixes: cbc21dc1cfe9 ("amt: add data plane of amt interface")
Signed-off-by: Taehee Yoo <ap420073@gmail.com>
---
 drivers/net/amt.c | 5 +++++
 1 file changed, 5 insertions(+)

diff --git a/drivers/net/amt.c b/drivers/net/amt.c
index da2023d44da4..6a12c32fb3a1 100644
--- a/drivers/net/amt.c
+++ b/drivers/net/amt.c
@@ -2260,6 +2260,10 @@ static bool amt_advertisement_handler(struct amt_dev *amt, struct sk_buff *skb)
 	    ipv4_is_zeronet(amta->ip4))
 		return true;
 
+	if (amt->status != AMT_STATUS_SENT_DISCOVERY ||
+	    amt->nonce != amta->nonce)
+		return true;
+
 	amt->remote_ip = amta->ip4;
 	netdev_dbg(amt->dev, "advertised remote ip = %pI4\n", &amt->remote_ip);
 	mod_delayed_work(amt_wq, &amt->req_wq, 0);
@@ -2972,6 +2976,7 @@ static int amt_dev_open(struct net_device *dev)
 
 	amt->req_cnt = 0;
 	amt->remote_ip = 0;
+	amt->nonce = 0;
 	get_random_bytes(&amt->key, sizeof(siphash_key_t));
 
 	amt->status = AMT_STATUS_INIT;
-- 
2.17.1

[PATCH net 6/8] amt: drop unexpected query message

[Taehee Yoo]

AMT gateway interface should not receive unexpected query messages.
In order to drop unexpected query messages, it checks nonce.
And it also checks ready4 and ready6 variables to drop duplicated messages.

Fixes: cbc21dc1cfe9 ("amt: add data plane of amt interface")
Signed-off-by: Taehee Yoo <ap420073@gmail.com>
---
 drivers/net/amt.c | 14 +++++++++++++-
 1 file changed, 13 insertions(+), 1 deletion(-)

diff --git a/drivers/net/amt.c b/drivers/net/amt.c
index 6a12c32fb3a1..dbaf490cc33f 100644
--- a/drivers/net/amt.c
+++ b/drivers/net/amt.c
@@ -967,8 +967,11 @@ static void amt_event_send_request(struct amt_dev *amt)
 		goto out;
 	}
 
-	if (!amt->req_cnt)
+	if (!amt->req_cnt) {
+		WRITE_ONCE(amt->ready4, false);
+		WRITE_ONCE(amt->ready6, false);
 		get_random_bytes(&amt->nonce, sizeof(__be32));
+	}
 
 	amt_send_request(amt, false);
 	amt_send_request(amt, true);
@@ -2353,6 +2356,9 @@ static bool amt_membership_query_handler(struct amt_dev *amt,
 	if (amtmq->reserved || amtmq->version)
 		return true;
 
+	if (amtmq->nonce != amt->nonce)
+		return true;
+
 	hdr_size -= sizeof(*eth);
 	if (iptunnel_pull_header(skb, hdr_size, htons(ETH_P_TEB), false))
 		return true;
@@ -2367,6 +2373,9 @@ static bool amt_membership_query_handler(struct amt_dev *amt,
 
 	iph = ip_hdr(skb);
 	if (iph->version == 4) {
+		if (READ_ONCE(amt->ready4))
+			return true;
+
 		if (!pskb_may_pull(skb, sizeof(*iph) + AMT_IPHDR_OPTS +
 				   sizeof(*ihv3)))
 			return true;
@@ -2389,6 +2398,9 @@ static bool amt_membership_query_handler(struct amt_dev *amt,
 		struct mld2_query *mld2q;
 		struct ipv6hdr *ip6h;
 
+		if (READ_ONCE(amt->ready6))
+			return true;
+
 		if (!pskb_may_pull(skb, sizeof(*ip6h) + AMT_IP6HDR_OPTS +
 				   sizeof(*mld2q)))
 			return true;
-- 
2.17.1

[PATCH net 7/8] amt: drop unexpected multicast data

[Taehee Yoo]

AMT gateway interface should not receive unexpected multicast data.
Multicast data message type should be received after sending an update
message, which means all establishment between gateway and relay is
finished.
So, amt_multicast_data_handler() checks amt->status.

Fixes: cbc21dc1cfe9 ("amt: add data plane of amt interface")
Signed-off-by: Taehee Yoo <ap420073@gmail.com>
---
 drivers/net/amt.c | 3 +++
 1 file changed, 3 insertions(+)

diff --git a/drivers/net/amt.c b/drivers/net/amt.c
index dbaf490cc33f..03decb3caa5c 100644
--- a/drivers/net/amt.c
+++ b/drivers/net/amt.c
@@ -2282,6 +2282,9 @@ static bool amt_multicast_data_handler(struct amt_dev *amt, struct sk_buff *skb)
 	struct ethhdr *eth;
 	struct iphdr *iph;
 
+	if (READ_ONCE(amt->status) != AMT_STATUS_SENT_UPDATE)
+		return true;
+
 	hdr_size = sizeof(*amtmd) + sizeof(struct udphdr);
 	if (!pskb_may_pull(skb, hdr_size))
 		return true;
-- 
2.17.1

[PATCH net 8/8] amt: do not use amt->nr_tunnels outside of lock

[Taehee Yoo]

amt->nr_tunnels is protected by amt->lock.
But, amt_request_handler() has been using this variable without the
amt->lock.
So, it expands context of amt->lock in the amt_request_handler() to
protect amt->nr_tunnels variable.

Fixes: cbc21dc1cfe9 ("amt: add data plane of amt interface")
Signed-off-by: Taehee Yoo <ap420073@gmail.com>
---
 drivers/net/amt.c | 9 ++++++---
 1 file changed, 6 insertions(+), 3 deletions(-)

diff --git a/drivers/net/amt.c b/drivers/net/amt.c
index 03decb3caa5c..f23d2d270895 100644
--- a/drivers/net/amt.c
+++ b/drivers/net/amt.c
@@ -2679,7 +2679,9 @@ static bool amt_request_handler(struct amt_dev *amt, struct sk_buff *skb)
 		if (tunnel->ip4 == iph->saddr)
 			goto send;
 
+	spin_lock_bh(&amt->lock);
 	if (amt->nr_tunnels >= amt->max_tunnels) {
+		spin_unlock_bh(&amt->lock);
 		icmp_ndo_send(skb, ICMP_DEST_UNREACH, ICMP_HOST_UNREACH, 0);
 		return true;
 	}
@@ -2687,8 +2689,10 @@ static bool amt_request_handler(struct amt_dev *amt, struct sk_buff *skb)
 	tunnel = kzalloc(sizeof(*tunnel) +
 			 (sizeof(struct hlist_head) * amt->hash_buckets),
 			 GFP_ATOMIC);
-	if (!tunnel)
+	if (!tunnel) {
+		spin_unlock_bh(&amt->lock);
 		return true;
+	}
 
 	tunnel->source_port = udph->source;
 	tunnel->ip4 = iph->saddr;
@@ -2701,10 +2705,9 @@ static bool amt_request_handler(struct amt_dev *amt, struct sk_buff *skb)
 
 	INIT_DELAYED_WORK(&tunnel->gc_wq, amt_tunnel_expire);
 
-	spin_lock_bh(&amt->lock);
 	list_add_tail_rcu(&tunnel->list, &amt->tunnel_list);
 	tunnel->key = amt->key;
-	amt_update_relay_status(tunnel, AMT_STATUS_RECEIVED_REQUEST, true);
+	__amt_update_relay_status(tunnel, AMT_STATUS_RECEIVED_REQUEST, true);
 	amt->nr_tunnels++;
 	mod_delayed_work(amt_wq, &tunnel->gc_wq,
 			 msecs_to_jiffies(amt_gmi(amt)));
-- 
2.17.1