tipc-discussion — User questions & problem reports, project news, developer discussion

Re: [tipc-discussion] [net 1/1] tipc: fix cancellation of topology subscriptions

[David M. <da...@da...>]

From: Jon Maloy <jon...@er...>
Date: Wed, 20 Mar 2019 10:59:03 +0100

> From: Erik Hugne <eri...@gm...>
> 
> When cancelling a subscription, we have to clear the cancel bit in the
> request before iterating over any established subscriptions with memcmp.
> Otherwise no subscription will ever be found, and it will not be
> possible to explicitly unsubscribe individual subscriptions.
 ...
> diff --git a/net/tipc/topsrv.c b/net/tipc/topsrv.c
> index 4a708a4..b05a89a 100644
> --- a/net/tipc/topsrv.c
> +++ b/net/tipc/topsrv.c
> @@ -363,6 +363,7 @@ static int tipc_conn_rcv_sub(struct tipc_topsrv *srv,
>  	struct tipc_subscription *sub;
>  
>  	if (tipc_sub_read(s, filter) & TIPC_SUB_CANCEL) {
> +		s->filter ^= __constant_ntohl(TIPC_SUB_CANCEL);
>  		tipc_conn_delete_sub(con, s);

It is confusing to say we have to "clear" some bit in a certain
situation and then use a xor operation which "inverts" a bit.

Please use "&= ~VAL" instead.

Thank you.

Re: [tipc-discussion] general protection fault in fib6_purge_rt

[Jon M. <jon...@er...>]

This one identifies the same culprit as syz...@sy..., but points to a different bug.
That bug has also been fixed, in commit adba75be0d23  ("tipc: fix lockdep warning when reinitilaizing sockets"), applied in 4.20 but not present in 4.16, - the source of the dump.
Once again, a dump from 4.20/5.0 might be a help.

///jon


> -----Original Message-----
> From: syzbot <syz...@sy...>
> Sent: 18-Mar-19 08:28
> To: da...@da...; Jon Maloy <jon...@er...>;
> ku...@ms...; lin...@vg...;
> ne...@vg...; syz...@go...; tipc-
> dis...@li...; yin...@wi...; yoshfuji@linux-
> ipv6.org
> Subject: Re: general protection fault in fib6_purge_rt
> 
> syzbot has bisected this bug to:
> 
> commit 52dfae5c85a4c1078e9f1d5e8947d4a25f73dd81
> Author: Jon Maloy <jon...@er...>
> Date:   Thu Mar 22 19:42:52 2018 +0000
> 
>      tipc: obtain node identity from interface by default
> 
> bisection log:  https://syzkaller.appspot.com/x/bisect.txt?x=1116d2a3200000
> start commit:   52dfae5c tipc: obtain node identity from interface by defa..
> git tree:       linux-next
> final crash:    https://syzkaller.appspot.com/x/report.txt?x=1316d2a3200000
> console output: https://syzkaller.appspot.com/x/log.txt?x=1516d2a3200000
> kernel config:  https://syzkaller.appspot.com/x/.config?x=c8b6073d992e8217
> dashboard link:
> https://syzkaller.appspot.com/bug?extid=a25307ad099309f1c2b9
> syz repro:      https://syzkaller.appspot.com/x/repro.syz?x=16b2c56f200000
> C reproducer:   https://syzkaller.appspot.com/x/repro.c?x=13b8890b200000
> 
> Reported-by: syz...@sy...
> Fixes: 52dfae5c ("tipc: obtain node identity from interface by default")

Re: [tipc-discussion] general protection fault in dst_dev_put (2)

[Jon M. <jon...@er...>]

This one points to a bug that was fixed in commit dc4501ff2875 ("tipc: fix a double free in tipc_enable_bearer()"). 
That commit was applied to 4.20, but is not present in 4.16, which caused this dump. 
To get further with this I need a dump of the actual crash in 4.20 or 5.0.
Maybe we are looking at two different problems here?

///jon



> -----Original Message-----
> From: net...@vg... <net...@vg...>
> On Behalf Of syzbot
> Sent: 19-Mar-19 18:24
> To: da...@da...; ds...@gm...; Jon Maloy
> <jon...@er...>; lin...@vg...;
> ne...@vg...; pa...@ne...; syzkaller-
> bu...@go...; tip...@li...;
> yin...@wi...
> Subject: Re: general protection fault in dst_dev_put (2)
> 
> syzbot has bisected this bug to:
> 
> commit 52dfae5c85a4c1078e9f1d5e8947d4a25f73dd81
> Author: Jon Maloy <jon...@er...>
> Date:   Thu Mar 22 19:42:52 2018 +0000
> 
>      tipc: obtain node identity from interface by default
> 
> bisection log:  https://syzkaller.appspot.com/x/bisect.txt?x=1631fc2b200000
> start commit:   52dfae5c tipc: obtain node identity from interface by defa..
> git tree:       upstream
> final crash:    https://syzkaller.appspot.com/x/report.txt?x=1531fc2b200000
> console output: https://syzkaller.appspot.com/x/log.txt?x=1131fc2b200000
> kernel config:  https://syzkaller.appspot.com/x/.config?x=8f00801d7b7c4fe6
> dashboard link:
> https://syzkaller.appspot.com/bug?extid=9d4c12bfd45a58738d0a
> syz repro:      https://syzkaller.appspot.com/x/repro.syz?x=17254878c00000
> C reproducer:   https://syzkaller.appspot.com/x/repro.c?x=171ca5df400000
> 
> Reported-by: syz...@sy...
> Fixes: 52dfae5c ("tipc: obtain node identity from interface by default")

Re: [tipc-discussion] Random quick fixes

[Jon M. <jon...@er...>]

This looks ok to me. If you are interested I can add you as developer in the project, and you can check in the changes to SF yourself.

BR
///jon


> -----Original Message-----
> From: Peter Fröhlich <pet...@gm...>
> Sent: 20-Mar-19 12:31
> To: tip...@li...
> Subject: [tipc-discussion] Random quick fixes
> 
> Hi all,
> 
> Sorry, I am not sure what the "official" process is but I had to make a few
> changes to the current tipcutils to get things to compile without warnings.
> The diffs are below. Note that I didn't really "test" any of this, I am just
> starting out with tipc but I like to start from a "clean slate" when I look into a
> new codebase. Feel free to ignore if it's useless.
> 
> Best,
> Peter
> 
> ---
> diff --git a/test/group_test/worker.c b/test/group_test/worker.c index
> 95a5234..2e1f7c6 100644
> --- a/test/group_test/worker.c
> +++ b/test/group_test/worker.c
> @@ -168,7 +168,7 @@ static void print_stats(struct context *ctx)
>      unsigned long long msg_per_sec;
>          unsigned long long thruput;
>      unsigned long long snt, rcvd;
> -    char str[64];
> +    char str[128];
> 
>      if (!ctx->stats && (vlevel < 2))
>          return;
> @@ -349,7 +349,7 @@ static void rcv_traffic_msgs(struct context *ctx)
>              ctx->error = BAD_SRC_ADDR;
>          else if (hdr->check_length && rc != ctx->msg_len)
>              ctx->error = BAD_LENGTH;
> -
> +
>          if (mtyp == UNICAST)
>              ctx->rcv_uc++;
>                  else  if (mtyp == ANYCAST) diff --git a/test/iovec/iovec_client.c
> b/test/iovec/iovec_client.c index 4eabb2d..ca5600a 100644
> --- a/test/iovec/iovec_client.c
> +++ b/test/iovec/iovec_client.c
> @@ -2,6 +2,7 @@
>   *
>   */
> 
> +#include <sys/uio.h>
>  #include <sys/types.h>
>  #include <sys/socket.h>
>  #include <stdlib.h>
> @@ -93,7 +94,7 @@ static struct msghdr *create_msg(int msgsize, int
> numiov) {
> 
>      struct msghdr *m;
>      struct iovec *iov;
> -    int p, i, iovlen, total = 0;
> +    int i, iovlen, total = 0;
>      uint16_t cksum;
>      char *data;
>      char *iovp;
> diff --git a/test/iovec/iovec_server.c b/test/iovec/iovec_server.c index
> ee3fd1d..d19bc4f 100644
> --- a/test/iovec/iovec_server.c
> +++ b/test/iovec/iovec_server.c
> @@ -243,9 +243,10 @@ void sigcatcher(int sig)
>      case SIGKILL:
>      case SIGHUP:
>          for (i=0; i < 3; i++) {
> -            if (server[i] > 0)
> +            if (server[i] > 0) {
>                  printf("Killing test server %d\n", i);
>                  kill(server[i], SIGTERM);
> +            }
>          }
>          printf("Iovec server killed\n");
>          fflush(stdout);
> diff --git a/test/ptts/tipc_ts_common.c b/test/ptts/tipc_ts_common.c index
> d7da0b7..8d73a7a 100644
> --- a/test/ptts/tipc_ts_common.c
> +++ b/test/ptts/tipc_ts_common.c
> @@ -403,7 +403,7 @@ void setOption
>          int value       /* value to set */
>  )
>  {
> -    char failString[50]; /* string for failure return code */
> +    char failString[128]; /* string for failure return code */
> 
>      if (setsockopt (sockfd, SOL_TIPC, opt,
>                      (char*)&value, sizeof (value))) { @@ -423,7 +423,7 @@ void
> getOption
>          int *value     /* returned value */
>  )
>  {
> -    char failString[50];  /* string for failure return code */
> +    char failString[128];  /* string for failure return code */
>      socklen_t size;       /* size of the socket option value */
> 
>      size = sizeof(*value);
> @@ -934,7 +934,7 @@ void common_test_socketOptions(void)
>      int sol;                                 /* socket to listen on */
>      struct sockaddr_tipc saddr;              /* address structure for socket */
>      char buf[2048];                          /* buffer for message */
> -    char failString[100];                     /* string for the
> failure return code */
> +    char failString[128];                     /* string for the
> failure return code */
> 
>      int ii;                                  /* loop index for the sockets */
>      int jj;                                  /* loop index for the
> option value */
> @@ -1112,7 +1112,7 @@ void common_test_recvfrom  {
> 
>      char buffer [MAX_STR];        /* buffer for message to be received in */
> -    char failStr [50 + MAX_STR];  /* string for failure return code */
> +    char failStr [128 + MAX_STR];  /* string for failure return code */
>      struct sockaddr_tipc addr;    /* address of socket */
>      int so;                       /* socket to use */
>      int size;                     /* size of the received message */
> diff --git a/utils/tipc-link-watcher/tipc-link-watcher.c
> b/utils/tipc-link-watcher/tipc-link-watcher.c
> index a24cf19..075d2eb 100644
> --- a/utils/tipc-link-watcher/tipc-link-watcher.c
> +++ b/utils/tipc-link-watcher/tipc-link-watcher.c
> @@ -1552,7 +1552,7 @@ static bool handle_getsubopt(char **optionp, char
> *const *tokens,
>      int sub_opt;
>      int tmp;
>      char *valuep;
> -    while (*optionp != '\0') {
> +    while (**optionp != '\0') {
>          sub_opt = getsubopt (optionp, tokens, &valuep);
>          if ((t_low < sub_opt) && (sub_opt < t_high) )
>          {
> 
> 
> _______________________________________________
> tipc-discussion mailing list
> tip...@li...
> https://lists.sourceforge.net/lists/listinfo/tipc-discussion

[tipc-discussion] Random quick fixes

[Peter F. <pet...@gm...>]

Hi all,

Sorry, I am not sure what the "official" process is but I had to make
a few changes to the current tipcutils to get things to compile
without warnings. The diffs are below. Note that I didn't really
"test" any of this, I am just starting out with tipc but I like to
start from a "clean slate" when I look into a new codebase. Feel free
to ignore if it's useless.

Best,
Peter

---
diff --git a/test/group_test/worker.c b/test/group_test/worker.c
index 95a5234..2e1f7c6 100644
--- a/test/group_test/worker.c
+++ b/test/group_test/worker.c
@@ -168,7 +168,7 @@ static void print_stats(struct context *ctx)
     unsigned long long msg_per_sec;
         unsigned long long thruput;
     unsigned long long snt, rcvd;
-    char str[64];
+    char str[128];

     if (!ctx->stats && (vlevel < 2))
         return;
@@ -349,7 +349,7 @@ static void rcv_traffic_msgs(struct context *ctx)
             ctx->error = BAD_SRC_ADDR;
         else if (hdr->check_length && rc != ctx->msg_len)
             ctx->error = BAD_LENGTH;
-
+
         if (mtyp == UNICAST)
             ctx->rcv_uc++;
                 else  if (mtyp == ANYCAST)
diff --git a/test/iovec/iovec_client.c b/test/iovec/iovec_client.c
index 4eabb2d..ca5600a 100644
--- a/test/iovec/iovec_client.c
+++ b/test/iovec/iovec_client.c
@@ -2,6 +2,7 @@
  *
  */

+#include <sys/uio.h>
 #include <sys/types.h>
 #include <sys/socket.h>
 #include <stdlib.h>
@@ -93,7 +94,7 @@ static struct msghdr *create_msg(int msgsize, int numiov) {

     struct msghdr *m;
     struct iovec *iov;
-    int p, i, iovlen, total = 0;
+    int i, iovlen, total = 0;
     uint16_t cksum;
     char *data;
     char *iovp;
diff --git a/test/iovec/iovec_server.c b/test/iovec/iovec_server.c
index ee3fd1d..d19bc4f 100644
--- a/test/iovec/iovec_server.c
+++ b/test/iovec/iovec_server.c
@@ -243,9 +243,10 @@ void sigcatcher(int sig)
     case SIGKILL:
     case SIGHUP:
         for (i=0; i < 3; i++) {
-            if (server[i] > 0)
+            if (server[i] > 0) {
                 printf("Killing test server %d\n", i);
                 kill(server[i], SIGTERM);
+            }
         }
         printf("Iovec server killed\n");
         fflush(stdout);
diff --git a/test/ptts/tipc_ts_common.c b/test/ptts/tipc_ts_common.c
index d7da0b7..8d73a7a 100644
--- a/test/ptts/tipc_ts_common.c
+++ b/test/ptts/tipc_ts_common.c
@@ -403,7 +403,7 @@ void setOption
         int value       /* value to set */
 )
 {
-    char failString[50]; /* string for failure return code */
+    char failString[128]; /* string for failure return code */

     if (setsockopt (sockfd, SOL_TIPC, opt,
                     (char*)&value, sizeof (value))) {
@@ -423,7 +423,7 @@ void getOption
         int *value     /* returned value */
 )
 {
-    char failString[50];  /* string for failure return code */
+    char failString[128];  /* string for failure return code */
     socklen_t size;       /* size of the socket option value */

     size = sizeof(*value);
@@ -934,7 +934,7 @@ void common_test_socketOptions(void)
     int sol;                                 /* socket to listen on */
     struct sockaddr_tipc saddr;              /* address structure for socket */
     char buf[2048];                          /* buffer for message */
-    char failString[100];                     /* string for the
failure return code */
+    char failString[128];                     /* string for the
failure return code */

     int ii;                                  /* loop index for the sockets */
     int jj;                                  /* loop index for the
option value */
@@ -1112,7 +1112,7 @@ void common_test_recvfrom
 {

     char buffer [MAX_STR];        /* buffer for message to be received in */
-    char failStr [50 + MAX_STR];  /* string for failure return code */
+    char failStr [128 + MAX_STR];  /* string for failure return code */
     struct sockaddr_tipc addr;    /* address of socket */
     int so;                       /* socket to use */
     int size;                     /* size of the received message */
diff --git a/utils/tipc-link-watcher/tipc-link-watcher.c
b/utils/tipc-link-watcher/tipc-link-watcher.c
index a24cf19..075d2eb 100644
--- a/utils/tipc-link-watcher/tipc-link-watcher.c
+++ b/utils/tipc-link-watcher/tipc-link-watcher.c
@@ -1552,7 +1552,7 @@ static bool handle_getsubopt(char **optionp,
char *const *tokens,
     int sub_opt;
     int tmp;
     char *valuep;
-    while (*optionp != '\0') {
+    while (**optionp != '\0') {
         sub_opt = getsubopt (optionp, tokens, &valuep);
         if ((t_low < sub_opt) && (sub_opt < t_high) )
         {

Re: [tipc-discussion] [PATCH net-next] tipc: fix cancellation of topology subscriptions

[Jon M. <jon...@er...>]

Yes. I already added it to the patch.

///jon


> -----Original Message-----
> From: Partha <par...@gm...>
> Sent: 20-Mar-19 09:34
> To: Erik Hugne <eri...@gm...>; Jon Maloy
> <jon...@er...>
> Cc: tip...@li...
> Subject: Re: [tipc-discussion] [PATCH net-next] tipc: fix cancellation of
> topology subscriptions
> 
> Hi Erik / Jon,
> 
> The bug was introduced by:
> commit 8985ecc7c1e07c42acc1e44ac56fa224f8a5c62f
> Author: Jon Maloy <jon...@er...>
> Date:   Thu Feb 15 10:40:46 2018 +0100
> 
>      tipc: simplify endianness handling in topology subscriber
> 
> 
> -       /* Detect & process a subscription cancellation request */
> -       if (s->filter & htohl(TIPC_SUB_CANCEL, swap)) {
> -               s->filter &= ~htohl(TIPC_SUB_CANCEL, swap);
> +       if (tipc_sub_read(s, filter) & TIPC_SUB_CANCEL) {
>                  tipc_con_delete_sub(con, s);
>                  return 0;
>          }
> 
> /Partha
> 
> On 2019-03-19 12:55, Erik Hugne wrote:
> > Well, i couldnt find any commit where the bug was introduced.. but i
> > agree that it should probably be in net.
> > Could you change the subject header and submit?
> >
> > //E
> >
> > On Tue, 19 Mar 2019, 11:56 Jon Maloy, <jon...@er...> wrote:
> >
> >> Acked-by: Jon.
> >> Why did you mark it "net-next"? To me this looks like a bug fix and
> >> should go to 'net'.
> >> I'll post it for you if you want.
> >>
> >> ///jon
> >>
> >>> -----Original Message-----
> >>> From: eri...@gm... <eri...@gm...>
> >>> Sent: 19-Mar-19 11:08
> >>> To: tip...@li...; Jon Maloy
> >>> <jon...@er...>; yin...@wi...
> >>> Cc: Erik Hugne <eri...@gm...>
> >>> Subject: [PATCH net-next] tipc: fix cancellation of topology
> >> subscriptions
> >>>
> >>> From: Erik Hugne <eri...@gm...>
> >>>
> >>> When cancelling a subscription, we have to clear the cancel bit in
> >>> the
> >> request
> >>> before iterating over any established subscriptions with memcmp.
> >> Otherwise
> >>> no subscription will ever be found, and it will not be possible to
> >> unsubscribe.
> >>>
> >>> Signed-off-by: Erik Hugne <eri...@gm...>
> >>> ---
> >>> Ignore my previous garbage patch, this solve the problem with one line.
> >>> Looking through the history of topsrv.c, i cannot find that this has
> >>> ever worked.
> >>>
> >>>   net/tipc/topsrv.c | 1 +
> >>>   1 file changed, 1 insertion(+)
> >>>
> >>> diff --git a/net/tipc/topsrv.c b/net/tipc/topsrv.c index
> >>> 4a708a4e8583..b05a89a723b1 100644
> >>> --- a/net/tipc/topsrv.c
> >>> +++ b/net/tipc/topsrv.c
> >>> @@ -363,6 +363,7 @@ static int tipc_conn_rcv_sub(struct tipc_topsrv
> *srv,
> >>>        struct tipc_subscription *sub;
> >>>
> >>>        if (tipc_sub_read(s, filter) & TIPC_SUB_CANCEL) {
> >>> +             s->filter ^= __constant_ntohl(TIPC_SUB_CANCEL);
> >>>                tipc_conn_delete_sub(con, s);
> >>>                return 0;
> >>>        }
> >>> --
> >>> 2.14.1
> >>
> >>
> >
> > _______________________________________________
> > tipc-discussion mailing list
> > tip...@li...
> > https://lists.sourceforge.net/lists/listinfo/tipc-discussion
> >

Re: [tipc-discussion] Topology subscriptions in bash

[Jon M. <jon...@er...>]

Nice.  Is this all for the bash project, or do we need to add something to tipc_utils?

///jon


> -----Original Message-----
> From: Erik Hugne <eri...@gm...>
> Sent: 19-Mar-19 22:33
> To: tip...@li...; Jon Maloy
> <jon...@er...>
> Subject: Topology subscriptions in bash
> 
> A small tool to get shell integrations to the TIPC topology server, maybe
> someone will find it useful https://github.com/Hugne/tipc.sh
> 
> The subscriber examples use part of the functionality that i plan to submit to
> the bash project for network redirection support, e.g:
> 
> [root@kdev ~] exec 5</dev/tipc/1000/1
> [root@kdev ~] echo hello world >/dev/tipc/1000/1 [root@kdev ~] cat <&5
> hello world ^C [root@kdev ~]
> 
> //E

[tipc-discussion] [net 1/1] tipc: fix cancellation of topology subscriptions

[Jon M. <jon...@er...>]

From: Erik Hugne <eri...@gm...>

When cancelling a subscription, we have to clear the cancel bit in the
request before iterating over any established subscriptions with memcmp.
Otherwise no subscription will ever be found, and it will not be
possible to explicitly unsubscribe individual subscriptions.

Fixes: 8985ecc7c1e0 ("tipc: simplify endianness handling in topology subscriber")
Signed-off-by: Erik Hugne <eri...@gm...>
Signed-off-by: Jon Maloy <jon...@er...>
---
 net/tipc/topsrv.c | 1 +
 1 file changed, 1 insertion(+)

diff --git a/net/tipc/topsrv.c b/net/tipc/topsrv.c
index 4a708a4..b05a89a 100644
--- a/net/tipc/topsrv.c
+++ b/net/tipc/topsrv.c
@@ -363,6 +363,7 @@ static int tipc_conn_rcv_sub(struct tipc_topsrv *srv,
 	struct tipc_subscription *sub;
 
 	if (tipc_sub_read(s, filter) & TIPC_SUB_CANCEL) {
+		s->filter ^= __constant_ntohl(TIPC_SUB_CANCEL);
 		tipc_conn_delete_sub(con, s);
 		return 0;
 	}
-- 
2.1.4

Re: [tipc-discussion] [PATCH net-next] tipc: fix cancellation of topology subscriptions

[Partha <par...@gm...>]

Hi Erik / Jon,

The bug was introduced by:
commit 8985ecc7c1e07c42acc1e44ac56fa224f8a5c62f
Author: Jon Maloy <jon...@er...>
Date:   Thu Feb 15 10:40:46 2018 +0100

     tipc: simplify endianness handling in topology subscriber


-       /* Detect & process a subscription cancellation request */
-       if (s->filter & htohl(TIPC_SUB_CANCEL, swap)) {
-               s->filter &= ~htohl(TIPC_SUB_CANCEL, swap);
+       if (tipc_sub_read(s, filter) & TIPC_SUB_CANCEL) {
                 tipc_con_delete_sub(con, s);
                 return 0;
         }

/Partha

On 2019-03-19 12:55, Erik Hugne wrote:
> Well, i couldnt find any commit where the bug was introduced.. but i agree
> that it should probably be in net.
> Could you change the subject header and submit?
> 
> //E
> 
> On Tue, 19 Mar 2019, 11:56 Jon Maloy, <jon...@er...> wrote:
> 
>> Acked-by: Jon.
>> Why did you mark it "net-next"? To me this looks like a bug fix and should
>> go to 'net'.
>> I'll post it for you if you want.
>>
>> ///jon
>>
>>> -----Original Message-----
>>> From: eri...@gm... <eri...@gm...>
>>> Sent: 19-Mar-19 11:08
>>> To: tip...@li...; Jon Maloy
>>> <jon...@er...>; yin...@wi...
>>> Cc: Erik Hugne <eri...@gm...>
>>> Subject: [PATCH net-next] tipc: fix cancellation of topology
>> subscriptions
>>>
>>> From: Erik Hugne <eri...@gm...>
>>>
>>> When cancelling a subscription, we have to clear the cancel bit in the
>> request
>>> before iterating over any established subscriptions with memcmp.
>> Otherwise
>>> no subscription will ever be found, and it will not be possible to
>> unsubscribe.
>>>
>>> Signed-off-by: Erik Hugne <eri...@gm...>
>>> ---
>>> Ignore my previous garbage patch, this solve the problem with one line.
>>> Looking through the history of topsrv.c, i cannot find that this has ever
>>> worked.
>>>
>>>   net/tipc/topsrv.c | 1 +
>>>   1 file changed, 1 insertion(+)
>>>
>>> diff --git a/net/tipc/topsrv.c b/net/tipc/topsrv.c index
>>> 4a708a4e8583..b05a89a723b1 100644
>>> --- a/net/tipc/topsrv.c
>>> +++ b/net/tipc/topsrv.c
>>> @@ -363,6 +363,7 @@ static int tipc_conn_rcv_sub(struct tipc_topsrv *srv,
>>>        struct tipc_subscription *sub;
>>>
>>>        if (tipc_sub_read(s, filter) & TIPC_SUB_CANCEL) {
>>> +             s->filter ^= __constant_ntohl(TIPC_SUB_CANCEL);
>>>                tipc_conn_delete_sub(con, s);
>>>                return 0;
>>>        }
>>> --
>>> 2.14.1
>>
>>
> 
> _______________________________________________
> tipc-discussion mailing list
> tip...@li...
> https://lists.sourceforge.net/lists/listinfo/tipc-discussion
> 

[tipc-discussion] [net-next 3/3] tipc: adapt link failover for new Gap-ACK algorithm

[Tuong L. <tuo...@de...>]

In commit 0ae955e2656d ("tipc: improve TIPC throughput by Gap ACK
blocks"), we enhance the link transmq by releasing as many packets as
possible with the multi-ACKs from peer node. This also means the queue
is now non-linear and the peer link deferdq becomes vital.

Whereas, in the case of link failover, all messages in the link transmq
need to be transmitted as tunnel messages in such a way that message
sequentiality and cardinality per sender is preserved. This requires us
to maintain the link deferdq somehow, so that when the tunnel messages
arrive, the inner user messages along with the ones in the deferdq will
be delivered to upper layer correctly.

The commit accomplishes this by defining a new queue in the TIPC link
structure to hold the old link deferdq when link failover happens and
process it upon receipt of tunnel messages.

Also, in the case of link syncing, the link deferdq will not be purged
to avoid unnecessary retransmissions that in the worst case will fail
because the packets might have been freed on the sending side.

Acked-by: Jon Maloy <jon...@er...>
Signed-off-by: Tuong Lien <tuo...@de...>
---
 net/tipc/link.c | 106 ++++++++++++++++++++++++++++++++++++++++++--------------
 1 file changed, 80 insertions(+), 26 deletions(-)

diff --git a/net/tipc/link.c b/net/tipc/link.c
index 1f2cde0d025f..3cb9f326ee6f 100644
--- a/net/tipc/link.c
+++ b/net/tipc/link.c
@@ -151,6 +151,7 @@ struct tipc_link {
 	/* Failover/synch */
 	u16 drop_point;
 	struct sk_buff *failover_reasm_skb;
+	struct sk_buff_head failover_deferdq;
 
 	/* Max packet negotiation */
 	u16 mtu;
@@ -498,6 +499,7 @@ bool tipc_link_create(struct net *net, char *if_name, int bearer_id,
 	__skb_queue_head_init(&l->transmq);
 	__skb_queue_head_init(&l->backlogq);
 	__skb_queue_head_init(&l->deferdq);
+	__skb_queue_head_init(&l->failover_deferdq);
 	skb_queue_head_init(&l->wakeupq);
 	skb_queue_head_init(l->inputq);
 	return true;
@@ -888,6 +890,7 @@ void tipc_link_reset(struct tipc_link *l)
 	__skb_queue_purge(&l->transmq);
 	__skb_queue_purge(&l->deferdq);
 	__skb_queue_purge(&l->backlogq);
+	__skb_queue_purge(&l->failover_deferdq);
 	l->backlog[TIPC_LOW_IMPORTANCE].len = 0;
 	l->backlog[TIPC_MEDIUM_IMPORTANCE].len = 0;
 	l->backlog[TIPC_HIGH_IMPORTANCE].len = 0;
@@ -1159,34 +1162,14 @@ static bool tipc_data_input(struct tipc_link *l, struct sk_buff *skb,
  * Consumes buffer
  */
 static int tipc_link_input(struct tipc_link *l, struct sk_buff *skb,
-			   struct sk_buff_head *inputq)
+			   struct sk_buff_head *inputq,
+			   struct sk_buff **reasm_skb)
 {
 	struct tipc_msg *hdr = buf_msg(skb);
-	struct sk_buff **reasm_skb = &l->reasm_buf;
 	struct sk_buff *iskb;
 	struct sk_buff_head tmpq;
 	int usr = msg_user(hdr);
-	int rc = 0;
 	int pos = 0;
-	int ipos = 0;
-
-	if (unlikely(usr == TUNNEL_PROTOCOL)) {
-		if (msg_type(hdr) == SYNCH_MSG) {
-			__skb_queue_purge(&l->deferdq);
-			goto drop;
-		}
-		if (!tipc_msg_extract(skb, &iskb, &ipos))
-			return rc;
-		kfree_skb(skb);
-		skb = iskb;
-		hdr = buf_msg(skb);
-		if (less(msg_seqno(hdr), l->drop_point))
-			goto drop;
-		if (tipc_data_input(l, skb, inputq))
-			return rc;
-		usr = msg_user(hdr);
-		reasm_skb = &l->failover_reasm_skb;
-	}
 
 	if (usr == MSG_BUNDLER) {
 		skb_queue_head_init(&tmpq);
@@ -1211,11 +1194,66 @@ static int tipc_link_input(struct tipc_link *l, struct sk_buff *skb,
 		tipc_link_bc_init_rcv(l->bc_rcvlink, hdr);
 		tipc_bcast_unlock(l->net);
 	}
-drop:
+
 	kfree_skb(skb);
 	return 0;
 }
 
+/* tipc_link_tnl_rcv() - receive TUNNEL_PROTOCOL message, drop or process the
+ *			 inner message along with the ones in the old link's
+ *			 deferdq
+ * @l: tunnel link
+ * @skb: TUNNEL_PROTOCOL message
+ * @inputq: queue to put messages ready for delivery
+ */
+static int tipc_link_tnl_rcv(struct tipc_link *l, struct sk_buff *skb,
+			     struct sk_buff_head *inputq)
+{
+	struct sk_buff **reasm_skb = &l->failover_reasm_skb;
+	struct sk_buff_head *fdefq = &l->failover_deferdq;
+	struct tipc_msg *hdr = buf_msg(skb);
+	struct sk_buff *iskb;
+	int ipos = 0;
+	int rc = 0;
+	u16 seqno;
+
+	/* SYNCH_MSG */
+	if (msg_type(hdr) == SYNCH_MSG)
+		goto drop;
+
+	/* FAILOVER_MSG */
+	if (!tipc_msg_extract(skb, &iskb, &ipos)) {
+		pr_warn_ratelimited("Cannot extract FAILOVER_MSG, defq: %d\n",
+				    skb_queue_len(fdefq));
+		return rc;
+	}
+
+	do {
+		seqno = buf_seqno(iskb);
+
+		if (unlikely(less(seqno, l->drop_point))) {
+			kfree_skb(iskb);
+			continue;
+		}
+
+		if (unlikely(seqno != l->drop_point)) {
+			__tipc_skb_queue_sorted(fdefq, seqno, iskb);
+			continue;
+		}
+
+		l->drop_point++;
+
+		if (!tipc_data_input(l, iskb, inputq))
+			rc |= tipc_link_input(l, iskb, inputq, reasm_skb);
+		if (unlikely(rc))
+			break;
+	} while ((iskb = __tipc_skb_dequeue(fdefq, l->drop_point)));
+
+drop:
+	kfree_skb(skb);
+	return rc;
+}
+
 static bool tipc_link_release_pkts(struct tipc_link *l, u16 acked)
 {
 	bool released = false;
@@ -1457,8 +1495,11 @@ int tipc_link_rcv(struct tipc_link *l, struct sk_buff *skb,
 		/* Deliver packet */
 		l->rcv_nxt++;
 		l->stats.recv_pkts++;
-		if (!tipc_data_input(l, skb, l->inputq))
-			rc |= tipc_link_input(l, skb, l->inputq);
+
+		if (unlikely(msg_user(hdr) == TUNNEL_PROTOCOL))
+			rc |= tipc_link_tnl_rcv(l, skb, l->inputq);
+		else if (!tipc_data_input(l, skb, l->inputq))
+			rc |= tipc_link_input(l, skb, l->inputq, &l->reasm_buf);
 		if (unlikely(++l->rcv_unacked >= TIPC_MIN_LINK_WIN))
 			rc |= tipc_link_build_state_msg(l, xmitq);
 		if (unlikely(rc & ~TIPC_LINK_SND_STATE))
@@ -1588,6 +1629,7 @@ void tipc_link_create_dummy_tnl_msg(struct tipc_link *l,
 void tipc_link_tnl_prepare(struct tipc_link *l, struct tipc_link *tnl,
 			   int mtyp, struct sk_buff_head *xmitq)
 {
+	struct sk_buff_head *fdefq = &tnl->failover_deferdq;
 	struct sk_buff *skb, *tnlskb;
 	struct tipc_msg *hdr, tnlhdr;
 	struct sk_buff_head *queue = &l->transmq;
@@ -1615,7 +1657,11 @@ void tipc_link_tnl_prepare(struct tipc_link *l, struct tipc_link *tnl,
 	/* Initialize reusable tunnel packet header */
 	tipc_msg_init(tipc_own_addr(l->net), &tnlhdr, TUNNEL_PROTOCOL,
 		      mtyp, INT_H_SIZE, l->addr);
-	pktcnt = skb_queue_len(&l->transmq) + skb_queue_len(&l->backlogq);
+	if (mtyp == SYNCH_MSG)
+		pktcnt = l->snd_nxt - buf_seqno(skb_peek(&l->transmq));
+	else
+		pktcnt = skb_queue_len(&l->transmq);
+	pktcnt += skb_queue_len(&l->backlogq);
 	msg_set_msgcnt(&tnlhdr, pktcnt);
 	msg_set_bearer_id(&tnlhdr, l->peer_bearer_id);
 tnl:
@@ -1646,6 +1692,14 @@ void tipc_link_tnl_prepare(struct tipc_link *l, struct tipc_link *tnl,
 		tnl->drop_point = l->rcv_nxt;
 		tnl->failover_reasm_skb = l->reasm_buf;
 		l->reasm_buf = NULL;
+
+		/* Failover the link's deferdq */
+		if (unlikely(!skb_queue_empty(fdefq))) {
+			pr_warn("Link failover deferdq not empty: %d!\n",
+				skb_queue_len(fdefq));
+			__skb_queue_purge(fdefq);
+		}
+		skb_queue_splice_init(&l->deferdq, fdefq);
 	}
 }
 
-- 
2.13.7

[tipc-discussion] [net-next 2/3] tipc: reduce duplicate packets for unicast traffic

[Tuong L. <tuo...@de...>]

For unicast transmission, the current NACK sending althorithm is over-
active that forces the sending side to retransmit a packet that is not
really lost but just arrived at the receiving side with some delay, or
even retransmit same packets that have already been retransmitted
before. As a result, many duplicates are observed also under normal
condition, ie. without packet loss.

One example case is: node1 transmits 1 2 3 4 10 5 6 7 8 9, when node2
receives packet #10, it puts into the deferdq. When the packet #5 comes
it sends NACK with gap [6 - 9]. However, shortly after that, when
packet #6 arrives, it pulls out packet #10 from the deferfq, but it is
still out of order, so it makes another NACK with gap [7 - 9] and so on
... Finally, node1 has to retransmit the packets 5 6 7 8 9 a number of
times, but in fact all the packets are not lost at all, so duplicates!

This commit reduces duplicates by changing the condition to send NACK,
also restricting the retransmissions on individual packets via a timer
of about 1ms. However, it also needs to say that too tricky condition
for NACKs or too long timeout value for retransmissions will result in
performance reducing! The criterias in this commit are found to be
effective for both the requirements to reduce duplicates but not affect
performance.

The tipc_link_rcv() is also improved to only dequeue skb from the link
deferdq if it is expected (ie. its seqno <= rcv_nxt).

Acked-by: Jon Maloy <jon...@er...>
Signed-off-by: Tuong Lien <tuo...@de...>
---
 net/tipc/link.c | 26 ++++++++++++++++----------
 net/tipc/msg.h  | 21 +++++++++++++++++++++
 2 files changed, 37 insertions(+), 10 deletions(-)

diff --git a/net/tipc/link.c b/net/tipc/link.c
index 5aee1ed23ba9..1f2cde0d025f 100644
--- a/net/tipc/link.c
+++ b/net/tipc/link.c
@@ -209,6 +209,7 @@ enum {
 };
 
 #define TIPC_BC_RETR_LIM msecs_to_jiffies(10)   /* [ms] */
+#define TIPC_UC_RETR_TIME (jiffies + msecs_to_jiffies(1))
 
 /*
  * Interval between NACKs when packets arrive out of order
@@ -1305,6 +1306,10 @@ static void tipc_link_advance_transmq(struct tipc_link *l, u16 acked, u16 gap,
 			kfree_skb(skb);
 		} else if (less_eq(seqno, acked + gap)) {
 			/* retransmit skb */
+			if (time_before(jiffies, TIPC_SKB_CB(skb)->nxt_retr))
+				continue;
+			TIPC_SKB_CB(skb)->nxt_retr = TIPC_UC_RETR_TIME;
+
 			_skb = __pskb_copy(skb, MIN_H_SIZE, GFP_ATOMIC);
 			if (!_skb)
 				continue;
@@ -1380,6 +1385,7 @@ static int tipc_link_build_nack_msg(struct tipc_link *l,
 				    struct sk_buff_head *xmitq)
 {
 	u32 def_cnt = ++l->stats.deferred_recv;
+	u32 defq_len = skb_queue_len(&l->deferdq);
 	int match1, match2;
 
 	if (link_is_bc_rcvlink(l)) {
@@ -1390,7 +1396,7 @@ static int tipc_link_build_nack_msg(struct tipc_link *l,
 		return 0;
 	}
 
-	if ((skb_queue_len(&l->deferdq) == 1) || !(def_cnt % TIPC_NACK_INTV))
+	if (defq_len >= 3 && !((defq_len - 3) % 16))
 		tipc_link_build_proto_msg(l, STATE_MSG, 0, 0, 0, 0, 0, xmitq);
 	return 0;
 }
@@ -1404,29 +1410,29 @@ int tipc_link_rcv(struct tipc_link *l, struct sk_buff *skb,
 		  struct sk_buff_head *xmitq)
 {
 	struct sk_buff_head *defq = &l->deferdq;
-	struct tipc_msg *hdr;
+	struct tipc_msg *hdr = buf_msg(skb);
 	u16 seqno, rcv_nxt, win_lim;
 	int rc = 0;
 
+	/* Verify and update link state */
+	if (unlikely(msg_user(hdr) == LINK_PROTOCOL))
+		return tipc_link_proto_rcv(l, skb, xmitq);
+
+	/* Don't send probe at next timeout expiration */
+	l->silent_intv_cnt = 0;
+
 	do {
 		hdr = buf_msg(skb);
 		seqno = msg_seqno(hdr);
 		rcv_nxt = l->rcv_nxt;
 		win_lim = rcv_nxt + TIPC_MAX_LINK_WIN;
 
-		/* Verify and update link state */
-		if (unlikely(msg_user(hdr) == LINK_PROTOCOL))
-			return tipc_link_proto_rcv(l, skb, xmitq);
-
 		if (unlikely(!link_is_up(l))) {
 			if (l->state == LINK_ESTABLISHING)
 				rc = TIPC_LINK_UP_EVT;
 			goto drop;
 		}
 
-		/* Don't send probe at next timeout expiration */
-		l->silent_intv_cnt = 0;
-
 		/* Drop if outside receive window */
 		if (unlikely(less(seqno, rcv_nxt) || more(seqno, win_lim))) {
 			l->stats.duplicates++;
@@ -1457,7 +1463,7 @@ int tipc_link_rcv(struct tipc_link *l, struct sk_buff *skb,
 			rc |= tipc_link_build_state_msg(l, xmitq);
 		if (unlikely(rc & ~TIPC_LINK_SND_STATE))
 			break;
-	} while ((skb = __skb_dequeue(defq)));
+	} while ((skb = __tipc_skb_dequeue(defq, l->rcv_nxt)));
 
 	return rc;
 drop:
diff --git a/net/tipc/msg.h b/net/tipc/msg.h
index dcb73421c19a..9e64d35db393 100644
--- a/net/tipc/msg.h
+++ b/net/tipc/msg.h
@@ -1150,4 +1150,25 @@ static inline void tipc_skb_queue_splice_tail_init(struct sk_buff_head *list,
 	tipc_skb_queue_splice_tail(&tmp, head);
 }
 
+/* __tipc_skb_dequeue() - dequeue the head skb according to expected seqno
+ * @list: list to be dequeued from
+ * @seqno: seqno of the expected msg
+ *
+ * returns skb dequeued from the list if its seqno is less than or equal to
+ * the expected one, otherwise the skb is still hold
+ *
+ * Note: must be used with appropriate locks held only
+ */
+static inline struct sk_buff *__tipc_skb_dequeue(struct sk_buff_head *list,
+						 u16 seqno)
+{
+	struct sk_buff *skb = skb_peek(list);
+
+	if (skb && less_eq(buf_seqno(skb), seqno)) {
+		__skb_unlink(skb, list);
+		return skb;
+	}
+	return NULL;
+}
+
 #endif
-- 
2.13.7

[tipc-discussion] [net-next 1/3] tipc: improve TIPC throughput by Gap ACK blocks

[Tuong L. <tuo...@de...>]

During unicast link transmission, it's observed very often that because
of one or a few lost/dis-ordered packets, the sending side will fastly
reach the send window limit and must wait for the packets to be arrived
at the receiving side or in the worst case, a retransmission must be
done first. The sending side cannot release a lot of subsequent packets
in its transmq even though all of them might have already been received
by the receiving side.
That is, one or two packets dis-ordered/lost and dozens of packets have
to wait, this obviously reduces the overall throughput!

This commit introduces an algorithm to overcome this by using "Gap ACK
blocks". Basically, a Gap ACK block will consist of <ack, gap> numbers
that describes the link deferdq where packets have been got by the
receiving side but with gaps, for example:

      link deferdq: [1 2 3 4      10 11      13 14 15       20]
--> Gap ACK blocks:       <4, 5>,   <11, 1>,      <15, 4>, <20, 0>

The Gap ACK blocks will be sent to the sending side along with the
traditional ACK or NACK message. Immediately when receiving the message
the sending side will now not only release from its transmq the packets
ack-ed by the ACK but also by the Gap ACK blocks! So, more packets can
be enqueued and transmitted.
In addition, the sending side can now do "multi-retransmissions"
according to the Gaps reported in the Gap ACK blocks.

The new algorithm as verified helps greatly improve the TIPC throughput
especially under packet loss condition.

So far, a maximum of 32 blocks is quite enough without any "Too few Gap
ACK blocks" reports with a 5.0% packet loss rate, however this number
can be increased in the furture if needed.

Also, the patch is backward compatible.

Acked-by: Jon Maloy <jon...@er...>
Signed-off-by: Tuong Lien <tuo...@de...>
---
 net/tipc/link.c | 134 +++++++++++++++++++++++++++++++++++++++++++++++++++-----
 net/tipc/msg.h  |  30 +++++++++++++
 net/tipc/node.h |   6 ++-
 3 files changed, 158 insertions(+), 12 deletions(-)

diff --git a/net/tipc/link.c b/net/tipc/link.c
index 52d23b3ffaf5..5aee1ed23ba9 100644
--- a/net/tipc/link.c
+++ b/net/tipc/link.c
@@ -246,6 +246,10 @@ static int tipc_link_build_nack_msg(struct tipc_link *l,
 static void tipc_link_build_bc_init_msg(struct tipc_link *l,
 					struct sk_buff_head *xmitq);
 static bool tipc_link_release_pkts(struct tipc_link *l, u16 to);
+static u16 tipc_build_gap_ack_blks(struct tipc_link *l, void *data);
+static void tipc_link_advance_transmq(struct tipc_link *l, u16 acked, u16 gap,
+				      struct tipc_gap_ack_blks *ga,
+				      struct sk_buff_head *xmitq);
 
 /*
  *  Simple non-static link routines (i.e. referenced outside this file)
@@ -1226,6 +1230,102 @@ static bool tipc_link_release_pkts(struct tipc_link *l, u16 acked)
 	return released;
 }
 
+/* tipc_build_gap_ack_blks - build Gap ACK blocks
+ * @l: tipc link that data have come with gaps in sequence if any
+ * @data: data buffer to store the Gap ACK blocks after built
+ *
+ * returns the actual allocated memory size
+ */
+static u16 tipc_build_gap_ack_blks(struct tipc_link *l, void *data)
+{
+	struct sk_buff *skb = skb_peek(&l->deferdq);
+	struct tipc_gap_ack_blks *ga = data;
+	u16 len, expect, seqno = 0;
+	u8 n = 0;
+
+	if (!skb)
+		goto exit;
+
+	expect = buf_seqno(skb);
+	skb_queue_walk(&l->deferdq, skb) {
+		seqno = buf_seqno(skb);
+		if (unlikely(more(seqno, expect))) {
+			ga->gacks[n].ack = htons(expect - 1);
+			ga->gacks[n].gap = htons(seqno - expect);
+			if (++n >= MAX_GAP_ACK_BLKS) {
+				pr_info_ratelimited("Too few Gap ACK blocks!\n");
+				goto exit;
+			}
+		} else if (unlikely(less(seqno, expect))) {
+			pr_warn("Unexpected skb in deferdq!\n");
+			continue;
+		}
+		expect = seqno + 1;
+	}
+
+	/* last block */
+	ga->gacks[n].ack = htons(seqno);
+	ga->gacks[n].gap = 0;
+	n++;
+
+exit:
+	len = tipc_gap_ack_blks_sz(n);
+	ga->len = htons(len);
+	ga->gack_cnt = n;
+	return len;
+}
+
+/* tipc_link_advance_transmq - advance TIPC link transmq queue by releasing
+ *			       acked packets, also doing retransmissions if
+ *			       gaps found
+ * @l: tipc link with transmq queue to be advanced
+ * @acked: seqno of last packet acked by peer without any gaps before
+ * @gap: # of gap packets
+ * @ga: buffer pointer to Gap ACK blocks from peer
+ * @xmitq: queue for accumulating the retransmitted packets if any
+ */
+static void tipc_link_advance_transmq(struct tipc_link *l, u16 acked, u16 gap,
+				      struct tipc_gap_ack_blks *ga,
+				      struct sk_buff_head *xmitq)
+{
+	struct sk_buff *skb, *_skb, *tmp;
+	struct tipc_msg *hdr;
+	u16 bc_ack = l->bc_rcvlink->rcv_nxt - 1;
+	u16 ack = l->rcv_nxt - 1;
+	u16 seqno;
+	u16 n = 0;
+
+	skb_queue_walk_safe(&l->transmq, skb, tmp) {
+		seqno = buf_seqno(skb);
+
+next_gap_ack:
+		if (less_eq(seqno, acked)) {
+			/* release skb */
+			__skb_unlink(skb, &l->transmq);
+			kfree_skb(skb);
+		} else if (less_eq(seqno, acked + gap)) {
+			/* retransmit skb */
+			_skb = __pskb_copy(skb, MIN_H_SIZE, GFP_ATOMIC);
+			if (!_skb)
+				continue;
+			hdr = buf_msg(_skb);
+			msg_set_ack(hdr, ack);
+			msg_set_bcast_ack(hdr, bc_ack);
+			_skb->priority = TC_PRIO_CONTROL;
+			__skb_queue_tail(xmitq, _skb);
+			l->stats.retransmitted++;
+		} else {
+			/* retry with Gap ACK blocks if any */
+			if (!ga || n >= ga->gack_cnt)
+				break;
+			acked = ntohs(ga->gacks[n].ack);
+			gap = ntohs(ga->gacks[n].gap);
+			n++;
+			goto next_gap_ack;
+		}
+	}
+}
+
 /* tipc_link_build_state_msg: prepare link state message for transmission
  *
  * Note that sending of broadcast ack is coordinated among nodes, to reduce
@@ -1378,6 +1478,7 @@ static void tipc_link_build_proto_msg(struct tipc_link *l, int mtyp, bool probe,
 	struct tipc_mon_state *mstate = &l->mon_state;
 	int dlen = 0;
 	void *data;
+	u16 glen = 0;
 
 	/* Don't send protocol message during reset or link failover */
 	if (tipc_link_is_blocked(l))
@@ -1390,8 +1491,8 @@ static void tipc_link_build_proto_msg(struct tipc_link *l, int mtyp, bool probe,
 		rcvgap = buf_seqno(skb_peek(dfq)) - l->rcv_nxt;
 
 	skb = tipc_msg_create(LINK_PROTOCOL, mtyp, INT_H_SIZE,
-			      tipc_max_domain_size, l->addr,
-			      tipc_own_addr(l->net), 0, 0, 0);
+			      tipc_max_domain_size + MAX_GAP_ACK_BLKS_SZ,
+			      l->addr, tipc_own_addr(l->net), 0, 0, 0);
 	if (!skb)
 		return;
 
@@ -1418,9 +1519,11 @@ static void tipc_link_build_proto_msg(struct tipc_link *l, int mtyp, bool probe,
 		msg_set_bc_gap(hdr, link_bc_rcv_gap(bcl));
 		msg_set_probe(hdr, probe);
 		msg_set_is_keepalive(hdr, probe || probe_reply);
-		tipc_mon_prep(l->net, data, &dlen, mstate, l->bearer_id);
-		msg_set_size(hdr, INT_H_SIZE + dlen);
-		skb_trim(skb, INT_H_SIZE + dlen);
+		if (l->peer_caps & TIPC_GAP_ACK_BLOCK)
+			glen = tipc_build_gap_ack_blks(l, data);
+		tipc_mon_prep(l->net, data + glen, &dlen, mstate, l->bearer_id);
+		msg_set_size(hdr, INT_H_SIZE + glen + dlen);
+		skb_trim(skb, INT_H_SIZE + glen + dlen);
 		l->stats.sent_states++;
 		l->rcv_unacked = 0;
 	} else {
@@ -1590,6 +1693,7 @@ static int tipc_link_proto_rcv(struct tipc_link *l, struct sk_buff *skb,
 			       struct sk_buff_head *xmitq)
 {
 	struct tipc_msg *hdr = buf_msg(skb);
+	struct tipc_gap_ack_blks *ga = NULL;
 	u16 rcvgap = 0;
 	u16 ack = msg_ack(hdr);
 	u16 gap = msg_seq_gap(hdr);
@@ -1600,6 +1704,7 @@ static int tipc_link_proto_rcv(struct tipc_link *l, struct sk_buff *skb,
 	u16 dlen = msg_data_sz(hdr);
 	int mtyp = msg_type(hdr);
 	bool reply = msg_probe(hdr);
+	u16 glen = 0;
 	void *data;
 	char *if_name;
 	int rc = 0;
@@ -1697,7 +1802,17 @@ static int tipc_link_proto_rcv(struct tipc_link *l, struct sk_buff *skb,
 				rc = TIPC_LINK_UP_EVT;
 			break;
 		}
-		tipc_mon_rcv(l->net, data, dlen, l->addr,
+
+		/* Receive Gap ACK blocks from peer if any */
+		if (l->peer_caps & TIPC_GAP_ACK_BLOCK) {
+			ga = (struct tipc_gap_ack_blks *)data;
+			glen = ntohs(ga->len);
+			/* sanity check: if failed, ignore Gap ACK blocks */
+			if (glen != tipc_gap_ack_blks_sz(ga->gack_cnt))
+				ga = NULL;
+		}
+
+		tipc_mon_rcv(l->net, data + glen, dlen - glen, l->addr,
 			     &l->mon_state, l->bearer_id);
 
 		/* Send NACK if peer has sent pkts we haven't received yet */
@@ -1706,13 +1821,12 @@ static int tipc_link_proto_rcv(struct tipc_link *l, struct sk_buff *skb,
 		if (rcvgap || reply)
 			tipc_link_build_proto_msg(l, STATE_MSG, 0, reply,
 						  rcvgap, 0, 0, xmitq);
-		tipc_link_release_pkts(l, ack);
+
+		tipc_link_advance_transmq(l, ack, gap, ga, xmitq);
 
 		/* If NACK, retransmit will now start at right position */
-		if (gap) {
-			rc = tipc_link_retrans(l, l, ack + 1, ack + gap, xmitq);
+		if (gap)
 			l->stats.recv_nacks++;
-		}
 
 		tipc_link_advance_backlog(l, xmitq);
 		if (unlikely(!skb_queue_empty(&l->wakeupq)))
diff --git a/net/tipc/msg.h b/net/tipc/msg.h
index 528ba9241acc..dcb73421c19a 100644
--- a/net/tipc/msg.h
+++ b/net/tipc/msg.h
@@ -117,6 +117,36 @@ struct tipc_msg {
 	__be32 hdr[15];
 };
 
+/* struct tipc_gap_ack - TIPC Gap ACK block
+ * @ack: seqno of the last consecutive packet in link deferdq
+ * @gap: number of gap packets since the last ack
+ *
+ * E.g:
+ *       link deferdq: 1 2 3 4      10 11      13 14 15       20
+ * --> Gap ACK blocks:      <4, 5>,   <11, 1>,      <15, 4>, <20, 0>
+ */
+struct tipc_gap_ack {
+	__be16 ack;
+	__be16 gap;
+};
+
+/* struct tipc_gap_ack_blks
+ * @len: actual length of the record
+ * @gack_cnt: number of Gap ACK blocks in the record
+ * @gacks: array of Gap ACK blocks
+ */
+struct tipc_gap_ack_blks {
+	__be16 len;
+	u8 gack_cnt;
+	struct tipc_gap_ack gacks[];
+};
+
+#define tipc_gap_ack_blks_sz(n) (sizeof(struct tipc_gap_ack_blks) + \
+				 sizeof(struct tipc_gap_ack) * (n))
+
+#define MAX_GAP_ACK_BLKS	32
+#define MAX_GAP_ACK_BLKS_SZ	tipc_gap_ack_blks_sz(MAX_GAP_ACK_BLKS)
+
 static inline struct tipc_msg *buf_msg(struct sk_buff *skb)
 {
 	return (struct tipc_msg *)skb->data;
diff --git a/net/tipc/node.h b/net/tipc/node.h
index 2404225c5d58..c0bf49ea3de4 100644
--- a/net/tipc/node.h
+++ b/net/tipc/node.h
@@ -52,7 +52,8 @@ enum {
 	TIPC_BCAST_RCAST      = (1 << 4),
 	TIPC_NODE_ID128       = (1 << 5),
 	TIPC_LINK_PROTO_SEQNO = (1 << 6),
-	TIPC_MCAST_RBCTL      = (1 << 7)
+	TIPC_MCAST_RBCTL      = (1 << 7),
+	TIPC_GAP_ACK_BLOCK    = (1 << 8)
 };
 
 #define TIPC_NODE_CAPABILITIES (TIPC_SYN_BIT           |  \
@@ -62,7 +63,8 @@ enum {
 				TIPC_BLOCK_FLOWCTL     |   \
 				TIPC_NODE_ID128        |   \
 				TIPC_LINK_PROTO_SEQNO  |   \
-				TIPC_MCAST_RBCTL)
+				TIPC_MCAST_RBCTL       |   \
+				TIPC_GAP_ACK_BLOCK)
 #define INVALID_BEARER_ID -1
 
 void tipc_node_stop(struct net *net);
-- 
2.13.7

[tipc-discussion] [net-next 0/3] tipc: improve unicast link transmission

[Tuong L. <tuo...@de...>]

The series introduces an algorithm to improve TIPC throughput especially
in terms of packet loss, also tries to reduce packet duplication due to
overactive NACK sending mechanism.

The link failover situation is also covered by the patches.

Tuong Lien (3):
  tipc: improve TIPC throughput by Gap ACK blocks
  tipc: reduce duplicate packets for unicast traffic
  tipc: adapt link failover for new Gap-ACK algorithm

 net/tipc/link.c | 266 ++++++++++++++++++++++++++++++++++++++++++++++----------
 net/tipc/msg.h  |  51 +++++++++++
 net/tipc/node.h |   6 +-
 3 files changed, 275 insertions(+), 48 deletions(-)

-- 
2.13.7

[tipc-discussion] Topology subscriptions in bash

[Erik H. <eri...@gm...>]

A small tool to get shell integrations to the TIPC topology server,
maybe someone will find it useful
https://github.com/Hugne/tipc.sh

The subscriber examples use part of the functionality that i plan to
submit to the bash project
for network redirection support, e.g:

[root@kdev ~] exec 5</dev/tipc/1000/1
[root@kdev ~] echo hello world >/dev/tipc/1000/1
[root@kdev ~] cat <&5
hello world
^C
[root@kdev ~]

//E

Re: [tipc-discussion] [net-next v3 3/3] tipc: smooth change between replicast and broadcast

[David M. <da...@da...>]

From: Hoang Le <hoa...@de...>
Date: Tue, 19 Mar 2019 18:49:50 +0700

> Currently, a multicast stream may start out using replicast, because
> there are few destinations, and then it should ideally switch to
> L2/broadcast IGMP/multicast when the number of destinations grows beyond
> a certain limit. The opposite should happen when the number decreases
> below the limit.
> 
> To eliminate the risk of message reordering caused by method change,
> a sending socket must stick to a previously selected method until it
> enters an idle period of 5 seconds. Means there is a 5 seconds pause
> in the traffic from the sender socket.
> 
> If the sender never makes such a pause, the method will never change,
> and transmission may become very inefficient as the cluster grows.
> 
> With this commit, we allow such a switch between replicast and
> broadcast without any need for a traffic pause.
> 
> Solution is to send a dummy message with only the header, also with
> the SYN bit set, via broadcast or replicast. For the data message,
> the SYN bit is set and sending via replicast or broadcast (inverse
> method with dummy).
> 
> Then, at receiving side any messages follow first SYN bit message
> (data or dummy message), they will be held in deferred queue until
> another pair (dummy or data message) arrived in other link.
> 
> v2: reverse christmas tree declaration
> 
> Acked-by: Jon Maloy <jon...@er...>
> Signed-off-by: Hoang Le <hoa...@de...>

Applied.

Re: [tipc-discussion] [net-next v3 2/3] tipc: introduce new capability flag for cluster

[David M. <da...@da...>]

From: Hoang Le <hoa...@de...>
Date: Tue, 19 Mar 2019 18:49:49 +0700

> As a preparation for introducing a smooth switching between replicast
> and broadcast method for multicast message, We have to introduce a new
> capability flag TIPC_MCAST_RBCTL to handle this new feature.
> 
> During a cluster upgrade a node can come back with this new capabilities
> which also must be reflected in the cluster capabilities field.
> The new feature is only applicable if all node in the cluster supports
> this new capability.
> 
> Acked-by: Jon Maloy <jon...@er...>
> Signed-off-by: Hoang Le <hoa...@de...>

Applied.

Re: [tipc-discussion] [net-next v3 1/3] tipc: support broadcast/replicast configurable for bc-link

[David M. <da...@da...>]

From: Hoang Le <hoa...@de...>
Date: Tue, 19 Mar 2019 18:49:48 +0700

> Currently, a multicast stream uses either broadcast or replicast as
> transmission method, based on the ratio between number of actual
> destinations nodes and cluster size.
> 
> However, when an L2 interface (e.g., VXLAN) provides pseudo
> broadcast support, this becomes very inefficient, as it blindly
> replicates multicast packets to all cluster/subnet nodes,
> irrespective of whether they host actual target sockets or not.
> 
> The TIPC multicast algorithm is able to distinguish real destination
> nodes from other nodes, and hence provides a smarter and more
> efficient method for transferring multicast messages than
> pseudo broadcast can do.
> 
> Because of this, we now make it possible for users to force
> the broadcast link to permanently switch to using replicast,
> irrespective of which capabilities the bearer provides,
> or pretend to provide.
> Conversely, we also make it possible to force the broadcast link
> to always use true broadcast. While maybe less useful in
> deployed systems, this may at least be useful for testing the
> broadcast algorithm in small clusters.
> 
> We retain the current AUTOSELECT ability, i.e., to let the broadcast link
> automatically select which algorithm to use, and to switch back and forth
> between broadcast and replicast as the ratio between destination
> node number and cluster size changes. This remains the default method.
> 
> Furthermore, we make it possible to configure the threshold ratio for
> such switches. The default ratio is now set to 10%, down from 25% in the
> earlier implementation.
> 
> Acked-by: Jon Maloy <jon...@er...>
> Signed-off-by: Hoang Le <hoa...@de...>

Applied.

Re: [tipc-discussion] [PATCH net-next] tipc: fix cancellation of topology subscriptions

[Jon M. <jon...@er...>]

No problem. I’ll do that.

///jon


From: Erik Hugne <eri...@gm...>
Sent: 19-Mar-19 12:55
To: Jon Maloy <jon...@er...>
Cc: tip...@li...; Xue, Ying <yin...@wi...>
Subject: Re: [PATCH net-next] tipc: fix cancellation of topology subscriptions

Well, i couldnt find any commit where the bug was introduced.. but i agree that it should probably be in net.
Could you change the subject header and submit?

//E

On Tue, 19 Mar 2019, 11:56 Jon Maloy, <jon...@er...<mailto:jon...@er...>> wrote:
Acked-by: Jon.
Why did you mark it "net-next"? To me this looks like a bug fix and should go to 'net'.
I'll post it for you if you want.

///jon

> -----Original Message-----
> From: eri...@gm...<mailto:eri...@gm...> <eri...@gm...<mailto:eri...@gm...>>
> Sent: 19-Mar-19 11:08
> To: tip...@li...<mailto:tip...@li...>; Jon Maloy
> <jon...@er...<mailto:jon...@er...>>; yin...@wi...<mailto:yin...@wi...>
> Cc: Erik Hugne <eri...@gm...<mailto:eri...@gm...>>
> Subject: [PATCH net-next] tipc: fix cancellation of topology subscriptions
>
> From: Erik Hugne <eri...@gm...<mailto:eri...@gm...>>
>
> When cancelling a subscription, we have to clear the cancel bit in the request
> before iterating over any established subscriptions with memcmp. Otherwise
> no subscription will ever be found, and it will not be possible to unsubscribe.
>
> Signed-off-by: Erik Hugne <eri...@gm...<mailto:eri...@gm...>>
> ---
> Ignore my previous garbage patch, this solve the problem with one line.
> Looking through the history of topsrv.c, i cannot find that this has ever
> worked.
>
>  net/tipc/topsrv.c | 1 +
>  1 file changed, 1 insertion(+)
>
> diff --git a/net/tipc/topsrv.c b/net/tipc/topsrv.c index
> 4a708a4e8583..b05a89a723b1 100644
> --- a/net/tipc/topsrv.c
> +++ b/net/tipc/topsrv.c
> @@ -363,6 +363,7 @@ static int tipc_conn_rcv_sub(struct tipc_topsrv *srv,
>       struct tipc_subscription *sub;
>
>       if (tipc_sub_read(s, filter) & TIPC_SUB_CANCEL) {
> +             s->filter ^= __constant_ntohl(TIPC_SUB_CANCEL);
>               tipc_conn_delete_sub(con, s);
>               return 0;
>       }
> --
> 2.14.1

Re: [tipc-discussion] [PATCH net-next] tipc: fix cancellation of topology subscriptions

[Erik H. <eri...@gm...>]

Well, i couldnt find any commit where the bug was introduced.. but i agree
that it should probably be in net.
Could you change the subject header and submit?

//E

On Tue, 19 Mar 2019, 11:56 Jon Maloy, <jon...@er...> wrote:

> Acked-by: Jon.
> Why did you mark it "net-next"? To me this looks like a bug fix and should
> go to 'net'.
> I'll post it for you if you want.
>
> ///jon
>
> > -----Original Message-----
> > From: eri...@gm... <eri...@gm...>
> > Sent: 19-Mar-19 11:08
> > To: tip...@li...; Jon Maloy
> > <jon...@er...>; yin...@wi...
> > Cc: Erik Hugne <eri...@gm...>
> > Subject: [PATCH net-next] tipc: fix cancellation of topology
> subscriptions
> >
> > From: Erik Hugne <eri...@gm...>
> >
> > When cancelling a subscription, we have to clear the cancel bit in the
> request
> > before iterating over any established subscriptions with memcmp.
> Otherwise
> > no subscription will ever be found, and it will not be possible to
> unsubscribe.
> >
> > Signed-off-by: Erik Hugne <eri...@gm...>
> > ---
> > Ignore my previous garbage patch, this solve the problem with one line.
> > Looking through the history of topsrv.c, i cannot find that this has ever
> > worked.
> >
> >  net/tipc/topsrv.c | 1 +
> >  1 file changed, 1 insertion(+)
> >
> > diff --git a/net/tipc/topsrv.c b/net/tipc/topsrv.c index
> > 4a708a4e8583..b05a89a723b1 100644
> > --- a/net/tipc/topsrv.c
> > +++ b/net/tipc/topsrv.c
> > @@ -363,6 +363,7 @@ static int tipc_conn_rcv_sub(struct tipc_topsrv *srv,
> >       struct tipc_subscription *sub;
> >
> >       if (tipc_sub_read(s, filter) & TIPC_SUB_CANCEL) {
> > +             s->filter ^= __constant_ntohl(TIPC_SUB_CANCEL);
> >               tipc_conn_delete_sub(con, s);
> >               return 0;
> >       }
> > --
> > 2.14.1
>
>

[tipc-discussion] [net-next v3 3/3] tipc: smooth change between replicast and broadcast

[Hoang Le <hoa...@de...>]

Currently, a multicast stream may start out using replicast, because
there are few destinations, and then it should ideally switch to
L2/broadcast IGMP/multicast when the number of destinations grows beyond
a certain limit. The opposite should happen when the number decreases
below the limit.

To eliminate the risk of message reordering caused by method change,
a sending socket must stick to a previously selected method until it
enters an idle period of 5 seconds. Means there is a 5 seconds pause
in the traffic from the sender socket.

If the sender never makes such a pause, the method will never change,
and transmission may become very inefficient as the cluster grows.

With this commit, we allow such a switch between replicast and
broadcast without any need for a traffic pause.

Solution is to send a dummy message with only the header, also with
the SYN bit set, via broadcast or replicast. For the data message,
the SYN bit is set and sending via replicast or broadcast (inverse
method with dummy).

Then, at receiving side any messages follow first SYN bit message
(data or dummy message), they will be held in deferred queue until
another pair (dummy or data message) arrived in other link.

v2: reverse christmas tree declaration

Acked-by: Jon Maloy <jon...@er...>
Signed-off-by: Hoang Le <hoa...@de...>
---
 net/tipc/bcast.c  | 165 +++++++++++++++++++++++++++++++++++++++++++++-
 net/tipc/bcast.h  |   5 ++
 net/tipc/msg.h    |  10 +++
 net/tipc/socket.c |   5 ++
 4 files changed, 184 insertions(+), 1 deletion(-)

diff --git a/net/tipc/bcast.c b/net/tipc/bcast.c
index 12b59268bdd6..5264a8ff6e01 100644
--- a/net/tipc/bcast.c
+++ b/net/tipc/bcast.c
@@ -220,9 +220,24 @@ static void tipc_bcast_select_xmit_method(struct net *net, int dests,
 	}
 	/* Can current method be changed ? */
 	method->expires = jiffies + TIPC_METHOD_EXPIRE;
-	if (method->mandatory || time_before(jiffies, exp))
+	if (method->mandatory)
 		return;
 
+	if (!(tipc_net(net)->capabilities & TIPC_MCAST_RBCTL) &&
+	    time_before(jiffies, exp))
+		return;
+
+	/* Configuration as force 'broadcast' method */
+	if (bb->force_bcast) {
+		method->rcast = false;
+		return;
+	}
+	/* Configuration as force 'replicast' method */
+	if (bb->force_rcast) {
+		method->rcast = true;
+		return;
+	}
+	/* Configuration as 'autoselect' or default method */
 	/* Determine method to use now */
 	method->rcast = dests <= bb->bc_threshold;
 }
@@ -285,6 +300,63 @@ static int tipc_rcast_xmit(struct net *net, struct sk_buff_head *pkts,
 	return 0;
 }
 
+/* tipc_mcast_send_sync - deliver a dummy message with SYN bit
+ * @net: the applicable net namespace
+ * @skb: socket buffer to copy
+ * @method: send method to be used
+ * @dests: destination nodes for message.
+ * @cong_link_cnt: returns number of encountered congested destination links
+ * Returns 0 if success, otherwise errno
+ */
+static int tipc_mcast_send_sync(struct net *net, struct sk_buff *skb,
+				struct tipc_mc_method *method,
+				struct tipc_nlist *dests,
+				u16 *cong_link_cnt)
+{
+	struct tipc_msg *hdr, *_hdr;
+	struct sk_buff_head tmpq;
+	struct sk_buff *_skb;
+
+	/* Is a cluster supporting with new capabilities ? */
+	if (!(tipc_net(net)->capabilities & TIPC_MCAST_RBCTL))
+		return 0;
+
+	hdr = buf_msg(skb);
+	if (msg_user(hdr) == MSG_FRAGMENTER)
+		hdr = msg_get_wrapped(hdr);
+	if (msg_type(hdr) != TIPC_MCAST_MSG)
+		return 0;
+
+	/* Allocate dummy message */
+	_skb = tipc_buf_acquire(MCAST_H_SIZE, GFP_KERNEL);
+	if (!skb)
+		return -ENOMEM;
+
+	/* Preparing for 'synching' header */
+	msg_set_syn(hdr, 1);
+
+	/* Copy skb's header into a dummy header */
+	skb_copy_to_linear_data(_skb, hdr, MCAST_H_SIZE);
+	skb_orphan(_skb);
+
+	/* Reverse method for dummy message */
+	_hdr = buf_msg(_skb);
+	msg_set_size(_hdr, MCAST_H_SIZE);
+	msg_set_is_rcast(_hdr, !msg_is_rcast(hdr));
+
+	skb_queue_head_init(&tmpq);
+	__skb_queue_tail(&tmpq, _skb);
+	if (method->rcast)
+		tipc_bcast_xmit(net, &tmpq, cong_link_cnt);
+	else
+		tipc_rcast_xmit(net, &tmpq, dests, cong_link_cnt);
+
+	/* This queue should normally be empty by now */
+	__skb_queue_purge(&tmpq);
+
+	return 0;
+}
+
 /* tipc_mcast_xmit - deliver message to indicated destination nodes
  *                   and to identified node local sockets
  * @net: the applicable net namespace
@@ -300,6 +372,9 @@ int tipc_mcast_xmit(struct net *net, struct sk_buff_head *pkts,
 		    u16 *cong_link_cnt)
 {
 	struct sk_buff_head inputq, localq;
+	bool rcast = method->rcast;
+	struct tipc_msg *hdr;
+	struct sk_buff *skb;
 	int rc = 0;
 
 	skb_queue_head_init(&inputq);
@@ -313,6 +388,18 @@ int tipc_mcast_xmit(struct net *net, struct sk_buff_head *pkts,
 	/* Send according to determined transmit method */
 	if (dests->remote) {
 		tipc_bcast_select_xmit_method(net, dests->remote, method);
+
+		skb = skb_peek(pkts);
+		hdr = buf_msg(skb);
+		if (msg_user(hdr) == MSG_FRAGMENTER)
+			hdr = msg_get_wrapped(hdr);
+		msg_set_is_rcast(hdr, method->rcast);
+
+		/* Switch method ? */
+		if (rcast != method->rcast)
+			tipc_mcast_send_sync(net, skb, method,
+					     dests, cong_link_cnt);
+
 		if (method->rcast)
 			rc = tipc_rcast_xmit(net, pkts, dests, cong_link_cnt);
 		else
@@ -672,3 +759,79 @@ u32 tipc_bcast_get_broadcast_ratio(struct net *net)
 
 	return bb->rc_ratio;
 }
+
+void tipc_mcast_filter_msg(struct sk_buff_head *defq,
+			   struct sk_buff_head *inputq)
+{
+	struct sk_buff *skb, *_skb, *tmp;
+	struct tipc_msg *hdr, *_hdr;
+	bool match = false;
+	u32 node, port;
+
+	skb = skb_peek(inputq);
+	hdr = buf_msg(skb);
+
+	if (likely(!msg_is_syn(hdr) && skb_queue_empty(defq)))
+		return;
+
+	node = msg_orignode(hdr);
+	port = msg_origport(hdr);
+
+	/* Has the twin SYN message already arrived ? */
+	skb_queue_walk(defq, _skb) {
+		_hdr = buf_msg(_skb);
+		if (msg_orignode(_hdr) != node)
+			continue;
+		if (msg_origport(_hdr) != port)
+			continue;
+		match = true;
+		break;
+	}
+
+	if (!match) {
+		if (!msg_is_syn(hdr))
+			return;
+		__skb_dequeue(inputq);
+		__skb_queue_tail(defq, skb);
+		return;
+	}
+
+	/* Deliver non-SYN message from other link, otherwise queue it */
+	if (!msg_is_syn(hdr)) {
+		if (msg_is_rcast(hdr) != msg_is_rcast(_hdr))
+			return;
+		__skb_dequeue(inputq);
+		__skb_queue_tail(defq, skb);
+		return;
+	}
+
+	/* Queue non-SYN/SYN message from same link */
+	if (msg_is_rcast(hdr) == msg_is_rcast(_hdr)) {
+		__skb_dequeue(inputq);
+		__skb_queue_tail(defq, skb);
+		return;
+	}
+
+	/* Matching SYN messages => return the one with data, if any */
+	__skb_unlink(_skb, defq);
+	if (msg_data_sz(hdr)) {
+		kfree_skb(_skb);
+	} else {
+		__skb_dequeue(inputq);
+		kfree_skb(skb);
+		__skb_queue_tail(inputq, _skb);
+	}
+
+	/* Deliver subsequent non-SYN messages from same peer */
+	skb_queue_walk_safe(defq, _skb, tmp) {
+		_hdr = buf_msg(_skb);
+		if (msg_orignode(_hdr) != node)
+			continue;
+		if (msg_origport(_hdr) != port)
+			continue;
+		if (msg_is_syn(_hdr))
+			break;
+		__skb_unlink(_skb, defq);
+		__skb_queue_tail(inputq, _skb);
+	}
+}
diff --git a/net/tipc/bcast.h b/net/tipc/bcast.h
index 37c55e7347a5..484bde289d3a 100644
--- a/net/tipc/bcast.h
+++ b/net/tipc/bcast.h
@@ -67,11 +67,13 @@ void tipc_nlist_del(struct tipc_nlist *nl, u32 node);
 /* Cookie to be used between socket and broadcast layer
  * @rcast: replicast (instead of broadcast) was used at previous xmit
  * @mandatory: broadcast/replicast indication was set by user
+ * @deferredq: defer queue to make message in order
  * @expires: re-evaluate non-mandatory transmit method if we are past this
  */
 struct tipc_mc_method {
 	bool rcast;
 	bool mandatory;
+	struct sk_buff_head deferredq;
 	unsigned long expires;
 };
 
@@ -99,6 +101,9 @@ int tipc_bclink_reset_stats(struct net *net);
 u32 tipc_bcast_get_broadcast_mode(struct net *net);
 u32 tipc_bcast_get_broadcast_ratio(struct net *net);
 
+void tipc_mcast_filter_msg(struct sk_buff_head *defq,
+			   struct sk_buff_head *inputq);
+
 static inline void tipc_bcast_lock(struct net *net)
 {
 	spin_lock_bh(&tipc_net(net)->bclock);
diff --git a/net/tipc/msg.h b/net/tipc/msg.h
index d7e4b8b93f9d..528ba9241acc 100644
--- a/net/tipc/msg.h
+++ b/net/tipc/msg.h
@@ -257,6 +257,16 @@ static inline void msg_set_src_droppable(struct tipc_msg *m, u32 d)
 	msg_set_bits(m, 0, 18, 1, d);
 }
 
+static inline bool msg_is_rcast(struct tipc_msg *m)
+{
+	return msg_bits(m, 0, 18, 0x1);
+}
+
+static inline void msg_set_is_rcast(struct tipc_msg *m, bool d)
+{
+	msg_set_bits(m, 0, 18, 0x1, d);
+}
+
 static inline void msg_set_size(struct tipc_msg *m, u32 sz)
 {
 	m->hdr[0] = htonl((msg_word(m, 0) & ~0x1ffff) | sz);
diff --git a/net/tipc/socket.c b/net/tipc/socket.c
index e482b342bfa8..b7c292d518e6 100644
--- a/net/tipc/socket.c
+++ b/net/tipc/socket.c
@@ -485,6 +485,7 @@ static int tipc_sk_create(struct net *net, struct socket *sock,
 		tsk_set_unreturnable(tsk, true);
 		if (sock->type == SOCK_DGRAM)
 			tsk_set_unreliable(tsk, true);
+		__skb_queue_head_init(&tsk->mc_method.deferredq);
 	}
 
 	trace_tipc_sk_create(sk, NULL, TIPC_DUMP_NONE, " ");
@@ -582,6 +583,7 @@ static int tipc_release(struct socket *sock)
 	sk->sk_shutdown = SHUTDOWN_MASK;
 	tipc_sk_leave(tsk);
 	tipc_sk_withdraw(tsk, 0, NULL);
+	__skb_queue_purge(&tsk->mc_method.deferredq);
 	sk_stop_timer(sk, &sk->sk_timer);
 	tipc_sk_remove(tsk);
 
@@ -2162,6 +2164,9 @@ static void tipc_sk_filter_rcv(struct sock *sk, struct sk_buff *skb,
 	if (unlikely(grp))
 		tipc_group_filter_msg(grp, &inputq, xmitq);
 
+	if (msg_type(hdr) == TIPC_MCAST_MSG)
+		tipc_mcast_filter_msg(&tsk->mc_method.deferredq, &inputq);
+
 	/* Validate and add to receive buffer if there is space */
 	while ((skb = __skb_dequeue(&inputq))) {
 		hdr = buf_msg(skb);
-- 
2.17.1

[tipc-discussion] [net-next v3 2/3] tipc: introduce new capability flag for cluster

[Hoang Le <hoa...@de...>]

As a preparation for introducing a smooth switching between replicast
and broadcast method for multicast message, We have to introduce a new
capability flag TIPC_MCAST_RBCTL to handle this new feature.

During a cluster upgrade a node can come back with this new capabilities
which also must be reflected in the cluster capabilities field.
The new feature is only applicable if all node in the cluster supports
this new capability.

Acked-by: Jon Maloy <jon...@er...>
Signed-off-by: Hoang Le <hoa...@de...>
---
 net/tipc/core.c |  2 ++
 net/tipc/core.h |  3 +++
 net/tipc/node.c | 18 ++++++++++++++++++
 net/tipc/node.h |  6 ++++--
 4 files changed, 27 insertions(+), 2 deletions(-)

diff --git a/net/tipc/core.c b/net/tipc/core.c
index 5b38f5164281..27cccd101ef6 100644
--- a/net/tipc/core.c
+++ b/net/tipc/core.c
@@ -43,6 +43,7 @@
 #include "net.h"
 #include "socket.h"
 #include "bcast.h"
+#include "node.h"
 
 #include <linux/module.h>
 
@@ -59,6 +60,7 @@ static int __net_init tipc_init_net(struct net *net)
 	tn->node_addr = 0;
 	tn->trial_addr = 0;
 	tn->addr_trial_end = 0;
+	tn->capabilities = TIPC_NODE_CAPABILITIES;
 	memset(tn->node_id, 0, sizeof(tn->node_id));
 	memset(tn->node_id_string, 0, sizeof(tn->node_id_string));
 	tn->mon_threshold = TIPC_DEF_MON_THRESHOLD;
diff --git a/net/tipc/core.h b/net/tipc/core.h
index 8020a6c360ff..7a68e1b6a066 100644
--- a/net/tipc/core.h
+++ b/net/tipc/core.h
@@ -122,6 +122,9 @@ struct tipc_net {
 	/* Topology subscription server */
 	struct tipc_topsrv *topsrv;
 	atomic_t subscription_count;
+
+	/* Cluster capabilities */
+	u16 capabilities;
 };
 
 static inline struct tipc_net *tipc_net(struct net *net)
diff --git a/net/tipc/node.c b/net/tipc/node.c
index 2dc4919ab23c..2717893e9dbe 100644
--- a/net/tipc/node.c
+++ b/net/tipc/node.c
@@ -383,6 +383,11 @@ static struct tipc_node *tipc_node_create(struct net *net, u32 addr,
 				tipc_link_update_caps(l, capabilities);
 		}
 		write_unlock_bh(&n->lock);
+		/* Calculate cluster capabilities */
+		tn->capabilities = TIPC_NODE_CAPABILITIES;
+		list_for_each_entry_rcu(temp_node, &tn->node_list, list) {
+			tn->capabilities &= temp_node->capabilities;
+		}
 		goto exit;
 	}
 	n = kzalloc(sizeof(*n), GFP_ATOMIC);
@@ -433,6 +438,11 @@ static struct tipc_node *tipc_node_create(struct net *net, u32 addr,
 			break;
 	}
 	list_add_tail_rcu(&n->list, &temp_node->list);
+	/* Calculate cluster capabilities */
+	tn->capabilities = TIPC_NODE_CAPABILITIES;
+	list_for_each_entry_rcu(temp_node, &tn->node_list, list) {
+		tn->capabilities &= temp_node->capabilities;
+	}
 	trace_tipc_node_create(n, true, " ");
 exit:
 	spin_unlock_bh(&tn->node_list_lock);
@@ -589,6 +599,7 @@ static void  tipc_node_clear_links(struct tipc_node *node)
  */
 static bool tipc_node_cleanup(struct tipc_node *peer)
 {
+	struct tipc_node *temp_node;
 	struct tipc_net *tn = tipc_net(peer->net);
 	bool deleted = false;
 
@@ -604,6 +615,13 @@ static bool tipc_node_cleanup(struct tipc_node *peer)
 		deleted = true;
 	}
 	tipc_node_write_unlock(peer);
+
+	/* Calculate cluster capabilities */
+	tn->capabilities = TIPC_NODE_CAPABILITIES;
+	list_for_each_entry_rcu(temp_node, &tn->node_list, list) {
+		tn->capabilities &= temp_node->capabilities;
+	}
+
 	spin_unlock_bh(&tn->node_list_lock);
 	return deleted;
 }
diff --git a/net/tipc/node.h b/net/tipc/node.h
index 4f59a30e989a..2404225c5d58 100644
--- a/net/tipc/node.h
+++ b/net/tipc/node.h
@@ -51,7 +51,8 @@ enum {
 	TIPC_BLOCK_FLOWCTL    = (1 << 3),
 	TIPC_BCAST_RCAST      = (1 << 4),
 	TIPC_NODE_ID128       = (1 << 5),
-	TIPC_LINK_PROTO_SEQNO = (1 << 6)
+	TIPC_LINK_PROTO_SEQNO = (1 << 6),
+	TIPC_MCAST_RBCTL      = (1 << 7)
 };
 
 #define TIPC_NODE_CAPABILITIES (TIPC_SYN_BIT           |  \
@@ -60,7 +61,8 @@ enum {
 				TIPC_BCAST_RCAST       |   \
 				TIPC_BLOCK_FLOWCTL     |   \
 				TIPC_NODE_ID128        |   \
-				TIPC_LINK_PROTO_SEQNO)
+				TIPC_LINK_PROTO_SEQNO  |   \
+				TIPC_MCAST_RBCTL)
 #define INVALID_BEARER_ID -1
 
 void tipc_node_stop(struct net *net);
-- 
2.17.1

[tipc-discussion] [net-next v3 1/3] tipc: support broadcast/replicast configurable for bc-link

[Hoang Le <hoa...@de...>]

Currently, a multicast stream uses either broadcast or replicast as
transmission method, based on the ratio between number of actual
destinations nodes and cluster size.

However, when an L2 interface (e.g., VXLAN) provides pseudo
broadcast support, this becomes very inefficient, as it blindly
replicates multicast packets to all cluster/subnet nodes,
irrespective of whether they host actual target sockets or not.

The TIPC multicast algorithm is able to distinguish real destination
nodes from other nodes, and hence provides a smarter and more
efficient method for transferring multicast messages than
pseudo broadcast can do.

Because of this, we now make it possible for users to force
the broadcast link to permanently switch to using replicast,
irrespective of which capabilities the bearer provides,
or pretend to provide.
Conversely, we also make it possible to force the broadcast link
to always use true broadcast. While maybe less useful in
deployed systems, this may at least be useful for testing the
broadcast algorithm in small clusters.

We retain the current AUTOSELECT ability, i.e., to let the broadcast link
automatically select which algorithm to use, and to switch back and forth
between broadcast and replicast as the ratio between destination
node number and cluster size changes. This remains the default method.

Furthermore, we make it possible to configure the threshold ratio for
such switches. The default ratio is now set to 10%, down from 25% in the
earlier implementation.

Acked-by: Jon Maloy <jon...@er...>
Signed-off-by: Hoang Le <hoa...@de...>
---
 include/uapi/linux/tipc_netlink.h |   2 +
 net/tipc/bcast.c                  | 104 ++++++++++++++++++++++++++++--
 net/tipc/bcast.h                  |   7 ++
 net/tipc/link.c                   |   8 +++
 net/tipc/netlink.c                |   4 +-
 5 files changed, 120 insertions(+), 5 deletions(-)

diff --git a/include/uapi/linux/tipc_netlink.h b/include/uapi/linux/tipc_netlink.h
index 0ebe02ef1a86..efb958fd167d 100644
--- a/include/uapi/linux/tipc_netlink.h
+++ b/include/uapi/linux/tipc_netlink.h
@@ -281,6 +281,8 @@ enum {
 	TIPC_NLA_PROP_TOL,		/* u32 */
 	TIPC_NLA_PROP_WIN,		/* u32 */
 	TIPC_NLA_PROP_MTU,		/* u32 */
+	TIPC_NLA_PROP_BROADCAST,	/* u32 */
+	TIPC_NLA_PROP_BROADCAST_RATIO,	/* u32 */
 
 	__TIPC_NLA_PROP_MAX,
 	TIPC_NLA_PROP_MAX = __TIPC_NLA_PROP_MAX - 1
diff --git a/net/tipc/bcast.c b/net/tipc/bcast.c
index d8026543bf4c..12b59268bdd6 100644
--- a/net/tipc/bcast.c
+++ b/net/tipc/bcast.c
@@ -54,7 +54,9 @@ const char tipc_bclink_name[] = "broadcast-link";
  * @dests: array keeping number of reachable destinations per bearer
  * @primary_bearer: a bearer having links to all broadcast destinations, if any
  * @bcast_support: indicates if primary bearer, if any, supports broadcast
+ * @force_bcast: forces broadcast for multicast traffic
  * @rcast_support: indicates if all peer nodes support replicast
+ * @force_rcast: forces replicast for multicast traffic
  * @rc_ratio: dest count as percentage of cluster size where send method changes
  * @bc_threshold: calculated from rc_ratio; if dests > threshold use broadcast
  */
@@ -64,7 +66,9 @@ struct tipc_bc_base {
 	int dests[MAX_BEARERS];
 	int primary_bearer;
 	bool bcast_support;
+	bool force_bcast;
 	bool rcast_support;
+	bool force_rcast;
 	int rc_ratio;
 	int bc_threshold;
 };
@@ -485,10 +489,63 @@ static int tipc_bc_link_set_queue_limits(struct net *net, u32 limit)
 	return 0;
 }
 
+static int tipc_bc_link_set_broadcast_mode(struct net *net, u32 bc_mode)
+{
+	struct tipc_bc_base *bb = tipc_bc_base(net);
+
+	switch (bc_mode) {
+	case BCLINK_MODE_BCAST:
+		if (!bb->bcast_support)
+			return -ENOPROTOOPT;
+
+		bb->force_bcast = true;
+		bb->force_rcast = false;
+		break;
+	case BCLINK_MODE_RCAST:
+		if (!bb->rcast_support)
+			return -ENOPROTOOPT;
+
+		bb->force_bcast = false;
+		bb->force_rcast = true;
+		break;
+	case BCLINK_MODE_SEL:
+		if (!bb->bcast_support || !bb->rcast_support)
+			return -ENOPROTOOPT;
+
+		bb->force_bcast = false;
+		bb->force_rcast = false;
+		break;
+	default:
+		return -EINVAL;
+	}
+
+	return 0;
+}
+
+static int tipc_bc_link_set_broadcast_ratio(struct net *net, u32 bc_ratio)
+{
+	struct tipc_bc_base *bb = tipc_bc_base(net);
+
+	if (!bb->bcast_support || !bb->rcast_support)
+		return -ENOPROTOOPT;
+
+	if (bc_ratio > 100 || bc_ratio <= 0)
+		return -EINVAL;
+
+	bb->rc_ratio = bc_ratio;
+	tipc_bcast_lock(net);
+	tipc_bcbase_calc_bc_threshold(net);
+	tipc_bcast_unlock(net);
+
+	return 0;
+}
+
 int tipc_nl_bc_link_set(struct net *net, struct nlattr *attrs[])
 {
 	int err;
 	u32 win;
+	u32 bc_mode;
+	u32 bc_ratio;
 	struct nlattr *props[TIPC_NLA_PROP_MAX + 1];
 
 	if (!attrs[TIPC_NLA_LINK_PROP])
@@ -498,12 +555,28 @@ int tipc_nl_bc_link_set(struct net *net, struct nlattr *attrs[])
 	if (err)
 		return err;
 
-	if (!props[TIPC_NLA_PROP_WIN])
+	if (!props[TIPC_NLA_PROP_WIN] &&
+	    !props[TIPC_NLA_PROP_BROADCAST] &&
+	    !props[TIPC_NLA_PROP_BROADCAST_RATIO]) {
 		return -EOPNOTSUPP;
+	}
+
+	if (props[TIPC_NLA_PROP_BROADCAST]) {
+		bc_mode = nla_get_u32(props[TIPC_NLA_PROP_BROADCAST]);
+		err = tipc_bc_link_set_broadcast_mode(net, bc_mode);
+	}
 
-	win = nla_get_u32(props[TIPC_NLA_PROP_WIN]);
+	if (!err && props[TIPC_NLA_PROP_BROADCAST_RATIO]) {
+		bc_ratio = nla_get_u32(props[TIPC_NLA_PROP_BROADCAST_RATIO]);
+		err = tipc_bc_link_set_broadcast_ratio(net, bc_ratio);
+	}
 
-	return tipc_bc_link_set_queue_limits(net, win);
+	if (!err && props[TIPC_NLA_PROP_WIN]) {
+		win = nla_get_u32(props[TIPC_NLA_PROP_WIN]);
+		err = tipc_bc_link_set_queue_limits(net, win);
+	}
+
+	return err;
 }
 
 int tipc_bcast_init(struct net *net)
@@ -529,7 +602,7 @@ int tipc_bcast_init(struct net *net)
 		goto enomem;
 	bb->link = l;
 	tn->bcl = l;
-	bb->rc_ratio = 25;
+	bb->rc_ratio = 10;
 	bb->rcast_support = true;
 	return 0;
 enomem:
@@ -576,3 +649,26 @@ void tipc_nlist_purge(struct tipc_nlist *nl)
 	nl->remote = 0;
 	nl->local = false;
 }
+
+u32 tipc_bcast_get_broadcast_mode(struct net *net)
+{
+	struct tipc_bc_base *bb = tipc_bc_base(net);
+
+	if (bb->force_bcast)
+		return BCLINK_MODE_BCAST;
+
+	if (bb->force_rcast)
+		return BCLINK_MODE_RCAST;
+
+	if (bb->bcast_support && bb->rcast_support)
+		return BCLINK_MODE_SEL;
+
+	return 0;
+}
+
+u32 tipc_bcast_get_broadcast_ratio(struct net *net)
+{
+	struct tipc_bc_base *bb = tipc_bc_base(net);
+
+	return bb->rc_ratio;
+}
diff --git a/net/tipc/bcast.h b/net/tipc/bcast.h
index 751530ab0c49..37c55e7347a5 100644
--- a/net/tipc/bcast.h
+++ b/net/tipc/bcast.h
@@ -48,6 +48,10 @@ extern const char tipc_bclink_name[];
 
 #define TIPC_METHOD_EXPIRE msecs_to_jiffies(5000)
 
+#define BCLINK_MODE_BCAST  0x1
+#define BCLINK_MODE_RCAST  0x2
+#define BCLINK_MODE_SEL    0x4
+
 struct tipc_nlist {
 	struct list_head list;
 	u32 self;
@@ -92,6 +96,9 @@ int tipc_nl_add_bc_link(struct net *net, struct tipc_nl_msg *msg);
 int tipc_nl_bc_link_set(struct net *net, struct nlattr *attrs[]);
 int tipc_bclink_reset_stats(struct net *net);
 
+u32 tipc_bcast_get_broadcast_mode(struct net *net);
+u32 tipc_bcast_get_broadcast_ratio(struct net *net);
+
 static inline void tipc_bcast_lock(struct net *net)
 {
 	spin_lock_bh(&tipc_net(net)->bclock);
diff --git a/net/tipc/link.c b/net/tipc/link.c
index 341ecd796aa4..52d23b3ffaf5 100644
--- a/net/tipc/link.c
+++ b/net/tipc/link.c
@@ -2197,6 +2197,8 @@ int tipc_nl_add_bc_link(struct net *net, struct tipc_nl_msg *msg)
 	struct nlattr *attrs;
 	struct nlattr *prop;
 	struct tipc_net *tn = net_generic(net, tipc_net_id);
+	u32 bc_mode = tipc_bcast_get_broadcast_mode(net);
+	u32 bc_ratio = tipc_bcast_get_broadcast_ratio(net);
 	struct tipc_link *bcl = tn->bcl;
 
 	if (!bcl)
@@ -2233,6 +2235,12 @@ int tipc_nl_add_bc_link(struct net *net, struct tipc_nl_msg *msg)
 		goto attr_msg_full;
 	if (nla_put_u32(msg->skb, TIPC_NLA_PROP_WIN, bcl->window))
 		goto prop_msg_full;
+	if (nla_put_u32(msg->skb, TIPC_NLA_PROP_BROADCAST, bc_mode))
+		goto prop_msg_full;
+	if (bc_mode & BCLINK_MODE_SEL)
+		if (nla_put_u32(msg->skb, TIPC_NLA_PROP_BROADCAST_RATIO,
+				bc_ratio))
+			goto prop_msg_full;
 	nla_nest_end(msg->skb, prop);
 
 	err = __tipc_nl_add_bc_link_stat(msg->skb, &bcl->stats);
diff --git a/net/tipc/netlink.c b/net/tipc/netlink.c
index 99ee419210ba..5240f64e8ccc 100644
--- a/net/tipc/netlink.c
+++ b/net/tipc/netlink.c
@@ -110,7 +110,9 @@ const struct nla_policy tipc_nl_prop_policy[TIPC_NLA_PROP_MAX + 1] = {
 	[TIPC_NLA_PROP_UNSPEC]		= { .type = NLA_UNSPEC },
 	[TIPC_NLA_PROP_PRIO]		= { .type = NLA_U32 },
 	[TIPC_NLA_PROP_TOL]		= { .type = NLA_U32 },
-	[TIPC_NLA_PROP_WIN]		= { .type = NLA_U32 }
+	[TIPC_NLA_PROP_WIN]		= { .type = NLA_U32 },
+	[TIPC_NLA_PROP_BROADCAST]	= { .type = NLA_U32 },
+	[TIPC_NLA_PROP_BROADCAST_RATIO]	= { .type = NLA_U32 }
 };
 
 const struct nla_policy tipc_nl_bearer_policy[TIPC_NLA_BEARER_MAX + 1]	= {
-- 
2.17.1

[tipc-discussion] [iproute2-next 1/2] tipc: add link broadcast set method and ratio

[Hoang Le <hoa...@de...>]

The command added here makes it possible to forcibly configure the
broadcast link to use either broadcast or replicast, in addition to
the already existing auto selection algorithm.

A sample usage is shown below:
$tipc link set broadcast BROADCAST
$tipc link set broadcast AUTOSELECT ratio 25

$tipc link set broadcast -h
Usage: tipc link set broadcast PROPERTY

PROPERTIES
 BROADCAST         - Forces all multicast traffic to be
                     transmitted via broadcast only,
                     irrespective of cluster size and number
                     of destinations

 REPLICAST         - Forces all multicast traffic to be
                     transmitted via replicast only,
                     irrespective of cluster size and number
                     of destinations

 AUTOSELECT        - Auto switching to broadcast or replicast
                     depending on cluster size and destination
                     node number

 ratio SIZE        - Set the AUTOSELECT criteria, percentage of
                     destination nodes vs cluster size

Acked-by: Jon Maloy <jon...@er...>
Signed-off-by: Hoang Le <hoa...@de...>
---
 include/uapi/linux/tipc_netlink.h |  2 +
 tipc/link.c                       | 96 ++++++++++++++++++++++++++++++-
 2 files changed, 97 insertions(+), 1 deletion(-)

diff --git a/include/uapi/linux/tipc_netlink.h b/include/uapi/linux/tipc_netlink.h
index 0ebe02ef1a86..efb958fd167d 100644
--- a/include/uapi/linux/tipc_netlink.h
+++ b/include/uapi/linux/tipc_netlink.h
@@ -281,6 +281,8 @@ enum {
 	TIPC_NLA_PROP_TOL,		/* u32 */
 	TIPC_NLA_PROP_WIN,		/* u32 */
 	TIPC_NLA_PROP_MTU,		/* u32 */
+	TIPC_NLA_PROP_BROADCAST,	/* u32 */
+	TIPC_NLA_PROP_BROADCAST_RATIO,	/* u32 */
 
 	__TIPC_NLA_PROP_MAX,
 	TIPC_NLA_PROP_MAX = __TIPC_NLA_PROP_MAX - 1
diff --git a/tipc/link.c b/tipc/link.c
index 43e26da3fa6b..e3b10bb7b3d4 100644
--- a/tipc/link.c
+++ b/tipc/link.c
@@ -28,6 +28,9 @@
 #define PRIORITY_STR "priority"
 #define TOLERANCE_STR "tolerance"
 #define WINDOW_STR "window"
+#define BROADCAST_STR "broadcast"
+
+static const char tipc_bclink_name[] = "broadcast-link";
 
 static int link_list_cb(const struct nlmsghdr *nlh, void *data)
 {
@@ -521,7 +524,8 @@ static void cmd_link_set_help(struct cmdl *cmdl)
 		"PROPERTIES\n"
 		" tolerance TOLERANCE   - Set link tolerance\n"
 		" priority PRIORITY     - Set link priority\n"
-		" window WINDOW         - Set link window\n",
+		" window WINDOW         - Set link window\n"
+		" broadcast BROADCAST   - Set link broadcast\n",
 		cmdl->argv[0]);
 }
 
@@ -585,6 +589,95 @@ static int cmd_link_set_prop(struct nlmsghdr *nlh, const struct cmd *cmd,
 	return msg_doit(nlh, link_get_cb, &prop);
 }
 
+static void cmd_link_set_bcast_help(struct cmdl *cmdl)
+{
+	fprintf(stderr, "Usage: %s link set broadcast PROPERTY\n\n"
+		"PROPERTIES\n"
+		" BROADCAST         - Forces all multicast traffic to be\n"
+		"                     transmitted via broadcast only,\n"
+		"                     irrespective of cluster size and number\n"
+		"                     of destinations\n\n"
+		" REPLICAST         - Forces all multicast traffic to be\n"
+		"                     transmitted via replicast only,\n"
+		"                     irrespective of cluster size and number\n"
+		"                     of destinations\n\n"
+		" AUTOSELECT        - Auto switching to broadcast or replicast\n"
+		"                     depending on cluster size and destination\n"
+		"                     node number\n\n"
+		" ratio SIZE        - Set the AUTOSELECT criteria, percentage of\n"
+		"                     destination nodes vs cluster size\n\n",
+		cmdl->argv[0]);
+}
+
+static int cmd_link_set_bcast(struct nlmsghdr *nlh, const struct cmd *cmd,
+			     struct cmdl *cmdl, void *data)
+{
+	char buf[MNL_SOCKET_BUFFER_SIZE];
+	struct nlattr *props;
+	struct nlattr *attrs;
+	struct opt *opt;
+	struct opt opts[] = {
+		{ "BROADCAST",	OPT_KEY, NULL },
+		{ "REPLICAST",	OPT_KEY, NULL },
+		{ "AUTOSELECT",	OPT_KEY, NULL },
+		{ "ratio",	OPT_KEYVAL,	NULL },
+		{ NULL }
+	};
+	int method = 0;
+
+	if (help_flag) {
+		(cmd->help)(cmdl);
+		return -EINVAL;
+	}
+
+	if (parse_opts(opts, cmdl) < 0)
+		return -EINVAL;
+
+	for (opt = opts; opt->key; opt++)
+		if (opt->val)
+			break;
+
+	if (!opt || !opt->key) {
+		(cmd->help)(cmdl);
+		return -EINVAL;
+	}
+
+	nlh = msg_init(buf, TIPC_NL_LINK_SET);
+	if (!nlh) {
+		fprintf(stderr, "error, message initialisation failed\n");
+		return -1;
+	}
+
+	attrs = mnl_attr_nest_start(nlh, TIPC_NLA_LINK);
+	/* Direct to broadcast-link setting */
+	mnl_attr_put_strz(nlh, TIPC_NLA_LINK_NAME, tipc_bclink_name);
+	props = mnl_attr_nest_start(nlh, TIPC_NLA_LINK_PROP);
+
+	if (get_opt(opts, "BROADCAST"))
+		method = 0x1;
+	else if (get_opt(opts, "REPLICAST"))
+		method = 0x2;
+	else if (get_opt(opts, "AUTOSELECT"))
+		method = 0x4;
+
+	opt = get_opt(opts, "ratio");
+	if (!method && !opt) {
+		(cmd->help)(cmdl);
+		return -EINVAL;
+	}
+
+	if (method)
+		mnl_attr_put_u32(nlh, TIPC_NLA_PROP_BROADCAST, method);
+
+	if (opt)
+		mnl_attr_put_u32(nlh, TIPC_NLA_PROP_BROADCAST_RATIO,
+				 atoi(opt->val));
+
+	mnl_attr_nest_end(nlh, props);
+	mnl_attr_nest_end(nlh, attrs);
+	return msg_doit(nlh, NULL, NULL);
+}
+
 static int cmd_link_set(struct nlmsghdr *nlh, const struct cmd *cmd,
 			struct cmdl *cmdl, void *data)
 {
@@ -592,6 +685,7 @@ static int cmd_link_set(struct nlmsghdr *nlh, const struct cmd *cmd,
 		{ PRIORITY_STR,	cmd_link_set_prop,	cmd_link_set_help },
 		{ TOLERANCE_STR,	cmd_link_set_prop,	cmd_link_set_help },
 		{ WINDOW_STR,	cmd_link_set_prop,	cmd_link_set_help },
+		{ BROADCAST_STR, cmd_link_set_bcast, cmd_link_set_bcast_help },
 		{ NULL }
 	};
 
-- 
2.17.1

[tipc-discussion] [iproute2-next 2/2] tipc: add link broadcast get

[Hoang Le <hoa...@de...>]

The command prints the actually method that multicast
is running in the system.
Also 'ratio' value for AUTOSELECT method.

A sample usage is shown below:
$tipc link get broadcast
BROADCAST

$tipc link get broadcast
AUTOSELECT ratio:30%

$tipc link get broadcast -j -p
[ {
        "method": "AUTOSELECT"
    },{
        "ratio": 30
    } ]

Acked-by: Jon Maloy <jon...@er...>
Signed-off-by: Hoang Le <hoa...@de...>
---
 tipc/link.c | 85 ++++++++++++++++++++++++++++++++++++++++++++++++++++-
 1 file changed, 84 insertions(+), 1 deletion(-)

diff --git a/tipc/link.c b/tipc/link.c
index e3b10bb7b3d4..e123c1863575 100644
--- a/tipc/link.c
+++ b/tipc/link.c
@@ -175,10 +175,92 @@ static void cmd_link_get_help(struct cmdl *cmdl)
 		"PROPERTIES\n"
 		" tolerance             - Get link tolerance\n"
 		" priority              - Get link priority\n"
-		" window                - Get link window\n",
+		" window                - Get link window\n"
+		" broadcast             - Get link broadcast\n",
 		cmdl->argv[0]);
 }
 
+static int cmd_link_get_bcast_cb(const struct nlmsghdr *nlh, void *data)
+{
+	int *prop = data;
+	int prop_ratio = TIPC_NLA_PROP_BROADCAST_RATIO;
+	struct genlmsghdr *genl = mnl_nlmsg_get_payload(nlh);
+	struct nlattr *info[TIPC_NLA_MAX + 1] = {};
+	struct nlattr *attrs[TIPC_NLA_LINK_MAX + 1] = {};
+	struct nlattr *props[TIPC_NLA_PROP_MAX + 1] = {};
+	int bc_mode;
+
+	mnl_attr_parse(nlh, sizeof(*genl), parse_attrs, info);
+	if (!info[TIPC_NLA_LINK])
+		return MNL_CB_ERROR;
+
+	mnl_attr_parse_nested(info[TIPC_NLA_LINK], parse_attrs, attrs);
+	if (!attrs[TIPC_NLA_LINK_PROP])
+		return MNL_CB_ERROR;
+
+	mnl_attr_parse_nested(attrs[TIPC_NLA_LINK_PROP], parse_attrs, props);
+	if (!props[*prop])
+		return MNL_CB_ERROR;
+
+	bc_mode = mnl_attr_get_u32(props[*prop]);
+
+	new_json_obj(json);
+	open_json_object(NULL);
+	switch (bc_mode) {
+	case 0x1:
+		print_string(PRINT_ANY, "method", "%s\n", "BROADCAST");
+		break;
+	case 0x2:
+		print_string(PRINT_ANY, "method", "%s\n", "REPLICAST");
+		break;
+	case 0x4:
+		print_string(PRINT_ANY, "method", "%s", "AUTOSELECT");
+		close_json_object();
+		open_json_object(NULL);
+		print_uint(PRINT_ANY, "ratio", " ratio:%u%\n",
+			   mnl_attr_get_u32(props[prop_ratio]));
+		break;
+	default:
+		print_string(PRINT_ANY, NULL, "UNKNOWN\n", NULL);
+		break;
+	}
+	close_json_object();
+	delete_json_obj();
+	return MNL_CB_OK;
+}
+
+static void cmd_link_get_bcast_help(struct cmdl *cmdl)
+{
+	fprintf(stderr, "Usage: %s link get PPROPERTY\n\n"
+		"PROPERTIES\n"
+		" broadcast             - Get link broadcast\n",
+		cmdl->argv[0]);
+}
+
+static int cmd_link_get_bcast(struct nlmsghdr *nlh, const struct cmd *cmd,
+			     struct cmdl *cmdl, void *data)
+{
+	int prop = TIPC_NLA_PROP_BROADCAST;
+	char buf[MNL_SOCKET_BUFFER_SIZE];
+	struct nlattr *attrs;
+
+	if (help_flag) {
+		(cmd->help)(cmdl);
+		return -EINVAL;
+	}
+
+	nlh = msg_init(buf, TIPC_NL_LINK_GET);
+	if (!nlh) {
+		fprintf(stderr, "error, message initialisation failed\n");
+		return -1;
+	}
+	attrs = mnl_attr_nest_start(nlh, TIPC_NLA_LINK);
+	/* Direct to broadcast-link setting */
+	mnl_attr_put_strz(nlh, TIPC_NLA_LINK_NAME, tipc_bclink_name);
+	mnl_attr_nest_end(nlh, attrs);
+	return msg_doit(nlh, cmd_link_get_bcast_cb, &prop);
+}
+
 static int cmd_link_get(struct nlmsghdr *nlh, const struct cmd *cmd,
 			struct cmdl *cmdl, void *data)
 {
@@ -186,6 +268,7 @@ static int cmd_link_get(struct nlmsghdr *nlh, const struct cmd *cmd,
 		{ PRIORITY_STR,	cmd_link_get_prop,	cmd_link_get_help },
 		{ TOLERANCE_STR,	cmd_link_get_prop,	cmd_link_get_help },
 		{ WINDOW_STR,	cmd_link_get_prop,	cmd_link_get_help },
+		{ BROADCAST_STR, cmd_link_get_bcast, cmd_link_get_bcast_help },
 		{ NULL }
 	};
 
-- 
2.17.1

Re: [tipc-discussion] [PATCH net-next] tipc: fix cancellation of topology subscriptions

[Jon M. <jon...@er...>]

Acked-by: Jon.
Why did you mark it "net-next"? To me this looks like a bug fix and should go to 'net'.
I'll post it for you if you want.

///jon

> -----Original Message-----
> From: eri...@gm... <eri...@gm...>
> Sent: 19-Mar-19 11:08
> To: tip...@li...; Jon Maloy
> <jon...@er...>; yin...@wi...
> Cc: Erik Hugne <eri...@gm...>
> Subject: [PATCH net-next] tipc: fix cancellation of topology subscriptions
> 
> From: Erik Hugne <eri...@gm...>
> 
> When cancelling a subscription, we have to clear the cancel bit in the request
> before iterating over any established subscriptions with memcmp. Otherwise
> no subscription will ever be found, and it will not be possible to unsubscribe.
> 
> Signed-off-by: Erik Hugne <eri...@gm...>
> ---
> Ignore my previous garbage patch, this solve the problem with one line.
> Looking through the history of topsrv.c, i cannot find that this has ever
> worked.
> 
>  net/tipc/topsrv.c | 1 +
>  1 file changed, 1 insertion(+)
> 
> diff --git a/net/tipc/topsrv.c b/net/tipc/topsrv.c index
> 4a708a4e8583..b05a89a723b1 100644
> --- a/net/tipc/topsrv.c
> +++ b/net/tipc/topsrv.c
> @@ -363,6 +363,7 @@ static int tipc_conn_rcv_sub(struct tipc_topsrv *srv,
>  	struct tipc_subscription *sub;
> 
>  	if (tipc_sub_read(s, filter) & TIPC_SUB_CANCEL) {
> +		s->filter ^= __constant_ntohl(TIPC_SUB_CANCEL);
>  		tipc_conn_delete_sub(con, s);
>  		return 0;
>  	}
> --
> 2.14.1