From: jmo Date: Tue, 20 Sep 2011 13:18:55 +0000 (+0000) Subject: WORKAROUND: NonInteractive console output stream buffers size configurable (default... X-Git-Url: http://g0dil.de/git?a=commitdiff_plain;h=1342c5d49ca8bd523c9fdf7d89b40448cdfd394b;p=senf.git WORKAROUND: NonInteractive console output stream buffers size configurable (default 1MB) command /sys/log/self { streamBuffer;}; 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 --- 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().shutdown(senf::TCPSocketProtocol::ShutRD); } - catch (...) {} - } - catch (std::exception & ex) { - SENF_LOG(("unexpected failure writing to socket:" << ex.what())); - try { handle().facet().shutdown(senf::TCPSocketProtocol::ShutRD); } - catch (...) {} - } - catch (...) { - SENF_LOG(("unexpected failure writing to socket: unknown exception")); - try { handle().facet().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().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().shutdown(senf::TCPSocketProtocol::ShutRD); } + catch (...) {} + } + catch (std::exception & ex) { + SENF_LOG(("unexpected failure writing to socket:" << ex.what())); + try { handle().facet().shutdown(senf::TCPSocketProtocol::ShutRD); } + catch (...) {} + } + catch (...) { + SENF_LOG(("unexpected failure writing to socket: unknown exception")); + try { handle().facet().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().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 --- 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(os)->client(); + return dynamic_cast(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 --- 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 --- 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 NonblockingSocketOStream; + typedef boost::iostreams::stream SocketStreamOStream; typedef senf::ServerSocketHandle< senf::MakeSocketPolicy< senf::TCPv4SocketProtocol::Policy, @@ -173,6 +173,11 @@ namespace detail { public: NoninteractiveClientReader(Client & client); + typedef std::vector 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_; };