Socket: Fix handle.state() on invalid handles

Utils/Console: Catch exceptions when closing client
PPI: Add throttle tracing
PPI: Fix Event and IOEvent throttling

PPI/Connectors.cc

@@ -84,7 +84,7 @@ prefix_ void senf::ppi::connector::Connector::trace(Packet const & p, char const
         return;
     SENF_LOG_BLOCK(({
                 std::string type (prettyName(p.typeId().id()));
-                log << "PPI trace: " << label << " 0x" << std::hex << p.id() << " " 
+                log << "PPI packet trace: " << label << " 0x" << std::hex << p.id() << " " 
                     << type.substr(21, type.size()-22) << " on " << & module() << " " 
                     << prettyName(typeid(module())) << " connector 0x" << this << "\n";
                 if (traceState_ == TRACE_CONTENTS)
@@ -92,6 +92,17 @@ prefix_ void senf::ppi::connector::Connector::trace(Packet const & p, char const
             }));
 }
 
+prefix_ void senf::ppi::connector::Connector::throttleTrace(char const * label,
+                                                            char const * type)
+{
+    if (traceState_ == NO_TRACING)
+        return;
+    SENF_LOG_BLOCK(({
+                log << "PPI throttling trace: " << label << " " << type << " on " << & module() 
+                    << " " << prettyName(typeid(module())) << " connector 0x" << this << "\n";
+            }));
+}
+
 namespace senf { namespace ppi { namespace connector {
 
     SENF_CONSOLE_REGISTER_ENUM_MEMBER( 
@@ -121,20 +132,23 @@ namespace {
                  "A log message is generated whenever the packet traverses a connector. The\n"
                  "TRACE_IDS log message has the following format:\n"
                  "\n"
-                 "    PPI trace: <packet-id> <packet-type> <direction>\n"
+                 "    PPI packet trace: <direction> <packet-id> <packet-type>\n"
+                 "                      on <module-id> <module-type> connector <connector-id>\n"
+                 "    PPI throttling trace: <direction> <throttle-msg>\n"
                  "                      on <module-id> <module-type> connector <connector-id>\n"
                  "\n"
                  "The fields are:\n"
                  "\n"
+                 "    direction       'IN' for packets/throttle notifications entering the module,\n"
+                 "                    'OUT' for packets/throttle notifications leaving it\n"
                  "    packet-id       Numeric unique packet id. This value is unique for packets\n"
                  "                    alive at the same time, packets at different times may (and\n"
                  "                    will) share id's\n"
                  "    packet-type     The type of the packet header\n"
-                 "    direction       'INCOMING' for packets entering the module, 'OUTGOING' for\n"
-                 "                    packets leaving it\n"
                  "    module-id       Unique module id\n"
                  "    module-type     Type of the module the packet is sent to/from\n"
-                 "    connector-id    Unique connector id\n");
+                 "    connector-id    Unique connector id\n"
+                 "    throttle-msg    Type of throttling event\n");
 
         senf::ppi::ModuleManager::instance().consoleDir()
             .add("tracing", SENF_FNP(void, senf::ppi::connector::Connector::tracing,
@@ -197,7 +211,8 @@ prefix_ void senf::ppi::connector::PassiveConnector::notifyUnthrottle()
         remoteThrottled_ = false;
         if (!nativeThrottled_)
             emitUnthrottle();
-    }
+    } else
+        throttleTrace("OUT", "not forwarding unthrottle event");
 }
 
 ///////////////////////////////////////////////////////////////////////////
@@ -214,6 +229,7 @@ prefix_ void senf::ppi::connector::ActiveConnector::v_init()
 
 prefix_ void senf::ppi::connector::ActiveConnector::notifyThrottle()
 {
+    throttleTrace("IN ", "throttle");
     if (! throttled_) {
         throttled_ = true;
         if (throttleCallback_)
@@ -227,6 +243,7 @@ prefix_ void senf::ppi::connector::ActiveConnector::notifyThrottle()
 
 prefix_ void senf::ppi::connector::ActiveConnector::notifyUnthrottle()
 {
+    throttleTrace("IN ", "unthrottle");
     if (throttled_) {
         throttled_ = false;
         if (unthrottleCallback_)
@@ -256,7 +273,7 @@ prefix_ senf::Packet senf::ppi::connector::InputConnector::operator()()
         queue_.pop_back();
         v_dequeueEvent();
     }
-    trace(p, "INCOMING");
+    trace(p, "IN ");
     return p;
 }
 

PPI/Connectors.cci

@@ -110,12 +110,14 @@ prefix_ bool senf::ppi::connector::PassiveConnector::throttled()
 
 prefix_ void senf::ppi::connector::PassiveConnector::emitThrottle()
 {
+    throttleTrace("OUT", "throttle");
     if (connected())
         peer().notifyThrottle();
 }
 
 prefix_ void senf::ppi::connector::PassiveConnector::emitUnthrottle()
 {
+    throttleTrace("OUT", "unthrottle");
     if (connected()) {
         peer().notifyUnthrottle();
         v_unthrottleEvent();
@@ -150,7 +152,8 @@ prefix_ void senf::ppi::connector::PassiveConnector::throttle()
     if (!throttled()) {
         nativeThrottled_ = true;
         emitThrottle();
-    }
+    } else
+        nativeThrottled_ = true;
 }
 
 prefix_ void senf::ppi::connector::PassiveConnector::unthrottle()
@@ -176,6 +179,8 @@ prefix_ void senf::ppi::connector::PassiveConnector::emit()
     SENF_ASSERT(callback_ && "senf::ppi::connector::PassiveConnector: missing onRequest()");
     if (!throttled())
         callback_();
+    else
+        throttleTrace("IN ", "queueing packet");
 }
 
 ///////////////////////////////////////////////////////////////////////////
@@ -286,7 +291,7 @@ prefix_ senf::ppi::connector::InputConnector & senf::ppi::connector::OutputConne
 
 prefix_ void senf::ppi::connector::OutputConnector::operator()(Packet const & p)
 {
-    trace(p, "OUTGOING");
+    trace(p, "OUT");
     if (connected())
         peer().enqueue(p);
 }

PPI/Connectors.hh

@@ -183,6 +183,7 @@ namespace connector {
         void connect(Connector & target);
 
         void trace(Packet const & p, char const * label);
+        void throttleTrace(char const * label, char const * type);
         
     private:
         virtual std::type_info const & packetTypeID();
@@ -247,11 +248,11 @@ namespace connector {
     private:
         virtual void v_init();
 
-        // Called by the routing to change the remote throttling state
+        // Called by the routing to change the throttling state from forwarding routes
         void notifyThrottle();          ///< Forward a throttle notification to this connector
         void notifyUnthrottle();        ///< Forward an unthrottle notification to this connector
 
-        // Internal members to emit throttling notifications
+        // Internal members to emit throttling notifications to the connected peer
         void emitThrottle();
         void emitUnthrottle();
 

PPI/Events.cc

@@ -43,13 +43,17 @@ prefix_ void senf::ppi::EventDescriptor::notifyUnthrottle()
     for (; i != i_end; ++i)
         if ((*i)->throttled())
             break;
-    if (i == i_end) 
-        enabled(true);
+    if (i != i_end) 
+        return;
+    throttled_ = false;
+    enabled(true);
 }
 
 prefix_ void senf::ppi::EventDescriptor::enabled(bool v)
 {
     SENF_ASSERT(v_isRegistered() && "Module::registerEvent() call missing");
+    if (throttled_ && v)
+        return;
     if (v && ! enabled_)
         v_enable();
     else if (! v && enabled_)

PPI/Events.cci

@@ -44,7 +44,7 @@ prefix_ bool senf::ppi::EventDescriptor::enabled()
 // protected members
 
 prefix_ senf::ppi::EventDescriptor::EventDescriptor()
-    : enabled_(false)
+    : enabled_(false), throttled_(false)
 {}
 
 ////////////////////////////////////////
@@ -52,6 +52,7 @@ prefix_ senf::ppi::EventDescriptor::EventDescriptor()
 
 prefix_ void senf::ppi::EventDescriptor::notifyThrottle()
 {
+    throttled_ = true;
     enabled(false);
 }
 

PPI/Events.hh

@@ -87,6 +87,7 @@ namespace ppi {
         void registerRoute(ForwardingRoute & route);
 
         bool enabled_;
+        bool throttled_;
 
         typedef std::vector<ForwardingRoute*> Routes;
         Routes routes_;

PPI/IOEvent.ct

@@ -36,7 +36,10 @@ prefix_ void senf::ppi::IOEvent::set(Handle handle, unsigned events)
     if (handle) {
         fd_ = senf::scheduler::get_descriptor(handle);
         event_.events(events).handle(fd_);
-        event_.enable();
+        if (enabled())
+            event_.enable();
+        else
+            event_.disable();
     }
     else {
         event_.disable();

PPI/ModuleManager.cc

@@ -80,7 +80,13 @@ prefix_ senf::ppi::ModuleManager::ModuleManager()
         .add("dump", senf::membind(&ModuleManager::dumpModules, this))
         .doc("Dump complete PPI structure\n"
              "The dump will contain one paragraph for each module. The first line gives module\n"
-             "information, additional lines list all connectors and their peers (if connected).");
+             "information, additional lines list all connectors and their peers (if connected).\n"
+             "\n"
+             "This information can be processed by 'PPI/drawmodules.py' and 'dot' (from the\n"
+             "graphviz package) to generate a graphic representation of the module structure:\n"
+             "\n"
+             "    $ echo /sys/ppi/dump | nc -q1 <host> <port> \\\n"
+             "          | python PPI/drawmodules.py | dot -Tpng /dev/fd/0 >modules.png\n");
 }
 
 prefix_ void senf::ppi::ModuleManager::dumpModules(std::ostream & os)

PPI/SocketSink.ct

@@ -125,7 +125,7 @@ prefix_ void senf::ppi::module::PassiveSocketSink<Writer>::write()
 template <class Writer>
 prefix_ void senf::ppi::module::PassiveSocketSink<Writer>::checkThrottle()
 {
-    if (handle_)
+    if (handle_.valid())
         input.unthrottle();
     else
         input.throttle();

Socket/ClientSocketHandle.cti

@@ -377,7 +377,11 @@ template <class SPolicy>
 prefix_ void senf::ClientSocketHandle<SPolicy>::state(SocketStateMap & map, unsigned lod)
 {
     map["handle"] = prettyName(typeid(*this));
-    this->body().state(map, lod);
+    if (this->valid()) {
+        map["valid"] << "true";
+        this->body().state(map,lod);
+    } else
+        map["valid"] << "false";
 }
 
 template <class SPolicy>

Socket/ProtocolClientSocketHandle.cti

@@ -91,7 +91,11 @@ senf::ProtocolClientSocketHandle<SocketProtocol>::state(SocketStateMap & map,
                                                                unsigned lod)
 {
     map["handle"] = prettyName(typeid(*this));
-    this->body().state(map,lod);
+    if (this->valid()) {
+        map["valid"] << "true";
+        this->body().state(map,lod);
+    } else
+        map["valid"] << "false";
 }
 
 template <class SocketProtocol>

Socket/ProtocolServerSocketHandle.cti

@@ -91,7 +91,11 @@ senf::ProtocolServerSocketHandle<SocketProtocol>::state(SocketStateMap & map,
                                                                unsigned lod)
 {
     map["handle"] = prettyName(typeid(*this));
-    this->body().state(map,lod);
+    if (this->valid()) {
+        map["valid"] << "true";
+        this->body().state(map,lod);
+    } else
+        map["valid"] << "false";
 }
 
 template <class SocketProtocol>

Socket/ServerSocketHandle.cti

@@ -147,7 +147,11 @@ template <class SPolicy>
 prefix_ void senf::ServerSocketHandle<SPolicy>::state(SocketStateMap & map, unsigned lod)
 {
     map["handle"] = prettyName(typeid(*this));
-    this->body().state(map,lod);
+    if (this->valid()) {
+        map["valid"] << "true";
+        this->body().state(map,lod);
+    } else
+        map["valid"] << "false";
 }
 
 template <class SPolicy>

Socket/SocketHandle.cti

@@ -175,7 +175,11 @@ prefix_ void senf::SocketHandle<SPolicy>::state(SocketStateMap & map, unsigned l
     // the type name and therefore show the \e static policy of the
     // socket handle.
     map["handle"] << prettyName(typeid(*this));
-    body().state(map,lod);
+    if (valid()) {
+        map["valid"] << "true";
+        body().state(map,lod);
+    } else
+        map["valid"] << "false";
 }
 
 template <class SPolicy>

Utils/Console/Server.cc

@@ -111,7 +111,15 @@ prefix_ void senf::console::Server::newClient(int event)
 
 prefix_ void senf::console::Server::removeClient(Client & client)
 {
-    SENF_LOG(( "Disposing client " << client.handle().peer() ));
+    SENF_LOG_BLOCK(({
+                log << "Disposing client ";
+                try {
+                    log << client.handle().peer();
+                }
+                catch (senf::SystemException ex) {
+                    log << "(unknown)";
+                }
+            }));
     // THIS DELETES THE CLIENT INSTANCE !!
     clients_.erase(boost::intrusive_ptr<Client>(&client));
 }