python/ovs/stream: teach stream.py tcp socket

Signed-off-by: Isaku Yamahata <yamahata@valinux.co.jp>
Signed-off-by: Ben Pfaff <blp@nicira.com>

python/ovs/stream.py

@@ -51,12 +51,28 @@ class Stream(object):
     W_RECV = 1                  # Data received.
     W_SEND = 2                  # Send buffer room available.
 
+    _SOCKET_METHODS = {}
+
+    @staticmethod
+    def register_method(method):
+        def _register_method(cls):
+            Stream._SOCKET_METHODS[method + ":"] = cls
+            return cls
+        return _register_method
+
+    @staticmethod
+    def _find_method(name):
+        for method, cls in Stream._SOCKET_METHODS.items():
+            if name.startswith(method):
+                return cls
+        return None
+
     @staticmethod
     def is_valid_name(name):
         """Returns True if 'name' is a stream name in the form "TYPE:ARGS" and
-        TYPE is a supported stream type (currently only "unix:"), otherwise
+        TYPE is a supported stream type (currently only "unix:" and "tcp:"),
-        False."""
+        otherwise False."""
-        return name.startswith("unix:")
+        return bool(Stream._find_method(name))
 
     def __init__(self, socket, name, status):
         self.socket = socket
@@ -70,12 +86,18 @@ class Stream(object):
 
         self.error = 0
 
+    # Default value of dscp bits for connection between controller and manager.
+    # Value of IPTOS_PREC_INTERNETCONTROL = 0xc0 which is defined
+    # in <netinet/ip.h> is used.
+    IPTOS_PREC_INTERNETCONTROL = 0xc0
+    DSCP_DEFAULT = IPTOS_PREC_INTERNETCONTROL >> 2
+
     @staticmethod
-    def open(name):
+    def open(name, dscp=DSCP_DEFAULT):
         """Attempts to connect a stream to a remote peer.  'name' is a
         connection name in the form "TYPE:ARGS", where TYPE is an active stream
         class's name and ARGS are stream class-specific.  Currently the only
-        supported TYPE is "unix".
+        supported TYPEs are "unix" and "tcp".
 
         Returns (error, stream): on success 'error' is 0 and 'stream' is the
         new Stream, on failure 'error' is a positive errno value and 'stream'
@@ -84,19 +106,22 @@ class Stream(object):
         Never returns errno.EAGAIN or errno.EINPROGRESS.  Instead, returns 0
         and a new Stream.  The connect() method can be used to check for
         successful connection completion."""
-        if not Stream.is_valid_name(name):
+        cls = Stream._find_method(name)
+        if not cls:
             return errno.EAFNOSUPPORT, None
 
-        connect_path = name[5:]
+        suffix = name.split(":", 1)[1]
-        error, sock = ovs.socket_util.make_unix_socket(socket.SOCK_STREAM,
+        error, sock = cls._open(suffix, dscp)
-                                                       True, None,
-                                                       connect_path)
         if error:
             return error, None
         else:
             status = ovs.socket_util.check_connection_completion(sock)
             return 0, Stream(sock, name, status)
 
+    @staticmethod
+    def _open(suffix, dscp):
+        raise NotImplementedError("This method must be overrided by subclass")
+
     @staticmethod
     def open_block((error, stream)):
         """Blocks until a Stream completes its connection attempt, either
@@ -313,6 +338,27 @@ def usage(name):
     return """
 Active %s connection methods:
   unix:FILE               Unix domain socket named FILE
+  tcp:IP:PORT             TCP socket to IP with port no of PORT
 
 Passive %s connection methods:
   punix:FILE              Listen on Unix domain socket FILE""" % (name, name)
+
+
+@Stream.register_method("unix")
+class UnixStream(Stream):
+    @staticmethod
+    def _open(suffix, dscp):
+        connect_path = suffix
+        return  ovs.socket_util.make_unix_socket(socket.SOCK_STREAM,
+                                                 True, None, connect_path)
+
+
+@Stream.register_method("tcp")
+class TCPStream(Stream):
+    @staticmethod
+    def _open(suffix, dscp):
+        error, sock = ovs.socket_util.inet_open_active(socket.SOCK_STREAM,
+                                                       suffix, 0, dscp)
+        if not error:
+            sock.setsockopt(socket.IPPROTO_TCP, socket.TCP_NODELAY, 1)
+        return error, sock

tests/ovsdb-idl.at

@@ -48,9 +48,29 @@ m4_define([OVSDB_CHECK_IDL_PY],
    OVSDB_SERVER_SHUTDOWN
    AT_CLEANUP])
 
+# same as OVSDB_CHECK_IDL but uses the Python IDL implementation with tcp
+m4_define([OVSDB_CHECK_IDL_TCP_PY],
+  [AT_SETUP([$1 - Python tcp])
+   AT_SKIP_IF([test $HAVE_PYTHON = no])
+   AT_KEYWORDS([ovsdb server idl positive Python with tcp socket $5])
+   AT_CHECK([ovsdb-tool create db $abs_srcdir/idltest.ovsschema],
+                  [0], [stdout], [ignore])
+   AT_CHECK([perl $srcdir/choose-port.pl], [0], [stdout])
+   TCP_PORT=`cat stdout`
+   AT_CHECK([ovsdb-server '-vPATTERN:console:ovsdb-server|%c|%m' --detach --no-chdir --pidfile="`pwd`"/pid --remote=ptcp:$TCP_PORT:127.0.0.1 --unixctl="`pwd`"/unixctl db], [0], [ignore], [ignore])
+   m4_if([$2], [], [],
+     [AT_CHECK([ovsdb-client transact tcp:127.0.0.1:$TCP_PORT $2], [0], [ignore], [ignore], [kill `cat pid`])])
+   AT_CHECK([$PYTHON $srcdir/test-ovsdb.py  -t10 idl $srcdir/idltest.ovsschema tcp:127.0.0.1:$TCP_PORT $3],
+            [0], [stdout], [ignore], [kill `cat pid`])
+   AT_CHECK([sort stdout | perl $srcdir/uuidfilt.pl]m4_if([$6],,, [[| $6]]),
+            [0], [$4], [], [kill `cat pid`])
+   OVSDB_SERVER_SHUTDOWN
+   AT_CLEANUP])
+
 m4_define([OVSDB_CHECK_IDL],
   [OVSDB_CHECK_IDL_C($@)
-   OVSDB_CHECK_IDL_PY($@)])
+   OVSDB_CHECK_IDL_PY($@)
+   OVSDB_CHECK_IDL_TCP_PY($@)])
 
 OVSDB_CHECK_IDL([simple idl, initially empty, no ops],
   [],