udp: fix for multiple workers and add test

Since the main thread is not used for session polling anymore, when vpp
is started with multiple wokers, allocate connections on the first. Also
add a simple udp make test.

Change-Id: Id869f5d89e0fced51048f0384fa86a5022258b7c
Signed-off-by: Florin Coras <fcoras@cisco.com>

diff --git a/src/vnet/session-apps/echo_client.c b/src/vnet/session-apps/echo_client.c
index 1d010f2..6ee91f9 100644 (file)
--- a/src/vnet/session-apps/echo_client.c
+++ b/src/vnet/session-apps/echo_client.c
@@ -369,7 +369,7 @@ echo_clients_session_connected_callback (u32 app_index, u32 api_context,
   echo_client_main_t *ecm = &echo_client_main;
   eclient_session_t *session;
   u32 session_index;
-  u8 thread_index = vlib_get_thread_index ();
+  u8 thread_index = s->thread_index;
 
   if (is_fail)
     {
@@ -378,7 +378,8 @@ echo_clients_session_connected_callback (u32 app_index, u32 api_context,
       return 0;
     }
 
-  ASSERT (s->thread_index == thread_index);
+  ASSERT (thread_index == vlib_get_thread_index ()
+         || session_transport_service_type (s) == TRANSPORT_SERVICE_CL);
 
   if (!ecm->vpp_event_queue[thread_index])
     ecm->vpp_event_queue[thread_index] =

diff --git a/src/vnet/session/session.c b/src/vnet/session/session.c
index ee02526..cfba31e 100644 (file)
--- a/src/vnet/session/session.c
+++ b/src/vnet/session/session.c
@@ -142,7 +142,8 @@ session_alloc_for_connection (transport_connection_t * tc)
   stream_session_t *s;
   u32 thread_index = tc->thread_index;
 
-  ASSERT (thread_index == vlib_get_thread_index ());
+  ASSERT (thread_index == vlib_get_thread_index ()
+         || transport_protocol_is_cl (tc->proto));
 
   s = session_alloc (thread_index);
   s->session_type = session_type_from_proto_and_ip (tc->proto, tc->is_ip4);

diff --git a/src/vnet/session/transport.c b/src/vnet/session/transport.c
index b3d42d0..a401190 100644 (file)
--- a/src/vnet/session/transport.c
+++ b/src/vnet/session/transport.c
@@ -207,6 +207,12 @@ transport_protocol_tx_fn_type (transport_proto_t tp)
   return tp_vfts[tp].tx_type;
 }
 
+u8
+transport_protocol_is_cl (transport_proto_t tp)
+{
+  return (tp_vfts[tp].service_type == TRANSPORT_SERVICE_CL);
+}
+
 #define PORT_MASK ((1 << 16)- 1)
 
 void

diff --git a/src/vnet/session/transport.h b/src/vnet/session/transport.h
index 8340fd8..e29f3ca 100644 (file)
--- a/src/vnet/session/transport.h
+++ b/src/vnet/session/transport.h
@@ -131,6 +131,7 @@ int transport_alloc_local_endpoint (u8 proto, transport_endpoint_t * rmt,
                                    ip46_address_t * lcl_addr,
                                    u16 * lcl_port);
 void transport_endpoint_cleanup (u8 proto, ip46_address_t * lcl_ip, u16 port);
+u8 transport_protocol_is_cl (transport_proto_t tp);
 void transport_init (void);
 
 #endif /* VNET_VNET_URI_TRANSPORT_H_ */

diff --git a/src/vnet/udp/udp.c b/src/vnet/udp/udp.c
index 947cc1e..888a6cb 100644 (file)
--- a/src/vnet/udp/udp.c
+++ b/src/vnet/udp/udp.c
@@ -288,6 +288,10 @@ udp_open_connection (transport_endpoint_t * rmt)
   node_index = rmt->is_ip4 ? udp4_input_node.index : udp6_input_node.index;
   udp_register_dst_port (vm, lcl_port, node_index, 1 /* is_ipv4 */ );
 
+  /* We don't poll main thread if we have workers */
+  if (vlib_num_workers ())
+    thread_index = 1;
+
   uc = udp_connection_alloc (thread_index);
   ip_copy (&uc->c_rmt_ip, &rmt->ip, rmt->is_ip4);
   ip_copy (&uc->c_lcl_ip, &lcl_addr, rmt->is_ip4);
@@ -301,10 +305,14 @@ udp_open_connection (transport_endpoint_t * rmt)
 }
 
 transport_connection_t *
-udp_half_open_session_get_transport (u32 conn_index)
+udp_session_get_half_open (u32 conn_index)
 {
   udp_connection_t *uc;
-  uc = udp_connection_get (conn_index, vlib_get_thread_index ());
+  u32 thread_index;
+
+  /* We don't poll main thread if we have workers */
+  thread_index = vlib_num_workers ()? 1 : 0;
+  uc = udp_connection_get (conn_index, thread_index);
   return &uc->connection;
 }
 
@@ -316,7 +324,7 @@ const static transport_proto_vft_t udp_proto = {
   .push_header = udp_push_header,
   .get_connection = udp_session_get,
   .get_listener = udp_session_get_listener,
-  .get_half_open = udp_half_open_session_get_transport,
+  .get_half_open = udp_session_get_half_open,
   .close = udp_session_close,
   .cleanup = udp_session_cleanup,
   .send_mss = udp_send_mss,
@@ -360,7 +368,7 @@ const static transport_proto_vft_t udpc_proto = {
   .push_header = udp_push_header,
   .get_connection = udp_session_get,
   .get_listener = udp_session_get_listener,
-  .get_half_open = udp_half_open_session_get_transport,
+  .get_half_open = udp_session_get_half_open,
   .close = udp_session_close,
   .cleanup = udp_session_cleanup,
   .send_mss = udp_send_mss,

diff --git a/test/test_udp.py b/test/test_udp.py
index 322d813..aabbbd3 100644 (file)
--- a/test/test_udp.py
+++ b/test/test_udp.py
@@ -223,5 +223,79 @@ class TestUdpEncap(VppTestCase):
             self.validate_inner4(p, p_4omo4, ttl=63)
 
 
+class TestUDP(VppTestCase):
+    """ UDP Test Case """
+
+    @classmethod
+    def setUpClass(cls):
+        super(TestUDP, cls).setUpClass()
+
+    def setUp(self):
+        super(TestUDP, self).setUp()
+        self.vapi.session_enable_disable(is_enabled=1)
+        self.create_loopback_interfaces(range(2))
+
+        table_id = 0
+
+        for i in self.lo_interfaces:
+            i.admin_up()
+
+            if table_id != 0:
+                tbl = VppIpTable(self, table_id)
+                tbl.add_vpp_config()
+
+            i.set_table_ip4(table_id)
+            i.config_ip4()
+            table_id += 1
+
+        # Configure namespaces
+        self.vapi.app_namespace_add(namespace_id="0",
+                                    sw_if_index=self.loop0.sw_if_index)
+        self.vapi.app_namespace_add(namespace_id="1",
+                                    sw_if_index=self.loop1.sw_if_index)
+
+    def tearDown(self):
+        for i in self.lo_interfaces:
+            i.unconfig_ip4()
+            i.set_table_ip4(0)
+            i.admin_down()
+        self.vapi.session_enable_disable(is_enabled=0)
+        super(TestUDP, self).tearDown()
+
+    def test_udp_transfer(self):
+        """ UDP echo client/server transfer """
+
+        # Add inter-table routes
+        ip_t01 = VppIpRoute(self, self.loop1.local_ip4, 32,
+                            [VppRoutePath("0.0.0.0",
+                                          0xffffffff,
+                                          nh_table_id=1)])
+        ip_t10 = VppIpRoute(self, self.loop0.local_ip4, 32,
+                            [VppRoutePath("0.0.0.0",
+                                          0xffffffff,
+                                          nh_table_id=0)], table_id=1)
+        ip_t01.add_vpp_config()
+        ip_t10.add_vpp_config()
+
+        # Start builtin server and client
+        uri = "udp://" + self.loop0.local_ip4 + "/1234"
+        error = self.vapi.cli("test echo server appns 0 fifo-size 4 no-echo" +
+                              "uri " + uri)
+        if error:
+            self.logger.critical(error)
+            self.assertEqual(error.find("failed"), -1)
+
+        error = self.vapi.cli("test echo client mbytes 10 appns 1 " +
+                              "fifo-size 4 no-output test-bytes " +
+                              "syn-timeout 2 no-return uri " + uri)
+        if error:
+            self.logger.critical(error)
+            self.assertEqual(error.find("failed"), -1)
+
+        # Delete inter-table routes
+        ip_t01.remove_vpp_config()
+        ip_t10.remove_vpp_config()
+
+
 if __name__ == '__main__':
     unittest.main(testRunner=VppTestRunner)