protocol.Serializer: Implement ping.

Author: mefyl

src/elle/protocol/Serializer.cc

@@ -156,7 +156,9 @@ namespace elle
       keep_going = 0,
       interrupt = 1,
       message = 2,
-      max = message,
+      ping = 3,
+      pong = 4,
+      max = pong,
     };
 
     static
@@ -174,14 +176,39 @@ namespace elle
       Impl(std::iostream& stream,
            elle::Buffer::Size chunk_size,
            bool checksum,
-           elle::Version const& version)
-        : _stream(stream)
+           elle::Version const& version,
+           boost::optional<std::chrono::milliseconds> ping_period,
+           boost::optional<std::chrono::milliseconds> ping_timeout)
+        : _scheduler(reactor::scheduler())
+        , _pings(0)
+        , _pongs(0)
+        , _ping_period(std::move(ping_period))
+        , _ping_delay(std::move(ping_timeout))
+        , _pinger(this->_scheduler.io_service())
+        , _pinger_handler(
+          [this] (boost::system::error_code const& error)
+          {
+            if (error == boost::system::errc::operation_canceled)
+              return;
+            else if (error)
+              ELLE_ABORT("%s: unexpected timer error: %s", this, error);
+            this->ping();
+            this->_pinger.expires_from_now(
+              boost::posix_time::milliseconds(this->_ping_period->count()));
+            this->_pinger.async_wait(this->_pinger_handler);
+          })
+        , _stream(stream)
         , _chunk_size(chunk_size)
         , _checksum(checksum)
         , _version(version)
         , _lock_write()
         , _lock_read()
-      {}
+      {
+        if (bool(this->_ping_period) != bool(this->_ping_delay))
+          elle::err("specify either both ping period and timeout or neither");
+        if (this->_ping_period)
+          this->_pinger_handler({});
+      }
 
       virtual
       ~Impl() = default;
@@ -274,11 +301,31 @@ namespace elle
       void
       write_control(Control control)
       {
+        if (control != Control::pong && control != Control::ping)
+          this->write_pings_pongs();
         ELLE_DUMP_SCOPE("send control %s", (int) control);
         char c = static_cast<char>(control);
         this->_stream.write(&c, 1);
       }
 
+      void
+      write_pings_pongs()
+      {
+        while (this->_pongs || this->_pings)
+        {
+          while (this->_pongs)
+          {
+            --this->_pongs;
+            this->write_control(Control::pong);
+          }
+          while (this->_pings)
+          {
+            --this->_pings;
+            this->write_control(Control::ping);
+          }
+        }
+      }
+
       bool
       read_control()
       {
@@ -305,10 +352,70 @@ namespace elle
             case Control::message:
               ignore_message(this->_stream, this->version());
               break;
+            case Control::ping:
+              this->pinged();
+              break;
+            case Control::pong:
+              if (this->_ping_timers.empty())
+                ELLE_ERR("%s: unexpected pong", this);
+              else
+                this->_ping_timers.pop_front();
+              break;
           }
         }
       }
 
+      void
+      ping()
+      {
+        ++this->_pings;
+        if (!this->_lock_write.locked())
+        {
+          new reactor::Thread(
+            this->_scheduler,
+            elle::sprintf("%s pinger", this),
+            [this]
+            {
+              elle::reactor::Lock lock(this->_lock_write);
+              this->write_pings_pongs();
+              this->_stream.flush();
+            },
+            true);
+        }
+        this->_ping_timers.emplace_back(this->_scheduler.io_service());
+        this->_ping_timers.back().expires_from_now(
+          boost::posix_time::milliseconds(this->_ping_delay->count()));
+        this->_ping_timers.back().async_wait(
+          [this] (boost::system::error_code const& error)
+          {
+            if (!error)
+              this->_ping_timeout();
+          });
+      }
+
+      void
+      pinged()
+      {
+        ++this->_pongs;
+        if (!this->_lock_write.locked())
+        {
+          elle::reactor::Lock lock(this->_lock_write);
+          this->write_pings_pongs();
+          this->_stream.flush();
+        }
+      }
+
+      ELLE_ATTRIBUTE(elle::reactor::Scheduler&, scheduler);
+      ELLE_ATTRIBUTE(int, pings);
+      ELLE_ATTRIBUTE(int, pongs);
+      ELLE_ATTRIBUTE(boost::optional<std::chrono::milliseconds>, ping_period);
+      ELLE_ATTRIBUTE(boost::optional<std::chrono::milliseconds>, ping_delay);
+      ELLE_ATTRIBUTE(boost::asio::deadline_timer, pinger);
+      ELLE_ATTRIBUTE(std::function<void (boost::system::error_code const&)>,
+                     pinger_handler);
+      ELLE_ATTRIBUTE(std::list<boost::asio::deadline_timer>, ping_timers);
+      ELLE_ATTRIBUTE_RX(boost::signals2::signal<void ()>, ping_timeout);
+
       void
       _write(elle::Buffer const& packet)
       {
@@ -360,6 +467,8 @@ namespace elle
                 this->write_control(Control::keep_going);
                 send();
               };
+              this->write_pings_pongs();
+              this->_stream.flush();
             }
           }
           catch (elle::reactor::Terminate const&)
@@ -369,6 +478,7 @@ namespace elle
               ELLE_DEBUG("interrupted after sending %s bytes over %s",
                          offset, packet.size());
               this->write_control(Control::interrupt);
+              this->write_pings_pongs();
               this->_stream.flush();
             }
             throw;
@@ -405,9 +515,12 @@ namespace elle
     | Construction |
     `-------------*/
 
-    Serializer::Serializer(std::iostream& stream,
-                           elle::Version const& version,
-                           bool checksum)
+    Serializer::Serializer(
+      std::iostream& stream,
+      elle::Version const& version,
+      bool checksum,
+      boost::optional<std::chrono::milliseconds> ping_period,
+      boost::optional<std::chrono::milliseconds> ping_timeout)
       : Super(*elle::reactor::Scheduler::scheduler())
       , _stream(stream)
       , _version(version)
@@ -430,7 +543,9 @@ namespace elle
       }
       ELLE_TRACE("using version: '%s'", this->version());
       this->_impl.reset(
-        new Impl(stream, this->_chunk_size, checksum, version));
+        new Impl(stream, this->_chunk_size, checksum, version,
+                 std::move(ping_period), std::move(ping_timeout)));
+      this->_impl->ping_timeout().connect(this->_ping_timeout);
     }
 
     Serializer::~Serializer()

src/elle/protocol/Serializer.hh

@@ -1,5 +1,6 @@
 #pragma once
 
+#include <chrono>
 #include <iostream>
 
 #include <elle/reactor/mutex.hh>
@@ -65,7 +66,9 @@ namespace elle
       ///                 packets sent.
       Serializer(std::iostream& stream,
                  elle::Version const& version = elle::Version(0, 1, 0),
-                 bool checksum = true);
+                 bool checksum = true,
+                 boost::optional<std::chrono::milliseconds> ping_period = {},
+                 boost::optional<std::chrono::milliseconds> ping_timeout = {});
       ~Serializer();
 
     /*----------.
@@ -102,6 +105,7 @@ namespace elle
       ELLE_ATTRIBUTE_R(elle::Version, version, override);
       ELLE_ATTRIBUTE_R(elle::Buffer::Size, chunk_size);
       ELLE_ATTRIBUTE_R(bool, checksum);
+      ELLE_ATTRIBUTE_RX(boost::signals2::signal<void ()>, ping_timeout);
     public:
       class Impl;
     private:

tests/elle/protocol/serializer.cc

@@ -191,6 +191,8 @@ class SocketInstrumentation
     , _alice_routed(0)
     , _bob_routed(0)
   {
+    this->_alice_barrier.open();
+    this->_bob_barrier.open();
     this->_a_server.listen();
     this->_b_server.listen();
     this->_alice.reset(
@@ -248,6 +250,7 @@ class SocketInstrumentation
     {
       auto route = [&] (elle::reactor::network::Socket* sender,
                         elle::reactor::network::Socket* recipient,
+                        elle::reactor::Barrier& barrier,
                         int64_t& routed,
                         Conf& conf)
         {
@@ -259,8 +262,7 @@ class SocketInstrumentation
               char buffer[1024];
               ELLE_DEBUG("reading");
               int64_t size =
-                sender->read_some(elle::WeakBuffer(buffer),
-                                  valgrind(250_ms, 10));
+                sender->read_some(elle::WeakBuffer(buffer));
               ELLE_DEBUG("read %s", size);
               conf(elle::ConstWeakBuffer(buffer, size));
               if (conf.corrupt_offset >= 0
@@ -281,6 +283,7 @@ class SocketInstrumentation
                 routed = conf.quota;
               }
               ELLE_DEBUG("%s: route %s bytes", *this, size);
+              elle::reactor::wait(barrier);
               recipient->write(elle::ConstWeakBuffer(buffer, size));
             }
           }
@@ -291,16 +294,20 @@ class SocketInstrumentation
         };
       scope.run_background("A to B",
                            std::bind(route, a.get(), b.get(),
+                                     std::ref(this->_alice_barrier),
                                      std::ref(this->_alice_routed),
                                      std::ref(*this->alice_conf)));
       scope.run_background("B to A",
                            std::bind(route, b.get(), a.get(),
+                                     std::ref(this->_bob_barrier),
                                      std::ref(this->_bob_routed),
                                      std::ref(*this->bob_conf)));
       scope.wait();
     };
   }
 
+  ELLE_ATTRIBUTE_RX(elle::reactor::Barrier, alice_barrier);
+  ELLE_ATTRIBUTE_RX(elle::reactor::Barrier, bob_barrier);
   elle::reactor::network::TCPServer _a_server;
   elle::reactor::network::TCPServer _b_server;
   std::unique_ptr<elle::reactor::network::TCPSocket> _alice;
@@ -348,7 +355,9 @@ dialog(elle::Version const& version,
        std::function<void (elle::protocol::Serializer&)> const& a,
        std::function<void (elle::protocol::Serializer&)> const& b,
        std::function<void (elle::reactor::Thread&, elle::reactor::Thread&,
-                           SocketProvider&)> const& f = {})
+                           SocketProvider&)> const& f = {},
+       boost::optional<std::chrono::milliseconds> ping_period = {},
+       boost::optional<std::chrono::milliseconds> ping_timeout = {})
 {
   SocketProvider sockets;
   std::unique_ptr<elle::protocol::Serializer> alice;
@@ -360,14 +369,16 @@ dialog(elle::Version const& version,
       [&]
       {
         alice.reset(new elle::protocol::Serializer(
-                      sockets.alice(), version, checksum));
+                      sockets.alice(), version, checksum,
+                      ping_period, ping_timeout));
       });
     scope.run_background(
       "setup bob's serializer",
       [&]
       {
         bob.reset(new elle::protocol::Serializer(
-                    sockets.bob(), version, checksum));
+                    sockets.bob(), version, checksum,
+                    ping_period, ping_timeout));
       });
     scope.wait();
   };
@@ -878,7 +889,6 @@ ELLE_TEST_SCHEDULED(interruption2)
   // Check that terminating a Channel.read() call does not lose an unrelated
   // packet.
   namespace ip = elle::protocol;
-
   for (auto version: {elle::Version{0, 2, 0}, elle::Version{0, 3, 0}})
   {
     ELLE_LOG("test version %s", version);
@@ -954,7 +964,6 @@ ELLE_TEST_SCHEDULED(interruption2)
         ELLE_TRACE("read on %s", c.id());
         auto buf = c.read();
         BOOST_CHECK_EQUAL(buf.string(), "foo");
-
         // just read some more to check we're still on
         ip::Channel c3(stream);
         c3.write(elle::Buffer("baz"));
@@ -964,7 +973,6 @@ ELLE_TEST_SCHEDULED(interruption2)
         ELLE_TRACE("read c3");
         buf = c3.read();
         BOOST_CHECK_EQUAL(buf.string(), "baz");
-
         // check killing reader thread before any data is read
         b.close();
         elle::reactor::Thread t2("read nothing", [&] {
@@ -995,6 +1003,55 @@ ELLE_TEST_SCHEDULED(interruption2)
   }
 }
 
+ELLE_TEST_SCHEDULED(ping)
+{
+  elle::reactor::Barrier alice;
+  elle::reactor::Barrier bob;
+  bool timeout_expected = false;
+  int timeouts = 0;
+  dialog<SocketInstrumentation>(
+    elle::Version(0, 3, 0),
+    true,
+    [] (SocketInstrumentation& socket)
+    {},
+    [&] (elle::protocol::Serializer& s)
+    {
+      s.ping_timeout().connect(
+        [&] { ++timeouts; BOOST_TEST(timeout_expected); });
+      s.write(elle::Buffer("alice"));
+      BOOST_TEST(s.read() == elle::Buffer("bob"));
+      alice.open();
+      BOOST_TEST(s.read() == elle::Buffer("bob"));
+    },
+    [&] (elle::protocol::Serializer& s)
+    {
+      s.ping_timeout().connect(
+        [&] { ++timeouts; BOOST_TEST(timeout_expected); });
+      s.write(elle::Buffer("bob"));
+      BOOST_TEST(s.read() == elle::Buffer("alice"));
+      bob.open();
+      BOOST_TEST(s.read() == elle::Buffer("alice"));
+    },
+    [&] (elle::reactor::Thread& a,
+         elle::reactor::Thread& b,
+         SocketInstrumentation& socket)
+    {
+      elle::reactor::wait(elle::reactor::Waitables{&alice, &bob});
+      // Let some pings go through
+      elle::reactor::sleep(500_ms);
+      timeout_expected = true;
+      socket.alice_barrier().close();
+      socket.bob_barrier().close();
+      // Let some timeouts go through
+      elle::reactor::sleep(500_ms);
+      a.terminate();
+      b.terminate();
+      BOOST_TEST(timeouts >= 2);
+    },
+    std::chrono::milliseconds(200),
+    std::chrono::milliseconds(100));
+}
+
 ELLE_TEST_SUITE()
 {
   auto& suite = boost::unit_test::framework::master_test_suite();
@@ -1008,4 +1065,5 @@ ELLE_TEST_SUITE()
   suite.add(BOOST_TEST_CASE(termination), 0, valgrind(3, 10));
   suite.add(BOOST_TEST_CASE(eof), 0, valgrind(3));
   suite.add(BOOST_TEST_CASE(message), 0, valgrind(3));
+  suite.add(BOOST_TEST_CASE(ping), 0, valgrind(3));
 }