WORKAROUND: NonInteractive console output stream buffers size configurable (default...

TODO: This should be put into in a more generalized interface

git-svn-id: https://svn.berlios.de/svnroot/repos/senf/trunk@1811 270642c3-0616-0410-b53a-bc976706d245

diff --git a/senf/Utils/Console/Server.cc b/senf/Utils/Console/Server.cc
index 639b76d..6bb4aae 100644 (file)
--- a/senf/Utils/Console/Server.cc
+++ b/senf/Utils/Console/Server.cc
@@ -64,16 +64,17 @@ namespace {
 }
 
 //-/////////////////////////////////////////////////////////////////////////////////////////////////
-// senf::console::detail::NonBlockingSocketSink
+// senf::console::detail::SocketStreamSink
 
-prefix_ std::streamsize senf::console::detail::NonblockingSocketSink::write(const char * s,
+prefix_ std::streamsize senf::console::detail::SocketStreamSink::write(const char * s,
                                                                             std::streamsize n)
 {
+// since handle is now non blocking we done check for writeable
     try {
-        if (client_.handle().writeable()) {
+//        if (client_.handle().writeable()) {
             std::string data (s, n);
             client_.write(data);
-        }
+//        }
     }
     catch (...) {}
     return n;
@@ -232,10 +233,13 @@ prefix_ unsigned senf::console::detail::DumbClientReader::v_width()
 
 prefix_
 senf::console::detail::NoninteractiveClientReader::NoninteractiveClientReader(Client & client)
-    : ClientReader (client),
+    : ClientReader (client), streamBufferMaxSize_( 1024*1024),
       readevent_ ("senf::console::detail::NoninteractiveClientReader",
                   senf::membind(&NoninteractiveClientReader::newData, this),
-                  handle(), senf::scheduler::FdEvent::EV_READ)
+                  handle(), senf::scheduler::FdEvent::EV_READ),
+      writeevent_ ("senf::console::detail::NoninteractiveClientReader",
+                  membind(&NoninteractiveClientReader::writeHandler, this), handle(),
+                  scheduler::FdEvent::EV_WRITE, false)
 {}
 
 prefix_ void senf::console::detail::NoninteractiveClientReader::v_disablePrompt()
@@ -244,26 +248,27 @@ prefix_ void senf::console::detail::NoninteractiveClientReader::v_disablePrompt(
 prefix_ void senf::console::detail::NoninteractiveClientReader::v_enablePrompt()
 {}
 
+prefix_ void senf::console::detail::NoninteractiveClientReader::streamBufferMaxSize(SendQueue::size_type size)
+{
+    streamBufferMaxSize_ = size;
+}
+
+prefix_ senf::console::detail::NoninteractiveClientReader::SendQueue::size_type
+senf::console::detail::NoninteractiveClientReader::streamBufferMaxSize()
+    const
+{
+    return streamBufferMaxSize_;
+}
+
 prefix_ void senf::console::detail::NoninteractiveClientReader::v_write(std::string const & data)
 {
-    try {
-        handle().write(data);
-    }
-    catch (senf::ExceptionMixin & ex) {
-        SENF_LOG(("unexpected failure writing to socket:" << ex.message()));
-        try { handle().facet<senf::TCPSocketProtocol>().shutdown(senf::TCPSocketProtocol::ShutRD); }
-        catch (...) {}
-    }
-    catch (std::exception & ex) {
-        SENF_LOG(("unexpected failure writing to socket:" << ex.what()));
-        try { handle().facet<senf::TCPSocketProtocol>().shutdown(senf::TCPSocketProtocol::ShutRD); }
-        catch (...) {}
-    }
-    catch (...) {
-        SENF_LOG(("unexpected failure writing to socket: unknown exception"));
-        try { handle().facet<senf::TCPSocketProtocol>().shutdown(senf::TCPSocketProtocol::ShutRD); }
-        catch (...) {}
-    }
+    if( sendQueue_.size() > streamBufferMaxSize_)
+        return;
+    sendQueue_.insert( sendQueue_.end(), data.begin(), data.end());
+    writeHandler(scheduler::FdEvent::EV_WRITE);
+    if (! sendQueue_.empty())
+        writeevent_.enable();
+
 }
 
 prefix_ unsigned senf::console::detail::NoninteractiveClientReader::v_width()
@@ -290,6 +295,40 @@ senf::console::detail::NoninteractiveClientReader::newData(int event)
     stream() << std::flush;
 }
 
+prefix_ void
+senf::console::detail::NoninteractiveClientReader::writeHandler(int event)
+{
+    if (event != senf::scheduler::FdEvent::EV_WRITE) {
+        writeevent_.disable();
+        readevent_.disable();
+        try { handle().facet<senf::TCPSocketProtocol>().shutdown(senf::TCPSocketProtocol::ShutRD); }
+        catch (...) {}
+        return;
+    }
+    try {
+        sendQueue_.erase(sendQueue_.begin(),
+                handle().write(boost::make_iterator_range(sendQueue_.begin(), sendQueue_.end())));
+    }
+    catch (senf::ExceptionMixin & ex) {
+        SENF_LOG(("unexpected failure writing to socket:" << ex.message()));
+        try { handle().facet<senf::TCPSocketProtocol>().shutdown(senf::TCPSocketProtocol::ShutRD); }
+        catch (...) {}
+    }
+    catch (std::exception & ex) {
+        SENF_LOG(("unexpected failure writing to socket:" << ex.what()));
+        try { handle().facet<senf::TCPSocketProtocol>().shutdown(senf::TCPSocketProtocol::ShutRD); }
+        catch (...) {}
+    }
+    catch (...) {
+        SENF_LOG(("unexpected failure writing to socket: unknown exception"));
+        try { handle().facet<senf::TCPSocketProtocol>().shutdown(senf::TCPSocketProtocol::ShutRD); }
+        catch (...) {}
+    }
+    if (sendQueue_.empty())
+        writeevent_.disable();
+
+}
+
 //-/////////////////////////////////////////////////////////////////////////////////////////////////
 // senf::console::Client
 
@@ -307,6 +346,7 @@ prefix_ senf::console::Client::Client(Server & server, ClientHandle handle)
       name_ (server.name()), reader_ (), mode_ (server.mode())
 {
     handle_.facet<senf::TCPSocketProtocol>().nodelay();
+    handle_.blocking(false);
     executor_.chroot(root());
     switch (mode_) {
     case Server::Interactive :
@@ -336,7 +376,14 @@ prefix_ void senf::console::Client::setNoninteractive()
     readevent_.disable();
     timer_.disable();
     mode_ = Server::Noninteractive;
-    reader_.reset(new detail::NoninteractiveClientReader(*this));
+    detail::NoninteractiveClientReader * newReader (new detail::NoninteractiveClientReader(*this));
+    reader_.reset( newReader);
+    consoleDir().add( "streamBuffer", senf::console::factory::Command( senf::membind(
+            SENF_MEMFNP( detail::NoninteractiveClientReader::SendQueue::size_type, detail::NoninteractiveClientReader, streamBufferMaxSize, () const ),
+            newReader )));
+    consoleDir().add( "streamBuffer", senf::console::factory::Command( senf::membind(
+            SENF_MEMFNP( void, detail::NoninteractiveClientReader, streamBufferMaxSize, (detail::NoninteractiveClientReader::SendQueue::size_type) ),
+            newReader )));
 }
 
 prefix_ std::string::size_type senf::console::Client::handleInput(std::string data,
@@ -369,7 +416,7 @@ prefix_ std::string::size_type senf::console::Client::handleInput(std::string da
             backtrace_ = msg.substr(0,i);
             msg = msg.substr(i+4);
         } else
-            backtrace_.clear();
+
         stream() << msg << std::endl;
     }
     catch (...) {
@@ -402,7 +449,6 @@ prefix_ unsigned senf::console::Client::getWidth(std::ostream & os, unsigned def
 
 //-/////////////////////////////////////////////////////////////////////////////////////////////////
 // senf::console::Client::SysBacktrace
-
 prefix_ senf::console::Client::SysBacktrace::SysBacktrace()
 {
     sysdir().add("backtrace", factory::Command(&SysBacktrace::backtrace)

diff --git a/senf/Utils/Console/Server.cci b/senf/Utils/Console/Server.cci
index e235c62..8f34fcc 100644 (file)
--- a/senf/Utils/Console/Server.cci
+++ b/senf/Utils/Console/Server.cci
@@ -55,13 +55,13 @@ prefix_ senf::console::detail::ServerManager & senf::console::detail::ServerMana
 }
 
 //-/////////////////////////////////////////////////////////////////////////////////////////////////
-// senf::console::detail::NonblockingSocketSink
+// senf::console::detail::SocketStreamSink
 
-prefix_ senf::console::detail::NonblockingSocketSink::NonblockingSocketSink(Client & client)
+prefix_ senf::console::detail::SocketStreamSink::SocketStreamSink(Client & client)
     : client_ (client)
 {}
 
-prefix_ senf::console::Client & senf::console::detail::NonblockingSocketSink::client()
+prefix_ senf::console::Client & senf::console::detail::SocketStreamSink::client()
     const
 {
     return client_;
@@ -176,7 +176,7 @@ prefix_ unsigned senf::console::Client::width()
 
 prefix_ senf::console::Client & senf::console::Client::get(std::ostream & os)
 {
-    return dynamic_cast<detail::NonblockingSocketOStream&>(os)->client();
+    return dynamic_cast<detail::SocketStreamOStream&>(os)->client();
 }
 
 prefix_ senf::console::Client::ClientHandle senf::console::Client::handle()

diff --git a/senf/Utils/Console/Server.hh b/senf/Utils/Console/Server.hh
index b2cdca6..ac26367 100644 (file)
--- a/senf/Utils/Console/Server.hh
+++ b/senf/Utils/Console/Server.hh
@@ -151,10 +151,10 @@ namespace console {
      */
     class Client
         : public senf::intrusive_refcount,
-          private boost::base_from_member< detail::NonblockingSocketOStream >,
+          private boost::base_from_member< detail::SocketStreamOStream >,
           public senf::log::IOStreamTarget
     {
-        typedef boost::base_from_member< detail::NonblockingSocketOStream > out_t;
+        typedef boost::base_from_member< detail::SocketStreamOStream > out_t;
 
         SENF_LOG_CLASS_AREA();
         SENF_LOG_DEFAULT_LEVEL( senf::log::NOTICE );
@@ -241,7 +241,7 @@ namespace console {
 
         friend class Server;
         friend class detail::ClientReader;
-        friend class detail::NonblockingSocketSink;
+        friend class detail::SocketStreamSink;
 
         class SysBacktrace
         {

diff --git a/senf/Utils/Console/Server.ih b/senf/Utils/Console/Server.ih
index 7c219da..cc66b51 100644 (file)
--- a/senf/Utils/Console/Server.ih
+++ b/senf/Utils/Console/Server.ih
@@ -73,11 +73,11 @@ namespace detail {
 
         \fixme Don't throw exceptions ... set stream error indicator (if at all)
      */
-    class NonblockingSocketSink
+    class SocketStreamSink
         : public boost::iostreams::sink
     {
     public:
-        NonblockingSocketSink(Client & client);
+        SocketStreamSink(Client & client);
         std::streamsize write(const char * s, std::streamsize n);
 
         Client & client() const;
@@ -86,7 +86,7 @@ namespace detail {
         Client & client_;
     };
 
-    typedef boost::iostreams::stream<NonblockingSocketSink> NonblockingSocketOStream;
+    typedef boost::iostreams::stream<SocketStreamSink> SocketStreamOStream;
 
     typedef senf::ServerSocketHandle<
         senf::MakeSocketPolicy< senf::TCPv4SocketProtocol::Policy,
@@ -173,6 +173,11 @@ namespace detail {
     public:
         NoninteractiveClientReader(Client & client);
 
+        typedef std::vector<char> SendQueue;
+
+        void streamBufferMaxSize(SendQueue::size_type size);
+        SendQueue::size_type streamBufferMaxSize() const;
+
     private:
         virtual void v_disablePrompt();
         virtual void v_enablePrompt();
@@ -180,8 +185,12 @@ namespace detail {
         virtual unsigned v_width() const;
 
         void newData(int event);
+        void writeHandler( int event);
 
+        SendQueue::size_type streamBufferMaxSize_;
+        SendQueue sendQueue_;
         scheduler::FdEvent readevent_;
+        scheduler::FdEvent writeevent_;
         std::string buffer_;
     };