initial portal code

8 files changed, 758 insertions, 0 deletions

diff --git a/vespalib/src/tests/portal/CMakeLists.txt b/vespalib/src/tests/portal/CMakeLists.txt
new file mode 100644
index 00000000000..3c39f286a15
--- /dev/null
+++ b/vespalib/src/tests/portal/CMakeLists.txt
@@ -0,0 +1,8 @@
+# Copyright 2018 Yahoo Holdings. Licensed under the terms of the Apache 2.0 license. See LICENSE in the project root.
+vespa_add_executable(vespalib_portal_test_app TEST
+    SOURCES
+    portal_test.cpp
+    DEPENDS
+    vespalib
+)
+vespa_add_test(NAME vespalib_portal_test_app COMMAND vespalib_portal_test_app)
diff --git a/vespalib/src/tests/portal/handle_manager/CMakeLists.txt b/vespalib/src/tests/portal/handle_manager/CMakeLists.txt
new file mode 100644
index 00000000000..b12b72f9ad6
--- /dev/null
+++ b/vespalib/src/tests/portal/handle_manager/CMakeLists.txt
@@ -0,0 +1,8 @@
+# Copyright 2018 Yahoo Holdings. Licensed under the terms of the Apache 2.0 license. See LICENSE in the project root.
+vespa_add_executable(vespalib_handle_manager_test_app TEST
+    SOURCES
+    handle_manager_test.cpp
+    DEPENDS
+    vespalib
+)
+vespa_add_test(NAME vespalib_handle_manager_test_app COMMAND vespalib_handle_manager_test_app)
diff --git a/vespalib/src/tests/portal/handle_manager/handle_manager_test.cpp b/vespalib/src/tests/portal/handle_manager/handle_manager_test.cpp
new file mode 100644
index 00000000000..4ce25aa0a7a
--- /dev/null
+++ b/vespalib/src/tests/portal/handle_manager/handle_manager_test.cpp
@@ -0,0 +1,146 @@
+// Copyright 2018 Yahoo Holdings. Licensed under the terms of the Apache 2.0 license. See LICENSE in the project root.
+
+#include <vespa/vespalib/testkit/test_kit.h>
+#include <vespa/vespalib/portal/handle_manager.h>
+#include <vespa/vespalib/util/gate.h>
+
+#include <thread>
+#include <chrono>
+#include <atomic>
+
+using namespace vespalib;
+using namespace vespalib::portal;
+
+TEST_F("require that handles can be created, locked and destroyed", TimeBomb(60)) {
+    HandleManager manager;
+    uint64_t handle = manager.create();
+    EXPECT_TRUE(handle != manager.null_handle());
+    {
+        HandleGuard guard = manager.lock(handle);
+        EXPECT_TRUE(guard.valid());
+        EXPECT_EQUAL(guard.handle(), handle);
+    }
+    EXPECT_TRUE(manager.destroy(handle));
+    {
+        HandleGuard guard = manager.lock(handle);
+        EXPECT_TRUE(!guard.valid());
+        EXPECT_EQUAL(guard.handle(), manager.null_handle());
+    }
+}
+
+TEST_F("require that multiple guards can be taken for the same handle", TimeBomb(60)) {
+    HandleManager manager;
+    uint64_t handle = manager.create();
+    EXPECT_TRUE(handle != manager.null_handle());
+    {
+        HandleGuard guard1 = manager.lock(handle);
+        HandleGuard guard2 = manager.lock(handle); // <- does not block
+        EXPECT_TRUE(guard1.valid());
+        EXPECT_EQUAL(guard1.handle(), handle);
+        EXPECT_TRUE(guard2.valid());
+        EXPECT_EQUAL(guard2.handle(), handle);
+    }
+    EXPECT_TRUE(manager.destroy(handle));    
+}
+
+TEST_F("require that handles are independent", TimeBomb(60)) {
+    HandleManager manager;
+    uint64_t handle1 = manager.create();
+    uint64_t handle2 = manager.create();
+    uint64_t handle3 = manager.create();
+    EXPECT_TRUE(handle1 != manager.null_handle());
+    EXPECT_TRUE(handle2 != manager.null_handle());
+    EXPECT_TRUE(handle3 != manager.null_handle());
+    EXPECT_TRUE(handle1 != handle2);
+    EXPECT_TRUE(handle1 != handle3);
+    EXPECT_TRUE(handle2 != handle3);
+    {
+        HandleGuard guard1 = manager.lock(handle1);
+        HandleGuard guard2 = manager.lock(handle2);
+        EXPECT_TRUE(guard1.valid());
+        EXPECT_EQUAL(guard1.handle(), handle1);
+        EXPECT_TRUE(guard2.valid());
+        EXPECT_EQUAL(guard2.handle(), handle2);
+        EXPECT_TRUE(manager.destroy(handle3)); // <- does not block
+        HandleGuard guard3 = manager.lock(handle3);
+        EXPECT_TRUE(!guard3.valid());
+        EXPECT_EQUAL(guard3.handle(), manager.null_handle());
+    }
+    EXPECT_TRUE(manager.destroy(handle1));
+    EXPECT_TRUE(manager.destroy(handle2));
+    EXPECT_TRUE(!manager.destroy(handle3));
+}
+
+struct Fixture {
+    HandleManager manager;
+    uint64_t handle;
+    Gate gate;
+    std::atomic<size_t> cnt1;
+    std::atomic<size_t> cnt2;
+    Fixture() : manager(), handle(manager.create()), gate(), cnt1(0), cnt2(0) {}
+};
+
+TEST_MT_FF("require that destroy waits for active handle guards", 2, Fixture(), TimeBomb(60)) {
+    if (thread_id == 0) {
+        {
+            auto guard = f1.manager.lock(f1.handle);
+            TEST_BARRIER(); // #1
+            EXPECT_TRUE(!f1.gate.await(20));
+        }
+        EXPECT_TRUE(f1.gate.await(60000));
+    } else {
+        TEST_BARRIER(); // #1
+        EXPECT_TRUE(f1.manager.destroy(f1.handle));
+        f1.gate.countDown();
+    }
+}
+
+TEST_MT_FF("require that destroy disables ability to lock handles", 3, Fixture(), TimeBomb(60)) {
+    if (thread_id == 0) {
+        auto guard = f1.manager.lock(f1.handle);
+        ASSERT_TRUE(guard.valid());
+        TEST_BARRIER(); // #1
+        while (f1.cnt1 == 0) {
+            std::this_thread::sleep_for(std::chrono::milliseconds(1));
+        }
+        EXPECT_EQUAL(f1.cnt2, 0u);
+    } else if (thread_id == 1) {
+        TEST_BARRIER(); // #1
+        EXPECT_TRUE(f1.manager.destroy(f1.handle));
+        EXPECT_EQUAL(f1.cnt1, 1u);
+        ++f1.cnt2;
+    } else {
+        TEST_BARRIER(); // #1
+        while (f1.cnt1 == 0) {
+            auto guard = f1.manager.lock(f1.handle);
+            if (guard.valid()) {
+                std::this_thread::sleep_for(std::chrono::milliseconds(1));
+            } else {
+                EXPECT_EQUAL(f1.cnt2, 0u);
+                ++f1.cnt1;
+            }
+        }
+    }
+}
+
+TEST_MT_FF("require that a single destroy call returns true", 10, Fixture(), TimeBomb(60)) {
+    if (thread_id == 0) { // 1 thread here
+        auto guard = f1.manager.lock(f1.handle);
+        ASSERT_TRUE(guard.valid());
+        TEST_BARRIER(); // #1
+        std::this_thread::sleep_for(std::chrono::milliseconds(1));
+    } else { // 'many' threads here
+        TEST_BARRIER(); // #1
+        if (f1.manager.destroy(f1.handle)) {
+            ++f1.cnt1;
+        } else {
+            ++f1.cnt2;
+        }
+    }
+    TEST_BARRIER(); // #2
+    EXPECT_EQUAL(f1.cnt1, 1u);
+    EXPECT_GREATER(num_threads, 5u);
+    EXPECT_EQUAL(f1.cnt2, (num_threads - 2));
+}
+
+TEST_MAIN() { TEST_RUN_ALL(); }
diff --git a/vespalib/src/tests/portal/http_request/CMakeLists.txt b/vespalib/src/tests/portal/http_request/CMakeLists.txt
new file mode 100644
index 00000000000..00dbf356634
--- /dev/null
+++ b/vespalib/src/tests/portal/http_request/CMakeLists.txt
@@ -0,0 +1,8 @@
+# Copyright 2018 Yahoo Holdings. Licensed under the terms of the Apache 2.0 license. See LICENSE in the project root.
+vespa_add_executable(vespalib_portal_http_request_test_app TEST
+    SOURCES
+    http_request_test.cpp
+    DEPENDS
+    vespalib
+)
+vespa_add_test(NAME vespalib_portal_http_request_test_app COMMAND vespalib_portal_http_request_test_app)
diff --git a/vespalib/src/tests/portal/http_request/http_request_test.cpp b/vespalib/src/tests/portal/http_request/http_request_test.cpp
new file mode 100644
index 00000000000..6e1527efa4b
--- /dev/null
+++ b/vespalib/src/tests/portal/http_request/http_request_test.cpp
@@ -0,0 +1,120 @@
+// Copyright 2018 Yahoo Holdings. Licensed under the terms of the Apache 2.0 license. See LICENSE in the project root.
+
+#include <vespa/vespalib/testkit/test_kit.h>
+#include <vespa/vespalib/portal/http_request.h>
+
+using namespace vespalib;
+using namespace vespalib::portal;
+
+vespalib::string simple_req("GET /my/path HTTP/1.1\r\n"
+                            "Host: my.host.com:80\r\n"
+                            "CustomHeader: CustomValue\r\n"
+                            "\r\n123456789");
+size_t simple_req_padding = 9;
+size_t simple_req_size = (simple_req.size() - simple_req_padding);
+
+void verify_simple_req(const HttpRequest &req) {
+    EXPECT_TRUE(!req.need_more_data());
+    EXPECT_TRUE(req.valid());
+    EXPECT_TRUE(req.is_get());
+    EXPECT_EQUAL(req.get_uri(), "/my/path");
+    EXPECT_EQUAL(req.get_header("host"), "my.host.com:80");
+    EXPECT_EQUAL(req.get_header("customheader"), "CustomValue");
+    EXPECT_EQUAL(req.get_header("non-existing-header"), "");
+}
+
+HttpRequest make_request(const vespalib::string &req) {
+    HttpRequest result;
+    ASSERT_EQUAL(result.handle_data(req.data(), req.size()), req.size());
+    ASSERT_TRUE(result.valid());
+    return result;
+}
+
+void verify_invalid_request(const vespalib::string &req) {
+    HttpRequest result;
+    EXPECT_EQUAL(result.handle_data(req.data(), req.size()), req.size());
+    EXPECT_TRUE(!result.need_more_data());
+    EXPECT_TRUE(!result.valid());
+}
+
+TEST("require that request can be parsed in one go") {
+    HttpRequest req;
+    EXPECT_EQUAL(req.handle_data(simple_req.data(), simple_req_size), simple_req_size);
+    TEST_DO(verify_simple_req(req));
+}
+
+TEST("require that trailing data is not consumed") {
+    HttpRequest req;
+    EXPECT_EQUAL(req.handle_data(simple_req.data(), simple_req.size()), simple_req_size);
+    TEST_DO(verify_simple_req(req));
+}
+
+TEST("require that request can be parsed incrementally") {
+    HttpRequest req;
+    size_t chunk = 7;
+    size_t done = 0;
+    while (done < simple_req_size) {
+        size_t expect = std::min(simple_req_size - done, chunk);
+        EXPECT_EQUAL(req.handle_data(simple_req.data() + done, chunk), expect);
+        done += expect;
+    }
+    EXPECT_EQUAL(done, simple_req_size);
+    TEST_DO(verify_simple_req(req));
+}
+
+TEST("require that header continuation is replaced by single space") {
+    auto req = make_request("GET /my/path HTTP/1.1\r\n"
+                            "test: one\r\n"
+                            " two\r\n"
+                            "\tthree\r\n"
+                            "\r\n");
+    EXPECT_EQUAL(req.get_header("test"), "one two three");
+}
+
+TEST("require that duplicate headers are combined as list") {
+    auto req = make_request("GET /my/path HTTP/1.1\r\n"
+                            "test: one\r\n"
+                            "test: two\r\n"
+                            "test: three\r\n"
+                            "\r\n");
+    EXPECT_EQUAL(req.get_header("test"), "one,two,three");
+}
+
+TEST("require that leading and trailing whitespaces are stripped") {
+    auto req = make_request("GET /my/path HTTP/1.1\r\n"
+                            "test:   one  \r\n"
+                            "        , two  \r\n"
+                            "test:   three   \r\n"
+                            "\r\n");
+    EXPECT_EQUAL(req.get_header("test"), "one , two,three");
+}
+
+TEST("require that non-GET requests are detected") {
+    auto req = make_request("POST /my/path HTTP/1.1\r\n"
+                            "\r\n");
+    EXPECT_TRUE(!req.is_get());
+}
+
+TEST("require that request line must contain all relevant parts") {
+    TEST_DO(verify_invalid_request("/my/path HTTP/1.1\r\n"));
+    TEST_DO(verify_invalid_request("GET HTTP/1.1\r\n"));
+    TEST_DO(verify_invalid_request("GET /my/path\r\n"));
+}
+
+TEST("require that first header line cannot be a continuation") {
+    TEST_DO(verify_invalid_request("GET /my/path HTTP/1.1\r\n"
+                                   " two\r\n"));
+}
+
+TEST("require that header name is not allowed to be empty") {
+    TEST_DO(verify_invalid_request("GET /my/path HTTP/1.1\r\n"
+                                   ": value\r\n"));
+}
+
+TEST("require that header line must contain separator") {
+    TEST_DO(verify_invalid_request("GET /my/path HTTP/1.1\r\n"
+                                   "ok-header: ok-value\r\n"
+                                   "missing separator\r\n"));
+}
+
+TEST_MAIN() { TEST_RUN_ALL(); }
diff --git a/vespalib/src/tests/portal/portal_test.cpp b/vespalib/src/tests/portal/portal_test.cpp
new file mode 100644
index 00000000000..cac4b78d3e2
--- /dev/null
+++ b/vespalib/src/tests/portal/portal_test.cpp
@@ -0,0 +1,298 @@
+// Copyright 2018 Yahoo Holdings. Licensed under the terms of the Apache 2.0 license. See LICENSE in the project root.
+
+#include <vespa/vespalib/testkit/test_kit.h>
+#include <vespa/vespalib/portal/portal.h>
+#include <vespa/vespalib/util/exceptions.h>
+#include <vespa/vespalib/util/stringfmt.h>
+#include <vespa/vespalib/net/socket_spec.h>
+#include <vespa/vespalib/net/crypto_engine.h>
+#include <vespa/vespalib/net/sync_crypto_socket.h>
+#include <vespa/vespalib/net/tls/tls_crypto_engine.h>
+#include <vespa/vespalib/net/tls/maybe_tls_crypto_engine.h>
+#include <vespa/vespalib/test/make_tls_options_for_testing.h>
+#include <vespa/vespalib/util/threadstackexecutor.h>
+
+using namespace vespalib;
+
+//-----------------------------------------------------------------------------
+
+vespalib::string do_http(int port, CryptoEngine::SP crypto, const vespalib::string &method, const vespalib::string &uri) {
+    auto socket = SocketSpec::from_port(port).client_address().connect();
+    ASSERT_TRUE(socket.valid());
+    auto conn = SyncCryptoSocket::create(*crypto, std::move(socket), false);
+    vespalib::string http_req = vespalib::make_string("%s %s HTTP/1.1\r\n"
+                                                      "Host: localhost:%d\r\n"
+                                                      "\r\n", method.c_str(), uri.c_str(), port);
+    ASSERT_EQUAL(conn->write(http_req.data(), http_req.size()), ssize_t(http_req.size()));
+    char buf[1024];
+    vespalib::string result;
+    ssize_t res = conn->read(buf, sizeof(buf));
+    while (res > 0) {
+        result.append(vespalib::stringref(buf, res));
+        res = conn->read(buf, sizeof(buf));
+    }
+    ASSERT_EQUAL(res, 0);
+    return result;
+}
+
+vespalib::string fetch(int port, CryptoEngine::SP crypto, const vespalib::string &path) {
+    return do_http(port, std::move(crypto), "GET", path);
+}
+
+//-----------------------------------------------------------------------------
+
+vespalib::string make_expected_response(const vespalib::string &content_type, const vespalib::string &content) {
+    return vespalib::make_string("HTTP/1.1 200 OK\r\n"
+                                 "Connection: close\r\n"
+                                 "Content-Type: %s\r\n"
+                                 "\r\n"
+                                 "%s", content_type.c_str(), content.c_str());
+}
+
+vespalib::string make_expected_error(int code, const vespalib::string &message) {
+    return vespalib::make_string("HTTP/1.1 %d %s\r\n"
+                                 "Connection: close\r\n"
+                                 "\r\n", code, message.c_str());
+}
+
+//-----------------------------------------------------------------------------
+
+struct Encryption {
+    vespalib::string name;
+    CryptoEngine::SP engine;
+    ~Encryption();
+};
+Encryption::~Encryption() = default;
+
+auto null_crypto() { return std::make_shared<NullCryptoEngine>(); }
+auto xor_crypto() { return std::make_shared<XorCryptoEngine>(); }
+auto tls_crypto() { return std::make_shared<TlsCryptoEngine>(vespalib::test::make_tls_options_for_testing()); }
+auto maybe_tls_crypto(bool client_tls) { return std::make_shared<MaybeTlsCryptoEngine>(tls_crypto(), client_tls); }
+
+std::vector<Encryption> crypto_list = {{"no encryption", null_crypto()},
+                                       {"ad-hoc xor", xor_crypto()},
+                                       {"always TLS", tls_crypto()},
+                                       {"maybe TLS; yes", maybe_tls_crypto(true)},
+                                       {"maybe TLS; no", maybe_tls_crypto(false)}};
+
+//-----------------------------------------------------------------------------
+
+struct MyGetHandler : public Portal::GetHandler {
+    std::function<void(Portal::GetRequest)> fun;
+    template <typename F>
+    MyGetHandler(F &&f) : fun(std::move(f)) {}
+    void get(Portal::GetRequest request) const override {
+        fun(std::move(request));
+    }
+    ~MyGetHandler();
+};
+MyGetHandler::~MyGetHandler() = default;
+
+//-----------------------------------------------------------------------------
+
+TEST("require that failed portal listening throws exception") {
+    EXPECT_EXCEPTION(Portal::create(null_crypto(), -37), PortListenException, "-37");
+}
+
+TEST("require that portal can listen to auto-selected port") {
+    auto portal = Portal::create(null_crypto(), 0);
+    EXPECT_GREATER(portal->listen_port(), 0);
+}
+
+TEST("require that simple GET works with various encryption strategies") {
+    vespalib::string path = "/test";
+    vespalib::string type = "application/json";
+    vespalib::string content = "[1,2,3]";
+    MyGetHandler handler([&](Portal::GetRequest request)
+                         {
+                             EXPECT_EQUAL(request.get_uri(), path);
+                             request.respond_with_content(type, content);
+                         });
+    for (const Encryption &crypto: crypto_list) {
+        fprintf(stderr, "... testing simple GET with encryption: '%s'\n", crypto.name.c_str());
+        auto portal = Portal::create(crypto.engine, 0);
+        auto bound = portal->bind(path, handler);
+        auto expect = make_expected_response(type, content);
+        auto result = fetch(portal->listen_port(), crypto.engine, path);
+        EXPECT_EQUAL(result, expect);
+        bound.reset();
+        result = fetch(portal->listen_port(), crypto.engine, path);
+        expect = make_expected_error(404, "Not Found");
+        EXPECT_EQUAL(result, expect);
+    }
+}
+
+//-----------------------------------------------------------------------------
+
+TEST("require that methods other than GET return not implemented error") {
+    auto portal = Portal::create(null_crypto(), 0);
+    auto expect_get = make_expected_error(404, "Not Found");
+    auto expect_other = make_expected_error(501, "Not Implemented");
+    for (const auto &method: {"OPTIONS", "GET", "HEAD", "POST", "PUT", "DELETE", "TRACE", "CONNECT"}) {
+        auto result = do_http(portal->listen_port(), null_crypto(), method, "/test");
+        if (vespalib::string("GET") == method) {
+            EXPECT_EQUAL(result, expect_get);
+        } else {
+            EXPECT_EQUAL(result, expect_other);
+        }
+    }
+}
+
+TEST("require that GET handler can return HTTP error") {
+    vespalib::string path = "/test";
+    auto portal = Portal::create(null_crypto(), 0);
+    auto expect = make_expected_error(123, "My Error");
+    MyGetHandler handler([](Portal::GetRequest request)
+                         {
+                             request.respond_with_error(123, "My Error");
+                         });
+    auto bound = portal->bind(path, handler);
+    auto result = fetch(portal->listen_port(), null_crypto(), path);
+    EXPECT_EQUAL(result, expect);
+}
+
+TEST("require that get requests dropped on the floor returns HTTP error") {
+    vespalib::string path = "/test";
+    auto portal = Portal::create(null_crypto(), 0);
+    auto expect = make_expected_error(500, "Internal Server Error");
+    MyGetHandler handler([](Portal::GetRequest){});
+    auto bound = portal->bind(path, handler);
+    auto result = fetch(portal->listen_port(), null_crypto(), path);
+    EXPECT_EQUAL(result, expect);
+}
+
+struct GetTask : public Executor::Task {
+    Portal::GetRequest request;
+    GetTask(Portal::GetRequest request_in) : request(std::move(request_in)) {}
+    void run() override {
+        request.respond_with_content("text/plain", "hello");
+    }
+};
+
+TEST("require that GET requests can be completed in another thread") {
+    vespalib::string path = "/test";
+    ThreadStackExecutor executor(1, 128 * 1024);
+    auto portal = Portal::create(null_crypto(), 0);
+    auto expect = make_expected_response("text/plain", "hello");
+    MyGetHandler handler([&executor](Portal::GetRequest request)
+                         {
+                             executor.execute(std::make_unique<GetTask>(std::move(request)));
+                         });
+    auto bound = portal->bind(path, handler);
+    auto result = fetch(portal->listen_port(), null_crypto(), path);
+    EXPECT_EQUAL(result, expect);
+    executor.shutdown().sync();
+}
+
+TEST("require that bogus request returns HTTP error") {
+    auto portal = Portal::create(null_crypto(), 0);
+    auto expect = make_expected_error(400, "Bad Request");
+    auto result = do_http(portal->listen_port(), null_crypto(), "this request is", " totally bogus\r\n");
+    EXPECT_EQUAL(result, expect);
+}
+
+TEST("require that the handler with the longest matching prefix is selected") {
+    auto portal = Portal::create(null_crypto(), 0);
+    MyGetHandler handler1([](Portal::GetRequest request){ request.respond_with_content("text/plain", "handler1"); });
+    MyGetHandler handler2([](Portal::GetRequest request){ request.respond_with_content("text/plain", "handler2"); });
+    MyGetHandler handler3([](Portal::GetRequest request){ request.respond_with_content("text/plain", "handler3"); });
+    auto bound1 = portal->bind("/foo", handler1);
+    auto bound3 = portal->bind("/foo/bar/baz", handler3);
+    auto bound2 = portal->bind("/foo/bar", handler2);
+    EXPECT_EQUAL(fetch(portal->listen_port(), null_crypto(), "/foo"), make_expected_response("text/plain", "handler1"));
+    EXPECT_EQUAL(fetch(portal->listen_port(), null_crypto(), "/foo/bar"), make_expected_response("text/plain", "handler2"));
+    EXPECT_EQUAL(fetch(portal->listen_port(), null_crypto(), "/foo/bar/baz"), make_expected_response("text/plain", "handler3"));
+    bound3.reset();
+    EXPECT_EQUAL(fetch(portal->listen_port(), null_crypto(), "/foo/bar/baz"), make_expected_response("text/plain", "handler2"));
+    bound2.reset();
+    EXPECT_EQUAL(fetch(portal->listen_port(), null_crypto(), "/foo/bar/baz"), make_expected_response("text/plain", "handler1"));
+}
+
+TEST("require that newer handlers with the same prefix shadows older ones") {
+    auto portal = Portal::create(null_crypto(), 0);
+    MyGetHandler handler1([](Portal::GetRequest request){ request.respond_with_content("text/plain", "handler1"); });
+    MyGetHandler handler2([](Portal::GetRequest request){ request.respond_with_content("text/plain", "handler2"); });
+    MyGetHandler handler3([](Portal::GetRequest request){ request.respond_with_content("text/plain", "handler3"); });
+    auto bound1 = portal->bind("/foo", handler1);
+    EXPECT_EQUAL(fetch(portal->listen_port(), null_crypto(), "/foo"), make_expected_response("text/plain", "handler1"));
+    auto bound2 = portal->bind("/foo", handler2);
+    EXPECT_EQUAL(fetch(portal->listen_port(), null_crypto(), "/foo"), make_expected_response("text/plain", "handler2"));
+    auto bound3 = portal->bind("/foo", handler3);
+    EXPECT_EQUAL(fetch(portal->listen_port(), null_crypto(), "/foo"), make_expected_response("text/plain", "handler3"));
+    bound3.reset();
+    EXPECT_EQUAL(fetch(portal->listen_port(), null_crypto(), "/foo"), make_expected_response("text/plain", "handler2"));
+    bound2.reset();
+    EXPECT_EQUAL(fetch(portal->listen_port(), null_crypto(), "/foo"), make_expected_response("text/plain", "handler1"));
+}
+
+TEST("require that connection errors do not block shutdown by leaking resources (also tests tight shutdown timing)") {
+    MyGetHandler handler([](Portal::GetRequest request)
+                         {
+                             std::this_thread::sleep_for(std::chrono::milliseconds(5));
+                             request.respond_with_content("application/json", "[1,2,3]");
+                         });
+    for (const Encryption &crypto: crypto_list) {
+        fprintf(stderr, "... testing connection errors with encryption: '%s'\n", crypto.name.c_str());
+        auto portal = Portal::create(crypto.engine, 0);
+        auto bound = portal->bind("/test", handler);
+        { // close before sending anything
+            auto socket = SocketSpec::from_port(portal->listen_port()).client_address().connect();
+            auto conn = SyncCryptoSocket::create(*crypto.engine, std::move(socket), false);
+        }
+        { // send partial request then close connection
+            auto socket = SocketSpec::from_port(portal->listen_port()).client_address().connect();
+            auto conn = SyncCryptoSocket::create(*crypto.engine, std::move(socket), false);
+            vespalib::string req = "GET /test HTTP/1.1\r\n"
+                                   "Host: local";
+            ASSERT_EQUAL(conn->write(req.data(), req.size()), ssize_t(req.size()));
+        }
+        { // send request then close without reading response
+            auto socket = SocketSpec::from_port(portal->listen_port()).client_address().connect();
+            auto conn = SyncCryptoSocket::create(*crypto.engine, std::move(socket), false);
+            vespalib::string req = "GET /test HTTP/1.1\r\n"
+                                   "Host: localhost\r\n"
+                                   "\r\n";
+            ASSERT_EQUAL(conn->write(req.data(), req.size()), ssize_t(req.size()));
+        }
+    }
+}
+
+struct WaitingFixture {
+    Portal::SP portal;
+    Gate enter_callback;
+    Gate exit_callback;
+    MyGetHandler handler;
+    Portal::Token::UP bound;
+    WaitingFixture() : portal(Portal::create(null_crypto(), 0)),
+                       enter_callback(),
+                       exit_callback(),
+                       handler([this](Portal::GetRequest request)
+                               {
+                                   enter_callback.countDown();
+                                   request.respond_with_content("application/json", "[1,2,3]");
+                                   exit_callback.await();
+                               }),
+                       bound(portal->bind("/test", handler)) {}
+};
+
+TEST_MT_FFF("require that bind token destruction waits for active requests", 3,
+            WaitingFixture(), Gate(), TimeBomb(60))
+{
+    if (thread_id == 0) {
+        f1.enter_callback.await();
+        EXPECT_TRUE(!f2.await(20));
+        f1.exit_callback.countDown();
+        EXPECT_TRUE(f2.await(60000));
+    } else if (thread_id == 1) {
+        f1.enter_callback.await();
+        f1.bound.reset();
+        f2.countDown();
+    } else {
+        auto result = fetch(f1.portal->listen_port(), null_crypto(), "/test");
+        EXPECT_EQUAL(result, make_expected_response("application/json", "[1,2,3]"));
+    }
+}
+
+//-----------------------------------------------------------------------------
+
+TEST_MAIN() { TEST_RUN_ALL(); }
diff --git a/vespalib/src/tests/portal/reactor/CMakeLists.txt b/vespalib/src/tests/portal/reactor/CMakeLists.txt
new file mode 100644
index 00000000000..2ed829fcfc0
--- /dev/null
+++ b/vespalib/src/tests/portal/reactor/CMakeLists.txt
@@ -0,0 +1,8 @@
+# Copyright 2018 Yahoo Holdings. Licensed under the terms of the Apache 2.0 license. See LICENSE in the project root.
+vespa_add_executable(vespalib_reactor_test_app TEST
+    SOURCES
+    reactor_test.cpp
+    DEPENDS
+    vespalib
+)
+vespa_add_test(NAME vespalib_reactor_test_app COMMAND vespalib_reactor_test_app)
diff --git a/vespalib/src/tests/portal/reactor/reactor_test.cpp b/vespalib/src/tests/portal/reactor/reactor_test.cpp
new file mode 100644
index 00000000000..31bda119fd9
--- /dev/null
+++ b/vespalib/src/tests/portal/reactor/reactor_test.cpp
@@ -0,0 +1,162 @@
+// Copyright 2018 Yahoo Holdings. Licensed under the terms of the Apache 2.0 license. See LICENSE in the project root.
+
+#include <vespa/vespalib/testkit/test_kit.h>
+#include <vespa/vespalib/net/socket_handle.h>
+#include <vespa/vespalib/portal/reactor.h>
+#include <vespa/vespalib/util/gate.h>
+
+#include <thread>
+#include <chrono>
+
+#include <sys/types.h>
+#include <sys/socket.h>
+#include <fcntl.h>
+
+using namespace vespalib;
+using namespace vespalib::portal;
+
+struct SocketPair {
+    SocketHandle main;
+    SocketHandle other;
+    SocketPair() : main(), other() {
+        int sockets[2];
+        ASSERT_EQUAL(0, socketpair(AF_UNIX, SOCK_STREAM | O_NONBLOCK, 0, sockets));
+        main.reset(sockets[0]);
+        other.reset(sockets[1]);
+        // make main socket both readable and writable
+        ASSERT_EQUAL(other.write("x", 1), 1);
+    }
+    ~SocketPair();
+};
+SocketPair::~SocketPair() = default;
+
+std::atomic<size_t> tick_cnt = 0;
+int tick() {
+    ++tick_cnt;
+    std::this_thread::sleep_for(std::chrono::milliseconds(1));
+    return 0;
+}
+
+void wait_tick() {
+    size_t sample = tick_cnt;
+    while (sample == tick_cnt) {
+        std::this_thread::sleep_for(std::chrono::milliseconds(1));
+    }
+}
+
+struct SimpleHandler : Reactor::EventHandler {
+    SocketPair          sockets;
+    std::atomic<size_t> read_cnt;
+    std::atomic<size_t> write_cnt;
+    Reactor::Token::UP  token;
+    SimpleHandler(Reactor &reactor, bool read, bool write)
+        : sockets(), read_cnt(0), write_cnt(0), token()
+    {
+        token = reactor.attach(*this, sockets.main.get(), read, write);
+    }
+    void handle_event(bool read, bool write) override {
+        if (read) {
+            ++read_cnt;
+        }
+        if (write) {
+            ++write_cnt;
+        }
+    }
+    void verify(bool read, bool write) {
+        size_t read_sample = read_cnt;
+        size_t write_sample = write_cnt;
+        wait_tick();
+        wait_tick();
+        EXPECT_EQUAL((read_sample != read_cnt), read);
+        EXPECT_EQUAL((write_sample != write_cnt), write);
+    }
+    ~SimpleHandler();
+};
+SimpleHandler::~SimpleHandler() = default;
+
+struct DeletingHandler : SimpleHandler {
+    Gate token_deleted;
+    DeletingHandler(Reactor &reactor) : SimpleHandler(reactor, true, true),
+                                        token_deleted() {}
+    void handle_event(bool read, bool write) override {
+        SimpleHandler::handle_event(read, write);
+        token.reset();
+        token_deleted.countDown();
+    }
+    ~DeletingHandler();
+};
+DeletingHandler::~DeletingHandler() = default;
+
+struct WaitingHandler : SimpleHandler {
+    Gate enter_callback;
+    Gate exit_callback;
+    WaitingHandler(Reactor &reactor) : SimpleHandler(reactor, true, true),
+                                       enter_callback(), exit_callback() {}
+    void handle_event(bool read, bool write) override {
+        enter_callback.countDown();
+        SimpleHandler::handle_event(read, write);
+        exit_callback.await();
+    }
+    ~WaitingHandler();
+};
+WaitingHandler::~WaitingHandler() = default;
+
+//-----------------------------------------------------------------------------
+
+TEST_FF("require that reactor can produce async io events", Reactor(tick), TimeBomb(60)) {
+    for (bool read: {true, false}) {
+        for (bool write: {true, false}) {
+            {
+                SimpleHandler handler(f1, read, write);
+                TEST_DO(handler.verify(read, write));
+            }
+        }
+    }
+}
+
+TEST_FF("require that reactor token can be used to change active io events", Reactor(tick), TimeBomb(60)) {
+    SimpleHandler handler(f1, false, false);
+    TEST_DO(handler.verify(false, false));
+    for (int i = 0; i < 2; ++i) {
+        for (bool read: {true, false}) {
+            for (bool write: {true, false}) {
+                handler.token->update(read, write);
+                wait_tick(); // avoid stale events
+                TEST_DO(handler.verify(read, write));
+            }
+        }
+    }
+}
+
+TEST_FF("require that deleting reactor token disables io events", Reactor(tick), TimeBomb(60)) {
+    SimpleHandler handler(f1, true, true);
+    TEST_DO(handler.verify(true, true));
+    handler.token.reset();
+    TEST_DO(handler.verify(false, false));
+}
+
+TEST_FF("require that reactor token can be destroyed during io event handling", Reactor(tick), TimeBomb(60)) {
+    DeletingHandler handler(f1);
+    handler.token_deleted.await();
+    TEST_DO(handler.verify(false, false));
+    EXPECT_EQUAL(handler.read_cnt, 1u);
+    EXPECT_EQUAL(handler.write_cnt, 1u);
+}
+
+TEST_MT_FFFF("require that reactor token destruction waits for io event handling", 2,
+             Reactor(), WaitingHandler(f1), Gate(), TimeBomb(60))
+{
+    if (thread_id == 0) {
+        f2.enter_callback.await();
+        TEST_BARRIER(); // #1
+        EXPECT_TRUE(!f3.await(20));
+        f2.exit_callback.countDown();
+        EXPECT_TRUE(f3.await(60000));
+    } else {
+        TEST_BARRIER(); // #1
+        f2.token.reset();
+        f3.countDown();
+    }
+}
+
+TEST_MAIN() { TEST_RUN_ALL(); }