[2916] added tests and implementation of the LocalSocket class

src/lib/asiolink/Makefile.am

@@ -32,6 +32,7 @@ libb10_asiolink_la_SOURCES += tcp_endpoint.h
 libb10_asiolink_la_SOURCES += tcp_socket.h
 libb10_asiolink_la_SOURCES += udp_endpoint.h
 libb10_asiolink_la_SOURCES += udp_socket.h
+libb10_asiolink_la_SOURCES += local_socket.h local_socket.cc
 
 # Note: the ordering matters: -Wno-... must follow -Wextra (defined in
 # B10_CXXFLAGS)

src/lib/asiolink/local_socket.cc

@@ -0,0 +1,82 @@
+// Copyright (C) 2013  Internet Systems Consortium, Inc. ("ISC")
+//
+// Permission to use, copy, modify, and/or distribute this software for any
+// purpose with or without fee is hereby granted, provided that the above
+// copyright notice and this permission notice appear in all copies.
+//
+// THE SOFTWARE IS PROVIDED "AS IS" AND ISC DISCLAIMS ALL WARRANTIES WITH
+// REGARD TO THIS SOFTWARE INCLUDING ALL IMPLIED WARRANTIES OF MERCHANTABILITY
+// AND FITNESS.  IN NO EVENT SHALL ISC BE LIABLE FOR ANY SPECIAL, DIRECT,
+// INDIRECT, OR CONSEQUENTIAL DAMAGES OR ANY DAMAGES WHATSOEVER RESULTING FROM
+// LOSS OF USE, DATA OR PROFITS, WHETHER IN AN ACTION OF CONTRACT, NEGLIGENCE
+// OR OTHER TORTIOUS ACTION, ARISING OUT OF OR IN CONNECTION WITH THE USE OR
+// PERFORMANCE OF THIS SOFTWARE.
+
+#include <asiolink/local_socket.h>
+#include <asiolink/io_service.h>
+
+#include <asio.hpp>
+
+#include <boost/bind.hpp>
+
+#include <string>
+#include <sys/socket.h>
+
+namespace isc {
+namespace asiolink {
+
+class LocalSocket::Impl {
+public:
+    Impl(IOService& io_service, int fd) :
+        asio_sock_(io_service.get_io_service())
+    {
+        asio_sock_.assign(asio::local::stream_protocol(), fd, ec_);
+    }
+
+    void readCompleted(const asio::error_code& ec, size_t read_len,
+                       ReadCallback user_callback);
+
+    asio::local::stream_protocol::socket asio_sock_;
+    asio::error_code ec_;
+};
+
+void
+LocalSocket::Impl::readCompleted(const asio::error_code& ec,
+                                 size_t, ReadCallback user_callback)
+{
+    // assumption check: we pass non empty string iff ec indicates an error.
+    const std::string& err_msg = ec ? ec.message() : std::string();
+    assert(ec || err_msg.empty());
+
+    user_callback(err_msg);
+}
+
+LocalSocket::LocalSocket(IOService& io_service, int fd) :
+    impl_(new Impl(io_service, fd))
+{}
+
+LocalSocket::~LocalSocket() {
+    delete impl_;
+}
+
+int
+LocalSocket::getNative() const {
+    return (impl_->asio_sock_.native());
+}
+
+int
+LocalSocket::getProtocol() const {
+    return (AF_UNIX);
+}
+
+void
+LocalSocket::asyncRead(const ReadCallback& callback, void* buf,
+                       size_t buflen)
+{
+    asio::async_read(impl_->asio_sock_, asio::buffer(buf, buflen),
+                     boost::bind(&Impl::readCompleted, impl_, _1, _2,
+                                 callback));
+}
+
+} // namespace asiolink
+} // namespace isc

src/lib/asiolink/local_socket.h

@@ -0,0 +1,55 @@
+// Copyright (C) 2013  Internet Systems Consortium, Inc. ("ISC")
+//
+// Permission to use, copy, modify, and/or distribute this software for any
+// purpose with or without fee is hereby granted, provided that the above
+// copyright notice and this permission notice appear in all copies.
+//
+// THE SOFTWARE IS PROVIDED "AS IS" AND ISC DISCLAIMS ALL WARRANTIES WITH
+// REGARD TO THIS SOFTWARE INCLUDING ALL IMPLIED WARRANTIES OF MERCHANTABILITY
+// AND FITNESS.  IN NO EVENT SHALL ISC BE LIABLE FOR ANY SPECIAL, DIRECT,
+// INDIRECT, OR CONSEQUENTIAL DAMAGES OR ANY DAMAGES WHATSOEVER RESULTING FROM
+// LOSS OF USE, DATA OR PROFITS, WHETHER IN AN ACTION OF CONTRACT, NEGLIGENCE
+// OR OTHER TORTIOUS ACTION, ARISING OUT OF OR IN CONNECTION WITH THE USE OR
+// PERFORMANCE OF THIS SOFTWARE.
+
+#ifndef LOCAL_SOCKET_H
+#define LOCAL_SOCKET_H 1
+
+#include <asiolink/io_socket.h>
+#include <asiolink/io_service.h>
+
+#include <boost/noncopyable.hpp>
+
+namespace isc {
+namespace asiolink {
+
+/// \brief TBD
+class LocalSocket : boost::noncopyable, public IOSocket {
+public:
+    /// \brief Constructor from a native file descriptor of AF_UNIX socket.
+    ///
+    /// \param io_service
+    LocalSocket(IOService& io_service, int fd);
+
+    virtual ~LocalSocket();
+
+    virtual int getNative() const;
+
+    virtual int getProtocol() const;
+
+    typedef boost::function<void(const std::string& error)> ReadCallback;
+    void asyncRead(const ReadCallback& callback, void* buf, size_t buflen);
+
+private:
+    class Impl;
+    Impl* impl_;
+};
+
+} // namespace asiolink
+} // namespace isc
+
+#endif // LOCAL_SOCKET_H
+
+// Local Variables:
+// mode: c++
+// End:

src/lib/asiolink/tests/Makefile.am

@@ -34,6 +34,7 @@ run_unittests_SOURCES += tcp_socket_unittest.cc
 run_unittests_SOURCES += udp_endpoint_unittest.cc
 run_unittests_SOURCES += udp_socket_unittest.cc
 run_unittests_SOURCES += io_service_unittest.cc
+run_unittests_SOURCES += local_socket_unittest.cc
 
 run_unittests_CPPFLAGS = $(AM_CPPFLAGS) $(GTEST_INCLUDES)
 

src/lib/asiolink/tests/local_socket_unittest.cc

@@ -0,0 +1,229 @@
+// Copyright (C) 2013  Internet Systems Consortium, Inc. ("ISC")
+//
+// Permission to use, copy, modify, and/or distribute this software for any
+// purpose with or without fee is hereby granted, provided that the above
+// copyright notice and this permission notice appear in all copies.
+//
+// THE SOFTWARE IS PROVIDED "AS IS" AND ISC DISCLAIMS ALL WARRANTIES WITH
+// REGARD TO THIS SOFTWARE INCLUDING ALL IMPLIED WARRANTIES OF MERCHANTABILITY
+// AND FITNESS.  IN NO EVENT SHALL ISC BE LIABLE FOR ANY SPECIAL, DIRECT,
+// INDIRECT, OR CONSEQUENTIAL DAMAGES OR ANY DAMAGES WHATSOEVER RESULTING FROM
+// LOSS OF USE, DATA OR PROFITS, WHETHER IN AN ACTION OF CONTRACT, NEGLIGENCE
+// OR OTHER TORTIOUS ACTION, ARISING OUT OF OR IN CONNECTION WITH THE USE OR
+// PERFORMANCE OF THIS SOFTWARE.
+
+#include <asiolink/local_socket.h>
+
+#include <gtest/gtest.h>
+
+#include <boost/noncopyable.hpp>
+#include <boost/scoped_ptr.hpp>
+#include <boost/bind.hpp>
+
+#include <csignal>
+#include <vector>
+
+#include <sys/socket.h>
+#include <stdint.h>
+#include <unistd.h>             // for alarm(3)
+
+using namespace isc::asiolink;
+
+namespace {
+
+// duration (in seconds) until we break possible hangup; value is an
+// arbitrary choice.
+const unsigned IO_TIMEOUT = 10;
+
+// A simple RAII wrapper for a file descriptor so test sockets are safely
+// closed in each test.
+class ScopedSocket : boost::noncopyable {
+public:
+    ScopedSocket() : fd_(-1) {}
+    ~ScopedSocket() {
+        if (fd_ >= 0) {
+            EXPECT_EQ(0, ::close(fd_));
+        }
+    }
+    void set(int fd) {
+        assert(fd_ == -1);
+        fd_ = fd;
+    }
+    int get() { return (fd_); }
+    int release() {
+        const int ret = fd_;
+        fd_ = -1;
+        return (ret);
+    }
+private:
+    int fd_;
+};
+
+class LocalSocketTest : public ::testing::Test {
+protected:
+    LocalSocketTest() {
+        int sock_pair[2];
+        EXPECT_EQ(0, socketpair(AF_UNIX, SOCK_STREAM, 0, sock_pair));
+        sock_pair_[0].set(sock_pair[0]);
+        sock_pair_[1].set(sock_pair[1]);
+
+        // For tests using actual I/O we use a timer to prevent hangup
+        // due to a bug.  Set up the signal handler for the timer here.
+        g_io_service_ = &io_service_;
+        prev_handler_ = std::signal(SIGALRM, stopIOService);
+    }
+
+    ~LocalSocketTest() {
+        std::signal(SIGALRM, prev_handler_);
+    }
+
+    // Common set of tests for async read
+    void checkAsyncRead(size_t data_len);
+
+    IOService io_service_;
+    ScopedSocket sock_pair_[2];
+    std::vector<uint8_t> read_buf_;
+private:
+    static IOService* g_io_service_; // will be set to &io_service_
+    void (*prev_handler_)(int);
+
+    // SIGALRM handler to prevent hangup.  This must be a static method
+    // so it can be passed to std::signal().
+    static void stopIOService(int) {
+        g_io_service_->stop();
+    }
+};
+
+IOService* LocalSocketTest::g_io_service_ = NULL;
+
+TEST_F(LocalSocketTest, construct) {
+    const int fd = sock_pair_[0].release();
+    LocalSocket sock(io_service_, fd);
+    EXPECT_EQ(fd, sock.getNative());
+    EXPECT_EQ(AF_UNIX, sock.getProtocol());
+}
+
+TEST_F(LocalSocketTest, autoClose) {
+    // Confirm that passed FD will be closed on destruction of LocalSocket
+    const int fd = sock_pair_[0].release();
+    {
+        LocalSocket sock(io_service_, fd);
+    }
+    // fd should have been closed, so close() should fail (we assume there's
+    // no other open() call since then)
+    EXPECT_EQ(-1, ::close(fd));
+}
+
+void
+callback(const std::string& error, IOService* io_service, bool* called,
+         bool expect_error)
+{
+    if (expect_error) {
+        EXPECT_NE("", error);
+    } else {
+        EXPECT_EQ("", error);
+    }
+    *called = true;
+    io_service->stop();
+}
+
+void
+LocalSocketTest::checkAsyncRead(size_t data_len) {
+    LocalSocket sock(io_service_, sock_pair_[0].release());
+    bool callback_called = false;
+    read_buf_.resize(data_len);
+    sock.asyncRead(boost::bind(&callback, _1, &io_service_, &callback_called,
+                               false), &read_buf_[0], data_len);
+
+    std::vector<uint8_t> expected_data(data_len);
+    for (size_t i = 0; i < data_len; ++i) {
+        expected_data[i] = i & 0xff;
+    }
+    alarm(IO_TIMEOUT);
+    EXPECT_EQ(data_len, write(sock_pair_[1].get(), &expected_data[0],
+                              data_len));
+    io_service_.run();
+    EXPECT_TRUE(callback_called);
+    EXPECT_EQ(0, std::memcmp(&expected_data[0], &read_buf_[0], data_len));
+    
+}
+
+TEST_F(LocalSocketTest, asyncRead) {
+    // A simple case of asynchronous read: wait for 1 byte and successfully
+    // read it in the run() loop.
+    checkAsyncRead(1);
+}
+
+TEST_F(LocalSocketTest, asyncLargeRead) {
+    // Similar to the previous case, but for moderately larger data.
+    // (for the moment) we don't expect to use this interface with much
+    // larger data that could cause blocking write.
+    checkAsyncRead(1024);
+}
+
+TEST_F(LocalSocketTest, asyncPartialRead) {
+    alarm(IO_TIMEOUT);
+
+    // specify reading 4 bytes of data, and send 3 bytes.  It shouldn't cause
+    // callback.
+    char recv_buf[4];
+    bool callback_called = false;
+    LocalSocket sock(io_service_, sock_pair_[0].release());
+    sock.asyncRead(boost::bind(&callback, _1, &io_service_, &callback_called,
+                               false), recv_buf, sizeof(recv_buf));
+    EXPECT_EQ(3, write(sock_pair_[1].get(), recv_buf, 3));
+
+    // open another pair of sockets so we can stop the IO service after run.
+    int socks[2];
+    char ch = 0;
+    EXPECT_EQ(0, socketpair(AF_UNIX, SOCK_STREAM, 0, socks));
+    ScopedSocket aux_sockpair[2];
+    aux_sockpair[0].set(socks[0]);
+    aux_sockpair[1].set(socks[1]);
+    LocalSocket aux_sock(io_service_, aux_sockpair[0].release());
+    bool aux_callback_called = false;
+    aux_sock.asyncRead(boost::bind(&callback, _1, &io_service_,
+                                   &aux_callback_called, false), &ch, 1);
+    EXPECT_EQ(1, write(aux_sockpair[1].get(), &ch, 1));
+
+    // run the IO service, it will soon be stopped via the auxiliary callback.
+    // the main callback shouldn't be called.
+    io_service_.run();
+    EXPECT_FALSE(callback_called);
+    EXPECT_TRUE(aux_callback_called);
+}
+
+TEST_F(LocalSocketTest, asyncReadError) {
+    const int sock_fd = sock_pair_[0].release();
+    LocalSocket sock(io_service_, sock_fd);
+    bool callback_called = false;
+    read_buf_.resize(1);
+    read_buf_.at(0) = 53;       // dummy data to check it later
+    const char ch = 35; // send different data to the read socket with data
+    EXPECT_EQ(1, write(sock_pair_[1].get(), &ch, 1));
+    close(sock_fd);             // invalidate the read socket
+    // we'll get callback with an error (e.g. 'bad file descriptor)
+    sock.asyncRead(boost::bind(&callback, _1, &io_service_, &callback_called,
+                               true), &read_buf_[0], 1);
+
+    io_service_.run();
+    EXPECT_TRUE(callback_called);
+    EXPECT_EQ(53, read_buf_.at(0));
+}
+
+TEST_F(LocalSocketTest, asyncReadThenDestroy) {
+    // destroy the socket before running the IO service.  we'll still get
+    // callback with an error.
+    boost::scoped_ptr<LocalSocket> sock(
+        new LocalSocket(io_service_, sock_pair_[0].release()));
+    read_buf_.resize(1);
+    bool callback_called = false;
+    sock->asyncRead(boost::bind(&callback, _1, &io_service_, &callback_called,
+                                true), &read_buf_[0], 1);
+    sock.reset();
+
+    io_service_.run();
+    EXPECT_TRUE(callback_called);
+}
+
+}