qpid-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From acon...@apache.org
Subject qpid-proton git commit: PROTON-1403: C libuv proactor fixes and tests
Date Tue, 21 Feb 2017 01:04:52 GMT
Repository: qpid-proton
Updated Branches:
  refs/heads/master 6b6dd8699 -> 204c8474d


PROTON-1403: C libuv proactor fixes and tests

- wake before connection prevented connection
- early errors (e.g. bad host/port) not handled correctly
- INACTIVE event not generated on listener closed


Project: http://git-wip-us.apache.org/repos/asf/qpid-proton/repo
Commit: http://git-wip-us.apache.org/repos/asf/qpid-proton/commit/204c8474
Tree: http://git-wip-us.apache.org/repos/asf/qpid-proton/tree/204c8474
Diff: http://git-wip-us.apache.org/repos/asf/qpid-proton/diff/204c8474

Branch: refs/heads/master
Commit: 204c8474d2b34f640189eb5c3063622b45fb5cc9
Parents: 6b6dd86
Author: Alan Conway <aconway@redhat.com>
Authored: Mon Feb 20 19:56:19 2017 -0500
Committer: Alan Conway <aconway@redhat.com>
Committed: Mon Feb 20 20:01:54 2017 -0500

----------------------------------------------------------------------
 proton-c/src/proactor/libuv.c   |  28 ++++----
 proton-c/src/tests/proactor.c   | 133 +++++++++++++++++++----------------
 proton-c/src/tests/test_tools.h |  14 ++++
 3 files changed, 102 insertions(+), 73 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/qpid-proton/blob/204c8474/proton-c/src/proactor/libuv.c
----------------------------------------------------------------------
diff --git a/proton-c/src/proactor/libuv.c b/proton-c/src/proactor/libuv.c
index c7322cd..61239ac 100644
--- a/proton-c/src/proactor/libuv.c
+++ b/proton-c/src/proactor/libuv.c
@@ -235,11 +235,8 @@ static void to_leader(psocket_t *ps, void (*action)(psocket_t*)) {
 /* Push to the worker thread */
 static void to_worker(psocket_t *ps) {
   uv_mutex_lock(&ps->proactor->lock);
-  /* If already ON_WORKER do nothing */
-  if (ps->next == &UNLISTED && ps->state != ON_WORKER) {
-    ps->state = ON_WORKER;
-    push_lh(&ps->proactor->worker_q, ps);
-  }
+  ps->state = ON_WORKER;
+  push_lh(&ps->proactor->worker_q, ps);
   uv_mutex_unlock(&ps->proactor->lock);
 }
 
@@ -310,7 +307,10 @@ static inline pconnection_t *batch_pconnection(pn_event_batch_t *batch)
{
 static void leader_count(pn_proactor_t *p, int change) {
   uv_mutex_lock(&p->lock);
   p->count += change;
-  p->inactive = (p->count == 0);
+  if (p->count == 0) {
+    p->inactive = true;
+    uv_async_send(&p->async); /* Wake leader */
+  }
   uv_mutex_unlock(&p->lock);
 }
 
@@ -329,8 +329,8 @@ static void on_close_pconnection_final(uv_handle_t *h) {
 /* Close event for uv_tcp_t of a psocket_t */
 static void on_close_psocket(uv_handle_t *h) {
   psocket_t *ps = (psocket_t*)h->data;
+  leader_count(ps->proactor, -1);
   if (ps->is_conn) {
-    leader_count(ps->proactor, -1);
     pconnection_t *pc = as_pconnection(ps);
     uv_timer_stop(&pc->timer);
     /* Delay the free till the timer handle is also closed */
@@ -465,7 +465,7 @@ static void leader_connect(psocket_t *ps) {
   if (!err) {
     ps->state = ON_UV;
   } else {
-    psocket_error(ps, err, "connecting to");
+    pconnection_error(pc, err, "connecting to");
   }
 }
 
@@ -737,16 +737,16 @@ static void leader_process_lh(pn_proactor_t *p) {
   }
   for (psocket_t *ps = pop_lh(&p->leader_q); ps; ps = pop_lh(&p->leader_q))
{
     assert(ps->state == ON_LEADER);
-    if (ps->wakeup) {
-      uv_mutex_unlock(&p->lock);
-      ps->wakeup(ps);
-      ps->wakeup = NULL;
-      uv_mutex_lock(&p->lock);
-    } else if (ps->action) {
+    if (ps->action) {
       uv_mutex_unlock(&p->lock);
       ps->action(ps);
       ps->action = NULL;
       uv_mutex_lock(&p->lock);
+    } else if (ps->wakeup) {
+      uv_mutex_unlock(&p->lock);
+      ps->wakeup(ps);
+      ps->wakeup = NULL;
+      uv_mutex_lock(&p->lock);
     }
   }
 }

http://git-wip-us.apache.org/repos/asf/qpid-proton/blob/204c8474/proton-c/src/tests/proactor.c
----------------------------------------------------------------------
diff --git a/proton-c/src/tests/proactor.c b/proton-c/src/tests/proactor.c
index 9f8b1fe..6b95c11 100644
--- a/proton-c/src/tests/proactor.c
+++ b/proton-c/src/tests/proactor.c
@@ -32,6 +32,11 @@ static pn_millis_t timeout = 5*1000; /* timeout for hanging tests */
 
 static const char *localhost = "127.0.0.1"; /* host for connect/listen */
 
+struct test_events {
+  pn_proactor_t *proactor;
+  pn_event_batch_t *events;
+};
+
 /* Wait for the next single event, return its type */
 static pn_event_type_t wait_next(pn_proactor_t *proactor) {
   pn_event_batch_t *events = pn_proactor_wait(proactor);
@@ -121,49 +126,40 @@ int proactor_test_run(proactor_test_t *pts, size_t n) {
 }
 
 
-/* Simple test of client connect to a listening server */
-handler_state_t listen_connect_server(test_t *t, pn_event_t *e) {
-  switch (pn_event_type(e)) {
-    /* Ignore these events */
-   case PN_LISTENER_OPEN:
-   case PN_CONNECTION_LOCAL_OPEN:
-   case PN_CONNECTION_REMOTE_OPEN:
-   case PN_CONNECTION_BOUND:
-    return H_CONTINUE;
-
-    /* Act on these events */
-   case PN_LISTENER_ACCEPT:
-     pn_listener_accept(pn_event_listener(e), pn_connection());
-     return H_CONTINUE;
-   case PN_CONNECTION_INIT:
-    pn_connection_open(pn_event_connection(e));
-    return H_CONTINUE;
-   case PN_CONNECTION_REMOTE_CLOSE:
-    return H_FINISHED;
+/* Handler for test_listen_connect, does both sides of the connection */
+handler_state_t listen_connect_handler(test_t *t, pn_event_t *e) {
+  pn_connection_t *c = pn_event_connection(e);
+  pn_listener_t *l = pn_event_listener(e);
 
-   default:
-    TEST_CHECK(t, false, "unexpected event %s", pn_event_type_name(pn_event_type(e)));
-    return H_FAILED;
-    break;
-  }
-}
-
-handler_state_t listen_connect_client(test_t *t, pn_event_t *e) {
   switch (pn_event_type(e)) {
     /* Ignore these events */
    case PN_CONNECTION_LOCAL_OPEN:
    case PN_CONNECTION_BOUND:
+   case PN_CONNECTION_INIT:
     return H_CONTINUE;
 
     /* Act on these events */
-   case PN_CONNECTION_INIT:
-    pn_connection_open(pn_event_connection(e));
+   case PN_LISTENER_ACCEPT: {
+    pn_connection_t *accepted = pn_connection();
+    pn_connection_open(accepted);
+    pn_listener_accept(l, accepted); /* Listener takes ownership of accepted */
     return H_CONTINUE;
+   }
+
    case PN_CONNECTION_REMOTE_OPEN:
-    pn_connection_close(pn_event_connection(e));
+    if (pn_connection_state(c) | PN_LOCAL_ACTIVE) { /* Client is fully open - the test is
done */
+      pn_connection_close(c);
+      return H_FINISHED;
+    }  else {                   /* Server returns the open */
+      pn_connection_open(c);
+    }
+
+   case PN_CONNECTION_REMOTE_CLOSE:
+    if (pn_connection_state(c) | PN_LOCAL_ACTIVE) {
+      pn_connection_close(c);    /* Return the close */
+    }
     return H_FINISHED;
 
-    /* Unexpected events */
    default:
     TEST_CHECK(t, false, "unexpected event %s", pn_event_type_name(pn_event_type(e)));
     return H_FAILED;
@@ -171,38 +167,18 @@ handler_state_t listen_connect_client(test_t *t, pn_event_t *e) {
   }
 }
 
-/* Simplest client/server interaction */
-static void test_listen_connect(test_t *t) {
-  proactor_test_t pts[] =  { { t, listen_connect_client }, { t, listen_connect_server } };
-  proactor_test_t *client = &pts[0], *server = &pts[1];
-  proactor_test_init(pts, 2);
-
-  sock_t sock = sock_bind0();          /* Hold a port */
-  char port_str[16];
-  snprintf(port_str, sizeof(port_str), "%d", sock_port(sock));
-  pn_proactor_listen(server->proactor, pn_listener(), localhost, port_str, 4);
-  pn_event_type_t etype = wait_for(server->proactor, PN_LISTENER_OPEN);
-  if (TEST_CHECK(t, PN_LISTENER_OPEN == etype, pn_event_type_name(etype))) {
-    pn_proactor_connect(client->proactor, pn_connection(), localhost, port_str);
-    proactor_test_run(pts, 2);
-  }
-  sock_close(sock);
-  pn_proactor_free(client->proactor);
-  pn_proactor_free(server->proactor);
-}
-
-/* Test error handling */
-static void test_listen_connect_error(test_t *t) {
+/* Test bad-address error handling for listen and connect */
+static void test_early_error(test_t *t) {
   pn_proactor_t *p = pn_proactor();
   pn_proactor_set_timeout(p, timeout); /* In case of hang */
   pn_connection_t *c = pn_connection();
-  pn_proactor_connect(p, c, "nosuchost", "nosuchport");
+  pn_proactor_connect(p, c, "badhost", "amqp");
   pn_event_type_t etype = wait_for(p, PN_TRANSPORT_CLOSED);
   TEST_CHECK(t, PN_TRANSPORT_CLOSED == etype, pn_event_type_name(etype));
   TEST_CHECK(t, pn_condition_is_set(pn_transport_condition(pn_connection_transport(c))),
"");
 
   pn_listener_t *l = pn_listener();
-  pn_proactor_listen(p, l, "nosuchost", "nosuchport", 1);
+  pn_proactor_listen(p, l, "badhost", "amqp", 1);
   etype = wait_for(p, PN_LISTENER_CLOSE);
   TEST_CHECK(t, PN_LISTENER_CLOSE == etype, pn_event_type_name(etype));
   TEST_CHECK(t, pn_condition_is_set(pn_listener_condition(l)), "");
@@ -210,12 +186,51 @@ static void test_listen_connect_error(test_t *t) {
   pn_proactor_free(p);
 }
 
+/* Simplest client/server interaction with 2 proactors */
+static void test_listen_connect(test_t *t) {
+  proactor_test_t pts[] =  { { t, listen_connect_handler }, { t, listen_connect_handler }
};
+  proactor_test_init(pts, 2);
+  pn_proactor_t *client = pts[0].proactor, *server = pts[1].proactor;
+  test_port_t port = test_port();          /* Hold a port */
+
+  pn_proactor_listen(server, pn_listener(), localhost, port.str, 4);
+  pn_event_type_t etype = wait_for(server, PN_LISTENER_OPEN);
+  if (TEST_CHECK(t, PN_LISTENER_OPEN == etype, pn_event_type_name(etype))) {
+    sock_close(port.sock);
+    pn_proactor_connect(client, pn_connection(), localhost, port.str);
+    proactor_test_run(pts, 2);
+  }
+  pn_proactor_free(client);
+  pn_proactor_free(server);
+}
+
+/* Test that INACTIVE event is generated when last connections/listeners closes. */
+static void test_inactive(test_t *t) {
+  proactor_test_t pts[] =  { { t, listen_connect_handler }, { t, listen_connect_handler }};
+  proactor_test_init(pts, 2);
+  pn_proactor_t *client = pts[0].proactor, *server = pts[1].proactor;
+  test_port_t port = test_port();          /* Hold a port */
+
+  pn_listener_t *l = pn_listener();
+  pn_proactor_listen(server, l, localhost, port.str,  4);
+  pn_event_type_t etype = wait_for(server, PN_LISTENER_OPEN);
+  if (TEST_CHECK(t, PN_LISTENER_OPEN == etype, pn_event_type_name(etype))) {
+    sock_close(port.sock);
+    pn_proactor_connect(client, pn_connection(), localhost, port.str);
+    proactor_test_run(pts, 2);
+    etype = wait_for(client, PN_PROACTOR_INACTIVE);
+    pn_listener_close(l);
+    etype = wait_for(server, PN_PROACTOR_INACTIVE);
+  }
+  pn_proactor_free(client);
+  pn_proactor_free(server);
+}
+
 int main(int argv, char** argc) {
   int failed = 0;
-  if (0) {
-    RUN_TEST(failed, t, test_interrupt_timeout(&t));
-    RUN_TEST(failed, t, test_listen_connect_error(&t));
-  }
+  RUN_TEST(failed, t, test_inactive(&t));
+  RUN_TEST(failed, t, test_interrupt_timeout(&t));
+  RUN_TEST(failed, t, test_early_error(&t));
   RUN_TEST(failed, t, test_listen_connect(&t));
   return failed;
 }

http://git-wip-us.apache.org/repos/asf/qpid-proton/blob/204c8474/proton-c/src/tests/test_tools.h
----------------------------------------------------------------------
diff --git a/proton-c/src/tests/test_tools.h b/proton-c/src/tests/test_tools.h
index 2619337..2663dcf 100644
--- a/proton-c/src/tests/test_tools.h
+++ b/proton-c/src/tests/test_tools.h
@@ -153,4 +153,18 @@ static int sock_port(sock_t sock) {
   return ntohs(port);
 }
 
+typedef struct test_port_t {
+  sock_t sock;
+  int port;
+  char str[256];
+} test_port_t;
+
+static inline test_port_t test_port(void) {
+  test_port_t tp = {0};
+  tp.sock = sock_bind0();
+  tp.port = sock_port(tp.sock);
+  snprintf(tp.str, sizeof(tp.str), "%d", tp.port);
+  return tp;
+}
+
 #endif // TESTS_TEST_TOOLS_H


---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@qpid.apache.org
For additional commands, e-mail: commits-help@qpid.apache.org


Mime
View raw message