RE: [Tipc-discussion] patch for socket.c

[Guo, M. <mi...@in...>]

Cool catch!
I have not test the code on the SMP, anyway, go on your excellent work!

Thanks
Guo Min
tip...@li... wrote:
> I found a race using tipc_queue_size.  Multiple processors
> can be accessing the variable at the same time causing it to
> become corrupted.
> I was seeing places where the tipc_queue_size was already
> zero but being decremented.  This caused all kinds of strange
> things to happen. Anyway, I converted tipc_queue_size to an
> atomic_t to protect access to the variable.  I can now run
> the tipc benchmark client/server test without hanging the
> network and/or computer.
>=20
> Index: socket.c
> =
=3D=3D=3D=3D=3D=3D=3D=3D=3D=3D=3D=3D=3D=3D=3D=3D=3D=3D=3D=3D=3D=3D=3D=3D=3D=
=3D=3D=3D=3D=3D=3D=3D=3D=3D=3D=3D=3D=3D=3D=3D=3D=3D=3D=3D=3D=3D=3D=3D=3D=3D=
=3D=3D=3D=3D=3D=3D=3D=3D=3D=3D=3D=3D=3D=3D=3D=3D=3D
> RCS file: /cvsroot/tipc/source/unstable/net/tipc/linux-2.6/socket.c,v
> retrieving revision 1.3 diff -u -r1.3 socket.c
> --- socket.c	15 Apr 2004 18:06:37 -0000	1.3
> +++ socket.c	19 Apr 2004 21:50:19 -0000
> @@ -99,6 +99,7 @@
>  #include <linux/version.h>
>  #include <asm/semaphore.h>
>  #include <asm/string.h>
> +#include <asm/atomic.h>
>  #include <linux/fcntl.h>
>  #include "tipc_adapt.h"
>  #include <tipc_msg.h>
> @@ -133,7 +134,7 @@
>          int  buf_rest;
>  };
>=20
> -static uint tipc_queue_size =3D 0;
> +static atomic_t tipc_queue_size =3D ATOMIC_INIT(0);
>  extern wait_queue_head_t tipc_signal_wq;  static uint
> dispatch(struct tipc_port*,struct sk_buff *buf);  static void
> wakeupdispatch(struct tipc_port *port); @@ -193,8 +194,8 @@
>          pmask |=3D pollmask(tsock->queue_head);
>          tsock->pollmask =3D pmask;
>  	tsock->queue_size--;
> -	tipc_queue_size--;
>          spin_unlock_bh(tsock->p->lock);
> +	atomic_dec(&tipc_queue_size);
>  	if (unlikely(pmask & POLLHUP))
>  		tipc_disconnect(tsock->p->ref);
>  }
> @@ -284,7 +285,7 @@
>  		struct sk_buff *next =3D buf_next(buf);
>  		tipc_reject_msg(buf, TIPC_ERR_NO_PORT);
>  		buf =3D next;
> -		tipc_queue_size--;
> +		atomic_dec(&tipc_queue_size);
>  	}
>=20
>  	while (tsock->ev_head) {
> @@ -371,7 +372,7 @@
>                  /* Remove for next poll/read */
>  		tsock->pollmask &=3D ~MSG_ERROR;
>                  /* Empty error msg?  */
> -		if (!(pmask & TIPC_MSG_PENDING))
> +		if (!(pmask & TIPC_MSG_PENDING))
>                          advance_queue(tsock);
>  	}
>  	return pmask;
> @@ -760,14 +761,16 @@
>  	uint pmask =3D 0;
>          uint res =3D TIPC_OK;;
>=20
> -	if (unlikely(tipc_queue_size > OVERLOAD_LIMIT_BASE)) {
> -		if (overload(tipc_queue_size,
> OVERLOAD_LIMIT_BASE, msg)){
> +	if (unlikely((uint)atomic_read(&tipc_queue_size) >
> +						OVERLOAD_LIMIT_BASE)) {
> +		if (overload(atomic_read(&tipc_queue_size),
> +					OVERLOAD_LIMIT_BASE, msg)){
>  			res =3D TIPC_ERR_OVERLOAD;
>                          goto error;
>                  }
>          }
>  	if (unlikely(tsock->queue_size > OVERLOAD_LIMIT_BASE / 2)) {
> -		if (overload(tsock->queue_size,
> OVERLOAD_LIMIT_BASE / 2, msg)){
> +		if (overload(tsock->queue_size,
> OVERLOAD_LIMIT_BASE / 2, msg)) {
>  			res =3D TIPC_ERR_OVERLOAD;
>                          goto error;
>                  }
> @@ -779,7 +782,7 @@
>                  }
>          }
>  	tsock->queue_size +=3D 1;
> -	tipc_queue_size +=3D 1;
> +	atomic_inc(&tipc_queue_size);
>  	msg_dbg(msg,"<DISP<: ");
>  	buf_set_next(buf, 0);
>  	if (tsock->queue_head) {