[PATCH v2 net-next 1/2] tipc: refactor name table updates out of named packet receive routine

[PATCH v2 net-next 1/2] tipc: refactor name table updates out of named packet receive routine

[erik.hugne]

From: Erik Hugne <erik.hugne@ericsson.com>

We need to perform the same actions when processing deferred name
table updates, so this functionality is moved to a separate
function.

Signed-off-by: Erik Hugne <erik.hugne@ericsson.com>
Signed-off-by: Jon Maloy <jon.maloy@ericsson.com>
---
 net/tipc/name_distr.c | 74 ++++++++++++++++++++++++++-------------------------
 1 file changed, 38 insertions(+), 36 deletions(-)

diff --git a/net/tipc/name_distr.c b/net/tipc/name_distr.c
index dcc15bc..0591f33 100644
--- a/net/tipc/name_distr.c
+++ b/net/tipc/name_distr.c
@@ -263,52 +263,54 @@ static void named_purge_publ(struct publication *publ)
 }
 
 /**
+ * tipc_update_nametbl - try to process a nametable update and notify
+ *			 subscribers
+ *
+ * tipc_nametbl_lock must be held.
+ * Returns the publication item if successful, otherwise NULL.
+ */
+struct publication *tipc_update_nametbl(struct distr_item *i, u32 node,
+					u32 dtype)
+{
+	struct publication *publ = NULL;
+
+	if (dtype == PUBLICATION) {
+		publ = tipc_nametbl_insert_publ(ntohl(i->type), ntohl(i->lower),
+						ntohl(i->upper),
+						TIPC_CLUSTER_SCOPE, node,
+						ntohl(i->ref), ntohl(i->key));
+		if (publ) {
+			tipc_nodesub_subscribe(&publ->subscr, node, publ,
+					       (net_ev_handler)
+					       named_purge_publ);
+		}
+	} else if (dtype == WITHDRAWAL) {
+		publ = tipc_nametbl_remove_publ(ntohl(i->type), ntohl(i->lower),
+						node, ntohl(i->ref),
+						ntohl(i->key));
+		if (publ) {
+			tipc_nodesub_unsubscribe(&publ->subscr);
+			kfree(publ);
+		}
+	} else {
+		pr_warn("Unrecognized name table message received\n");
+	}
+	return publ;
+}
+
+/**
  * tipc_named_rcv - process name table update message sent by another node
  */
 void tipc_named_rcv(struct sk_buff *buf)
 {
-	struct publication *publ;
 	struct tipc_msg *msg = buf_msg(buf);
 	struct distr_item *item = (struct distr_item *)msg_data(msg);
 	u32 count = msg_data_sz(msg) / ITEM_SIZE;
 
 	write_lock_bh(&tipc_nametbl_lock);
 	while (count--) {
-		if (msg_type(msg) == PUBLICATION) {
-			publ = tipc_nametbl_insert_publ(ntohl(item->type),
-							ntohl(item->lower),
-							ntohl(item->upper),
-							TIPC_CLUSTER_SCOPE,
-							msg_orignode(msg),
-							ntohl(item->ref),
-							ntohl(item->key));
-			if (publ) {
-				tipc_nodesub_subscribe(&publ->subscr,
-						       msg_orignode(msg),
-						       publ,
-						       (net_ev_handler)
-						       named_purge_publ);
-			}
-		} else if (msg_type(msg) == WITHDRAWAL) {
-			publ = tipc_nametbl_remove_publ(ntohl(item->type),
-							ntohl(item->lower),
-							msg_orignode(msg),
-							ntohl(item->ref),
-							ntohl(item->key));
-
-			if (publ) {
-				tipc_nodesub_unsubscribe(&publ->subscr);
-				kfree(publ);
-			} else {
-				pr_err("Unable to remove publication by node 0x%x\n"
-				       " (type=%u, lower=%u, ref=%u, key=%u)\n",
-				       msg_orignode(msg), ntohl(item->type),
-				       ntohl(item->lower), ntohl(item->ref),
-				       ntohl(item->key));
-			}
-		} else {
-			pr_warn("Unrecognized name table message received\n");
-		}
+		tipc_update_nametbl(item, msg_orignode(msg),
+				    msg_type(msg));
 		item++;
 	}
 	write_unlock_bh(&tipc_nametbl_lock);
-- 
1.8.3.2

[PATCH v2 net-next 2/2] tipc: add name distributor resiliency queue

[erik.hugne]

From: Erik Hugne <erik.hugne@ericsson.com>

TIPC name table updates are distributed asynchronously in a cluster,
entailing a risk of certain race conditions. E.g., if two nodes
simultaneously issue conflicting (overlapping) publications, this may
not be detected until both publications have reached a third node, in
which case one of the publications will be silently dropped on that
node. Hence, we end up with an inconsistent name table.

In most cases this conflict is just a temporary race, e.g., one
node is issuing a publication under the assumption that a previous,
conflicting, publication has already been withdrawn by the other node.
However, because of the (rtt related) distributed update delay, this
may not yet hold true on all nodes. The symptom of this failure is a
syslog message: "tipc: Cannot publish {%u,%u,%u}, overlap error".

In this commit we add a resiliency queue at the receiving end of
the name table distributor. When insertion of an arriving publication
fails, we retain it in this queue for a short amount of time, assuming
that another update will arrive very soon and clear the conflict. If so
happens, we insert the publication, otherwise we drop it.

The (configurable) retention value defaults to 2000 ms. Knowing from
experience that the situation described above is extremely rare, there
is no risk that the queue will accumulate any large number of items.

Signed-off-by: Erik Hugne <erik.hugne@ericsson.com>
Signed-off-by: Jon Maloy <jon.maloy@ericsson.com>
---

v2: Fixed phrasing and spelling in the sysctl documentation

 Documentation/sysctl/net.txt | 16 ++++++++++
 net/tipc/core.h              |  1 +
 net/tipc/name_distr.c        | 69 ++++++++++++++++++++++++++++++++++++++++++--
 net/tipc/name_distr.h        |  1 +
 net/tipc/name_table.c        |  8 ++---
 net/tipc/sysctl.c            |  7 +++++
 6 files changed, 95 insertions(+), 7 deletions(-)

diff --git a/Documentation/sysctl/net.txt b/Documentation/sysctl/net.txt
index 9a0319a..04892b8 100644
--- a/Documentation/sysctl/net.txt
+++ b/Documentation/sysctl/net.txt
@@ -241,6 +241,9 @@ address of the router (or Connected) for internal networks.
 6. TIPC
 -------------------------------------------------------
 
+tipc_rmem
+----------
+
 The TIPC protocol now has a tunable for the receive memory, similar to the
 tcp_rmem - i.e. a vector of 3 INTEGERs: (min, default, max)
 
@@ -252,3 +255,16 @@ The max value is set to CONN_OVERLOAD_LIMIT, and the default and min values
 are scaled (shifted) versions of that same value.  Note that the min value
 is not at this point in time used in any meaningful way, but the triplet is
 preserved in order to be consistent with things like tcp_rmem.
+
+named_timeout
+--------------
+
+TIPC name table updates are distributed asynchronously in a cluster, without
+any form of transaction handling. This means that different race scenarios are
+possible. One such is that a name withdrawal sent out by one node and received
+by another node may arrive after a second, overlapping name publication already
+has been accepted from a third node, although the conflicting updates
+originally may have been issued in the correct sequential order.
+If named_timeout is nonzero, failed topology updates will be placed on a defer
+queue until another event arrives that clears the error, or until the timeout
+expires. Value is in milliseconds.
diff --git a/net/tipc/core.h b/net/tipc/core.h
index d2607a8..f773b14 100644
--- a/net/tipc/core.h
+++ b/net/tipc/core.h
@@ -81,6 +81,7 @@ extern u32 tipc_own_addr __read_mostly;
 extern int tipc_max_ports __read_mostly;
 extern int tipc_net_id __read_mostly;
 extern int sysctl_tipc_rmem[3] __read_mostly;
+extern int sysctl_tipc_named_timeout __read_mostly;
 
 /*
  * Other global variables
diff --git a/net/tipc/name_distr.c b/net/tipc/name_distr.c
index 0591f33..0cbe5e1 100644
--- a/net/tipc/name_distr.c
+++ b/net/tipc/name_distr.c
@@ -1,7 +1,7 @@
 /*
  * net/tipc/name_distr.c: TIPC name distribution code
  *
- * Copyright (c) 2000-2006, Ericsson AB
+ * Copyright (c) 2000-2006, 2014, Ericsson AB
  * Copyright (c) 2005, 2010-2011, Wind River Systems
  * All rights reserved.
  *
@@ -71,6 +71,21 @@ static struct publ_list *publ_lists[] = {
 };
 
 
+int sysctl_tipc_named_timeout __read_mostly = 2000;
+
+/**
+ * struct tipc_dist_queue - queue holding deferred name table updates
+ */
+static struct list_head tipc_dist_queue = LIST_HEAD_INIT(tipc_dist_queue);
+
+struct distr_queue_item {
+	struct distr_item i;
+	u32 dtype;
+	u32 node;
+	u64 expiry;
+	struct list_head next;
+};
+
 /**
  * publ_to_item - add publication info to a publication message
  */
@@ -299,6 +314,52 @@ struct publication *tipc_update_nametbl(struct distr_item *i, u32 node,
 }
 
 /**
+ * tipc_named_add_backlog - add a failed name table update to the backlog
+ *
+ */
+static void tipc_named_add_backlog(struct distr_item *i, u32 type, u32 node)
+{
+	struct distr_queue_item *e;
+	u64 now = get_jiffies_64();
+
+	e = kzalloc(sizeof(*e), GFP_ATOMIC);
+	if (!e)
+		return;
+	e->dtype = type;
+	e->node = node;
+	e->expiry = now + msecs_to_jiffies(sysctl_tipc_named_timeout);
+	memcpy(e, i, sizeof(*i));
+	list_add_tail(&e->next, &tipc_dist_queue);
+}
+
+/**
+ * tipc_named_process_backlog - try to process any pending name table updates
+ * from the network.
+ */
+void tipc_named_process_backlog(void)
+{
+	struct distr_queue_item *e, *tmp;
+	char addr[16];
+	u64 now = get_jiffies_64();
+
+	list_for_each_entry_safe(e, tmp, &tipc_dist_queue, next) {
+		if (e->expiry > now) {
+			if (!tipc_update_nametbl(&e->i, e->node, e->dtype))
+				continue;
+		} else {
+			tipc_addr_string_fill(addr, e->node);
+			pr_warn_ratelimited("Dropping name table update (%d) of {%u, %u, %u} from %s key=%u\n",
+					    e->dtype, ntohl(e->i.type),
+					    ntohl(e->i.lower),
+					    ntohl(e->i.upper),
+					    addr, ntohl(e->i.key));
+		}
+		list_del(&e->next);
+		kfree(e);
+	}
+}
+
+/**
  * tipc_named_rcv - process name table update message sent by another node
  */
 void tipc_named_rcv(struct sk_buff *buf)
@@ -306,13 +367,15 @@ void tipc_named_rcv(struct sk_buff *buf)
 	struct tipc_msg *msg = buf_msg(buf);
 	struct distr_item *item = (struct distr_item *)msg_data(msg);
 	u32 count = msg_data_sz(msg) / ITEM_SIZE;
+	u32 node = msg_orignode(msg);
 
 	write_lock_bh(&tipc_nametbl_lock);
 	while (count--) {
-		tipc_update_nametbl(item, msg_orignode(msg),
-				    msg_type(msg));
+		if (!tipc_update_nametbl(item, node, msg_type(msg)))
+			tipc_named_add_backlog(item, msg_type(msg), node);
 		item++;
 	}
+	tipc_named_process_backlog();
 	write_unlock_bh(&tipc_nametbl_lock);
 	kfree_skb(buf);
 }
diff --git a/net/tipc/name_distr.h b/net/tipc/name_distr.h
index 8afe32b..b9e75fe 100644
--- a/net/tipc/name_distr.h
+++ b/net/tipc/name_distr.h
@@ -73,5 +73,6 @@ void named_cluster_distribute(struct sk_buff *buf);
 void tipc_named_node_up(u32 dnode);
 void tipc_named_rcv(struct sk_buff *buf);
 void tipc_named_reinit(void);
+void tipc_named_process_backlog(void);
 
 #endif
diff --git a/net/tipc/name_table.c b/net/tipc/name_table.c
index c058e30..3a6a0a7 100644
--- a/net/tipc/name_table.c
+++ b/net/tipc/name_table.c
@@ -261,8 +261,6 @@ static struct publication *tipc_nameseq_insert_publ(struct name_seq *nseq,
 
 		/* Lower end overlaps existing entry => need an exact match */
 		if ((sseq->lower != lower) || (sseq->upper != upper)) {
-			pr_warn("Cannot publish {%u,%u,%u}, overlap error\n",
-				type, lower, upper);
 			return NULL;
 		}
 
@@ -284,8 +282,6 @@ static struct publication *tipc_nameseq_insert_publ(struct name_seq *nseq,
 		/* Fail if upper end overlaps into an existing entry */
 		if ((inspos < nseq->first_free) &&
 		    (upper >= nseq->sseqs[inspos].lower)) {
-			pr_warn("Cannot publish {%u,%u,%u}, overlap error\n",
-				type, lower, upper);
 			return NULL;
 		}
 
@@ -677,6 +673,8 @@ struct publication *tipc_nametbl_publish(u32 type, u32 lower, u32 upper,
 	if (likely(publ)) {
 		table.local_publ_count++;
 		buf = tipc_named_publish(publ);
+		/* Any pending external events? */
+		tipc_named_process_backlog();
 	}
 	write_unlock_bh(&tipc_nametbl_lock);
 
@@ -698,6 +696,8 @@ int tipc_nametbl_withdraw(u32 type, u32 lower, u32 ref, u32 key)
 	if (likely(publ)) {
 		table.local_publ_count--;
 		buf = tipc_named_withdraw(publ);
+		/* Any pending external events? */
+		tipc_named_process_backlog();
 		write_unlock_bh(&tipc_nametbl_lock);
 		list_del_init(&publ->pport_list);
 		kfree(publ);
diff --git a/net/tipc/sysctl.c b/net/tipc/sysctl.c
index f3fef93..1a779b1 100644
--- a/net/tipc/sysctl.c
+++ b/net/tipc/sysctl.c
@@ -47,6 +47,13 @@ static struct ctl_table tipc_table[] = {
 		.mode		= 0644,
 		.proc_handler	= proc_dointvec,
 	},
+	{
+		.procname	= "named_timeout",
+		.data		= &sysctl_tipc_named_timeout,
+		.maxlen		= sizeof(sysctl_tipc_named_timeout),
+		.mode		= 0644,
+		.proc_handler	= proc_dointvec,
+	},
 	{}
 };
 
-- 
1.8.3.2

Re: [PATCH v2 net-next 2/2] tipc: add name distributor resiliency queue

[Ying Xue]

On 08/26/2014 04:57 PM, erik.hugne@ericsson.com wrote:
> From: Erik Hugne <erik.hugne@ericsson.com>
> 
> TIPC name table updates are distributed asynchronously in a cluster,
> entailing a risk of certain race conditions. E.g., if two nodes
> simultaneously issue conflicting (overlapping) publications, this may
> not be detected until both publications have reached a third node, in
> which case one of the publications will be silently dropped on that
> node. Hence, we end up with an inconsistent name table.
> 
> In most cases this conflict is just a temporary race, e.g., one
> node is issuing a publication under the assumption that a previous,
> conflicting, publication has already been withdrawn by the other node.
> However, because of the (rtt related) distributed update delay, this
> may not yet hold true on all nodes. The symptom of this failure is a
> syslog message: "tipc: Cannot publish {%u,%u,%u}, overlap error".
> 
> In this commit we add a resiliency queue at the receiving end of
> the name table distributor. When insertion of an arriving publication
> fails, we retain it in this queue for a short amount of time, assuming
> that another update will arrive very soon and clear the conflict. If so
> happens, we insert the publication, otherwise we drop it.
> 
> The (configurable) retention value defaults to 2000 ms. Knowing from
> experience that the situation described above is extremely rare, there
> is no risk that the queue will accumulate any large number of items.
> 
> Signed-off-by: Erik Hugne <erik.hugne@ericsson.com>
> Signed-off-by: Jon Maloy <jon.maloy@ericsson.com>

Acked-by: Ying Xue <ying.xue@windriver.com>

> ---
> 
> v2: Fixed phrasing and spelling in the sysctl documentation
> 
>  Documentation/sysctl/net.txt | 16 ++++++++++
>  net/tipc/core.h              |  1 +
>  net/tipc/name_distr.c        | 69 ++++++++++++++++++++++++++++++++++++++++++--
>  net/tipc/name_distr.h        |  1 +
>  net/tipc/name_table.c        |  8 ++---
>  net/tipc/sysctl.c            |  7 +++++
>  6 files changed, 95 insertions(+), 7 deletions(-)
> 
> diff --git a/Documentation/sysctl/net.txt b/Documentation/sysctl/net.txt
> index 9a0319a..04892b8 100644
> --- a/Documentation/sysctl/net.txt
> +++ b/Documentation/sysctl/net.txt
> @@ -241,6 +241,9 @@ address of the router (or Connected) for internal networks.
>  6. TIPC
>  -------------------------------------------------------
>  
> +tipc_rmem
> +----------
> +
>  The TIPC protocol now has a tunable for the receive memory, similar to the
>  tcp_rmem - i.e. a vector of 3 INTEGERs: (min, default, max)
>  
> @@ -252,3 +255,16 @@ The max value is set to CONN_OVERLOAD_LIMIT, and the default and min values
>  are scaled (shifted) versions of that same value.  Note that the min value
>  is not at this point in time used in any meaningful way, but the triplet is
>  preserved in order to be consistent with things like tcp_rmem.
> +
> +named_timeout
> +--------------
> +
> +TIPC name table updates are distributed asynchronously in a cluster, without
> +any form of transaction handling. This means that different race scenarios are
> +possible. One such is that a name withdrawal sent out by one node and received
> +by another node may arrive after a second, overlapping name publication already
> +has been accepted from a third node, although the conflicting updates
> +originally may have been issued in the correct sequential order.
> +If named_timeout is nonzero, failed topology updates will be placed on a defer
> +queue until another event arrives that clears the error, or until the timeout
> +expires. Value is in milliseconds.
> diff --git a/net/tipc/core.h b/net/tipc/core.h
> index d2607a8..f773b14 100644
> --- a/net/tipc/core.h
> +++ b/net/tipc/core.h
> @@ -81,6 +81,7 @@ extern u32 tipc_own_addr __read_mostly;
>  extern int tipc_max_ports __read_mostly;
>  extern int tipc_net_id __read_mostly;
>  extern int sysctl_tipc_rmem[3] __read_mostly;
> +extern int sysctl_tipc_named_timeout __read_mostly;
>  
>  /*
>   * Other global variables
> diff --git a/net/tipc/name_distr.c b/net/tipc/name_distr.c
> index 0591f33..0cbe5e1 100644
> --- a/net/tipc/name_distr.c
> +++ b/net/tipc/name_distr.c
> @@ -1,7 +1,7 @@
>  /*
>   * net/tipc/name_distr.c: TIPC name distribution code
>   *
> - * Copyright (c) 2000-2006, Ericsson AB
> + * Copyright (c) 2000-2006, 2014, Ericsson AB
>   * Copyright (c) 2005, 2010-2011, Wind River Systems
>   * All rights reserved.
>   *
> @@ -71,6 +71,21 @@ static struct publ_list *publ_lists[] = {
>  };
>  
>  
> +int sysctl_tipc_named_timeout __read_mostly = 2000;
> +
> +/**
> + * struct tipc_dist_queue - queue holding deferred name table updates
> + */
> +static struct list_head tipc_dist_queue = LIST_HEAD_INIT(tipc_dist_queue);
> +
> +struct distr_queue_item {
> +	struct distr_item i;
> +	u32 dtype;
> +	u32 node;
> +	u64 expiry;
> +	struct list_head next;
> +};
> +
>  /**
>   * publ_to_item - add publication info to a publication message
>   */
> @@ -299,6 +314,52 @@ struct publication *tipc_update_nametbl(struct distr_item *i, u32 node,
>  }
>  
>  /**
> + * tipc_named_add_backlog - add a failed name table update to the backlog
> + *
> + */
> +static void tipc_named_add_backlog(struct distr_item *i, u32 type, u32 node)
> +{
> +	struct distr_queue_item *e;
> +	u64 now = get_jiffies_64();
> +
> +	e = kzalloc(sizeof(*e), GFP_ATOMIC);
> +	if (!e)
> +		return;
> +	e->dtype = type;
> +	e->node = node;
> +	e->expiry = now + msecs_to_jiffies(sysctl_tipc_named_timeout);
> +	memcpy(e, i, sizeof(*i));
> +	list_add_tail(&e->next, &tipc_dist_queue);
> +}
> +
> +/**
> + * tipc_named_process_backlog - try to process any pending name table updates
> + * from the network.
> + */
> +void tipc_named_process_backlog(void)
> +{
> +	struct distr_queue_item *e, *tmp;
> +	char addr[16];
> +	u64 now = get_jiffies_64();
> +
> +	list_for_each_entry_safe(e, tmp, &tipc_dist_queue, next) {
> +		if (e->expiry > now) {
> +			if (!tipc_update_nametbl(&e->i, e->node, e->dtype))
> +				continue;
> +		} else {
> +			tipc_addr_string_fill(addr, e->node);
> +			pr_warn_ratelimited("Dropping name table update (%d) of {%u, %u, %u} from %s key=%u\n",
> +					    e->dtype, ntohl(e->i.type),
> +					    ntohl(e->i.lower),
> +					    ntohl(e->i.upper),
> +					    addr, ntohl(e->i.key));
> +		}
> +		list_del(&e->next);
> +		kfree(e);
> +	}
> +}
> +
> +/**
>   * tipc_named_rcv - process name table update message sent by another node
>   */
>  void tipc_named_rcv(struct sk_buff *buf)
> @@ -306,13 +367,15 @@ void tipc_named_rcv(struct sk_buff *buf)
>  	struct tipc_msg *msg = buf_msg(buf);
>  	struct distr_item *item = (struct distr_item *)msg_data(msg);
>  	u32 count = msg_data_sz(msg) / ITEM_SIZE;
> +	u32 node = msg_orignode(msg);
>  
>  	write_lock_bh(&tipc_nametbl_lock);
>  	while (count--) {
> -		tipc_update_nametbl(item, msg_orignode(msg),
> -				    msg_type(msg));
> +		if (!tipc_update_nametbl(item, node, msg_type(msg)))
> +			tipc_named_add_backlog(item, msg_type(msg), node);
>  		item++;
>  	}
> +	tipc_named_process_backlog();
>  	write_unlock_bh(&tipc_nametbl_lock);
>  	kfree_skb(buf);
>  }
> diff --git a/net/tipc/name_distr.h b/net/tipc/name_distr.h
> index 8afe32b..b9e75fe 100644
> --- a/net/tipc/name_distr.h
> +++ b/net/tipc/name_distr.h
> @@ -73,5 +73,6 @@ void named_cluster_distribute(struct sk_buff *buf);
>  void tipc_named_node_up(u32 dnode);
>  void tipc_named_rcv(struct sk_buff *buf);
>  void tipc_named_reinit(void);
> +void tipc_named_process_backlog(void);
>  
>  #endif
> diff --git a/net/tipc/name_table.c b/net/tipc/name_table.c
> index c058e30..3a6a0a7 100644
> --- a/net/tipc/name_table.c
> +++ b/net/tipc/name_table.c
> @@ -261,8 +261,6 @@ static struct publication *tipc_nameseq_insert_publ(struct name_seq *nseq,
>  
>  		/* Lower end overlaps existing entry => need an exact match */
>  		if ((sseq->lower != lower) || (sseq->upper != upper)) {
> -			pr_warn("Cannot publish {%u,%u,%u}, overlap error\n",
> -				type, lower, upper);
>  			return NULL;
>  		}
>  
> @@ -284,8 +282,6 @@ static struct publication *tipc_nameseq_insert_publ(struct name_seq *nseq,
>  		/* Fail if upper end overlaps into an existing entry */
>  		if ((inspos < nseq->first_free) &&
>  		    (upper >= nseq->sseqs[inspos].lower)) {
> -			pr_warn("Cannot publish {%u,%u,%u}, overlap error\n",
> -				type, lower, upper);
>  			return NULL;
>  		}
>  
> @@ -677,6 +673,8 @@ struct publication *tipc_nametbl_publish(u32 type, u32 lower, u32 upper,
>  	if (likely(publ)) {
>  		table.local_publ_count++;
>  		buf = tipc_named_publish(publ);
> +		/* Any pending external events? */
> +		tipc_named_process_backlog();
>  	}
>  	write_unlock_bh(&tipc_nametbl_lock);
>  
> @@ -698,6 +696,8 @@ int tipc_nametbl_withdraw(u32 type, u32 lower, u32 ref, u32 key)
>  	if (likely(publ)) {
>  		table.local_publ_count--;
>  		buf = tipc_named_withdraw(publ);
> +		/* Any pending external events? */
> +		tipc_named_process_backlog();
>  		write_unlock_bh(&tipc_nametbl_lock);
>  		list_del_init(&publ->pport_list);
>  		kfree(publ);
> diff --git a/net/tipc/sysctl.c b/net/tipc/sysctl.c
> index f3fef93..1a779b1 100644
> --- a/net/tipc/sysctl.c
> +++ b/net/tipc/sysctl.c
> @@ -47,6 +47,13 @@ static struct ctl_table tipc_table[] = {
>  		.mode		= 0644,
>  		.proc_handler	= proc_dointvec,
>  	},
> +	{
> +		.procname	= "named_timeout",
> +		.data		= &sysctl_tipc_named_timeout,
> +		.maxlen		= sizeof(sysctl_tipc_named_timeout),
> +		.mode		= 0644,
> +		.proc_handler	= proc_dointvec,
> +	},
>  	{}
>  };
>  
> 


------------------------------------------------------------------------------
Slashdot TV.  
Video for Nerds.  Stuff that matters.
http://tv.slashdot.org/

Re: [PATCH v2 net-next 2/2] tipc: add name distributor resiliency queue

[Erik Hugne]

if (e->expiry > now) should in fact be
if (time_after(e->expiry, now) to handle timer wraps correctly.

Thanks Niklas Söderlund for pointing this out.

v2 removed from patchwork

//E

On Tue, Aug 26, 2014 at 10:57:57AM +0200, erik.hugne@ericsson.com wrote:
> +void tipc_named_process_backlog(void)
> +{
> +	struct distr_queue_item *e, *tmp;
> +	char addr[16];
> +	u64 now = get_jiffies_64();
> +
> +	list_for_each_entry_safe(e, tmp, &tipc_dist_queue, next) {
> +		if (e->expiry > now) {
> +			if (!tipc_update_nametbl(&e->i, e->node, e->dtype))
> +				continue;
> +		} else {
> +			tipc_addr_string_fill(addr, e->node);
> +			pr_warn_ratelimited("Dropping name table update (%d) of {%u, %u, %u} from %s key=%u\n",
> +					    e->dtype, ntohl(e->i.type),
> +					    ntohl(e->i.lower),
> +					    ntohl(e->i.upper),
> +					    addr, ntohl(e->i.key));
> +		}
> +		list_del(&e->next);
> +		kfree(e);
> +	}
> +}
> +

RE: [PATCH v2 net-next 2/2] tipc: add name distributor resiliency queue

[Jon Maloy]

(netdev removed from receiver list)

> -----Original Message-----
> From: Erik Hugne
> Sent: August-26-14 10:33 AM
> To: Jon Maloy; ying.xue@windriver.com; Richard Alpe;
> netdev@vger.kernel.org
> Cc: tipc-discussion@lists.sourceforge.net
> Subject: Re: [PATCH v2 net-next 2/2] tipc: add name distributor resiliency
> queue
> 
> if (e->expiry > now) should in fact be
> if (time_after(e->expiry, now) to handle timer wraps correctly.
> 
> Thanks Niklas Söderlund for pointing this out.
> 
> v2 removed from patchwork
> 
> //E

Also, don't use the version label an comment when you send this
to David. This has only relevance to us who are reviewing the
series. (I assume it was  a mistake)

///jon

> 
> On Tue, Aug 26, 2014 at 10:57:57AM +0200, erik.hugne@ericsson.com wrote:
> > +void tipc_named_process_backlog(void) {
> > +	struct distr_queue_item *e, *tmp;
> > +	char addr[16];
> > +	u64 now = get_jiffies_64();
> > +
> > +	list_for_each_entry_safe(e, tmp, &tipc_dist_queue, next) {
> > +		if (e->expiry > now) {
> > +			if (!tipc_update_nametbl(&e->i, e->node, e-
> >dtype))
> > +				continue;
> > +		} else {
> > +			tipc_addr_string_fill(addr, e->node);
> > +			pr_warn_ratelimited("Dropping name table update
> (%d) of {%u, %u, %u} from %s key=%u\n",
> > +					    e->dtype, ntohl(e->i.type),
> > +					    ntohl(e->i.lower),
> > +					    ntohl(e->i.upper),
> > +					    addr, ntohl(e->i.key));
> > +		}
> > +		list_del(&e->next);
> > +		kfree(e);
> > +	}
> > +}
> > +

Re: [PATCH v2 net-next 2/2] tipc: add name distributor resiliency queue

[Erik Hugne]

On Tue, Aug 26, 2014 at 04:44:34PM +0200, Jon Maloy wrote:
> (netdev removed from receiver list)
> 
> > -----Original Message-----
> > From: Erik Hugne
> > Sent: August-26-14 10:33 AM
> > To: Jon Maloy; ying.xue@windriver.com; Richard Alpe;
> > netdev@vger.kernel.org
> > Cc: tipc-discussion@lists.sourceforge.net
> > Subject: Re: [PATCH v2 net-next 2/2] tipc: add name distributor resiliency
> > queue
> > 
> > if (e->expiry > now) should in fact be
> > if (time_after(e->expiry, now) to handle timer wraps correctly.
> > 
> > Thanks Niklas Söderlund for pointing this out.
> > 
> > v2 removed from patchwork
> > 
> > //E
> 
> Also, don't use the version label an comment when you send this
> to David. This has only relevance to us who are reviewing the
> series. (I assume it was  a mistake)

No, this is in the general guidelines for submitting patches section 15.
https://www.kernel.org/doc/Documentation/SubmittingPatches

//E

RE: [PATCH v2 net-next 2/2] tipc: add name distributor resiliency queue

[Jon Maloy]

> -----Original Message-----
> From: Erik Hugne
> Sent: August-26-14 10:48 AM
> To: Jon Maloy
> Cc: ying.xue@windriver.com; Richard Alpe; netdev@vger.kernel.org; tipc-
> discussion@lists.sourceforge.net
> Subject: Re: [PATCH v2 net-next 2/2] tipc: add name distributor resiliency
> queue
> 
> On Tue, Aug 26, 2014 at 04:44:34PM +0200, Jon Maloy wrote:
> > (netdev removed from receiver list)
> >
> > > -----Original Message-----
> > > From: Erik Hugne
> > > Sent: August-26-14 10:33 AM
> > > To: Jon Maloy; ying.xue@windriver.com; Richard Alpe;
> > > netdev@vger.kernel.org
> > > Cc: tipc-discussion@lists.sourceforge.net
> > > Subject: Re: [PATCH v2 net-next 2/2] tipc: add name distributor
> > > resiliency queue
> > >
> > > if (e->expiry > now) should in fact be if (time_after(e->expiry,
> > > now) to handle timer wraps correctly.
> > >
> > > Thanks Niklas Söderlund for pointing this out.
> > >
> > > v2 removed from patchwork
> > >
> > > //E
> >
> > Also, don't use the version label an comment when you send this to
> > David. This has only relevance to us who are reviewing the series. (I
> > assume it was  a mistake)
> 
> No, this is in the general guidelines for submitting patches section 15.
> https://www.kernel.org/doc/Documentation/SubmittingPatches
> 
> //E

Ok. I missed that this actually was the second version you sent to
netdev.  What I meant was the versioning we are using during our
own, internal discussions at tipc-discussion. This certainly has no
relevance for David & co.

Regards
///jon

Re: [PATCH v2 net-next 1/2] tipc: refactor name table updates out of named packet receive routine

[Ying Xue]

On 08/26/2014 04:57 PM, erik.hugne@ericsson.com wrote:
> From: Erik Hugne <erik.hugne@ericsson.com>
> 
> We need to perform the same actions when processing deferred name
> table updates, so this functionality is moved to a separate
> function.
> 
> Signed-off-by: Erik Hugne <erik.hugne@ericsson.com>
> Signed-off-by: Jon Maloy <jon.maloy@ericsson.com>

Acked-by: Ying Xue <ying.xue@windriver.com>

> ---
>  net/tipc/name_distr.c | 74 ++++++++++++++++++++++++++-------------------------
>  1 file changed, 38 insertions(+), 36 deletions(-)
> 
> diff --git a/net/tipc/name_distr.c b/net/tipc/name_distr.c
> index dcc15bc..0591f33 100644
> --- a/net/tipc/name_distr.c
> +++ b/net/tipc/name_distr.c
> @@ -263,52 +263,54 @@ static void named_purge_publ(struct publication *publ)
>  }
>  
>  /**
> + * tipc_update_nametbl - try to process a nametable update and notify
> + *			 subscribers
> + *
> + * tipc_nametbl_lock must be held.
> + * Returns the publication item if successful, otherwise NULL.
> + */
> +struct publication *tipc_update_nametbl(struct distr_item *i, u32 node,
> +					u32 dtype)
> +{
> +	struct publication *publ = NULL;
> +
> +	if (dtype == PUBLICATION) {
> +		publ = tipc_nametbl_insert_publ(ntohl(i->type), ntohl(i->lower),
> +						ntohl(i->upper),
> +						TIPC_CLUSTER_SCOPE, node,
> +						ntohl(i->ref), ntohl(i->key));
> +		if (publ) {
> +			tipc_nodesub_subscribe(&publ->subscr, node, publ,
> +					       (net_ev_handler)
> +					       named_purge_publ);
> +		}
> +	} else if (dtype == WITHDRAWAL) {
> +		publ = tipc_nametbl_remove_publ(ntohl(i->type), ntohl(i->lower),
> +						node, ntohl(i->ref),
> +						ntohl(i->key));
> +		if (publ) {
> +			tipc_nodesub_unsubscribe(&publ->subscr);
> +			kfree(publ);
> +		}
> +	} else {
> +		pr_warn("Unrecognized name table message received\n");
> +	}
> +	return publ;
> +}
> +
> +/**
>   * tipc_named_rcv - process name table update message sent by another node
>   */
>  void tipc_named_rcv(struct sk_buff *buf)
>  {
> -	struct publication *publ;
>  	struct tipc_msg *msg = buf_msg(buf);
>  	struct distr_item *item = (struct distr_item *)msg_data(msg);
>  	u32 count = msg_data_sz(msg) / ITEM_SIZE;
>  
>  	write_lock_bh(&tipc_nametbl_lock);
>  	while (count--) {
> -		if (msg_type(msg) == PUBLICATION) {
> -			publ = tipc_nametbl_insert_publ(ntohl(item->type),
> -							ntohl(item->lower),
> -							ntohl(item->upper),
> -							TIPC_CLUSTER_SCOPE,
> -							msg_orignode(msg),
> -							ntohl(item->ref),
> -							ntohl(item->key));
> -			if (publ) {
> -				tipc_nodesub_subscribe(&publ->subscr,
> -						       msg_orignode(msg),
> -						       publ,
> -						       (net_ev_handler)
> -						       named_purge_publ);
> -			}
> -		} else if (msg_type(msg) == WITHDRAWAL) {
> -			publ = tipc_nametbl_remove_publ(ntohl(item->type),
> -							ntohl(item->lower),
> -							msg_orignode(msg),
> -							ntohl(item->ref),
> -							ntohl(item->key));
> -
> -			if (publ) {
> -				tipc_nodesub_unsubscribe(&publ->subscr);
> -				kfree(publ);
> -			} else {
> -				pr_err("Unable to remove publication by node 0x%x\n"
> -				       " (type=%u, lower=%u, ref=%u, key=%u)\n",
> -				       msg_orignode(msg), ntohl(item->type),
> -				       ntohl(item->lower), ntohl(item->ref),
> -				       ntohl(item->key));
> -			}
> -		} else {
> -			pr_warn("Unrecognized name table message received\n");
> -		}
> +		tipc_update_nametbl(item, msg_orignode(msg),
> +				    msg_type(msg));
>  		item++;
>  	}
>  	write_unlock_bh(&tipc_nametbl_lock);
>