Running the example is a little bit more complicated since we need to provide example UDP
packets so we can see the application at work. We do this using \c netcat. We open several shell
- windows and run the folling commands, each in it's own window
+ windows and run the following commands, each in it's own window
The first command listens for incoming UDP packets on port 44345:
<pre>
</pre>
Whenever we send out a packet with CR in the last window we should see it appear in the first
- one. If we send out packets faster than 1 packet per seccond, they will still only be forwarded
- with a speed of 1 packet / seccond. If the kernel UDP buffer overflows, packets would be dropped
- (however, this buffer is prohibitively large)
+ one. If we send out packets faster than 1 packet per second, they will start to be discarded if
+ more than two packets are in flight.
\image html screenshot.png
senf::ppi::module::ActiveSocketReader reads the incoming UDP packets and sends them into a
senf::ppi::module::PassiveQueue.
- The \a queue feeds the packets into a senf::ppi::module::PriorityJoin. The CopyPacketGenerator
+ The \a queue feeds the packets into a senf::ppi::module::PriorityJoin. The CloneSource
\a generator is fed as second input into the \a join to provide the stuffing packets.
The RateFilter \a rateFilter reads packets from it's input at a fixed rate and writes them into
\skip #include
\until Setup
- We also define some namepsace aliases
+ We also define some namespace aliases
\skip namespace
\until senf::ppi;
- The RateStuffer application is based on two additional modules.
+ The RateStuffer application is based on one additional application module.
\subsection ratefilter The RateFilter module
\until }
- The event is initialized to fire eery \a interval nanoseconds. The traffic is routed 'accross'
+ The event is initialized to fire eery \a interval nanoseconds. The traffic is routed 'across'
the timer which controls the traffic. The timer is then registered to call
RateFilter::timeout().
automatically be disabled if either \a input or \a output become throttled (which is not even
needed here).
- \subsection generator The CopyPacketGenerator
+ \subsection ratestuffer The RateStuffer subnet
- We need an additional module to provide the stuffing packets. The CopyPacketGenerator provides
- an unlimited stream of clone's of a template packet.
+ We decide to implement the RateStuffer as a subnet or collection. This is a simple struct or
+ class which contains all the modules necessary for a specific functionality. The modules are
+ initialized and connected in the class's constructor. External connectors are exported as
+ references to the corresponding module connectors:
\skip class
- \until };
+ \until rateFilter
+
+ First the needed modules are declared. We have
+ \li the \a barrier to discard incoming packets sent to fast
+ \li the \a queue to receive incoming packets and create throttling notifications
+ \li the \a generator to create the stuffing packets
+ \li the \a join to combine the input stream from the \a queue with the stuffing packet stream
+ \li the \a rateFilter to generate the fixed rate output stream
- This module is very simple.
+ \until output
- \until }
+ Here we declare the external connectors. The subnetwork exports a single input and output
+ connector. The external connectors are declared as \e references.
- We save the template packet and initialize the \a output connector to call
- CopyPacketGenerator::request() whenever a packet is requested.
+ \until output
- \until }
+ The constructor now initializes all the local objects. We pass the template \a packet to the \a
+ generator and set the timing \a interval of the \a rateFilter.
+
+ The \a input and \a output connector references are bound to the corresponding connectors we
+ want to expose: \a input to the \a queue's \a input and \a output to the \a rateStuffer's \a
+ output.
+
+ \until };
- The handler just provides \a packet on each request. This is ok as long as the packets are not
- changed, in which case we would return <tt>packet->clone()</tt> instead.
+ The constructor body sets up the connections within the subnetwork. Finally, we set the queueing
+ discipline of the \a queue. This Completes the RateStuffer. This subnetwork can now be used like
+ a module.
\subsection main Connecting the modules
\until udpWriter
- Next all the modules are allocated:
+ Here we allocate the components:
\li \a udpReader to read the packets from \a inputSocket
- \li \a queue to convert the active output of senf::ppi::module::ActiveSocketReader into a
- passive output
- \li \a generator to provide the stuffing in the form of packets containing <tt>"<idle>\n"</tt>
- \li \a join to combine the two packet streams (from \a udpReader and from \a generator) into a
- single stream
- \li \a rateFilter to generate the fixed rate packet stream of 1 packet every 1000000000
- nanoseconds and
+ \li \a stuffer for the real work and
\li \a udpWriter to send the packets to \a outputSocket
\until udpWriter
- The \ref senf::ppi::connect() calls now setup the necessary connections.
+ The \ref senf::ppi::connect() calls setup the necessary connections.
\until run
#include "PPI/Module.hh"
#include "PPI/IntervalTimer.hh"
#include "PPI/Joins.hh"
+#include "PPI/ThrottleBarrier.hh"
#include "PPI/PassiveQueue.hh"
+#include "PPI/Queueing.hh"
+#include "PPI/CloneSource.hh"
#include "PPI/Setup.hh"
//#include "ppitest.mpp"
}
// ////////////////////////////////////////////////////////////////////////
-// CopyPacketGenerator
-class CopyPacketGenerator
- : public module::Module
+class RateStuffer
{
- SENF_PPI_MODULE(CopyPacketGenerator);
-public:
-
- connector::PassiveOutput output;
-
- CopyPacketGenerator(senf::Packet p);
+ module::ThrottleBarrier barrier;
+ module::PassiveQueue queue;
+ module::CloneSource generator;
+ module::PriorityJoin join;
+ RateFilter rateFilter;
-private:
- void request();
-
- senf::Packet packet;
+public:
+ connector::PassiveInput & input;
+ connector::ActiveOutput & output;
+
+ RateStuffer(senf::ClockService::clock_type interval,
+ senf::Packet packet,
+ unsigned high = 1,
+ unsigned low = 0)
+ : barrier (),
+ queue (),
+ generator ( packet ),
+ join (),
+ rateFilter ( interval ),
+ input ( barrier.input ),
+ output ( rateFilter.output )
+ {
+ ppi::connect( barrier, queue );
+ ppi::connect( queue, join );
+ ppi::connect( generator, join );
+ ppi::connect( join, rateFilter );
+
+ queue.qdisc(ppi::ThresholdQueueing(high,low));
+ }
};
-
-CopyPacketGenerator::CopyPacketGenerator(senf::Packet p)
- : packet(p)
-{
- noroute(output);
- output.onRequest(&CopyPacketGenerator::request);
-}
-
-void CopyPacketGenerator::request()
-{
- output(packet);
-}
-
+
// ////////////////////////////////////////////////////////////////////////
// ////////////////////////////////////////////////////////////////////////
// 'O' = active connector
// '>' or '<' = input connector
//
-// [ udpReader ] O--> [ queue ] -->O [ ]
-// [ join ] -->O [ rateFilter] O--> [ udpWriter ]
-// [ generator ] -->O [ ]
+// +----------------------------------------------------+
+// | stuffer |
+// | |
+// [ udpReader ] O-->:---> [ queue ] -->O [ ] |
+// | [ join ] -->O [ rateFilter] O-->:O--> [ udpWriter ]
+// | [ generator ] -->O [ ] |
+// | |
+// +----------------------------------------------------+
int main(int argc, char * argv[])
{
senf::ConnectedUDPv4ClientSocketHandle outputSocket(
senf::INet4SocketAddress("localhost:44345"));
- module::ActiveSocketReader<> udpReader (inputSocket);
- module::PassiveQueue queue;
- CopyPacketGenerator generator (senf::DataPacket::create(std::string("<idle>\n")));
- module::PriorityJoin join;
- RateFilter rateFilter (1000000000ul);
- module::PassiveSocketWriter<> udpWriter (outputSocket);
-
- ppi::connect( udpReader, queue );
- ppi::connect( queue, join );
- ppi::connect( generator, join );
- ppi::connect( join, rateFilter );
- ppi::connect( rateFilter, udpWriter );
+ module::ActiveSocketReader<> udpReader ( inputSocket );
+ RateStuffer stuffer ( 1000000000ul,
+ senf::DataPacket::create(std::string("<idle>\n")),
+ 2u, 1u );
+ module::PassiveSocketWriter<> udpWriter ( outputSocket );
+
+ ppi::connect( udpReader, stuffer );
+ ppi::connect( stuffer, udpWriter );
ppi::run();
\note For this Module to work correctly, it is very important for the connectors to be
correctly throttled. Otherwise the system might well hang in an endless loop.
+
+ \ingroup adapter_modules
*/
class ActiveFeeder
: public Module
--- /dev/null
+// $Id$
+//
+// Copyright (C) 2007
+// Fraunhofer Institut fuer offene Kommunikationssysteme (FOKUS)
+// Kompetenzzentrum fuer Satelitenkommunikation (SatCom)
+// Stefan Bund <g0dil@berlios.de>
+//
+// This program is free software; you can redistribute it and/or modify
+// it under the terms of the GNU General Public License as published by
+// the Free Software Foundation; either version 2 of the License, or
+// (at your option) any later version.
+//
+// This program is distributed in the hope that it will be useful,
+// but WITHOUT ANY WARRANTY; without even the implied warranty of
+// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+// GNU General Public License for more details.
+//
+// You should have received a copy of the GNU General Public License
+// along with this program; if not, write to the
+// Free Software Foundation, Inc.,
+// 59 Temple Place - Suite 330, Boston, MA 02111-1307, USA.
+
+/** \file
+ \brief CloneSource non-inline non-template implementation */
+
+#include "CloneSource.hh"
+//#include "CloneSource.ih"
+
+// Custom includes
+
+//#include "CloneSource.mpp"
+#define prefix_
+///////////////////////////////cc.p////////////////////////////////////////
+
+senf::ppi::module::CloneSource::CloneSource(senf::Packet packet)
+ : packet_(packet)
+{
+ noroute(output);
+ output.onRequest(&CloneSource::request);
+}
+
+void senf::ppi::module::CloneSource::request()
+{
+ output(packet_.clone());
+}
+
+
+///////////////////////////////cc.e////////////////////////////////////////
+#undef prefix_
+//#include "CloneSource.mpp"
+
+\f
+// Local Variables:
+// mode: c++
+// fill-column: 100
+// comment-column: 40
+// c-file-style: "senf"
+// indent-tabs-mode: nil
+// ispell-local-dictionary: "american"
+// compile-command: "scons -u test"
+// End:
--- /dev/null
+// $Id$
+//
+// Copyright (C) 2007
+// Fraunhofer Institut fuer offene Kommunikationssysteme (FOKUS)
+// Kompetenzzentrum fuer Satelitenkommunikation (SatCom)
+// Stefan Bund <g0dil@berlios.de>
+//
+// This program is free software; you can redistribute it and/or modify
+// it under the terms of the GNU General Public License as published by
+// the Free Software Foundation; either version 2 of the License, or
+// (at your option) any later version.
+//
+// This program is distributed in the hope that it will be useful,
+// but WITHOUT ANY WARRANTY; without even the implied warranty of
+// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+// GNU General Public License for more details.
+//
+// You should have received a copy of the GNU General Public License
+// along with this program; if not, write to the
+// Free Software Foundation, Inc.,
+// 59 Temple Place - Suite 330, Boston, MA 02111-1307, USA.
+
+/** \file
+ \brief CloneSource public header */
+
+#ifndef HH_CloneSource_
+#define HH_CloneSource_ 1
+
+// Custom includes
+#include "Packets/Packets.hh"
+#include "Module.hh"
+#include "Connectors.hh"
+
+//#include "CloneSource.mpp"
+///////////////////////////////hh.p////////////////////////////////////////
+
+namespace senf {
+namespace ppi {
+namespace module {
+
+ /** \brief Generate clone's of a template packet
+
+ CloneSource will provide clone's of a template \a packet on it's \a output.
+
+ \ingroup sourcesink_modules
+ */
+ class CloneSource
+ : public Module
+ {
+ SENF_PPI_MODULE(CloneSource);
+ public:
+
+ connector::PassiveOutput output;
+
+ CloneSource(senf::Packet packet);
+
+ private:
+ void request();
+
+ senf::Packet packet_;
+ };
+
+}}}
+
+///////////////////////////////hh.e////////////////////////////////////////
+//#include "CloneSource.cci"
+//#include "CloneSource.ct"
+//#include "CloneSource.cti"
+#endif
+
+\f
+// Local Variables:
+// mode: c++
+// fill-column: 100
+// comment-column: 40
+// c-file-style: "senf"
+// indent-tabs-mode: nil
+// ispell-local-dictionary: "american"
+// compile-command: "scons -u test"
+// End:
--- /dev/null
+// $Id$
+//
+// Copyright (C) 2007
+// Fraunhofer Institut fuer offene Kommunikationssysteme (FOKUS)
+// Kompetenzzentrum fuer Satelitenkommunikation (SatCom)
+// Stefan Bund <g0dil@berlios.de>
+//
+// This program is free software; you can redistribute it and/or modify
+// it under the terms of the GNU General Public License as published by
+// the Free Software Foundation; either version 2 of the License, or
+// (at your option) any later version.
+//
+// This program is distributed in the hope that it will be useful,
+// but WITHOUT ANY WARRANTY; without even the implied warranty of
+// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+// GNU General Public License for more details.
+//
+// You should have received a copy of the GNU General Public License
+// along with this program; if not, write to the
+// Free Software Foundation, Inc.,
+// 59 Temple Place - Suite 330, Boston, MA 02111-1307, USA.
+
+/** \file
+ \brief CloneSource.test unit tests */
+
+//#include "CloneSource.test.hh"
+//#include "CloneSource.test.ih"
+
+// Custom includes
+#include "CloneSource.hh"
+#include "DebugModules.hh"
+#include "Setup.hh"
+#include "Packets/Packets.hh"
+
+#include <boost/test/auto_unit_test.hpp>
+#include <boost/test/test_tools.hpp>
+
+#define prefix_
+///////////////////////////////cc.p////////////////////////////////////////
+
+BOOST_AUTO_UNIT_TEST(cloneSource)
+{
+ char data[] = { 0xAB };
+ senf::Packet p (senf::DataPacket::create(data));
+
+ senf::ppi::module::CloneSource source (p);
+ senf::ppi::module::debug::ActiveSink sink;
+
+ senf::ppi::connect(source,sink);
+ senf::ppi::init();
+
+ BOOST_CHECK( sink.request() != p );
+ BOOST_CHECK( sink.request().data()[0] == p.data()[0] );
+ BOOST_CHECK( sink.request().data()[0] == p.data()[0] );
+}
+
+
+///////////////////////////////cc.e////////////////////////////////////////
+#undef prefix_
+
+\f
+// Local Variables:
+// mode: c++
+// fill-column: 100
+// comment-column: 40
+// c-file-style: "senf"
+// indent-tabs-mode: nil
+// ispell-local-dictionary: "american"
+// compile-command: "scons -u test"
+// End:
/** \file
\brief Connectors public header */
-/** \defgroup connectors Connector classes
-
- A connector has two independent properties
- \li it may be \e active or \e passive
- \li it may be an \e input or an \e output
-
- \e Active connectors are activated from within the module, \e passive connectors are signaled by
- the external framework. \e Input modules receive packets, \e output modules send packets.
-
- All passive connectors call some onRequest callback whenever I/O needs to be performed. All
- input modules possess a packet queue.
-
- We therefore have 4 connector types: senf::ppi::ActiveInput, senf::ppi::ActiveOutput,
- senf::ppi::PassiveInput and senf::ppi::PassiveOutput.
- */
-
#ifndef HH_Connectors_
#define HH_Connectors_ 1
/** \namespace senf::ppi::connector
\brief Connector classes
+
+ A connector has two independent properties
+ \li it may be \e active or \e passive
+ \li it may be an \e input or an \e output
+
+ \e Active connectors are activated from within the module, \e passive connectors are
+ signaled by the external framework. \e Input modules receive packets, \e output modules send
+ packets.
+
+ All passive connectors call some onRequest callback whenever I/O needs to be performed. All
+ input modules possess a packet queue.
+
+ We therefore have 4 connector types: senf::ppi::ActiveInput, senf::ppi::ActiveOutput,
+ senf::ppi::PassiveInput and senf::ppi::PassiveOutput.
*/
/** \brief Connector base-class
--- /dev/null
+// $Id$
+//
+// Copyright (C) 2007
+// Fraunhofer Institut fuer offene Kommunikationssysteme (FOKUS)
+// Kompetenzzentrum fuer Satelitenkommunikation (SatCom)
+// Stefan Bund <g0dil@berlios.de>
+//
+// This program is free software; you can redistribute it and/or modify
+// it under the terms of the GNU General Public License as published by
+// the Free Software Foundation; either version 2 of the License, or
+// (at your option) any later version.
+//
+// This program is distributed in the hope that it will be useful,
+// but WITHOUT ANY WARRANTY; without even the implied warranty of
+// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+// GNU General Public License for more details.
+//
+// You should have received a copy of the GNU General Public License
+// along with this program; if not, write to the
+// Free Software Foundation, Inc.,
+// 59 Temple Place - Suite 330, Boston, MA 02111-1307, USA.
+
+/** \file
+ \brief DebugEvent.test unit tests */
+
+//#include "DebugEvent.test.hh"
+//#include "DebugEvent.test.ih"
+
+// Custom includes
+#include "DebugEvent.hh"
+
+#include <boost/test/auto_unit_test.hpp>
+#include <boost/test/test_tools.hpp>
+
+#define prefix_
+///////////////////////////////cc.p////////////////////////////////////////
+
+BOOST_AUTO_UNIT_TEST(debugEvent)
+{
+ // Tested in Module.test.cc
+}
+
+///////////////////////////////cc.e////////////////////////////////////////
+#undef prefix_
+
+\f
+// Local Variables:
+// mode: c++
+// fill-column: 100
+// comment-column: 40
+// c-file-style: "senf"
+// indent-tabs-mode: nil
+// ispell-local-dictionary: "american"
+// compile-command: "scons -u test"
+// End:
output.onRequest(&PassiveSource::request);
}
+prefix_ void senf::ppi::module::debug::PassiveSource::throttle()
+{
+ output.throttle();
+}
+
+prefix_ void senf::ppi::module::debug::PassiveSource::unthrottle()
+{
+ output.unthrottle();
+}
+
prefix_ void senf::ppi::module::debug::PassiveSource::submit(Packet packet)
{
packets_.push_back(packet);
input.onRequest(&PassiveSink::request);
}
+prefix_ void senf::ppi::module::debug::PassiveSink::throttle()
+{
+ input.throttle();
+}
+
+prefix_ void senf::ppi::module::debug::PassiveSink::unthrottle()
+{
+ input.unthrottle();
+}
+
prefix_ bool senf::ppi::module::debug::PassiveSink::empty()
{
return packets_.empty();
connector::PassiveOutput output;
PassiveSource();
+
+ void throttle(); ///< Throttle output connector
+ void unthrottle(); ///< Unthrottle output connector
void submit(Packet packet); ///< Enqueue packet
typedef Queue::const_iterator iterator;
connector::PassiveInput input;
-
+
PassiveSink();
+ void throttle(); ///< Throttle input connection
+ void unthrottle(); ///< Unthrottle input connection
+
bool empty(); ///< \c true, if queue is empty
size_type size(); ///< number of packets in the queue
iterator begin(); ///< begin iterator of packets in the queue
--- /dev/null
+// $Id$
+//
+// Copyright (C) 2007
+// Fraunhofer Institut fuer offene Kommunikationssysteme (FOKUS)
+// Kompetenzzentrum fuer Satelitenkommunikation (SatCom)
+// Stefan Bund <g0dil@berlios.de>
+//
+// This program is free software; you can redistribute it and/or modify
+// it under the terms of the GNU General Public License as published by
+// the Free Software Foundation; either version 2 of the License, or
+// (at your option) any later version.
+//
+// This program is distributed in the hope that it will be useful,
+// but WITHOUT ANY WARRANTY; without even the implied warranty of
+// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+// GNU General Public License for more details.
+//
+// You should have received a copy of the GNU General Public License
+// along with this program; if not, write to the
+// Free Software Foundation, Inc.,
+// 59 Temple Place - Suite 330, Boston, MA 02111-1307, USA.
+
+/** \file
+ \brief DiscardSink non-inline non-template implementation */
+
+#include "DiscardSink.hh"
+//#include "DiscardSink.ih"
+
+// Custom includes
+
+//#include "DiscardSink.mpp"
+#define prefix_
+///////////////////////////////cc.p////////////////////////////////////////
+
+prefix_ senf::ppi::module::DiscardSink::DiscardSink()
+{
+ noroute(input);
+ input.onRequest(&DiscardSink::request);
+}
+
+prefix_ void senf::ppi::module::DiscardSink::request()
+{
+ (void) input();
+}
+
+///////////////////////////////cc.e////////////////////////////////////////
+#undef prefix_
+//#include "DiscardSink.mpp"
+
+\f
+// Local Variables:
+// mode: c++
+// fill-column: 100
+// comment-column: 40
+// c-file-style: "senf"
+// indent-tabs-mode: nil
+// ispell-local-dictionary: "american"
+// compile-command: "scons -u test"
+// End:
--- /dev/null
+// $Id$
+//
+// Copyright (C) 2007
+// Fraunhofer Institut fuer offene Kommunikationssysteme (FOKUS)
+// Kompetenzzentrum fuer Satelitenkommunikation (SatCom)
+// Stefan Bund <g0dil@berlios.de>
+//
+// This program is free software; you can redistribute it and/or modify
+// it under the terms of the GNU General Public License as published by
+// the Free Software Foundation; either version 2 of the License, or
+// (at your option) any later version.
+//
+// This program is distributed in the hope that it will be useful,
+// but WITHOUT ANY WARRANTY; without even the implied warranty of
+// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+// GNU General Public License for more details.
+//
+// You should have received a copy of the GNU General Public License
+// along with this program; if not, write to the
+// Free Software Foundation, Inc.,
+// 59 Temple Place - Suite 330, Boston, MA 02111-1307, USA.
+
+/** \file
+ \brief DiscardSink public header */
+
+#ifndef HH_DiscardSink_
+#define HH_DiscardSink_ 1
+
+// Custom includes
+#include "Connectors.hh"
+#include "Module.hh"
+
+//#include "DiscardSink.mpp"
+///////////////////////////////hh.p////////////////////////////////////////
+
+namespace senf {
+namespace ppi {
+namespace module {
+
+ /** \brief Module discarding all received packets
+
+ DiscardSink will accept any number of packets and will silently discard them.
+
+ \ingroup sourcesink_modules
+ */
+ class DiscardSink
+ : public Module
+ {
+ SENF_PPI_MODULE(DiscardSink);
+ public:
+ connector::PassiveInput input;
+
+ DiscardSink();
+
+ private:
+ void request();
+ };
+
+
+}}}
+
+///////////////////////////////hh.e////////////////////////////////////////
+//#include "DiscardSink.cci"
+//#include "DiscardSink.ct"
+//#include "DiscardSink.cti"
+#endif
+
+\f
+// Local Variables:
+// mode: c++
+// fill-column: 100
+// comment-column: 40
+// c-file-style: "senf"
+// indent-tabs-mode: nil
+// ispell-local-dictionary: "american"
+// compile-command: "scons -u test"
+// End:
--- /dev/null
+// $Id$
+//
+// Copyright (C) 2007
+// Fraunhofer Institut fuer offene Kommunikationssysteme (FOKUS)
+// Kompetenzzentrum fuer Satelitenkommunikation (SatCom)
+// Stefan Bund <g0dil@berlios.de>
+//
+// This program is free software; you can redistribute it and/or modify
+// it under the terms of the GNU General Public License as published by
+// the Free Software Foundation; either version 2 of the License, or
+// (at your option) any later version.
+//
+// This program is distributed in the hope that it will be useful,
+// but WITHOUT ANY WARRANTY; without even the implied warranty of
+// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+// GNU General Public License for more details.
+//
+// You should have received a copy of the GNU General Public License
+// along with this program; if not, write to the
+// Free Software Foundation, Inc.,
+// 59 Temple Place - Suite 330, Boston, MA 02111-1307, USA.
+
+/** \file
+ \brief DiscardSink.test unit tests */
+
+//#include "DiscardSink.test.hh"
+//#include "DiscardSink.test.ih"
+
+// Custom includes
+#include "DiscardSink.hh"
+#include "DebugModules.hh"
+#include "Setup.hh"
+
+#include <boost/test/auto_unit_test.hpp>
+#include <boost/test/test_tools.hpp>
+
+#define prefix_
+///////////////////////////////cc.p////////////////////////////////////////
+
+BOOST_AUTO_UNIT_TEST(discardSink)
+{
+ // here we really can just validate that everything compiles ...
+
+ senf::ppi::module::debug::ActiveSource source;
+ senf::ppi::module::DiscardSink sink;
+
+ senf::ppi::connect(source,sink);
+ senf::ppi::init();
+
+ senf::Packet p (senf::DataPacket::create());
+
+ source.submit(p);
+ source.submit(p);
+}
+
+///////////////////////////////cc.e////////////////////////////////////////
+#undef prefix_
+
+\f
+// Local Variables:
+// mode: c++
+// fill-column: 100
+// comment-column: 40
+// c-file-style: "senf"
+// indent-tabs-mode: nil
+// ispell-local-dictionary: "american"
+// compile-command: "scons -u test"
+// End:
namespace senf {
namespace ppi {
+
+ /** \defgroup event_group Events
+
+ Events provide notification of events outside the PPI framework: I/O activity, Timers
+ etc. Events are very important since they drive the PPI: Without events, nothing will
+ happen.
+
+ \section event_impl Implementing Events
+
+ All events are derived from EventImplementation which is based on EventDescriptor.
+ \see EventImplementation
+ */
// Implementation: The concrete EventDescriptor implementation will need to set things up so
// some callback (within the EventDescriptor implementation) will be called when the event
type of event is specified using the \a events mask with values from EventFlags.
\fixme Implement error/EOF handling
+
+ \see IOEventInfo
+
+ \ingroup event_group
*/
class IOEvent
: public EventImplementation<IOEventInfo>
An IdleEvent is signaled continually and repeatedly while enabled. It will consume 100% of
available CPU resources. The resource usage is controlled by adequate event throttling.
+
+ \ingroup event_group
*/
class IdleEvent
: public EventImplementation<>
--- /dev/null
+// $Id$
+//
+// Copyright (C) 2007
+// Fraunhofer Institut fuer offene Kommunikationssysteme (FOKUS)
+// Kompetenzzentrum fuer Satelitenkommunikation (SatCom)
+// Stefan Bund <g0dil@berlios.de>
+//
+// This program is free software; you can redistribute it and/or modify
+// it under the terms of the GNU General Public License as published by
+// the Free Software Foundation; either version 2 of the License, or
+// (at your option) any later version.
+//
+// This program is distributed in the hope that it will be useful,
+// but WITHOUT ANY WARRANTY; without even the implied warranty of
+// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+// GNU General Public License for more details.
+//
+// You should have received a copy of the GNU General Public License
+// along with this program; if not, write to the
+// Free Software Foundation, Inc.,
+// 59 Temple Place - Suite 330, Boston, MA 02111-1307, USA.
+
+/** \file
+ \brief IdleEvent.test unit tests */
+
+//#include "IdleEvent.test.hh"
+//#include "IdleEvent.test.ih"
+
+// Custom includes
+#include "IdleEvent.hh"
+
+#include <boost/test/auto_unit_test.hpp>
+#include <boost/test/test_tools.hpp>
+
+#define prefix_
+///////////////////////////////cc.p////////////////////////////////////////
+
+BOOST_AUTO_UNIT_TEST(idleEvent)
+{
+ // Tested by ActiveFeeder.test.cc
+}
+
+///////////////////////////////cc.e////////////////////////////////////////
+#undef prefix_
+
+\f
+// Local Variables:
+// mode: c++
+// fill-column: 100
+// comment-column: 40
+// c-file-style: "senf"
+// indent-tabs-mode: nil
+// ispell-local-dictionary: "american"
+// compile-command: "scons -u test"
+// End:
An IntervalTimer signals an event \a eventsPerInterval times each \a interval
nanoseconds. The event counter and timer are reset, whenever the event is disabled.
+
+ \see IntervalTimerEventInfo
+
+ \ingroup event_group
*/
class IntervalTimer
: public EventImplementation<IntervalTimerEventInfo>
--- /dev/null
+// $Id$
+//
+// Copyright (C) 2007
+// Fraunhofer Institut fuer offene Kommunikationssysteme (FOKUS)
+// Kompetenzzentrum fuer Satelitenkommunikation (SatCom)
+// Stefan Bund <g0dil@berlios.de>
+//
+// This program is free software; you can redistribute it and/or modify
+// it under the terms of the GNU General Public License as published by
+// the Free Software Foundation; either version 2 of the License, or
+// (at your option) any later version.
+//
+// This program is distributed in the hope that it will be useful,
+// but WITHOUT ANY WARRANTY; without even the implied warranty of
+// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+// GNU General Public License for more details.
+//
+// You should have received a copy of the GNU General Public License
+// along with this program; if not, write to the
+// Free Software Foundation, Inc.,
+// 59 Temple Place - Suite 330, Boston, MA 02111-1307, USA.
+
+/** \file
+ \brief IntervalTimer.test unit tests */
+
+//#include "IntervalTimer.test.hh"
+//#include "IntervalTimer.test.ih"
+
+// Custom includes
+#include "IntervalTimer.hh"
+#include "Module.hh"
+#include "Setup.hh"
+
+#include <boost/test/auto_unit_test.hpp>
+#include <boost/test/test_tools.hpp>
+
+#define prefix_
+///////////////////////////////cc.p////////////////////////////////////////
+
+namespace module = senf::ppi::module;
+namespace ppi = senf::ppi;
+
+namespace {
+
+ class TimerTest : public module::Module
+ {
+ SENF_PPI_MODULE(TimerTest);
+
+ ppi::IntervalTimer timer;
+
+ void tick(ppi::IntervalTimer::Event const & event) {
+ if ( --n == 0 )
+ timer.enabled(false);
+ }
+
+ unsigned n;
+
+ public:
+ TimerTest(senf::ClockService::clock_type d_, unsigned n_)
+ : timer ( senf::ClockService::milliseconds(d_) ),
+ n ( n_ )
+ {
+ registerEvent( &TimerTest::tick, timer );
+ }
+ };
+
+ bool is_close_clock(senf::ClockService::clock_type a, senf::ClockService::clock_type b,
+ unsigned long delta = 50000000ul)
+ {
+ return (a<b ? b-a : a-b ) < delta;
+ }
+}
+
+BOOST_AUTO_UNIT_TEST(intervalTimer)
+{
+ TimerTest timer (100,4);
+ senf::ClockService::clock_type start (senf::ClockService::now());
+ senf::ppi::run();
+ BOOST_CHECK_PREDICATE( is_close_clock,
+ (senf::ClockService::now())
+ (start+senf::ClockService::milliseconds(400)) );
+}
+
+///////////////////////////////cc.e////////////////////////////////////////
+#undef prefix_
+
+\f
+// Local Variables:
+// mode: c++
+// fill-column: 100
+// comment-column: 40
+// c-file-style: "senf"
+// indent-tabs-mode: nil
+// ispell-local-dictionary: "american"
+// compile-command: "scons -u test"
+// End:
namespace module {
+ /** \brief Join multiple packet streams with passive inputs
+
+ The PassiveJoin will combine any number of packet streams. You may connect any number of
+ ActiveOutput's to the PassiveJoin instance. The combined stream is then provided on the
+ ActiveOutput \a output.
+
+ Since PassiveJoin allows any number of incoming packet streams, the input connectors are
+ dynamically managed. A special senf::ppi::connect() overload is used to dynamically create
+ the needed input connectors. This hides this extra functionality from the user.
+ \code
+ senf::ppi::module::PassiveJoin join;
+
+ ppi::connect(module1,join); // Connect first module to join's input
+ ppi::connect(module2.some_output,join); // Connect another module to join's input
+ ppi::connect(join,module3); // Forward combined stream to module3
+ \endcode
+
+ \ingroup routing_modules
+ */
class PassiveJoin
: public Module
{
PassiveJoin();
+ private:
+ connector::PassiveInput & newInput();
+
+#ifndef DOXYGEN
+ // I didn't get template friend functions to work ...
+ public:
+#endif
template <class Source>
connector::PassiveInput & connect(Source & source);
private:
- connector::PassiveInput & newInput();
-
void request(connector::PassiveInput & input);
void onThrottle();
void onUnthrottle();
Inputs inputs_;
};
+ /** \brief Join multiple packet streams with active inputs
+
+ The PriorityJoin will combine any number of packet streams. You may connect any number of
+ PassiveInput's to the PassiveJoin instance. The combined stream is then provided on the
+ PassiveOutput \a output.
+
+ When a packet request is received on Priorityjoin's \a output, The request will be serviced
+ from the first unthrottled input. The order, in which connectors are connected to the
+ PriorityJoin's input is important: The earlier connected peer has the higher priority and
+ will be serviced first.
+
+ Since PriorityJoin allows any number of incoming packet streams, the input connectors are
+ dynamically managed. A special senf::ppi::connect() overload is used to dynamically create
+ the needed input connectors. This hides this extra functionality from the user.
+ \code
+ senf::ppi::module::PriorityJoin join;
+
+ ppi::connect(module1,join); // Connect first module to join's input
+ ppi::connect(module2.some_output,join); // Connect another module to join's input
+ ppi::connect(join,module3); // Forward combined stream to module3
+ \endcode
+ Here, \a module1 has higher priority than \a module2 which will only be queried if \a
+ module1 is throttled.
+
+ \ingroup routing_modules
+ */
class PriorityJoin
: public Module
{
PriorityJoin();
+ private:
+ connector::ActiveInput & newInput();
+
+#ifndef DOXYGEN
+ public:
+#endif
template <class Source>
connector::ActiveInput & connect(Source & source);
private:
- connector::ActiveInput & newInput();
-
void request();
void onThrottle();
void onUnthrottle();
/** \mainpage libPPI : The Packet Processing Infrastructure
- The PPI provides an infrastructure to create packet oriented network processing
- applications. A PPI application is built by combining processing modules in a very flexible
- manner.
+ The PPI provides an infrastructure to create packet oriented network processing applications. A
+ PPI application is built by combining processing modules in a very flexible manner.
\image html scenario.png Target Scenario
performance optimizations for TCP traffic (PEP). This router is built by combining several
modules.
+ \see \ref overview \n
+ <a href="../../../Examples/RateStuffer/doc/html/index.html">PPI Example Application:
+ RateStuffer</a> \n
+ \ref senf::ppi::module "Modules" \n
+ \ref senf::ppi::connector "Connectors" \n
+ \ref event_group
+ */
+
+/** \page overview PPI Overview and Concepts
+
\section design Design considerations
The PPI interface is designed to be as simple as possible. It provides sane defaults for all
route statement as defining the 'conceptual data flow' since this is also how control messages
should flow (sans the direction, which is defined by the connectors active/passive property).
- \see \ref ppi_implementation \n
- <a href="http://openfacts.berlios.de/index-en.phtml?title=SENF:_Packet_Processing_Infrastructure">Implementation plan</a>
+ \see \ref ppi_implementation
*/
-/** \page ppi_implementation Implementation Overview
+/** \page ppi_implementation Implementation Notes
\section processing Data Processing
\fixme Implement Spliters: PassiveSplitter, PrioritySplitter, CloneSplitter
\fixme Implement DiscardSink, CloneSource
+ \fixme Implement ThrottleBarrier
*/
\f
namespace ppi {
namespace module {
+ /** \namespace senf::ppi::module
+ \brief PPI Modules
+
+ The modules build the PPI core. The PPI provides a set of general purpose infrastructure
+ modules. For concrete applications, additional application specific processing modules need
+ to be implemented.
+
+ \section module_impl Implementing Modules
+
+ All modules derive from senf::ppi::module::Module. See this class for a documentation on how
+ to write new modules.
+
+ \section infrastructure_modules General Purpose Modules
+
+ The PPI provided general purpose modules can be grouped into several categories
+
+ \li \ref io_modules receive external data or forward packets out of the PPI
+ \li \ref sourcesink_modules generate or absorb packets internally
+ \li \ref routing_modules forward packets within the network
+ \li \ref adapter_modules are used to connect incompatible connectors to each other
+ */
+
+ /** \defgroup io_modules Input/Output Modules
+
+ Input/Output Modules receive data from external sources or forward data from the PPI to
+ outside targets.
+ */
+
+ /** \defgroup sourcesink_modules Source/Sink Modules
+
+ Source and Sink modules generate or absorb packets internally. In contrast to \ref
+ io_modules, they do not communicate outside the PPI.
+ */
+
+ /** \defgroup routing_modules Routing Modules
+
+ Routing modules perform packet forwarding within the network. They do not process the packet
+ data, they just route it.
+ */
+
+ /** \defgroup adapter_modules Adapter Modules
+
+ Adapter modules adapt incompatible connectors to each other. They allow connection a pair of
+ active or passive connectors.
+ */
+
/** \brief Module base-class
senf::ppi::Module is the base-class of all PPI modules. It provides the module implementation
virtual void macro_SENF_PPI_MODULE_missing() = 0;
#endif
+#ifndef DOXYGEN
private:
- virtual void init();
+#endif
+ virtual void init(); ///< Called just before the network is run
#ifndef DOXYGEN
public:
friend class senf::ppi::ModuleManager;
};
+ /** \brief Define PPI Module
+
+ Every module must begin by using this macro.
+
+ \see senf::ppi::module::Module
+ */
# define SENF_PPI_MODULE(name) \
public: \
~ name() { destroy(); } \
namespace senf {
namespace ppi {
- /** \brief
+ /** \brief Internal: Module management
+
+ Every module is registered with the ModuleManager. The ModuleManager controls module
+ initialization and execution of the network.
*/
class ModuleManager
{
///@}
///////////////////////////////////////////////////////////////////////////
- void registerModule(module::Module & module);
- void unregisterModule(module::Module & module);
-
- void init();
- void run();
+ void init(); ///< Called by senf::ppi::init()
+ void run(); ///< Called by senf::ppi::run()
- bool running() const;
+ bool running() const; ///< \c true, if the network is running
private:
ModuleManager();
+ void registerModule(module::Module & module);
+ void unregisterModule(module::Module & module);
+
typedef std::vector<module::Module *> ModuleRegistry;
#ifndef DOXYGEN
ModuleRegistry moduleRegistry_;
bool running_;
bool terminate_;
+
+ friend class module::Module;
};
The PassiveQueue will automatically throttle in both directions. Throttling on the input
connector is the standard throttling as implemented in connector::PassiveInput. Additional,
forward throttling notifications are sent out whenever the queue is empty.
+
+ \ingroup adapter_modules
*/
class PassiveQueue
: public module::Module
\param[in] event Type of event triggering the update */
};
+ /** \brief Simple queueing discipline with high and low threshold
+
+ The ThresholdQueueing QueueingDiscipline is a simple queueing discipline which throttles the
+ input as soon the number of packets in the queue reaches the \a high threshold. The input
+ will be unthrottled when the number of packets drops to the \a low threshold.
+
+ The default queueing discipline is ThresholdQueueing(1,0).
+ */
class ThresholdQueueing
: public QueueingDiscipline
{
input.onRequest(&QueueTester::nop);
}
- void nop() {}
+ void nop() { /* no operation */ }
void forward() {
if (input && output)
#else
+ /** \brief Connect modules
+
+ senf::ppi::connect() establishes a connection between two modules or, to be more precise,
+ between two connectors. For enhanced usability, \a source and \a target may be a Connector,
+ a Module or a collection/subnetwork. Passing a Module or collection/subnetwork as \a source
+ will originate the connection on the \c output member of that Module or collection while
+ passing a module or collection/subnetwork as \a target will terminate the connection on that
+ Module or collections \c input member. For most simple modules, the specification of the
+ connector is therefore obsolete.
+
+ Furthermore, the connect() call may be extended by special modules (e.g. PassiveJoin which
+ allows an arbitrary of input connections).
+ */
template <class Source, class Target>
void connect(Source & source, Target & target);
#endif
+
+ /** \brief Start the network
+
+ Calling senf::ppi::run() will start processing the network. The main event loop is managed
+ by the Scheduler. Before starting the Scheduler main loop, all Module init() members are
+ called.
+ senf::ppi::run() will return when no more work is to be done, that is when no events are
+ enabled (Since the events are enabled and disabled by the throttle notifications which
+ depend among other things on the packet queues, this is the same as checking for packets in
+ any queue). It is Ok to call senf::ppi::run() multiple times during the program lifetime.
+ */
void run();
+
+ /** \brief Manually initialize the network
+
+ For debugging purposes, it is sometimes simpler to not use senf::ppi::run() but instead
+ drive the network via explicit calls using the debug modules. However, it is still necessary
+ to initialize the network. This operation is performed by senf::ppi::init().
+ */
void init();
}}
--- /dev/null
+// $Id$
+//
+// Copyright (C) 2007
+// Fraunhofer Institut fuer offene Kommunikationssysteme (FOKUS)
+// Kompetenzzentrum fuer Satelitenkommunikation (SatCom)
+// Stefan Bund <g0dil@berlios.de>
+//
+// This program is free software; you can redistribute it and/or modify
+// it under the terms of the GNU General Public License as published by
+// the Free Software Foundation; either version 2 of the License, or
+// (at your option) any later version.
+//
+// This program is distributed in the hope that it will be useful,
+// but WITHOUT ANY WARRANTY; without even the implied warranty of
+// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+// GNU General Public License for more details.
+//
+// You should have received a copy of the GNU General Public License
+// along with this program; if not, write to the
+// Free Software Foundation, Inc.,
+// 59 Temple Place - Suite 330, Boston, MA 02111-1307, USA.
+
+/** \file
+ \brief Setup.test unit tests */
+
+//#include "Setup.test.hh"
+//#include "Setup.test.ih"
+
+// Custom includes
+#include "Setup.hh"
+
+#include <boost/test/auto_unit_test.hpp>
+#include <boost/test/test_tools.hpp>
+
+#define prefix_
+///////////////////////////////cc.p////////////////////////////////////////
+
+BOOST_AUTO_UNIT_TEST(setup)
+{
+ // Tested in nearly all other modules ...
+}
+
+///////////////////////////////cc.e////////////////////////////////////////
+#undef prefix_
+
+\f
+// Local Variables:
+// mode: c++
+// fill-column: 100
+// comment-column: 40
+// c-file-style: "senf"
+// indent-tabs-mode: nil
+// ispell-local-dictionary: "american"
+// compile-command: "scons -u test"
+// End:
Whenever the FileHandle object is ready for reading, the \a Reader's \c operator() is called
to read a packet. The default \a Reader is \c PacketReader<>, which will read packets from a
datagram SocketHandle into DataPacket's. You may
-
+
+ \ingroup io_modules
*/
template <class Reader=PacketReader<> >
class ActiveSocketReader
void operator()(Handle handle, Packet packet); // insertion function
};
\endcode
+
+ \ingroup io_modules
*/
template <class Writer=PacketWriter>
class ActiveSocketWriter : public Module
void operator()(Handle handle, Packet packet); // insertion function
};
\endcode
+
+ \ingroup io_modules
*/
template <class Writer=PacketWriter>
class PassiveSocketWriter : public Module
--- /dev/null
+// $Id$
+//
+// Copyright (C) 2007
+// Fraunhofer Institut fuer offene Kommunikationssysteme (FOKUS)
+// Kompetenzzentrum fuer Satelitenkommunikation (SatCom)
+// Stefan Bund <g0dil@berlios.de>
+//
+// This program is free software; you can redistribute it and/or modify
+// it under the terms of the GNU General Public License as published by
+// the Free Software Foundation; either version 2 of the License, or
+// (at your option) any later version.
+//
+// This program is distributed in the hope that it will be useful,
+// but WITHOUT ANY WARRANTY; without even the implied warranty of
+// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+// GNU General Public License for more details.
+//
+// You should have received a copy of the GNU General Public License
+// along with this program; if not, write to the
+// Free Software Foundation, Inc.,
+// 59 Temple Place - Suite 330, Boston, MA 02111-1307, USA.
+
+/** \file
+ \brief ThrottleBarrier non-inline non-template implementation */
+
+#include "ThrottleBarrier.hh"
+//#include "ThrottleBarrier.ih"
+
+// Custom includes
+
+//#include "ThrottleBarrier.mpp"
+#define prefix_
+///////////////////////////////cc.p////////////////////////////////////////
+
+prefix_ senf::ppi::module::ThrottleBarrier::ThrottleBarrier()
+{
+ route( input, output ).autoThrottling( false );
+ input.onRequest( &ThrottleBarrier::request );
+}
+
+prefix_ void senf::ppi::module::ThrottleBarrier::request()
+{
+ Packet p (input());
+ if (output)
+ output(p);
+}
+
+///////////////////////////////cc.e////////////////////////////////////////
+#undef prefix_
+//#include "ThrottleBarrier.mpp"
+
+\f
+// Local Variables:
+// mode: c++
+// fill-column: 100
+// comment-column: 40
+// c-file-style: "senf"
+// indent-tabs-mode: nil
+// ispell-local-dictionary: "american"
+// compile-command: "scons -u test"
+// End:
--- /dev/null
+// $Id$
+//
+// Copyright (C) 2007
+// Fraunhofer Institut fuer offene Kommunikationssysteme (FOKUS)
+// Kompetenzzentrum fuer Satelitenkommunikation (SatCom)
+// Stefan Bund <g0dil@berlios.de>
+//
+// This program is free software; you can redistribute it and/or modify
+// it under the terms of the GNU General Public License as published by
+// the Free Software Foundation; either version 2 of the License, or
+// (at your option) any later version.
+//
+// This program is distributed in the hope that it will be useful,
+// but WITHOUT ANY WARRANTY; without even the implied warranty of
+// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+// GNU General Public License for more details.
+//
+// You should have received a copy of the GNU General Public License
+// along with this program; if not, write to the
+// Free Software Foundation, Inc.,
+// 59 Temple Place - Suite 330, Boston, MA 02111-1307, USA.
+
+/** \file
+ \brief ThrottleBarrier public header */
+
+#ifndef HH_ThrottleBarrier_
+#define HH_ThrottleBarrier_ 1
+
+// Custom includes
+#include "Connectors.hh"
+#include "Module.hh"
+
+//#include "ThrottleBarrier.mpp"
+///////////////////////////////hh.p////////////////////////////////////////
+
+namespace senf {
+namespace ppi {
+namespace module {
+
+ /** \brief Packet discarding barrier for throttle notifications
+
+ A ThrottleBarrier will \e not forward any throttle notifications received on it's output. It
+ will instead discard any packet received on it's input while the output is throttled. This
+ module is used to circumvent any operating system based queueing. This allows the queueing
+ to be controlled explicitly within the PPI.
+
+ \ingroup routing_modules
+ */
+ class ThrottleBarrier
+ : public Module
+ {
+ SENF_PPI_MODULE(ThrottleBarrier);
+ public:
+
+ connector::PassiveInput input;
+ connector::ActiveOutput output;
+
+ ThrottleBarrier();
+
+ private:
+ void request();
+ };
+
+
+}}}
+
+///////////////////////////////hh.e////////////////////////////////////////
+//#include "ThrottleBarrier.cci"
+//#include "ThrottleBarrier.ct"
+//#include "ThrottleBarrier.cti"
+#endif
+
+\f
+// Local Variables:
+// mode: c++
+// fill-column: 100
+// comment-column: 40
+// c-file-style: "senf"
+// indent-tabs-mode: nil
+// ispell-local-dictionary: "american"
+// compile-command: "scons -u test"
+// End:
--- /dev/null
+// $Id$
+//
+// Copyright (C) 2007
+// Fraunhofer Institut fuer offene Kommunikationssysteme (FOKUS)
+// Kompetenzzentrum fuer Satelitenkommunikation (SatCom)
+// Stefan Bund <g0dil@berlios.de>
+//
+// This program is free software; you can redistribute it and/or modify
+// it under the terms of the GNU General Public License as published by
+// the Free Software Foundation; either version 2 of the License, or
+// (at your option) any later version.
+//
+// This program is distributed in the hope that it will be useful,
+// but WITHOUT ANY WARRANTY; without even the implied warranty of
+// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+// GNU General Public License for more details.
+//
+// You should have received a copy of the GNU General Public License
+// along with this program; if not, write to the
+// Free Software Foundation, Inc.,
+// 59 Temple Place - Suite 330, Boston, MA 02111-1307, USA.
+
+/** \file
+ \brief ThrottleBarrier.test unit tests */
+
+//#include "ThrottleBarrier.test.hh"
+//#include "ThrottleBarrier.test.ih"
+
+// Custom includes
+#include "ThrottleBarrier.hh"
+#include "DebugModules.hh"
+#include "Packets/Packets.hh"
+
+#include <boost/test/auto_unit_test.hpp>
+#include <boost/test/test_tools.hpp>
+
+#define prefix_
+///////////////////////////////cc.p////////////////////////////////////////
+
+BOOST_AUTO_UNIT_TEST(throttleBarrier)
+{
+ senf::ppi::module::debug::ActiveSource source;
+ senf::ppi::module::ThrottleBarrier barrier;
+ senf::ppi::module::debug::PassiveSink sink;
+
+ senf::ppi::connect(source,barrier);
+ senf::ppi::connect(barrier,sink);
+
+ senf::Packet p (senf::DataPacket::create());
+
+ BOOST_CHECK( source );
+ source.submit(p);
+ BOOST_CHECK_EQUAL( sink.size(), 1u );
+ sink.throttle();
+ BOOST_CHECK( source );
+ source.submit(p);
+ BOOST_CHECK_EQUAL( sink.size(), 1u );
+}
+
+///////////////////////////////cc.e////////////////////////////////////////
+#undef prefix_
+
+\f
+// Local Variables:
+// mode: c++
+// fill-column: 100
+// comment-column: 40
+// c-file-style: "senf"
+// indent-tabs-mode: nil
+// ispell-local-dictionary: "american"
+// compile-command: "scons -u test"
+// End:
prefix_ typename senf::ppi::detail::Callback<Arg>::type
senf::ppi::detail::Callback<Arg>::make(void (FnClass::* memfn )(FnArg arg), Owner & owner)
{
- return boost::bind(memfn,static_cast<FnClass *>(&owner),1);
+ return boost::bind(memfn,static_cast<FnClass *>(&owner),_1);
}
template <class Arg>
#define prefix_
///////////////////////////////cc.p////////////////////////////////////////
-#define CheckErrno(op,args) if (op args < 0) throw SystemException(# op, errno)
+#define CheckError(op,args) if (op args < 0) throw SystemException(# op, errno)
+
+///////////////////////////////////////////////////////////////////////////
+// senf::ClockService::Impl
struct senf::ClockService::Impl
{
prefix_ senf::ClockService::Impl::Impl()
{
- CheckErrno( sigemptyset, (&alrm_set) );
- CheckErrno( sigaddset, (&alrm_set, SIGALRM) );
+ CheckError( sigemptyset, (&alrm_set) );
+ CheckError( sigaddset, (&alrm_set, SIGALRM) );
}
prefix_ void senf::ClockService::Impl::block()
{
- CheckErrno( sigprocmask, (SIG_BLOCK, &alrm_set, 0) );
+ CheckError( sigprocmask, (SIG_BLOCK, &alrm_set, 0) );
}
prefix_ void senf::ClockService::Impl::unblock()
{
- CheckErrno( sigprocmask, (SIG_UNBLOCK, &alrm_set, 0) );
+ CheckError( sigprocmask, (SIG_UNBLOCK, &alrm_set, 0) );
}
prefix_ void senf::ClockService::Impl::timer(int)
{
- boost::posix_time::ptime time (boost::posix_time::microsec_clock::universal_time());
- if (ClockService::instance().checkSkew(time))
- ClockService::instance().clockSkew(
- time, ClockService::instance().heartbeat_ + boost::posix_time::seconds(
- ClockService::CheckInterval));
- ClockService::instance().heartbeat_ = time;
+ ClockService::instance().timer();
}
///////////////////////////////////////////////////////////////////////////
// private members
prefix_ senf::ClockService::ClockService()
- : base_ (boost::posix_time::microsec_clock::universal_time()),
- heartbeat_ (base_), impl_(new ClockService::Impl())
+ : impl_(new ClockService::Impl())
{
- struct sigaction action;
- action.sa_handler = & senf::ClockService::Impl::timer;
- CheckErrno( sigemptyset, (&action.sa_mask) );
- action.sa_flags = SA_RESTART;
- CheckErrno( sigaction, (SIGALRM, &action, &impl_->oldaction) );
+ restart_m(false);
+}
- struct itimerval itimer;
- itimer.it_interval.tv_sec = CheckInterval;
- itimer.it_interval.tv_usec = 0;
- itimer.it_value.tv_sec = CheckInterval;
- itimer.it_value.tv_usec = 0;
- CheckErrno( setitimer, (ITIMER_REAL, &itimer, &impl_->olditimer) );
- impl_->unblock();
+prefix_ void senf::ClockService::timer()
+{
+ boost::posix_time::ptime time (boost::posix_time::microsec_clock::universal_time());
+ if (checkSkew(time))
+ clockSkew(time, heartbeat_ + boost::posix_time::seconds(
+ ClockService::CheckInterval));
+ heartbeat_ = time;
}
-prefix_ void senf::ClockService::restart_i()
+prefix_ void senf::ClockService::restart_m(bool restart)
{
- impl_->block(); // if any syscall fails, the alarm signal stays blocked which is correct
+ if (restart)
+ // if any syscall fails, the alarm signal stays blocked which is correct
+ impl_->block();
+
base_ = boost::posix_time::microsec_clock::universal_time();
heartbeat_ = base_;
struct sigaction action;
action.sa_handler = & senf::ClockService::Impl::timer;
- CheckErrno( sigemptyset, (&action.sa_mask) );
+ CheckError( sigemptyset, (&action.sa_mask) );
action.sa_flags = SA_RESTART;
- CheckErrno( sigaction, (SIGALRM, &action, 0) );
+ CheckError( sigaction, (SIGALRM, &action, restart ? 0 : &impl_->oldaction) );
struct itimerval itimer;
itimer.it_interval.tv_sec = CheckInterval;
itimer.it_interval.tv_usec = 0;
itimer.it_value.tv_sec = CheckInterval;
itimer.it_value.tv_usec = 0;
- CheckErrno( setitimer, (ITIMER_REAL, &itimer, 0) );
+ CheckError( setitimer, (ITIMER_REAL, &itimer, restart ? 0 : &impl_->olditimer) );
+
impl_->unblock();
}
prefix_ void senf::ClockService::updateSkew(boost::posix_time::ptime time)
{
Impl::Blocker alrmBlocker (impl_.get());
- struct itimerval itimer;
- CheckErrno( getitimer, (ITIMER_REAL, &itimer) );
- clockSkew(time, (heartbeat_
- + boost::posix_time::seconds(CheckInterval)
- - boost::posix_time::seconds(itimer.it_value.tv_sec)
- - boost::posix_time::microseconds(itimer.it_value.tv_usec)));
+
+ // Make a second 'checkSkew' test, this time with SIGALRM blocked. See
+ // senf::ClockService::now_i()
+
+ if (checkSkew(time)) {
+ struct itimerval itimer;
+ CheckError( getitimer, (ITIMER_REAL, &itimer) );
+ clockSkew(time, (heartbeat_
+ + boost::posix_time::seconds(CheckInterval)
+ - boost::posix_time::seconds(itimer.it_value.tv_sec)
+ - boost::posix_time::microseconds(itimer.it_value.tv_usec)));
+ }
}
///////////////////////////////cc.e////////////////////////////////////////
prefix_ senf::ClockService::clock_type senf::ClockService::now()
{
- // We must make sure to call instance() before fetching the current time since the first call to
- // instance() will construct the singleton and initialize heartbeat_. If this happens *after*
- // fetching the current time checkSkew() will detect clock skew since heartbeat_ will be <
- // current time.
- return instance().now_i();
+ return instance().now_m();
}
prefix_ senf::ClockService::abstime_type senf::ClockService::abstime(clock_type clock)
{
-#ifdef BOOST_DATE_TIME_POSIX_TIME_STD_CONFIG
- return instance().base_ + boost::posix_time::nanoseconds(clock);
-#else
- return instance().base_ + boost::posix_time::microseconds((clock+500)/1000);
-#endif
+ return instance().abstime_m(clock);
}
prefix_ senf::ClockService::clock_type senf::ClockService::clock(abstime_type time)
{
- boost::posix_time::time_duration delta (time - instance().base_);
- return clock_type( delta.ticks() )
- * clock_type( 1000000000UL / boost::posix_time::time_duration::ticks_per_second() );
+ return instance().clock_m(time);
}
prefix_ senf::ClockService::clock_type senf::ClockService::nanoseconds(clock_type v)
prefix_ senf::ClockService::clock_type senf::ClockService::microseconds(clock_type v)
{
- return nanoseconds(1000*v);
+ return v * nanoseconds(1000);
}
prefix_ senf::ClockService::clock_type senf::ClockService::milliseconds(clock_type v)
{
- return microseconds(1000*v);
+ return v * microseconds(1000);
}
prefix_ senf::ClockService::clock_type senf::ClockService::seconds(clock_type v)
{
- return milliseconds(1000*v);
+ return v * milliseconds(1000);
}
prefix_ senf::ClockService::clock_type senf::ClockService::minutes(clock_type v)
{
- return seconds(60*v);
+ return v * seconds(60);
}
prefix_ senf::ClockService::clock_type senf::ClockService::hours(clock_type v)
{
- return minutes(60*v);
+ return v * minutes(60);
}
prefix_ senf::ClockService::clock_type senf::ClockService::days(clock_type v)
{
- return hours(24*v);
+ return v * hours(24);
}
prefix_ void senf::ClockService::restart()
{
- instance().restart_i();
+ instance().restart_m();
}
////////////////////////////////////////
// private members
-prefix_ senf::ClockService & senf::ClockService::instance()
+prefix_ senf::ClockService::clock_type senf::ClockService::now_m()
{
- static ClockService instance;
- return instance;
-}
+ // We want to make the normal case (no skew) really fast. This first 'checkSkew' *might*
+ // transiently fail if a SIGALRM is delivered in the midst of the test. updateSkew will
+ // therefore block signals and do the check again to make sure.
+ //
+ // The opposite case (the test returns 'false' even though it should return 'true') is so highly
+ // improbable that it is treated as academic. (it will be catched by the next SIGALRM)
-prefix_ senf::ClockService::clock_type senf::ClockService::now_i()
-{
boost::posix_time::ptime time (boost::posix_time::microsec_clock::universal_time());
if (checkSkew(time))
updateSkew(time);
- return clock(time);
+
+ // 'clock' will pick up the corrected base_ value if needed.
+ return clock_m(time);
+}
+
+prefix_ senf::ClockService::abstime_type senf::ClockService::abstime_m(clock_type clock)
+{
+#ifdef BOOST_DATE_TIME_POSIX_TIME_STD_CONFIG
+ return base_ + boost::posix_time::nanoseconds(clock);
+#else
+ return base_ + boost::posix_time::microseconds((clock+500)/1000);
+#endif
+}
+
+prefix_ senf::ClockService::clock_type senf::ClockService::clock_m(abstime_type time)
+{
+ boost::posix_time::time_duration delta (time - base_);
+ return clock_type( delta.ticks() )
+ * clock_type( 1000000000UL / boost::posix_time::time_duration::ticks_per_second() );
}
prefix_ bool senf::ClockService::checkSkew(boost::posix_time::ptime time)
{
- return time < heartbeat_ || (time - heartbeat_) > boost::posix_time::seconds(2*CheckInterval);
+ boost::posix_time::ptime h (heartbeat_); // reduce chance for race condition
+ return time < h || (time - h) > boost::posix_time::seconds(2*CheckInterval);
}
prefix_ void senf::ClockService::clockSkew(boost::posix_time::ptime time,
#include <boost/utility.hpp>
#include <boost/date_time/posix_time/posix_time.hpp>
#include <boost/scoped_ptr.hpp>
+#include "Utils/singleton.hh"
//#include "ClockService.mpp"
///////////////////////////////hh.p////////////////////////////////////////
The ClockService provides a highly accurate monotonous clock source based on
gettimeofday(). However, it takes additional precautions to detect clock skew.
+
+ \implementation We use a mix of static and non-static members to achieve high performance
+ in the normal case (no clock skew) and still encapsulate the dependency on legacy C
+ headers. Using the senf::singleton mixin ensures, that the instance is constructed
+ before main even when instance() is not called.
*/
class ClockService
- : boost::noncopyable
+ : singleton<ClockService>
{
public:
///////////////////////////////////////////////////////////////////////////
*/
typedef boost::posix_time::ptime abstime_type;
- static unsigned const CheckInterval = 1;
+ static unsigned const CheckInterval = 10;
///////////////////////////////////////////////////////////////////////////
///\name Structors and default members
static void restart();
- protected:
-
private:
-
ClockService();
- static ClockService & instance();
- clock_type now_i();
- void restart_i();
+ void timer();
+
+ clock_type now_m();
+ abstime_type abstime_m(clock_type clock);
+ clock_type clock_m(abstime_type time);
+ void restart_m(bool restart = true);
bool checkSkew(boost::posix_time::ptime time);
- void clockSkew(boost::posix_time::ptime time, boost::posix_time::ptime expected);
void updateSkew(boost::posix_time::ptime time);
-
+ void clockSkew(boost::posix_time::ptime time, boost::posix_time::ptime expected);
+
boost::posix_time::ptime base_;
boost::posix_time::ptime heartbeat_;
+ // I don't want this header to depend on the legacy C headers.
struct Impl;
boost::scoped_ptr<Impl> impl_;
friend class Impl;
#ifndef DOXYGEN
friend class senf::detail::ClockServiceTest;
+ friend class singleton<ClockService>;
#endif
};
-
}
///////////////////////////////hh.e////////////////////////////////////////
{
terminate_ = false;
eventTime_ = ClockService::now();
- while (! terminate_ && ( ! timerQueue_.empty() || ! fdTable_.empty())) {
- while ( ! timerQueue_.empty() && timerQueue_.top()->second.timeout <= eventTime_ ) {
+ while (! terminate_) {
+
+ while (! timerQueue_.empty()) {
TimerMap::iterator i (timerQueue_.top());
if (! i->second.canceled)
- i->second.cb();
+ break;
timerMap_.erase(i);
timerQueue_.pop();
}
- if (terminate_)
- return;
-
int timeout (-1);
- if (! timerQueue_.empty()) {
+ if (timerQueue_.empty()) {
+ if (fdTable_.empty())
+ break;
+ }
+ else {
ClockService::clock_type delta (
(timerQueue_.top()->second.timeout - eventTime_)/1000000UL);
timeout = delta < 0 ? 0 : delta;
}
- ///\fixme Handle more than one epoll_event per call
+ ///\todo Handle more than one epoll_event per call
struct epoll_event ev;
int events = epoll_wait(epollFd_, &ev, 1, timeout);
if (events<0)
- // even though 'man epoll' does not mention EINTR the reality is different ...
if (errno != EINTR)
throw SystemException(errno);
- /// \fixme Fix unneeded timer delays
- // Hmm ... I remember, I purposely moved the timeout-handlers to the loop top ... but why?
- // This delays possible time-critical handlers even further ...
-
eventTime_ = ClockService::now();
+
+ // We always run event handlers. This is important, even if a file-descriptor is signaled
+ // since some descriptors (e.g. real files) will *always* be ready and we still may want to
+ // handle timers
+
+ while (! timerQueue_.empty()) {
+ TimerMap::iterator i (timerQueue_.top());
+ if (i->second.canceled)
+ ;
+ else if (i->second.timeout <= eventTime_)
+ i->second.cb();
+ else
+ break;
+ timerMap_.erase(i);
+ timerQueue_.pop();
+ }
+
if (events <= 0)
- // Timeout .. the handler will be run when going back to the loop top
continue;
FdTable::iterator i = fdTable_.find(ev.data.fd);
variable. The C++ standard then provides above guarantee. The instance will be
initialized the first time, the code flow passes the variable declaration found in
the instance() body.
-
- \fixme TimerQueue as \c map \e and \c priority_queue doesn't make sense ...
*/
static Scheduler & instance();
};
typedef std::map<int,EventSpec> FdTable;
- typedef std::map<unsigned,TimerSpec> TimerMap;
+ typedef std::map<unsigned,TimerSpec> TimerMap; // sorted by id
struct TimerSpecCompare
{
};
typedef std::priority_queue<TimerMap::iterator, std::vector<TimerMap::iterator>,
- TimerSpecCompare> TimerQueue;
+ TimerSpecCompare> TimerQueue; // sorted by time
FdTable fdTable_;
unsigned timerIdCounter_;
case EINTR:
break;
case EAGAIN:
+ case ECONNREFUSED:
+ // Writing to a UDP socket seems return this error code if a corresponding ICMP
+ // error code has been received before (at least on linux). This is inconsistent
+ // since I cannot rely on getting ECONNREFUSED. I therefore ignore this error. TCP
+ // sockets will return this error on connect() and not on write(). Therefore we can
+ // unconditionally ignore this error here.
rv = 0;
break;
default:
prefix_ Self & senf::singleton<Self>::instance()
{
static Self instance_;
- creator_.nop(); // Force instantiation of force_creation
+ // Force instantiation of force_creation (at static object creation time)
+ creator_.nop();
return instance_;
}
template <class Self>
prefix_ senf::singleton<Self>::force_creation::force_creation()
{
- senf::singleton<Self>::instance(); // Force execution of instance() thereby creating instance
+ // Force execution of instance() thereby creating instance
+ senf::singleton<Self>::instance();
}
template <class Self>
}
template <class Self>
-typename senf::singleton<Self>::force_creation senf::singleton<Self>::create_;
+typename senf::singleton<Self>::force_creation senf::singleton<Self>::creator_;
///////////////////////////////cti.e///////////////////////////////////////
#undef prefix_
advertised:
\li There must be only a single thread executing before main() starts. (This should always
be the case)
- \li The singleton class must have a default constructor
+ \li There must be only a single thread executing after main() ends. (This is always
+ important, otherwise global object destruction might fail)
+ \li The singleton class must have a non throwing default constructor and destructor
If these conditions are met, this mixin will ensure that the singleton is constructed \e
before main even starts executing. If static construction code calls the instance() member,
- it is ensured, that a valid instance is returned.
+ it is ensured, that the instance is constructed no later than the first call to instance().
Usage example:
\code
class SomeClass
: public senf::singleton<SomeClass>
{
- public:
- SomeClass(); // Must have default constructor
+ // Must have default constructor
+ SomeClass();
+
+ // Give singleton access to the constructor
+ friend class senf::singleton<SomeClass>;
+
+ public:
+ // By default 'instance()' is protected. If you want, you may make it public:
+ using senf::singleton<SomeClass>::instance;
// ...
};
SomeClass::instance().doSomething();
}
\endcode
+
+ \warning The singleton class should \e not have any static data members since it cannot be
+ guaranteed, that these members will be constructed before the singleton instance.
- \note This implementation is directly taken from
+ \implementation This implementation is directly taken from
<tt>boost/pool/detail/singleton.hpp</tt>. See that file for a description of the
technique. The only difference is, that I prefer to advertise this class as a mixin
(though it may be used the same way as the original too).
class singleton
: boost::noncopyable
{
- public:
- static Self & instance(); ///< Return singleton instance
+ protected:
+ static Self & instance(); ///< Return singleton instance
private:
/** \brief Internal
///////////////////////////////hh.e////////////////////////////////////////
//#include "singleton.cci"
//#include "singleton.ct"
-//#include "singleton.cti"
+#include "singleton.cti"
#endif
\f
DETAILS_AT_TOP = YES
BUILTIN_STL_SUPPORT = YES
EXTRACT_ALL = YES
-EXTRACT_PRIVATE = YES
-EXTRACT_STATIC = YES
+EXTRACT_PRIVATE = NO
+EXTRACT_STATIC = NO
+HIDE_FRIEND_COMPOUNDS = YES
INTERNAL_DOCS = YES
SOURCE_BROWSER = YES
ALPHABETICAL_INDEX = YES
ActiveInput
ActiveOutput
ActiveSink
+ActiveSocketReader
ActiveSource
addr
AddressingPolicy
bund
callback
callbacks
+catched
cb
+cc
+cd
cerr
cfi
CIDR
ClientSocketHandle
+CloneSource
CommunicationPolicy
CommunicationPolicyBase
ConcretePacket
ConnectedCommunicationPolicy
+ConnectedUDPv
const
createAfter
createBefore
initSize
inline
InputConnector
+inputSocket
Institut
Int
IntervalTimer
IpV
ipv
iterator
+join
key
Kommunikationssysteme
Kompetenzzentrum
ListB
ListN
ListPolicy
+localhost
loopback
MACAddress
mainpage
MyList
MyVector
namespace
+nc
+netcat
NextPacket
nextPacketKey
nextPacketRange
OtherPacketType
OutputConnector
outputRequest
+outputSocket
OverlayField
PacketData
PacketImpl
ParseVec
PassiveConnector
PassiveInput
+PassiveJoin
PassiveOutput
+PassiveQueue
PassiveSink
+PassiveSocketWriter
PassiveSource
png
ppi
pre
prev
+PriorityJoin
protocolbundle
protocolbundles
QueueingDiscipline
queueSize
+RateFilter
+rateFilter
+ratefilter
+RateStuffer
+ratestuffer
refcount
registerEvent
registerPacket
SafePacketParser
SatCom
Satelitenkommunikation
+scons
ScopeId
+screenshot
senf
ServerSocketHandle
setBegin
SizeParser
skipline
SocketHandle
+SocketProtocol
SomeEvent
SomeEventArg
someField
typeField
udp
UDPPacket
+udpReader
+udpWriter
UInt
UIntField
unicast
unthrottle
+unthrottled
unthrottles
unthrottling
VectorN
projects / senf.git / commitdiff