Issue #7777: socket: Add Reliable Datagram Sockets (PF_RDS) support.

Doc/library/socket.rst

@@ -236,6 +236,19 @@ The module :mod:`socket` exports the following constants and functions:
    .. versionadded:: 3.3
 
 
+.. data:: AF_RDS
+          PF_RDS
+          SOL_RDS
+          RDS_*
+
+   Many constants of these forms, documented in the Linux documentation, are
+   also defined in the socket module.
+
+   Availability: Linux >= 2.6.30.
+
+   .. versionadded:: 3.3
+
+
 .. data:: SIO_*
           RCVALL_*
 
@@ -407,14 +420,15 @@ The module :mod:`socket` exports the following constants and functions:
 
    Create a new socket using the given address family, socket type and protocol
    number.  The address family should be :const:`AF_INET` (the default),
-   :const:`AF_INET6`, :const:`AF_UNIX` or :const:`AF_CAN`. The socket type
+   :const:`AF_INET6`, :const:`AF_UNIX`, :const:`AF_CAN` or :const:`AF_RDS`. The
-   should be :const:`SOCK_STREAM` (the default), :const:`SOCK_DGRAM`,
+   socket type should be :const:`SOCK_STREAM` (the default),
-   :const:`SOCK_RAW` or perhaps one of the other ``SOCK_`` constants. The
+   :const:`SOCK_DGRAM`, :const:`SOCK_RAW` or perhaps one of the other ``SOCK_``
-   protocol number is usually zero and may be omitted in that case or
+   constants. The protocol number is usually zero and may be omitted in that
-   :const:`CAN_RAW` in case the address family is :const:`AF_CAN`.
+   case or :const:`CAN_RAW` in case the address family is :const:`AF_CAN`.
 
    .. versionchanged:: 3.3
       The AF_CAN family was added.
+      The AF_RDS family was added.
 
 
 .. function:: socketpair([family[, type[, proto]]])

Doc/whatsnew/3.3.rst

@@ -495,6 +495,9 @@ socket
 
   (Contributed by Matthias Fuchs, updated by Tiago Gonçalves in :issue:`10141`)
 
+* The :class:`~socket.socket` class now supports the PF_RDS protocol family
+  (http://en.wikipedia.org/wiki/Reliable_Datagram_Sockets and
+  http://oss.oracle.com/projects/rds/).
 
 ssl
 ---

Lib/test/test_socket.py

@@ -47,8 +47,20 @@ def _have_socket_can():
         s.close()
     return True
 
+def _have_socket_rds():
+    """Check whether RDS sockets are supported on this host."""
+    try:
+        s = socket.socket(socket.PF_RDS, socket.SOCK_SEQPACKET, 0)
+    except (AttributeError, OSError):
+        return False
+    else:
+        s.close()
+    return True
+
 HAVE_SOCKET_CAN = _have_socket_can()
 
+HAVE_SOCKET_RDS = _have_socket_rds()
+
 # Size in bytes of the int type
 SIZEOF_INT = array.array("i").itemsize
 
@@ -113,6 +125,23 @@ def setUp(self):
             self.skipTest('network interface `%s` does not exist' %
                            self.interface)
 
+
+class SocketRDSTest(unittest.TestCase):
+
+    """To be able to run this test, the `rds` kernel module must be loaded:
+    # modprobe rds
+    """
+    bufsize = 8192
+
+    def setUp(self):
+        self.serv = socket.socket(socket.PF_RDS, socket.SOCK_SEQPACKET, 0)
+        self.addCleanup(self.serv.close)
+        try:
+            self.port = support.bind_port(self.serv)
+        except OSError:
+            self.skipTest('unable to bind RDS socket')
+
+
 class ThreadableTest:
     """Threadable Test class
 
@@ -271,6 +300,29 @@ def clientTearDown(self):
         self.cli = None
         ThreadableTest.clientTearDown(self)
 
+class ThreadedRDSSocketTest(SocketRDSTest, ThreadableTest):
+
+    def __init__(self, methodName='runTest'):
+        SocketRDSTest.__init__(self, methodName=methodName)
+        ThreadableTest.__init__(self)
+        self.evt = threading.Event()
+
+    def clientSetUp(self):
+        self.cli = socket.socket(socket.PF_RDS, socket.SOCK_SEQPACKET, 0)
+        try:
+            # RDS sockets must be bound explicitly to send or receive data
+            self.cli.bind((HOST, 0))
+            self.cli_addr = self.cli.getsockname()
+        except OSError:
+            # skipTest should not be called here, and will be called in the
+            # server instead
+            pass
+
+    def clientTearDown(self):
+        self.cli.close()
+        self.cli = None
+        ThreadableTest.clientTearDown(self)
+
 class SocketConnectedTest(ThreadedTCPSocketTest):
     """Socket tests for client-server connection.
 
@@ -1239,6 +1291,112 @@ def _testSendMultiFrames(self):
         self.cli.send(self.cf2)
 
 
+@unittest.skipUnless(HAVE_SOCKET_RDS, 'RDS sockets required for this test.')
+class BasicRDSTest(unittest.TestCase):
+
+    def testCrucialConstants(self):
+        socket.AF_RDS
+        socket.PF_RDS
+
+    def testCreateSocket(self):
+        with socket.socket(socket.PF_RDS, socket.SOCK_SEQPACKET, 0) as s:
+            pass
+
+    def testSocketBufferSize(self):
+        bufsize = 16384
+        with socket.socket(socket.PF_RDS, socket.SOCK_SEQPACKET, 0) as s:
+            s.setsockopt(socket.SOL_SOCKET, socket.SO_RCVBUF, bufsize)
+            s.setsockopt(socket.SOL_SOCKET, socket.SO_SNDBUF, bufsize)
+
+
+@unittest.skipUnless(HAVE_SOCKET_RDS, 'RDS sockets required for this test.')
+@unittest.skipUnless(thread, 'Threading required for this test.')
+class RDSTest(ThreadedRDSSocketTest):
+
+    def __init__(self, methodName='runTest'):
+        ThreadedRDSSocketTest.__init__(self, methodName=methodName)
+
+    def testSendAndRecv(self):
+        data, addr = self.serv.recvfrom(self.bufsize)
+        self.assertEqual(self.data, data)
+        self.assertEqual(self.cli_addr, addr)
+
+    def _testSendAndRecv(self):
+        self.data = b'spam'
+        self.cli.sendto(self.data, 0, (HOST, self.port))
+
+    def testPeek(self):
+        data, addr = self.serv.recvfrom(self.bufsize, socket.MSG_PEEK)
+        self.assertEqual(self.data, data)
+        data, addr = self.serv.recvfrom(self.bufsize)
+        self.assertEqual(self.data, data)
+
+    def _testPeek(self):
+        self.data = b'spam'
+        self.cli.sendto(self.data, 0, (HOST, self.port))
+
+    @requireAttrs(socket.socket, 'recvmsg')
+    def testSendAndRecvMsg(self):
+        data, ancdata, msg_flags, addr = self.serv.recvmsg(self.bufsize)
+        self.assertEqual(self.data, data)
+
+    @requireAttrs(socket.socket, 'sendmsg')
+    def _testSendAndRecvMsg(self):
+        self.data = b'hello ' * 10
+        self.cli.sendmsg([self.data], (), 0, (HOST, self.port))
+
+    def testSendAndRecvMulti(self):
+        data, addr = self.serv.recvfrom(self.bufsize)
+        self.assertEqual(self.data1, data)
+
+        data, addr = self.serv.recvfrom(self.bufsize)
+        self.assertEqual(self.data2, data)
+
+    def _testSendAndRecvMulti(self):
+        self.data1 = b'bacon'
+        self.cli.sendto(self.data1, 0, (HOST, self.port))
+
+        self.data2 = b'egg'
+        self.cli.sendto(self.data2, 0, (HOST, self.port))
+
+    def testSelect(self):
+        r, w, x = select.select([self.serv], [], [], 3.0)
+        self.assertIn(self.serv, r)
+        data, addr = self.serv.recvfrom(self.bufsize)
+        self.assertEqual(self.data, data)
+
+    def _testSelect(self):
+        self.data = b'select'
+        self.cli.sendto(self.data, 0, (HOST, self.port))
+
+    def testCongestion(self):
+        # wait until the sender is done
+        self.evt.wait()
+
+    def _testCongestion(self):
+        # test the behavior in case of congestion
+        self.data = b'fill'
+        self.cli.setblocking(False)
+        try:
+            # try to lower the receiver's socket buffer size
+            self.cli.setsockopt(socket.SOL_SOCKET, socket.SO_RCVBUF, 16384)
+        except OSError:
+            pass
+        with self.assertRaises(OSError) as cm:
+            try:
+                # fill the receiver's socket buffer
+                while True:
+                    self.cli.sendto(self.data, 0, (HOST, self.port))
+            finally:
+                # signal the receiver we're done
+                self.evt.set()
+        # sendto() should have failed with ENOBUFS
+        self.assertEqual(cm.exception.errno, errno.ENOBUFS)
+        # and we should have received a congestion notification through poll
+        r, w, x = select.select([self.serv], [], [], 3.0)
+        self.assertIn(self.serv, r)
+
+
 @unittest.skipUnless(thread, 'Threading required for this test.')
 class BasicTCPTest(SocketConnectedTest):
 
@@ -4362,6 +4520,7 @@ def test_main():
         tests.append(TIPCTest)
         tests.append(TIPCThreadableTest)
     tests.extend([BasicCANTest, CANTest])
+    tests.extend([BasicRDSTest, RDSTest])
     tests.extend([
         CmsgMacroTests,
         SendmsgUDPTest,

Misc/NEWS

@@ -1516,6 +1516,8 @@ Tools/Demos
 Extension Modules
 -----------------
 
+- Issue #7777: socket: Add Reliable Datagram Sockets (PF_RDS) support.
+
 - Issue #13159: FileIO and BZ2Compressor/BZ2Decompressor now use a linear-time
   buffer growth strategy instead of a quadratic-time one.
 

Modules/socketmodule.c

@@ -1328,6 +1328,11 @@ getsockaddrarg(PySocketSockObject *s, PyObject *args,
     }
 #endif
 
+#ifdef AF_RDS
+    case AF_RDS:
+        /* RDS sockets use sockaddr_in: fall-through */
+#endif
+
     case AF_INET:
     {
         struct sockaddr_in* addr;
@@ -1686,6 +1691,11 @@ getsockaddrlen(PySocketSockObject *s, socklen_t *len_ret)
        }
 #endif
 
+#ifdef AF_RDS
+    case AF_RDS:
+        /* RDS sockets use sockaddr_in: fall-through */
+#endif
+
     case AF_INET:
     {
         *len_ret = sizeof (struct sockaddr_in);
@@ -5614,6 +5624,14 @@ PyInit__socket(void)
     PyModule_AddIntConstant(m, "PF_CAN", PF_CAN);
 #endif
 
+/* Reliable Datagram Sockets */
+#ifdef AF_RDS
+    PyModule_AddIntConstant(m, "AF_RDS", AF_RDS);
+#endif
+#ifdef PF_RDS
+    PyModule_AddIntConstant(m, "PF_RDS", PF_RDS);
+#endif
+
 #ifdef AF_PACKET
     PyModule_AddIntMacro(m, AF_PACKET);
 #endif
@@ -5909,6 +5927,27 @@ PyInit__socket(void)
     PyModule_AddIntConstant(m, "CAN_RAW_LOOPBACK", CAN_RAW_LOOPBACK);
     PyModule_AddIntConstant(m, "CAN_RAW_RECV_OWN_MSGS", CAN_RAW_RECV_OWN_MSGS);
 #endif
+#ifdef SOL_RDS
+    PyModule_AddIntConstant(m, "SOL_RDS", SOL_RDS);
+#endif
+#ifdef RDS_CANCEL_SENT_TO
+    PyModule_AddIntConstant(m, "RDS_CANCEL_SENT_TO", RDS_CANCEL_SENT_TO);
+#endif
+#ifdef RDS_GET_MR
+    PyModule_AddIntConstant(m, "RDS_GET_MR", RDS_GET_MR);
+#endif
+#ifdef RDS_FREE_MR
+    PyModule_AddIntConstant(m, "RDS_FREE_MR", RDS_FREE_MR);
+#endif
+#ifdef RDS_RECVERR
+    PyModule_AddIntConstant(m, "RDS_RECVERR", RDS_RECVERR);
+#endif
+#ifdef RDS_CONG_MONITOR
+    PyModule_AddIntConstant(m, "RDS_CONG_MONITOR", RDS_CONG_MONITOR);
+#endif
+#ifdef RDS_GET_MR_FOR_DEST
+    PyModule_AddIntConstant(m, "RDS_GET_MR_FOR_DEST", RDS_GET_MR_FOR_DEST);
+#endif
 #ifdef  IPPROTO_IP
     PyModule_AddIntConstant(m, "IPPROTO_IP", IPPROTO_IP);
 #else
@@ -6261,6 +6300,44 @@ PyInit__socket(void)
     PyModule_AddIntConstant(m, "IPX_TYPE", IPX_TYPE);
 #endif
 
+/* Reliable Datagram Sockets */
+#ifdef RDS_CMSG_RDMA_ARGS
+    PyModule_AddIntConstant(m, "RDS_CMSG_RDMA_ARGS", RDS_CMSG_RDMA_ARGS);
+#endif
+#ifdef RDS_CMSG_RDMA_DEST
+    PyModule_AddIntConstant(m, "RDS_CMSG_RDMA_DEST", RDS_CMSG_RDMA_DEST);
+#endif
+#ifdef RDS_CMSG_RDMA_MAP
+    PyModule_AddIntConstant(m, "RDS_CMSG_RDMA_MAP", RDS_CMSG_RDMA_MAP);
+#endif
+#ifdef RDS_CMSG_RDMA_STATUS
+    PyModule_AddIntConstant(m, "RDS_CMSG_RDMA_STATUS", RDS_CMSG_RDMA_STATUS);
+#endif
+#ifdef RDS_CMSG_RDMA_UPDATE
+    PyModule_AddIntConstant(m, "RDS_CMSG_RDMA_UPDATE", RDS_CMSG_RDMA_UPDATE);
+#endif
+#ifdef RDS_RDMA_READWRITE
+    PyModule_AddIntConstant(m, "RDS_RDMA_READWRITE", RDS_RDMA_READWRITE);
+#endif
+#ifdef RDS_RDMA_FENCE
+    PyModule_AddIntConstant(m, "RDS_RDMA_FENCE", RDS_RDMA_FENCE);
+#endif
+#ifdef RDS_RDMA_INVALIDATE
+    PyModule_AddIntConstant(m, "RDS_RDMA_INVALIDATE", RDS_RDMA_INVALIDATE);
+#endif
+#ifdef RDS_RDMA_USE_ONCE
+    PyModule_AddIntConstant(m, "RDS_RDMA_USE_ONCE", RDS_RDMA_USE_ONCE);
+#endif
+#ifdef RDS_RDMA_DONTWAIT
+    PyModule_AddIntConstant(m, "RDS_RDMA_DONTWAIT", RDS_RDMA_DONTWAIT);
+#endif
+#ifdef RDS_RDMA_NOTIFY_ME
+    PyModule_AddIntConstant(m, "RDS_RDMA_NOTIFY_ME", RDS_RDMA_NOTIFY_ME);
+#endif
+#ifdef RDS_RDMA_SILENT
+    PyModule_AddIntConstant(m, "RDS_RDMA_SILENT", RDS_RDMA_SILENT);
+#endif
+
     /* get{addr,name}info parameters */
 #ifdef EAI_ADDRFAMILY
     PyModule_AddIntConstant(m, "EAI_ADDRFAMILY", EAI_ADDRFAMILY);