[PATCH 0/5] RFC: Multicast and filtering features on AF_UNIX

[PATCH 0/5] RFC: Multicast and filtering features on AF_UNIX

[Alban Crequy]

Hi,

I am working on improving the performances of D-Bus [1]. dbus-daemon is
connected to the peers with Unix sockets and delivers D-Bus messages
according to match rules set by the peers themselves. Every D-Bus
message goes through dbus-daemon. The number of recipients if often 1
but it can be 0, 1, or more. With this architecture, dbus-daemon has to
wake up for every message on the bus and the context switches are
expensive.

/--------\ unix socket  /-------------\ unix socket  /-----------\
| sender |------------->| dbus-daemon |------------->| recipient |
\--------/  AF_UNIX     \-------------/  AF_UNIX     \-----------/

Ian and I wrote a kernel module "dbus" to improve the performances (the
sources are not attached to this email but available there: [2]). This
is a proof-of-concept. It implements a new address family AF_DBUS,
similar to AF_UNIX, but the kernel is smart enough to deliver the
messages directly to the correct recipients, bypassing dbus-daemon. The
only change in userspace is to use AF_DBUS instead of AF_UNIX. I
compared the performances here [2]. It is about 5000 lines of code (a
big part is duplicated from net/unix/af_unix.c).

Another possibility is to add the needed features directly in AF_UNIX
sockets (and avoid to create a new address family for D-Bus):

- multicast
- some kind of BSD socket filters?
- untables?

And hopefully the new features would be useful for other userspace
applications in addition to D-Bus. Wireshark could read and display
AF_UNIX packets.

I am looking for feedback on what is the best way to do this.

The following patches implements multicast on stream Unix socket.
Applications have to enable multicast on the socket with setsockopt()
and then every packets will be delivered to all sockets connected to
the multicast address. The patches are also available on [3].

[1] http://www.freedesktop.org/wiki/Software/dbus
[2] http://alban.apinc.org/blog/2010/09/15/d-bus-in-the-kernel-faster/
[3]
http://git.collabora.co.uk/?p=user/alban/linux-2.6.35.y/.git;a=shortlog;h=refs/heads/unix-multicast

-- 
Alban Crequy

[PATCH 1/5] AF_UNIX: add getsockopt and setsockopt on stream sockets

[Alban Crequy]

Signed-off-by: Alban Crequy <alban.crequy@collabora.co.uk>
---
 net/unix/af_unix.c |   21 +++++++++++++++++++--
 1 files changed, 19 insertions(+), 2 deletions(-)

diff --git a/net/unix/af_unix.c b/net/unix/af_unix.c
index fef2cc5..47d9f77 100644
--- a/net/unix/af_unix.c
+++ b/net/unix/af_unix.c
@@ -492,6 +492,10 @@ static unsigned int unix_dgram_poll(struct file *, struct socket *,
 				    poll_table *);
 static int unix_ioctl(struct socket *, unsigned int, unsigned long);
 static int unix_shutdown(struct socket *, int);
+static int unix_stream_setsockopt(struct socket *, int, int,
+				  char __user *, unsigned int);
+static int unix_stream_getsockopt(struct socket *, int, int,
+				  char __user *, int __user *);
 static int unix_stream_sendmsg(struct kiocb *, struct socket *,
 			       struct msghdr *, size_t);
 static int unix_stream_recvmsg(struct kiocb *, struct socket *,
@@ -518,8 +522,8 @@ static const struct proto_ops unix_stream_ops = {
 	.ioctl =	unix_ioctl,
 	.listen =	unix_listen,
 	.shutdown =	unix_shutdown,
-	.setsockopt =	sock_no_setsockopt,
-	.getsockopt =	sock_no_getsockopt,
+	.setsockopt =	unix_stream_setsockopt,
+	.getsockopt =	unix_stream_getsockopt,
 	.sendmsg =	unix_stream_sendmsg,
 	.recvmsg =	unix_stream_recvmsg,
 	.mmap =		sock_no_mmap,
@@ -1494,6 +1498,19 @@ out:
 }
 
 
+static int unix_stream_setsockopt(struct socket *sock, int level, int optname,
+				  char __user *optval, unsigned int optlen)
+{
+	return -EOPNOTSUPP;
+}
+
+int unix_stream_getsockopt(struct socket *sock, int level, int optname,
+			   char __user *optval, int __user *optlen)
+{
+	return -EOPNOTSUPP;
+}
+
+
 static int unix_stream_sendmsg(struct kiocb *kiocb, struct socket *sock,
 			       struct msghdr *msg, size_t len)
 {
-- 
1.7.1

[PATCH 2/5] AF_UNIX: enable/disable multicast with getsockopt/setsockopt

[Alban Crequy]

Multicast can be enabled or disabled after a socket is allocated but this
cannot be changed once the socket is bound or connected.

Userspace applications can enable multicast on an Unix stream socket:
  sockfd = socket(AF_UNIX, SOCK_STREAM, 0);
  #define UNIX_MULTICAST 1
  val = 1;
  len = sizeof(val);
  ret = setsockopt(sockfd, 0, UNIX_MULTICAST, &val, len);

Signed-off-by: Alban Crequy <alban.crequy@collabora.co.uk>
---
 include/net/af_unix.h |    4 +++
 net/unix/af_unix.c    |   57 +++++++++++++++++++++++++++++++++++++++++++++++-
 2 files changed, 59 insertions(+), 2 deletions(-)

diff --git a/include/net/af_unix.h b/include/net/af_unix.h
index 20725e2..4c77c69 100644
--- a/include/net/af_unix.h
+++ b/include/net/af_unix.h
@@ -40,6 +40,9 @@ struct unix_skb_parms {
 				spin_lock_nested(&unix_sk(s)->lock, \
 				SINGLE_DEPTH_NESTING)
 
+/* UNIX socket options */
+#define UNIX_MULTICAST	1	/* Enable multicast Unix sockets */
+
 #ifdef __KERNEL__
 /* The AF_UNIX socket */
 struct unix_sock {
@@ -56,6 +59,7 @@ struct unix_sock {
 	spinlock_t		lock;
 	unsigned int		gc_candidate : 1;
 	unsigned int		gc_maybe_cycle : 1;
+	unsigned int		multicast : 1;
 	struct socket_wq	peer_wq;
 };
 #define unix_sk(__sk) ((struct unix_sock *)__sk)
diff --git a/net/unix/af_unix.c b/net/unix/af_unix.c
index 47d9f77..c766e88 100644
--- a/net/unix/af_unix.c
+++ b/net/unix/af_unix.c
@@ -1501,13 +1501,66 @@ out:
 static int unix_stream_setsockopt(struct socket *sock, int level, int optname,
 				  char __user *optval, unsigned int optlen)
 {
-	return -EOPNOTSUPP;
+	struct unix_sock *u = unix_sk(sock->sk);
+	int val;
+	int err = 0;
+
+	if (optlen < sizeof(int))
+		return -EINVAL;
+
+	if (get_user(val, (int __user *)optval))
+		return -EFAULT;
+
+	switch (optname) {
+	case UNIX_MULTICAST:
+		/* Multicast feature can only be changed when the socket is
+		 * not used yet */
+		if (u->addr || sock->sk->sk_state != TCP_CLOSE)
+			return -EINVAL;
+
+		if (val != 0) {
+			u->multicast = 1;
+		} else {
+			u->multicast = 0;
+		}
+		break;
+
+	default:
+		err = -ENOPROTOOPT;
+		break;
+	}
+
+	return err;
 }
 
 int unix_stream_getsockopt(struct socket *sock, int level, int optname,
 			   char __user *optval, int __user *optlen)
 {
-	return -EOPNOTSUPP;
+	struct unix_sock *u = unix_sk(sock->sk);
+	int val, len;
+
+	if (get_user(len, optlen))
+		return -EFAULT;
+
+	len = min_t(unsigned int, len, sizeof(int));
+
+	if (len < 0)
+		return -EINVAL;
+
+	switch (optname) {
+	case UNIX_MULTICAST:
+		val = u->multicast;
+		break;
+
+	default:
+		return -ENOPROTOOPT;
+	}
+
+	if (put_user(len, optlen))
+		return -EFAULT;
+	if (copy_to_user(optval, &val, len))
+		return -EFAULT;
+	return 0;
 }
 
 
-- 
1.7.1

[PATCH 3/5] AF_UNIX: implement connect() on multicast Unix stream socket

[Alban Crequy]

All multicast peers use connect() without any bound socket.

Multicast Unix socket addresses and non-multicast Unix socket addresses live in
the same namespace. An userspace application cannot connect() a non-multicast
socket if the address is already used for multicast sockets.

Signed-off-by: Alban Crequy <alban.crequy@collabora.co.uk>
---
 net/unix/af_unix.c |   22 ++++++++++++++++++++--
 1 files changed, 20 insertions(+), 2 deletions(-)

diff --git a/net/unix/af_unix.c b/net/unix/af_unix.c
index c766e88..a8d9de7 100644
--- a/net/unix/af_unix.c
+++ b/net/unix/af_unix.c
@@ -1064,9 +1064,20 @@ static int unix_stream_connect(struct socket *sock, struct sockaddr *uaddr,
 restart:
 	/*  Find listening sock. */
 	other = unix_find_other(net, sunaddr, addr_len, sk->sk_type, hash, &err);
-	if (!other)
+	if (!other && !u->multicast)
 		goto out;
 
+	if (other && u->multicast) {
+		err = -ECONNREFUSED;
+		goto out;
+	}
+
+	if (u->multicast) {
+		sock->state = SS_CONNECTED;
+		sk->sk_state  = TCP_ESTABLISHED;
+		return 0;
+	}
+
 	/* Latch state of peer */
 	unix_state_lock(other);
 
@@ -1567,6 +1578,7 @@ int unix_stream_getsockopt(struct socket *sock, int level, int optname,
 static int unix_stream_sendmsg(struct kiocb *kiocb, struct socket *sock,
 			       struct msghdr *msg, size_t len)
 {
+	struct unix_sock *u = unix_sk(sock->sk);
 	struct sock_iocb *siocb = kiocb_to_siocb(kiocb);
 	struct sock *sk = sock->sk;
 	struct sock *other = NULL;
@@ -1591,12 +1603,18 @@ static int unix_stream_sendmsg(struct kiocb *kiocb, struct socket *sock,
 	if (msg->msg_namelen) {
 		err = sk->sk_state == TCP_ESTABLISHED ? -EISCONN : -EOPNOTSUPP;
 		goto out_err;
-	} else {
+	} else if (!u->multicast) {
 		sunaddr = NULL;
 		err = -ENOTCONN;
 		other = unix_peer(sk);
 		if (!other)
 			goto out_err;
+	} else {
+		sunaddr = NULL;
+		err = -ENOTCONN;
+		other = NULL; /* FIXME: get the list of other connection */
+		if (!other)
+			goto out_err;
 	}
 
 	if (sk->sk_shutdown & SEND_SHUTDOWN)
-- 
1.7.1

[PATCH 4/5] AF_UNIX: find peers on multicast Unix stream sockets

[Alban Crequy]

Multicast sockets are stored in the hash table unix_multicast_socket_table.

unix_find_socket_byname() is extended to return an array of sockets matching
the name instead of only one socket. Then unix_stream_sendmsg() can find all
the multicast peers.

Signed-off-by: Alban Crequy <alban.crequy@collabora.co.uk>
---
 net/unix/af_unix.c |  134 ++++++++++++++++++++++++++++++++++++++--------------
 1 files changed, 99 insertions(+), 35 deletions(-)

diff --git a/net/unix/af_unix.c b/net/unix/af_unix.c
index a8d9de7..f259849 100644
--- a/net/unix/af_unix.c
+++ b/net/unix/af_unix.c
@@ -115,11 +115,13 @@
 #include <net/checksum.h>
 #include <linux/security.h>
 
-static struct hlist_head unix_socket_table[UNIX_HASH_SIZE + 1];
+static struct hlist_head unix_socket_table[2 * UNIX_HASH_SIZE + 1];
 static DEFINE_SPINLOCK(unix_table_lock);
 static atomic_t unix_nr_socks = ATOMIC_INIT(0);
+static atomic_t unix_nr_multicast_socks = ATOMIC_INIT(0);
 
-#define unix_sockets_unbound	(&unix_socket_table[UNIX_HASH_SIZE])
+#define unix_multicast_socket_table	(&unix_socket_table[UNIX_HASH_SIZE])
+#define unix_sockets_unbound		(&unix_socket_table[2 * UNIX_HASH_SIZE])
 
 #define UNIX_ABSTRACT(sk)	(unix_sk(sk)->addr->hash != UNIX_HASH_SIZE)
 
@@ -227,7 +229,6 @@ static void __unix_remove_socket(struct sock *sk)
 
 static void __unix_insert_socket(struct hlist_head *list, struct sock *sk)
 {
-	WARN_ON(!sk_unhashed(sk));
 	sk_add_node(sk, list);
 }
 
@@ -247,12 +248,14 @@ static inline void unix_insert_socket(struct hlist_head *list, struct sock *sk)
 
 static struct sock *__unix_find_socket_byname(struct net *net,
 					      struct sockaddr_un *sunname,
-					      int len, int type, unsigned hash)
+					      int len, int type,
+					      unsigned hash, int multicast)
 {
 	struct sock *s;
 	struct hlist_node *node;
+	unsigned int index = (multicast ? UNIX_HASH_SIZE : 0) + (hash ^ type);
 
-	sk_for_each(s, node, &unix_socket_table[hash ^ type]) {
+	sk_for_each(s, node, &unix_socket_table[index]) {
 		struct unix_sock *u = unix_sk(s);
 
 		if (!net_eq(sock_net(s), net))
@@ -267,29 +270,50 @@ found:
 	return s;
 }
 
-static inline struct sock *unix_find_socket_byname(struct net *net,
-						   struct sockaddr_un *sunname,
-						   int len, int type,
-						   unsigned hash)
+static inline void unix_find_socket_byname(struct net *net,
+					   struct sockaddr_un *sunname,
+					   int len, int type,
+					   unsigned hash,
+					   int multicast,
+					   struct sock **others, int max_others)
 {
 	struct sock *s;
+	struct hlist_node *node;
+	int i = 0;
+	unsigned int index = (multicast ? UNIX_HASH_SIZE : 0) + (hash ^ type);
 
 	spin_lock(&unix_table_lock);
-	s = __unix_find_socket_byname(net, sunname, len, type, hash);
-	if (s)
-		sock_hold(s);
+
+	sk_for_each(s, node, &unix_socket_table[index]) {
+		struct unix_sock *u = unix_sk(s);
+
+		if (!net_eq(sock_net(s), net))
+			continue;
+
+		if (u->addr->len == len &&
+		    !memcmp(u->addr->name, sunname, len)) {
+
+			others[i++] = s;
+			sock_hold(s);
+			if (i == max_others)
+				break;
+		}
+	}
+
 	spin_unlock(&unix_table_lock);
-	return s;
 }
 
-static struct sock *unix_find_socket_byinode(struct net *net, struct inode *i)
+static struct sock *unix_find_socket_byinode(struct net *net, struct inode *i,
+					     int multicast)
 {
 	struct sock *s;
 	struct hlist_node *node;
+	unsigned int index = (multicast ? UNIX_HASH_SIZE : 0)
+	  + (i->i_ino & (UNIX_HASH_SIZE - 1));
 
 	spin_lock(&unix_table_lock);
 	sk_for_each(s, node,
-		    &unix_socket_table[i->i_ino & (UNIX_HASH_SIZE - 1)]) {
+		    &unix_socket_table[index]) {
 		struct dentry *dentry = unix_sk(s)->dentry;
 
 		if (!net_eq(sock_net(s), net))
@@ -363,6 +387,9 @@ static void unix_sock_destructor(struct sock *sk)
 	if (u->addr)
 		unix_release_addr(u->addr);
 
+	if (u->multicast)
+		atomic_dec(&unix_nr_multicast_socks);
+
 	atomic_dec(&unix_nr_socks);
 	local_bh_disable();
 	sock_prot_inuse_add(sock_net(sk), sk->sk_prot, -1);
@@ -700,7 +727,7 @@ retry:
 	ordernum = (ordernum+1)&0xFFFFF;
 
 	if (__unix_find_socket_byname(net, addr->name, addr->len, sock->type,
-				      addr->hash)) {
+				      addr->hash, 0)) {
 		spin_unlock(&unix_table_lock);
 		/* Sanity yield. It is unusual case, but yet... */
 		if (!(ordernum&0xFF))
@@ -719,9 +746,11 @@ out:	mutex_unlock(&u->readlock);
 	return err;
 }
 
-static struct sock *unix_find_other(struct net *net,
-				    struct sockaddr_un *sunname, int len,
-				    int type, unsigned hash, int *error)
+static void unix_find_other(struct net *net,
+			    struct sockaddr_un *sunname, int len,
+			    int type, unsigned hash, int multicast,
+			    struct sock **others, int max_others,
+			    int *error)
 {
 	struct sock *u;
 	struct path path;
@@ -740,7 +769,7 @@ static struct sock *unix_find_other(struct net *net,
 		err = -ECONNREFUSED;
 		if (!S_ISSOCK(inode->i_mode))
 			goto put_fail;
-		u = unix_find_socket_byinode(net, inode);
+		u = unix_find_socket_byinode(net, inode, multicast);
 		if (!u)
 			goto put_fail;
 
@@ -754,24 +783,26 @@ static struct sock *unix_find_other(struct net *net,
 			sock_put(u);
 			goto fail;
 		}
+		others[0] = u;
 	} else {
+		int i;
 		err = -ECONNREFUSED;
-		u = unix_find_socket_byname(net, sunname, len, type, hash);
-		if (u) {
+		unix_find_socket_byname(net, sunname, len, type, hash,
+                    multicast, others, max_others);
+		for(i = 0 ; i < max_others && others[i] != NULL ; i++) {
 			struct dentry *dentry;
-			dentry = unix_sk(u)->dentry;
+			dentry = unix_sk(others[i])->dentry;
 			if (dentry)
-				touch_atime(unix_sk(u)->mnt, dentry);
-		} else
-			goto fail;
+				touch_atime(unix_sk(others[i])->mnt, dentry);
+		}
 	}
-	return u;
+	return;
 
 put_fail:
 	path_put(&path);
 fail:
 	*error = err;
-	return NULL;
+	return;
 }
 
 
@@ -862,7 +893,7 @@ out_mknod_drop_write:
 	if (!sunaddr->sun_path[0]) {
 		err = -EADDRINUSE;
 		if (__unix_find_socket_byname(net, sunaddr, addr_len,
-					      sk->sk_type, hash)) {
+					      sk->sk_type, hash, 0)) {
 			unix_release_addr(addr);
 			goto out_unlock;
 		}
@@ -929,7 +960,7 @@ static int unix_dgram_connect(struct socket *sock, struct sockaddr *addr,
 	struct sock *sk = sock->sk;
 	struct net *net = sock_net(sk);
 	struct sockaddr_un *sunaddr = (struct sockaddr_un *)addr;
-	struct sock *other;
+	struct sock *other = NULL;
 	unsigned hash;
 	int err;
 
@@ -944,7 +975,8 @@ static int unix_dgram_connect(struct socket *sock, struct sockaddr *addr,
 			goto out;
 
 restart:
-		other = unix_find_other(net, sunaddr, alen, sock->type, hash, &err);
+		unix_find_other(net, sunaddr, alen, sock->type, hash,
+                    0, &other, 1, &err);
 		if (!other)
 			goto out;
 
@@ -1063,7 +1095,8 @@ static int unix_stream_connect(struct socket *sock, struct sockaddr *uaddr,
 
 restart:
 	/*  Find listening sock. */
-	other = unix_find_other(net, sunaddr, addr_len, sk->sk_type, hash, &err);
+	unix_find_other(net, sunaddr, addr_len, sk->sk_type, hash, 0, &other,
+            1, &err);
 	if (!other && !u->multicast)
 		goto out;
 
@@ -1075,6 +1108,26 @@ restart:
 	if (u->multicast) {
 		sock->state = SS_CONNECTED;
 		sk->sk_state  = TCP_ESTABLISHED;
+
+		unix_find_other(net, sunaddr, addr_len, sk->sk_type,
+		    hash, 1, &other, 1, &err);
+		if (other) {
+			otheru = unix_sk(other);
+			atomic_inc(&otheru->addr->refcnt);
+			u->addr = otheru->addr;
+		} else {
+			err = -ENOMEM;
+			u->addr = kmalloc(sizeof(*u->addr)+addr_len, GFP_KERNEL);
+			if (!u->addr)
+				return err;
+
+			memcpy(u->addr->name, sunaddr, addr_len);
+			u->addr->len = addr_len;
+			u->addr->hash = hash ^ sk->sk_type;
+			atomic_set(&u->addr->refcnt, 1);
+		}
+
+		unix_insert_socket(&unix_multicast_socket_table[u->addr->hash], sk);
 		return 0;
 	}
 
@@ -1427,8 +1480,8 @@ restart:
 		if (sunaddr == NULL)
 			goto out_free;
 
-		other = unix_find_other(net, sunaddr, namelen, sk->sk_type,
-					hash, &err);
+		unix_find_other(net, sunaddr, namelen, sk->sk_type,
+					hash, 0, &other, 1, &err);
 		if (other == NULL)
 			goto out_free;
 	}
@@ -1530,8 +1583,12 @@ static int unix_stream_setsockopt(struct socket *sock, int level, int optname,
 			return -EINVAL;
 
 		if (val != 0) {
+			if (!u->multicast)
+				atomic_inc(&unix_nr_multicast_socks);
 			u->multicast = 1;
 		} else {
+			if (u->multicast)
+				atomic_dec(&unix_nr_multicast_socks);
 			u->multicast = 0;
 		}
 		break;
@@ -1582,6 +1639,8 @@ static int unix_stream_sendmsg(struct kiocb *kiocb, struct socket *sock,
 	struct sock_iocb *siocb = kiocb_to_siocb(kiocb);
 	struct sock *sk = sock->sk;
 	struct sock *other = NULL;
+	struct sock **others = NULL;
+	int max_others;
 	struct sockaddr_un *sunaddr = msg->msg_name;
 	int err, size;
 	struct sk_buff *skb;
@@ -1612,7 +1671,12 @@ static int unix_stream_sendmsg(struct kiocb *kiocb, struct socket *sock,
 	} else {
 		sunaddr = NULL;
 		err = -ENOTCONN;
-		other = NULL; /* FIXME: get the list of other connection */
+		max_others = atomic_read(&unix_nr_multicast_socks);
+		others = kzalloc((max_others + 1) * sizeof(void *), GFP_KERNEL);
+		unix_find_other(sock_net(sk), u->addr->name,
+		    u->addr->len, 0, u->addr->hash, 1, others, max_others, &err);
+		other = others[0];
+		kfree(others);
 		if (!other)
 			goto out_err;
 	}
-- 
1.7.1

[PATCH 5/5] AF_UNIX: deliver the data to all the multicast peers

[Alban Crequy]

With multicast sockets, unix_stream_sendmsg() needs to deliver the data to
zero, one or several recipients. skb_clone() is used to get as many sk_buff as
recipients.

If a multicast peer is too slow to receive the packets, it will block the
sender at some point in sock_alloc_send_skb().

A sender can receive SIGPIPE caused by an unique peer on the multicast group.
This is probably not what we want.

I tested this with a few peers and either sending data, receiving data or just
sleeping.

Signed-off-by: Alban Crequy <alban.crequy@collabora.co.uk>
---
 net/unix/af_unix.c |   77 +++++++++++++++++++++++++++++++++++----------------
 1 files changed, 53 insertions(+), 24 deletions(-)

diff --git a/net/unix/af_unix.c b/net/unix/af_unix.c
index f259849..01cb603 100644
--- a/net/unix/af_unix.c
+++ b/net/unix/af_unix.c
@@ -1640,10 +1640,10 @@ static int unix_stream_sendmsg(struct kiocb *kiocb, struct socket *sock,
 	struct sock *sk = sock->sk;
 	struct sock *other = NULL;
 	struct sock **others = NULL;
+	struct sock **others_cur = NULL;
 	int max_others;
 	struct sockaddr_un *sunaddr = msg->msg_name;
 	int err, size;
-	struct sk_buff *skb;
 	int sent = 0;
 	struct scm_cookie tmp_scm;
 	bool fds_sent = false;
@@ -1668,15 +1668,22 @@ static int unix_stream_sendmsg(struct kiocb *kiocb, struct socket *sock,
 		other = unix_peer(sk);
 		if (!other)
 			goto out_err;
+		err = -ENOMEM;
+		others = kzalloc(2 * sizeof(void *), GFP_KERNEL);
+		if (!others)
+			goto out_err;
+		others[0] = other;
 	} else {
 		sunaddr = NULL;
-		err = -ENOTCONN;
 		max_others = atomic_read(&unix_nr_multicast_socks);
+		err = -ENOMEM;
 		others = kzalloc((max_others + 1) * sizeof(void *), GFP_KERNEL);
+		if (!others)
+			goto out_err;
 		unix_find_other(sock_net(sk), u->addr->name,
 		    u->addr->len, 0, u->addr->hash, 1, others, max_others, &err);
+		err = -ENOTCONN;
 		other = others[0];
-		kfree(others);
 		if (!other)
 			goto out_err;
 	}
@@ -1685,6 +1692,8 @@ static int unix_stream_sendmsg(struct kiocb *kiocb, struct socket *sock,
 		goto pipe_err;
 
 	while (sent < len) {
+		struct sk_buff *skb;
+		struct sk_buff *skb_cloned;
 		/*
 		 *	Optimisation for the fact that under 0.01% of X
 		 *	messages typically need breaking up.
@@ -1718,43 +1727,61 @@ static int unix_stream_sendmsg(struct kiocb *kiocb, struct socket *sock,
 		 */
 		size = min_t(int, size, skb_tailroom(skb));
 
-		memcpy(UNIXCREDS(skb), &siocb->scm->creds, sizeof(struct ucred));
-		/* Only send the fds in the first buffer */
-		if (siocb->scm->fp && !fds_sent) {
-			err = unix_attach_fds(siocb->scm, skb);
-			if (err) {
-				kfree_skb(skb);
-				goto out_err;
-			}
-			fds_sent = true;
-		}
-
 		err = memcpy_fromiovec(skb_put(skb, size), msg->msg_iov, size);
 		if (err) {
 			kfree_skb(skb);
 			goto out_err;
 		}
 
-		unix_state_lock(other);
+		others_cur = others;
+		while (*others_cur != NULL) {
+			skb_cloned = skb_clone(skb, GFP_KERNEL);
+			if (!skb_cloned) {
+				kfree_skb(skb);
+				goto out_err;
+			}
+			skb_set_owner_w(skb_cloned, sk);
+
+			memcpy(UNIXCREDS(skb_cloned), &siocb->scm->creds, sizeof(struct ucred));
+			/* Only send the fds in the first buffer of each
+ 			 * recipient */
+			if (siocb->scm->fp && !fds_sent) {
+				err = unix_attach_fds(siocb->scm, skb_cloned);
+				if (err) {
+					kfree_skb(skb);
+					kfree_skb(skb_cloned);
+					goto out_err;
+				}
+			}
+
+			unix_state_lock(*others_cur);
 
-		if (sock_flag(other, SOCK_DEAD) ||
-		    (other->sk_shutdown & RCV_SHUTDOWN))
-			goto pipe_err_free;
+			if (sock_flag(*others_cur, SOCK_DEAD) ||
+			    ((*others_cur)->sk_shutdown & RCV_SHUTDOWN)) {
+				unix_state_unlock(*others_cur);
+				kfree_skb(skb);
+				kfree_skb(skb_cloned);
+				goto pipe_err;
+			}
 
-		skb_queue_tail(&other->sk_receive_queue, skb);
-		unix_state_unlock(other);
-		other->sk_data_ready(other, size);
+			skb_queue_tail(&(*others_cur)->sk_receive_queue,
+			    skb_cloned);
+			unix_state_unlock(*others_cur);
+			(*others_cur)->sk_data_ready(*others_cur, size);
+			others_cur++;
+		}
+		kfree_skb(skb);
+		fds_sent = true;
 		sent += size;
 	}
 
 	scm_destroy(siocb->scm);
 	siocb->scm = NULL;
+	if (others)
+		kfree(others);
 
 	return sent;
 
-pipe_err_free:
-	unix_state_unlock(other);
-	kfree_skb(skb);
 pipe_err:
 	if (sent == 0 && !(msg->msg_flags&MSG_NOSIGNAL))
 		send_sig(SIGPIPE, current, 0);
@@ -1762,6 +1789,8 @@ pipe_err:
 out_err:
 	scm_destroy(siocb->scm);
 	siocb->scm = NULL;
+	if (others)
+		kfree(others);
 	return sent ? : err;
 }
 
-- 
1.7.1

Re: [PATCH 2/5] AF_UNIX: enable/disable multicast with getsockopt/setsockopt

[Eric Dumazet]

Le vendredi 24 septembre 2010 à 18:25 +0100, Alban Crequy a écrit :
> Multicast can be enabled or disabled after a socket is allocated but this
> cannot be changed once the socket is bound or connected.
> 
> Userspace applications can enable multicast on an Unix stream socket:
>   sockfd = socket(AF_UNIX, SOCK_STREAM, 0);
>   #define UNIX_MULTICAST 1
>   val = 1;
>   len = sizeof(val);
>   ret = setsockopt(sockfd, 0, UNIX_MULTICAST, &val, len);
> 
> Signed-off-by: Alban Crequy <alban.crequy@collabora.co.uk>


> +		if (val != 0) {
> +			u->multicast = 1;
> +		} else {
> +			u->multicast = 0;
> +		}

u->multicast = !!val;

Re: [PATCH 4/5] AF_UNIX: find peers on multicast Unix stream sockets

[Eric Dumazet]

Le vendredi 24 septembre 2010 à 18:25 +0100, Alban Crequy a écrit :

> @@ -1612,7 +1671,12 @@ static int unix_stream_sendmsg(struct kiocb *kiocb, struct socket *sock,
>  	} else {
>  		sunaddr = NULL;
>  		err = -ENOTCONN;
> -		other = NULL; /* FIXME: get the list of other connection */
> +		max_others = atomic_read(&unix_nr_multicast_socks);
> +		others = kzalloc((max_others + 1) * sizeof(void *), GFP_KERNEL);
> +		unix_find_other(sock_net(sk), u->addr->name,
> +		    u->addr->len, 0, u->addr->hash, 1, others, max_others, &err);
> +		other = others[0];
> +		kfree(others);
>  		if (!other)
>  			goto out_err;
>  	}

Seriously, this block sizing against unix_nr_multicast_socks is not
scalable. What happens if we have 1000 sockets ?
kzalloc() to clear 8000 bytes ?
Its also unsafe.

(say you kzalloc() a buffer for 2 sockets, and another cpu inserts a new
socket. unix_find_socket_byname() can overflow the buffer)


You should use a list, and allocates elements in
unix_find_socket_byname()

struct item {
	struct item *next;
	struct sock *s;
};

Re: [PATCH 2/5] AF_UNIX: enable/disable multicast with getsockopt/setsockopt

[Eric Dumazet]

Le vendredi 24 septembre 2010 à 19:46 +0200, Eric Dumazet a écrit :

Hmm

dbus-owner@lists.freedesktop.org is a subscriber only list ....

Please dont use it

Re: [PATCH 4/5] AF_UNIX: find peers on multicast Unix stream sockets

[Alban Crequy]

Le Fri, 24 Sep 2010 20:00:37 +0200,
Eric Dumazet <eric.dumazet@gmail.com> a écrit :

> Le vendredi 24 septembre 2010 à 18:25 +0100, Alban Crequy a écrit :
> 
> > @@ -1612,7 +1671,12 @@ static int unix_stream_sendmsg(struct kiocb
> > *kiocb, struct socket *sock, } else {
> >  		sunaddr = NULL;
> >  		err = -ENOTCONN;
> > -		other = NULL; /* FIXME: get the list of other
> > connection */
> > +		max_others = atomic_read(&unix_nr_multicast_socks);
> > +		others = kzalloc((max_others + 1) * sizeof(void
> > *), GFP_KERNEL);
> > +		unix_find_other(sock_net(sk), u->addr->name,
> > +		    u->addr->len, 0, u->addr->hash, 1, others,
> > max_others, &err);
> > +		other = others[0];
> > +		kfree(others);
> >  		if (!other)
> >  			goto out_err;
> >  	}
> 
> Seriously, this block sizing against unix_nr_multicast_socks is not
> scalable. What happens if we have 1000 sockets ?
> kzalloc() to clear 8000 bytes ?
> Its also unsafe.
> 
> (say you kzalloc() a buffer for 2 sockets, and another cpu inserts a
> new socket. unix_find_socket_byname() can overflow the buffer)
> 
> 
> You should use a list, and allocates elements in
> unix_find_socket_byname()
> 
> struct item {
> 	struct item *next;
> 	struct sock *s;
> };

Thanks for your review.

I cannot allocate elements directly in unix_find_socket_byname()
iteration after iteration because the spinlock "unix_table_lock" is
held. If I release the spinlock to allocate, the number of sockets in
the table may change.

I changed the code to count the sockets with the lock held and then
allocate. In the unfortunate case where the allocation is not big
enough (if another process inserts a new socket), it just tries again.

The code is available here. Please pull from:

git://git.collabora.co.uk/git/user/alban/linux-2.6.35.y/.git unix-multicast2

It is still a work in progress. Missing pieces:

- The flow control does not work correctly: poll/select does not match
  the reality

- Atomic delivery: if a process is killed or interrupted in the middle
  of a delivery, only a subset of the recipients will get the message 

- Some locking to provide the same delivery order to all the recipients
  when several senders run concurrently.

Feedback welcome,

Alban Crequy