-Start integrating the sampler into vermont

o Add sampler/src into vermont/sampler
o Remove sampler.conf


git-svn-id: file:///Users/braun/svn/vermont/trunk/vermont@70 aef3b71b-58ee-0310-9ba9-8811b9f0742f

sampler/ConcurrentQueue.h

@@ -0,0 +1,87 @@
+/*
+ * PSAMP Reference Implementation
+ *
+ * ConcurrentQueue.h
+ *
+ * Thread-safe (concurrent) queue implementation
+ *
+ * Author: Michael Drueing <michael@drueing.de>
+ *
+ */
+
+#ifndef CONCURRENT_QUEUE_H
+#define CONCURRENT_QUEUE_H
+
+#include <queue>
+#include "Globals.h"
+#include "Lock.h"
+#include "TimeoutLock.h"
+#include "Semaphore.h"
+
+template<class T>
+class ConcurrentQueue
+{
+public:
+  ConcurrentQueue() : queue(), count(0), lock(), semaphore()
+  {
+  };
+
+  ~ConcurrentQueue()
+  {
+    if (count != 0)
+      LOG("Queue: WARNING: freeing non-empty queue!\n");
+  };
+
+  inline void push(T *t)
+  {
+    lock.lock();
+    queue.push(t);
+    lock.unlock();
+
+    semaphore.post();
+  };
+
+  inline T *pop()
+  {
+    T *result;
+
+    semaphore.wait();
+
+    lock.lock();
+    result = queue.front();
+    queue.pop();
+    lock.unlock();
+
+    return result;
+  };
+
+  inline bool try_pop(T **t)
+  {
+    T *result;
+
+    if (!semaphore.try_wait())
+      return false;
+    
+    lock.lock();
+    result = queue.front();
+    queue.pop();
+    lock.unlock();
+
+    *t = result;
+
+    return true;
+  };
+
+  inline int getCount() const
+  {
+    return count;
+  };
+
+protected:
+  std::queue<T *> queue;
+  int count;
+  TimeoutLock lock;
+  Semaphore semaphore;
+};
+
+#endif

sampler/ConfigFile.h

@@ -0,0 +1,41 @@
+/*
+ * PSAMP Reference Implementation
+ *
+ * ConfigFile.h
+ *
+ * Structures that describe a config file (Sections, Keys, ...)
+ *
+ * Author: Michael Drueing <michael@drueing.de>
+ *
+ */
+ 
+#ifndef CONFIG_FILE_H
+#define CONFIG_FILE_H
+
+class ConfigKey
+{
+public:
+  enum KeyType {
+    CFG_STRING = 1;
+    CFG_BOOL,
+    CFG_INT,
+    CFG_DOUBLE,
+    CFG_BINARY
+  };
+  
+  ConfigKey(char *name, char *value);
+  ConfigKey(char *name, bool value);
+  ConfigKey(char *name, int value);
+  ConfigKey(char *name, double value);
+  ConfigKey(char *name, void *value, int length);
+
+  KeyType getType() const
+  {
+    return m_type;
+  }
+    
+private:
+  KeyType m_type;
+};
+
+#endif

sampler/ConfigManager.cpp

@@ -0,0 +1,13 @@
+/*
+ * PSAMP Reference Implementation
+ *
+ * ConfigManager.cpp
+ *
+ * Manages a config file
+ *
+ * Author: Michael Drueing <michael@drueing.de>
+ *
+ */
+
+#include "ConfigManager.h"
+

sampler/ConfigManager.h

@@ -0,0 +1,43 @@
+/*
+ * PSAMP Reference Implementation
+ *
+ * ConfigManager.h
+ *
+ * Manages a config file
+ *
+ * Author: Michael Drueing <michael@drueing.de>
+ *
+ */
+ 
+#ifndef CONFIG_MANAGER_H
+#define CONFIG_MANAGER_H
+
+#include <iostream>
+#include <fstream>
+#include <string>
+
+// Singleton class
+class ConfigManager
+{
+private:
+  int numPaths;
+  
+public:
+  ConfigManager(const char *fileName)
+  {
+    ReloadConfig(fileName);
+  }
+
+  void ReloadConfig(const char *fileName)
+  {
+    std::ifstream f(fileName);
+    std::string l;
+
+    while (!f.eof())
+    {
+    }
+  }
+        
+};
+
+#endif

sampler/ExporterSink.cpp

@@ -0,0 +1,51 @@
+/*
+ * PSAMP Reference Implementation
+ *
+ * ExporterSink.cpp
+ *
+ * Implementation of an IPFIX exporter packet sink
+ * using Jan Petranek's ipfixlolib
+ *
+ * Author: Michael Drueing <michael@drueing.de>
+ *
+ */
+ 
+#include "ExporterSink.h"
+#include "Globals.h"
+
+extern "C" {
+#include "ipfixlolib.h"
+}
+
+using namespace std;
+
+void *ExporterSink::exporterSinkProcess(void *arg)
+{
+  ExporterSink *sink = (ExporterSink *)arg;
+  ConcurrentQueue<Packet> *queue = sink->getQueue();
+  Packet *p;
+  int pckCount = 0;
+  int deadline = 400; // timeout in msec after first packet has been added
+  
+  LOG("ExporterSink started\n");
+  while (!sink->exitFlag)
+  {
+    pckCount = 1;
+    
+    // first we need to get a packet
+    p = queue->pop();
+    sink->StartNewPacketStream();
+    sink->AddPacket(p);
+    
+    while (pckCount < sink->ipfix_maxpackets)
+    {
+      // TODO: add time constraint here (max. wait time)
+      p = queue->pop(); 
+      // if (timeout) break;
+      sink->AddPacket(p);
+      pckCount++;
+    }
+    // TODO: add packets here with time constraints
+    sink->FlushPacketStream();
+  }
+}

sampler/ExporterSink.h

@@ -0,0 +1,167 @@
+/*
+ * PSAMP Reference Implementation
+ *
+ * ExporterSink.h
+ *
+ * Implementation of an IPFIX exporter packet sink
+ * using Jan Petranek's ipfixlolib
+ *
+ * Author: Michael Drueing <michael@drueing.de>
+ *
+ */
+ 
+#ifndef EXPORTER_SINK_H
+#define EXPORTER_SINK_H
+
+#include "PacketReceiver.h"
+#include "Thread.h"
+#include "Template.h"
+extern "C" {
+#include "ipfixlolib.h"
+}
+
+// the maximum number of packets to be queued
+#define MAX_PACKETS 1024
+// the default maximum of IPFIX packet per big IPFIX packet sent
+#define IPFIX_PACKETS_MAX 10
+
+class ExporterSink : public PacketReceiver
+{
+public:
+    ExporterSink(Template *tmpl, int sID) : sourceID(sID),
+        templ(tmpl), thread(ExporterSink::exporterSinkProcess), exitFlag(false),
+        numPacketsToRelease(0), ipfix_maxpackets(IPFIX_PACKETS_MAX)
+  {
+    int ret, i, tmplid;
+    unsigned short ttype, tlength, toffset;
+    
+    // generate the exporter
+    ret = ipfix_init_exporter(sourceID, &exporter);
+    if (ret)
+    {
+      LOG("Error initializing IPFIX exporter\n");
+      exit(1);
+    }
+    
+    // generate the ipfix template
+    tmplid = templ->getTemplateID();
+    ret =  ipfix_start_template_set(exporter, tmplid, templ->getFieldCount());
+    for (i = 0; i < templ->getFieldCount(); i++)
+    {
+      templ->getFieldInfo(i, &ttype, &tlength, &toffset);
+      ipfix_put_template_field(exporter, tmplid, tlength, ttype, 0);
+    }
+    ipfix_end_template_set(exporter, tmplid);
+  };
+  
+  void AddCollector(char *address, unsigned short port, const char *protocol)
+  {
+    ipfix_transport_protocol proto;
+    
+    if (strcasecmp(protocol, "TCP") == 0)
+      proto = TCP;
+    else if (strcasecmp(protocol, "UDP") == 0)
+      proto = UDP;
+    else
+      LOG("Invalid protocol %s\n", protocol);
+      
+    LOG("Adding %s://%s:%d\n", protocol, address, port);
+    ipfix_add_collector(exporter, address, port, proto);
+  }
+
+  ~ExporterSink()
+  {
+    ipfix_deinit_exporter(exporter);
+  };
+
+  inline void runSink()
+  {
+    thread.run(this);
+  };
+
+  inline void terminateSink()
+  {
+    exitFlag = true;
+  };
+  
+  // the usage of the following functions is:
+  //
+  // StartPacketStream();
+  // while (more_packets AND timeout_not_met)
+  // {
+  //   AddPaket(getNextPacket());
+  // }
+  // FlushPacketStream();
+  
+  // start a new IPFIX packet stream
+  void StartNewPacketStream()
+  {
+    unsigned short net_tmplid = htons(templ->getTemplateID());
+    LOG("Starting new packet stream\n");
+    numPacketsToRelease = 0;
+    ipfix_start_data_set(exporter, &net_tmplid);
+  }
+  
+  // Add this packet to the packet stream
+  void AddPacket(Packet *pck)
+  {
+    unsigned short ttype, tlength, toffset;
+    // first, store the packet to be released later, after we have sent the data
+    LOG("Adding packet to stream\n");
+    packetsToRelease[numPacketsToRelease++] = pck;
+    
+    for (int i = 0; i < templ->getFieldCount(); i++)
+    {
+      templ->getFieldInfo(i, &ttype, &tlength, &toffset);
+      ipfix_put_data_field(exporter, pck->getPacketData(toffset), tlength);
+    }
+  }
+  
+  // send out the IPFIX packet stream and reset
+  void FlushPacketStream()
+  {
+    // end the packet stream and send the IPFIX packet out through the wire
+    ipfix_end_data_set(exporter);
+    ipfix_send(exporter);
+    
+    LOG("Flushing %d packets from stream\n", numPacketsToRelease);
+    for (int i = 0; i < numPacketsToRelease; i++)
+    {
+      (packetsToRelease[i])->release();
+    }
+    numPacketsToRelease = 0;
+  }
+
+
+  /* max packets per big IPFIX packet */
+  bool setMaxIpfixPP(int x)
+  {
+      ipfix_maxpackets=x;
+
+      return true;
+  }
+
+  int getMaxIpfixPP()
+  {
+      return ipfix_maxpackets;
+  }
+
+protected:
+  int sourceID;
+  Template *templ;
+  Thread thread;
+  static void *exporterSinkProcess(void *);
+  
+  ipfix_exporter *exporter;
+
+  // these packets need to be release()'d after we send the current IPFIX packet
+  int numPacketsToRelease;
+  Packet *packetsToRelease[MAX_PACKETS];
+
+  int ipfix_maxpackets;
+
+public:
+  bool exitFlag;
+};
+
+#endif

sampler/Filter.cpp

@@ -0,0 +1,53 @@
+/*
+ * PSAMP Reference Implementation
+ *
+ * Filter.cpp
+ *
+ * Main filtering loop. Accepts a packet and applies filters on it
+ *
+ * Author: Michael Drueing <michael@drueing.de>
+ *
+ */
+ 
+#include "Filter.h"
+
+using namespace std;
+
+Filter *Filter::instance = 0;
+
+void *Filter::filterProcess(void *arg)
+{
+  Filter *filter = Filter::instance;
+  Packet *p;
+  bool keepPacket;
+  vector<PacketProcessor *>::iterator it;
+
+  while (!filter->exitFlag)
+  {
+    // get a packet...
+    p = filter->getQueue()->pop();
+    filter->pktIn++;
+    
+    // run packet through all packetProcessors
+    for (it = filter->processors.begin(); it != filter->processors.end(); ++it)
+    {
+      keepPacket = (*it)->processPacket(p);
+      if (!keepPacket)
+        break;
+    }
+    
+    //check if we passed all filters
+    if (keepPacket)
+    {
+      // get the packet to the receiver
+      filter->receiver->push(p);
+      filter->pktOut++;
+    }
+    else
+    {
+      // immediately drop the packet
+      p->release();
+    }
+  }
+};
+

sampler/Filter.h

@@ -0,0 +1,77 @@
+/*
+ * PSAMP Reference Implementation
+ *
+ * Filter.h
+ *
+ * Main filtering loop. Accepts a packet and applies filters on it
+ *
+ * Author: Michael Drueing <michael@drueing.de>
+ *
+ */
+ 
+#ifndef FILTER_H
+#define FILTER_H
+
+#include <vector>
+#include "Globals.h"
+#include "ConcurrentQueue.h"
+#include "Packet.h"
+#include "Thread.h"
+
+#include "PacketReceiver.h"
+#include "PacketProcessor.h"
+#include "SystematicSampler.h"
+#include "RandomSampler.h"
+
+class Filter : public PacketReceiver
+{
+public:
+  Filter() : thread(Filter::filterProcess), exitFlag(false),pktIn(0), pktOut(0)
+  {
+  }
+   
+  ~Filter()
+  {
+  }
+  
+  void startFilter()
+  {
+    Filter::instance = this;
+    thread.run(0);
+  }
+  
+  void terminate()
+  {
+    exitFlag = true;
+  }
+  
+  // adds a new output queue to the receivers
+  void setReceiver(PacketReceiver *recv)
+  {
+    receiver = recv->getQueue();
+  };
+  
+  // add a new filter or sampler
+  void addProcessor(PacketProcessor *p)
+  {
+    processors.push_back(p);
+  };
+
+protected:
+  Thread thread;
+  static void *filterProcess(void *);
+  std::vector<PacketProcessor *> processors;
+  //std::vector<ConcurrentQueue<Packet> *> receivers;
+  ConcurrentQueue<Packet> *receiver;
+  
+public:
+  bool exitFlag;
+  
+  //debug variables
+  unsigned long pktIn;
+  unsigned long pktOut;
+  
+  static Filter *Filter::instance;
+};
+
+#endif

sampler/Globals.h

@@ -0,0 +1,22 @@
+/*
+ * PSAMP Reference Implementation
+ *
+ * Globals.h
+ *
+ * global definitions used almost everywhere
+ *
+ * Author: Michael Drueing <michael@drueing.de>
+ *
+ */
+
+#ifndef GLOBALS_H
+#define GLOBALS_H
+
+#ifdef DEBUG
+#include <cstdio>
+#define LOG(fmt, ...)      fprintf(stderr, fmt, ## __VA_ARGS__);
+#else
+#define LOG(fmt, ...)      do { } while (0);
+#endif
+
+#endif

sampler/IPHeaderFilter.cpp

@@ -0,0 +1,56 @@
+/*
+ * PSAMP Reference Implementation
+ *
+ * IPHeaderFilter.cpp
+ *
+ * Filter by data from IP packet given by (offset, size, comparison, targetvalue)
+ *
+ * Author: Michael Drueing <michael@drueing.de>
+ *
+ */
+ 
+#include "IPHeaderFilter.h"
+
+bool IPHeaderFilter::compareValues(int srcvalue, int dstvalue)
+{
+  switch(m_comparison)
+  {
+  case CMP_LT:
+    return srcvalue < dstvalue;
+  case CMP_LE:
+    return srcvalue <= dstvalue;
+  case CMP_EQ:
+    return srcvalue == dstvalue;
+  case CMP_GE:
+    return srcvalue >= dstvalue;
+  case CMP_GT:
+    return srcvalue > dstvalue;
+  case CMP_NE:
+    return srcvalue != dstvalue;
+  default:
+    return 0;
+  }
+}
+
+int IPHeaderFilter::getData(void *data, int size)
+{
+  switch(size)
+  {
+  case 1:
+    return *((unsigned char *)data);
+  case 2:
+    return *((unsigned short *)data);
+  case 4:
+    return *((int *)data);
+  default:
+    LOG("Invalid Data Size %d\n", size);
+    return 0;
+  }
+}
+
+bool IPHeaderFilter::processPacket(const Packet *p)
+{
+  int srcvalue = getData(p->getPacketData(m_offset), m_size);
+  return compareValues(srcvalue, m_value);
+}
+

sampler/IPHeaderFilter.h

@@ -0,0 +1,57 @@
+/*
+ * PSAMP Reference Implementation
+ *
+ * IPHeaderFilter.h
+ *
+ * Filter by data from IP packet given by (offset, size, comparison, targetvalue)
+ *
+ * Author: Michael Drueing <michael@drueing.de>
+ *
+ */
+
+/* this is a multi-purpose filter which filters 32-bit values from packets
+   example usage:
+   a) filter all packets with IP Protocol = 23
+      offset=9 (it's located at offset 9)
+      size=1 (it's 1 byte)
+      comparison=CMP_EQ
+      targetvalue=23
+  
+   ... you should get the idea    
+*/
+
+#ifndef IP_HEADER_FILTER_H
+#define IP_HEADER_FILTER_H
+
+#include "Globals.h"
+#include "PacketProcessor.h"
+
+#define CMP_LT  0x00  // lower than
+#define CMP_LE  0x01  // lower or equal
+#define CMP_EQ  0x02  // equal
+#define CMP_GE  0x03  // greater or equal
+#define CMP_GT  0x04  // greater than
+#define CMP_NE  0x05  // not equal
+
+class IPHeaderFilter : public PacketProcessor
+{
+public:
+  IPHeaderFilter(int offset, int size, int comparison, int value) 
+    : m_offset(offset), m_size(size), m_comparison(comparison), m_value(value)
+  {
+  }
+  
+  virtual bool processPacket(const Packet *p);
+   
+protected:
+  bool compareValues(int srcvalue, int dstvalue);
+  int getData(void *data, int size);
+
+  int m_offset;
+  int m_size;
+  int m_comparison;
+  int m_value;
+};
+
+#endif
+

sampler/Lock.h

@@ -0,0 +1,54 @@
+/*
+ * PSAMP Reference Implementation
+ *
+ * Lock.h
+ *
+ * Class which encapsulates a simple lock
+ *
+ * Author: Michael Drueing <michael@drueing.de>
+ *
+ */
+
+#ifndef LOCK_H
+#define LOCK_H
+
+#include <pthread.h>
+
+class Lock
+{
+private:
+  // the system mutex used (currently using pthread mutexes)
+  pthread_mutex_t mutex;
+  
+public:
+  // constructs a new Lock
+  inline Lock()
+  {
+    pthread_mutex_init(&mutex, NULL);
+  };
+  
+  // deletes a Lock.
+  // TODO: use try_lock to see if lock is currently held. If so,
+  // then results are unpredictable!
+  inline ~Lock()
+  {
+    pthread_mutex_destroy(&mutex);
+  };
+  
+  inline void lock()
+  {
+    pthread_mutex_lock(&mutex);
+  };
+
+  inline void unlock()
+  {
+    pthread_mutex_unlock(&mutex);
+  };
+
+  inline bool try_lock()
+  {
+    return (pthread_mutex_trylock(&mutex) == 0);
+  };
+};
+
+#endif

sampler/Makefile.am

@@ -0,0 +1,14 @@
+bin_PROGRAMS = sampler
+
+
+# set the include path found by configure
+INCLUDES = $(all_includes)
+
+# the library search path.
+sampler_LDFLAGS = $(all_libraries) 
+sampler_SOURCES = Observer.cpp PacketSink.cpp Test.cpp SystematicSampler.cpp Filter.cpp ConfigManager.cpp RandomSampler.cpp Template.cpp Packet.cpp IPHeaderFilter.cpp ExporterSink.cpp
+sampler_LDADD = -lpcap -lpthread -lipfixlo -L../../ipfixlolib/src
+
+AM_CFLAGS = -DDEBUG -I../../ipfixlolib/src
+AM_CXXFLAGS = -DDEBUG -I../../ipfixlolib/src
+noinst_HEADERS = PacketProcessor.h Thread.h Globals.h Lock.h Observer.h Packet.h PacketSink.h Semaphore.h ConcurrentQueue.h SystematicSampler.h Filter.h ConfigManager.h TimeoutLock.h RandomSampler.h ConfigFile.h Template.h IPHeaderFilter.h IPHeaderFilter.h ExporterSink.h

sampler/Makefile.in

@@ -0,0 +1,477 @@
+# Makefile.in generated by automake 1.8.3 from Makefile.am.
+# @configure_input@
+
+# Copyright (C) 1994, 1995, 1996, 1997, 1998, 1999, 2000, 2001, 2002,
+# 2003, 2004  Free Software Foundation, Inc.
+# This Makefile.in is free software; the Free Software Foundation
+# gives unlimited permission to copy and/or distribute it,
+# with or without modifications, as long as this notice is preserved.
+
+# This program is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY, to the extent permitted by law; without
+# even the implied warranty of MERCHANTABILITY or FITNESS FOR A
+# PARTICULAR PURPOSE.
+
+@SET_MAKE@
+
+
+SOURCES = $(sampler_SOURCES)
+
+srcdir = @srcdir@
+top_srcdir = @top_srcdir@
+VPATH = @srcdir@
+pkgdatadir = $(datadir)/@PACKAGE@
+pkglibdir = $(libdir)/@PACKAGE@
+pkgincludedir = $(includedir)/@PACKAGE@
+top_builddir = ..
+am__cd = CDPATH="$${ZSH_VERSION+.}$(PATH_SEPARATOR)" && cd
+INSTALL = @INSTALL@
+install_sh_DATA = $(install_sh) -c -m 644
+install_sh_PROGRAM = $(install_sh) -c
+install_sh_SCRIPT = $(install_sh) -c
+INSTALL_HEADER = $(INSTALL_DATA)
+transform = $(program_transform_name)
+NORMAL_INSTALL = :
+PRE_INSTALL = :
+POST_INSTALL = :
+NORMAL_UNINSTALL = :
+PRE_UNINSTALL = :
+POST_UNINSTALL = :
+host_triplet = @host@
+bin_PROGRAMS = sampler$(EXEEXT)
+subdir = src
+DIST_COMMON = $(noinst_HEADERS) $(srcdir)/Makefile.am \
+	$(srcdir)/Makefile.in
+ACLOCAL_M4 = $(top_srcdir)/aclocal.m4
+am__aclocal_m4_deps = $(top_srcdir)/configure.in
+am__configure_deps = $(am__aclocal_m4_deps) $(CONFIGURE_DEPENDENCIES) \
+	$(ACLOCAL_M4)
+mkinstalldirs = $(SHELL) $(top_srcdir)/mkinstalldirs
+CONFIG_HEADER = $(top_builddir)/config.h
+CONFIG_CLEAN_FILES =
+am__installdirs = "$(DESTDIR)$(bindir)"
+binPROGRAMS_INSTALL = $(INSTALL_PROGRAM)
+PROGRAMS = $(bin_PROGRAMS)
+am_sampler_OBJECTS = Observer.$(OBJEXT) PacketSink.$(OBJEXT) \
+	Test.$(OBJEXT) SystematicSampler.$(OBJEXT) Filter.$(OBJEXT) \
+	ConfigManager.$(OBJEXT) RandomSampler.$(OBJEXT) \
+	Template.$(OBJEXT) Packet.$(OBJEXT) IPHeaderFilter.$(OBJEXT) \
+	ExporterSink.$(OBJEXT)
+sampler_OBJECTS = $(am_sampler_OBJECTS)
+sampler_DEPENDENCIES =
+DEFAULT_INCLUDES = -I. -I$(srcdir) -I$(top_builddir)
+depcomp = $(SHELL) $(top_srcdir)/depcomp
+am__depfiles_maybe = depfiles
+@AMDEP_TRUE@DEP_FILES = ./$(DEPDIR)/ConfigManager.Po \
+@AMDEP_TRUE@	./$(DEPDIR)/ExporterSink.Po ./$(DEPDIR)/Filter.Po \
+@AMDEP_TRUE@	./$(DEPDIR)/IPHeaderFilter.Po \
+@AMDEP_TRUE@	./$(DEPDIR)/Observer.Po ./$(DEPDIR)/Packet.Po \
+@AMDEP_TRUE@	./$(DEPDIR)/PacketSink.Po \
+@AMDEP_TRUE@	./$(DEPDIR)/RandomSampler.Po \
+@AMDEP_TRUE@	./$(DEPDIR)/SystematicSampler.Po \
+@AMDEP_TRUE@	./$(DEPDIR)/Template.Po ./$(DEPDIR)/Test.Po
+CXXCOMPILE = $(CXX) $(DEFS) $(DEFAULT_INCLUDES) $(INCLUDES) \
+	$(AM_CPPFLAGS) $(CPPFLAGS) $(AM_CXXFLAGS) $(CXXFLAGS)
+LTCXXCOMPILE = $(LIBTOOL) --mode=compile $(CXX) $(DEFS) \
+	$(DEFAULT_INCLUDES) $(INCLUDES) $(AM_CPPFLAGS) $(CPPFLAGS) \
+	$(AM_CXXFLAGS) $(CXXFLAGS)
+CXXLD = $(CXX)
+CXXLINK = $(LIBTOOL) --mode=link $(CXXLD) $(AM_CXXFLAGS) $(CXXFLAGS) \
+	$(AM_LDFLAGS) $(LDFLAGS) -o $@
+SOURCES = $(sampler_SOURCES)
+DIST_SOURCES = $(sampler_SOURCES)
+HEADERS = $(noinst_HEADERS)
+ETAGS = etags
+CTAGS = ctags
+DISTFILES = $(DIST_COMMON) $(DIST_SOURCES) $(TEXINFOS) $(EXTRA_DIST)
+ACLOCAL = @ACLOCAL@
+AMDEP_FALSE = @AMDEP_FALSE@
+AMDEP_TRUE = @AMDEP_TRUE@
+AMTAR = @AMTAR@
+AR = @AR@
+AUTOCONF = @AUTOCONF@
+AUTOHEADER = @AUTOHEADER@
+AUTOMAKE = @AUTOMAKE@
+AWK = @AWK@
+CC = @CC@
+CCDEPMODE = @CCDEPMODE@
+CFLAGS = @CFLAGS@
+CPP = @CPP@
+CPPFLAGS = @CPPFLAGS@
+CXX = @CXX@
+CXXCPP = @CXXCPP@
+CXXDEPMODE = @CXXDEPMODE@
+CXXFLAGS = @CXXFLAGS@
+CYGPATH_W = @CYGPATH_W@
+DEFS = @DEFS@
+DEPDIR = @DEPDIR@
+ECHO = @ECHO@
+ECHO_C = @ECHO_C@
+ECHO_N = @ECHO_N@
+ECHO_T = @ECHO_T@
+EGREP = @EGREP@
+EXEEXT = @EXEEXT@
+F77 = @F77@
+FFLAGS = @FFLAGS@
+INSTALL_DATA = @INSTALL_DATA@
+INSTALL_PROGRAM = @INSTALL_PROGRAM@
+INSTALL_SCRIPT = @INSTALL_SCRIPT@
+INSTALL_STRIP_PROGRAM = @INSTALL_STRIP_PROGRAM@
+LDFLAGS = @LDFLAGS@
+LIBOBJS = @LIBOBJS@
+LIBS = @LIBS@
+LIBTOOL = @LIBTOOL@
+LN_S = @LN_S@
+LTLIBOBJS = @LTLIBOBJS@
+MAKEINFO = @MAKEINFO@
+OBJEXT = @OBJEXT@
+PACKAGE = @PACKAGE@
+PACKAGE_BUGREPORT = @PACKAGE_BUGREPORT@
+PACKAGE_NAME = @PACKAGE_NAME@
+PACKAGE_STRING = @PACKAGE_STRING@
+PACKAGE_TARNAME = @PACKAGE_TARNAME@
+PACKAGE_VERSION = @PACKAGE_VERSION@
+PATH_SEPARATOR = @PATH_SEPARATOR@
+RANLIB = @RANLIB@
+SET_MAKE = @SET_MAKE@
+SHELL = @SHELL@
+STRIP = @STRIP@
+VERSION = @VERSION@
+ac_ct_AR = @ac_ct_AR@
+ac_ct_CC = @ac_ct_CC@
+ac_ct_CXX = @ac_ct_CXX@
+ac_ct_F77 = @ac_ct_F77@
+ac_ct_RANLIB = @ac_ct_RANLIB@
+ac_ct_STRIP = @ac_ct_STRIP@
+am__fastdepCC_FALSE = @am__fastdepCC_FALSE@
+am__fastdepCC_TRUE = @am__fastdepCC_TRUE@
+am__fastdepCXX_FALSE = @am__fastdepCXX_FALSE@
+am__fastdepCXX_TRUE = @am__fastdepCXX_TRUE@
+am__include = @am__include@
+am__leading_dot = @am__leading_dot@
+am__quote = @am__quote@
+bindir = @bindir@
+build = @build@
+build_alias = @build_alias@
+build_cpu = @build_cpu@
+build_os = @build_os@
+build_vendor = @build_vendor@
+datadir = @datadir@
+exec_prefix = @exec_prefix@
+host = @host@
+host_alias = @host_alias@
+host_cpu = @host_cpu@
+host_os = @host_os@
+host_vendor = @host_vendor@
+includedir = @includedir@
+infodir = @infodir@
+install_sh = @install_sh@
+libdir = @libdir@
+libexecdir = @libexecdir@
+localstatedir = @localstatedir@
+mandir = @mandir@
+mkdir_p = @mkdir_p@
+oldincludedir = @oldincludedir@
+prefix = @prefix@
+program_transform_name = @program_transform_name@
+sbindir = @sbindir@
+sharedstatedir = @sharedstatedir@
+sysconfdir = @sysconfdir@
+target_alias = @target_alias@
+
+# set the include path found by configure
+INCLUDES = $(all_includes)
+
+# the library search path.
+sampler_LDFLAGS = $(all_libraries) 
+sampler_SOURCES = Observer.cpp PacketSink.cpp Test.cpp SystematicSampler.cpp Filter.cpp ConfigManager.cpp RandomSampler.cpp Template.cpp Packet.cpp IPHeaderFilter.cpp ExporterSink.cpp
+sampler_LDADD = -lpcap -lpthread -lipfixlo -L../../ipfixlolib/src
+AM_CFLAGS = -DDEBUG -I../../ipfixlolib/src
+AM_CXXFLAGS = -DDEBUG -I../../ipfixlolib/src
+noinst_HEADERS = PacketProcessor.h Thread.h Globals.h Lock.h Observer.h Packet.h PacketSink.h Semaphore.h ConcurrentQueue.h SystematicSampler.h Filter.h ConfigManager.h TimeoutLock.h RandomSampler.h ConfigFile.h Template.h IPHeaderFilter.h IPHeaderFilter.h ExporterSink.h
+all: all-am
+
+.SUFFIXES:
+.SUFFIXES: .cpp .lo .o .obj
+$(srcdir)/Makefile.in:  $(srcdir)/Makefile.am  $(am__configure_deps)
+	@for dep in $?; do \
+	  case '$(am__configure_deps)' in \
+	    *$$dep*) \
+	      cd $(top_builddir) && $(MAKE) $(AM_MAKEFLAGS) am--refresh \
+		&& exit 0; \
+	      exit 1;; \
+	  esac; \
+	done; \
+	echo ' cd $(top_srcdir) && $(AUTOMAKE) --gnu  src/Makefile'; \
+	cd $(top_srcdir) && \
+	  $(AUTOMAKE) --gnu  src/Makefile
+.PRECIOUS: Makefile
+Makefile: $(srcdir)/Makefile.in $(top_builddir)/config.status
+	@case '$?' in \
+	  *config.status*) \
+	    cd $(top_builddir) && $(MAKE) $(AM_MAKEFLAGS) am--refresh;; \
+	  *) \
+	    echo ' cd $(top_builddir) && $(SHELL) ./config.status $(subdir)/$@ $(am__depfiles_maybe)'; \
+	    cd $(top_builddir) && $(SHELL) ./config.status $(subdir)/$@ $(am__depfiles_maybe);; \
+	esac;
+
+$(top_builddir)/config.status: $(top_srcdir)/configure $(CONFIG_STATUS_DEPENDENCIES)
+	cd $(top_builddir) && $(MAKE) $(AM_MAKEFLAGS) am--refresh
+
+$(top_srcdir)/configure:  $(am__configure_deps)
+	cd $(top_builddir) && $(MAKE) $(AM_MAKEFLAGS) am--refresh
+$(ACLOCAL_M4):  $(am__aclocal_m4_deps)
+	cd $(top_builddir) && $(MAKE) $(AM_MAKEFLAGS) am--refresh
+install-binPROGRAMS: $(bin_PROGRAMS)
+	@$(NORMAL_INSTALL)
+	test -z "$(bindir)" || $(mkdir_p) "$(DESTDIR)$(bindir)"
+	@list='$(bin_PROGRAMS)'; for p in $$list; do \
+	  p1=`echo $$p|sed 's/$(EXEEXT)$$//'`; \
+	  if test -f $$p \
+	     || test -f $$p1 \
+	  ; then \
+	    f=`echo "$$p1" | sed 's,^.*/,,;$(transform);s/$$/$(EXEEXT)/'`; \
+	   echo " $(INSTALL_PROGRAM_ENV) $(LIBTOOL) --mode=install $(binPROGRAMS_INSTALL) '$$p' '$(DESTDIR)$(bindir)/$$f'"; \
+	   $(INSTALL_PROGRAM_ENV) $(LIBTOOL) --mode=install $(binPROGRAMS_INSTALL) "$$p" "$(DESTDIR)$(bindir)/$$f" || exit 1; \
+	  else :; fi; \
+	done
+
+uninstall-binPROGRAMS:
+	@$(NORMAL_UNINSTALL)
+	@list='$(bin_PROGRAMS)'; for p in $$list; do \
+	  f=`echo "$$p" | sed 's,^.*/,,;s/$(EXEEXT)$$//;$(transform);s/$$/$(EXEEXT)/'`; \
+	  echo " rm -f '$(DESTDIR)$(bindir)/$$f'"; \
+	  rm -f "$(DESTDIR)$(bindir)/$$f"; \
+	done
+
+clean-binPROGRAMS:
+	@list='$(bin_PROGRAMS)'; for p in $$list; do \
+	  f=`echo $$p|sed 's/$(EXEEXT)$$//'`; \
+	  echo " rm -f $$p $$f"; \
+	  rm -f $$p $$f ; \
+	done
+sampler$(EXEEXT): $(sampler_OBJECTS) $(sampler_DEPENDENCIES) 
+	@rm -f sampler$(EXEEXT)
+	$(CXXLINK) $(sampler_LDFLAGS) $(sampler_OBJECTS) $(sampler_LDADD) $(LIBS)
+
+mostlyclean-compile:
+	-rm -f *.$(OBJEXT)
+
+distclean-compile:
+	-rm -f *.tab.c
+
+@AMDEP_TRUE@@am__include@ @am__quote@./$(DEPDIR)/ConfigManager.Po@am__quote@
+@AMDEP_TRUE@@am__include@ @am__quote@./$(DEPDIR)/ExporterSink.Po@am__quote@
+@AMDEP_TRUE@@am__include@ @am__quote@./$(DEPDIR)/Filter.Po@am__quote@
+@AMDEP_TRUE@@am__include@ @am__quote@./$(DEPDIR)/IPHeaderFilter.Po@am__quote@
+@AMDEP_TRUE@@am__include@ @am__quote@./$(DEPDIR)/Observer.Po@am__quote@
+@AMDEP_TRUE@@am__include@ @am__quote@./$(DEPDIR)/Packet.Po@am__quote@
+@AMDEP_TRUE@@am__include@ @am__quote@./$(DEPDIR)/PacketSink.Po@am__quote@
+@AMDEP_TRUE@@am__include@ @am__quote@./$(DEPDIR)/RandomSampler.Po@am__quote@
+@AMDEP_TRUE@@am__include@ @am__quote@./$(DEPDIR)/SystematicSampler.Po@am__quote@
+@AMDEP_TRUE@@am__include@ @am__quote@./$(DEPDIR)/Template.Po@am__quote@
+@AMDEP_TRUE@@am__include@ @am__quote@./$(DEPDIR)/Test.Po@am__quote@
+
+.cpp.o:
+@am__fastdepCXX_TRUE@	if $(CXXCOMPILE) -MT $@ -MD -MP -MF "$(DEPDIR)/$*.Tpo" -c -o $@ $<; \
+@am__fastdepCXX_TRUE@	then mv -f "$(DEPDIR)/$*.Tpo" "$(DEPDIR)/$*.Po"; else rm -f "$(DEPDIR)/$*.Tpo"; exit 1; fi
+@AMDEP_TRUE@@am__fastdepCXX_FALSE@	source='$<' object='$@' libtool=no @AMDEPBACKSLASH@
+@AMDEP_TRUE@@am__fastdepCXX_FALSE@	depfile='$(DEPDIR)/$*.Po' tmpdepfile='$(DEPDIR)/$*.TPo' @AMDEPBACKSLASH@
+@AMDEP_TRUE@@am__fastdepCXX_FALSE@	$(CXXDEPMODE) $(depcomp) @AMDEPBACKSLASH@
+@am__fastdepCXX_FALSE@	$(CXXCOMPILE) -c -o $@ $<
+
+.cpp.obj:
+@am__fastdepCXX_TRUE@	if $(CXXCOMPILE) -MT $@ -MD -MP -MF "$(DEPDIR)/$*.Tpo" -c -o $@ `$(CYGPATH_W) '$<'`; \
+@am__fastdepCXX_TRUE@	then mv -f "$(DEPDIR)/$*.Tpo" "$(DEPDIR)/$*.Po"; else rm -f "$(DEPDIR)/$*.Tpo"; exit 1; fi
+@AMDEP_TRUE@@am__fastdepCXX_FALSE@	source='$<' object='$@' libtool=no @AMDEPBACKSLASH@
+@AMDEP_TRUE@@am__fastdepCXX_FALSE@	depfile='$(DEPDIR)/$*.Po' tmpdepfile='$(DEPDIR)/$*.TPo' @AMDEPBACKSLASH@
+@AMDEP_TRUE@@am__fastdepCXX_FALSE@	$(CXXDEPMODE) $(depcomp) @AMDEPBACKSLASH@
+@am__fastdepCXX_FALSE@	$(CXXCOMPILE) -c -o $@ `$(CYGPATH_W) '$<'`
+
+.cpp.lo:
+@am__fastdepCXX_TRUE@	if $(LTCXXCOMPILE) -MT $@ -MD -MP -MF "$(DEPDIR)/$*.Tpo" -c -o $@ $<; \
+@am__fastdepCXX_TRUE@	then mv -f "$(DEPDIR)/$*.Tpo" "$(DEPDIR)/$*.Plo"; else rm -f "$(DEPDIR)/$*.Tpo"; exit 1; fi
+@AMDEP_TRUE@@am__fastdepCXX_FALSE@	source='$<' object='$@' libtool=yes @AMDEPBACKSLASH@
+@AMDEP_TRUE@@am__fastdepCXX_FALSE@	depfile='$(DEPDIR)/$*.Plo' tmpdepfile='$(DEPDIR)/$*.TPlo' @AMDEPBACKSLASH@
+@AMDEP_TRUE@@am__fastdepCXX_FALSE@	$(CXXDEPMODE) $(depcomp) @AMDEPBACKSLASH@
+@am__fastdepCXX_FALSE@	$(LTCXXCOMPILE) -c -o $@ $<
+
+mostlyclean-libtool:
+	-rm -f *.lo
+
+clean-libtool:
+	-rm -rf .libs _libs
+
+distclean-libtool:
+	-rm -f libtool
+uninstall-info-am:
+
+ID: $(HEADERS) $(SOURCES) $(LISP) $(TAGS_FILES)
+	list='$(SOURCES) $(HEADERS) $(LISP) $(TAGS_FILES)'; \
+	unique=`for i in $$list; do \
+	    if test -f "$$i"; then echo $$i; else echo $(srcdir)/$$i; fi; \
+	  done | \
+	  $(AWK) '    { files[$$0] = 1; } \
+	       END { for (i in files) print i; }'`; \
+	mkid -fID $$unique
+tags: TAGS
+
+TAGS:  $(HEADERS) $(SOURCES)  $(TAGS_DEPENDENCIES) \
+		$(TAGS_FILES) $(LISP)
+	tags=; \
+	here=`pwd`; \
+	list='$(SOURCES) $(HEADERS)  $(LISP) $(TAGS_FILES)'; \
+	unique=`for i in $$list; do \
+	    if test -f "$$i"; then echo $$i; else echo $(srcdir)/$$i; fi; \
+	  done | \
+	  $(AWK) '    { files[$$0] = 1; } \
+	       END { for (i in files) print i; }'`; \
+	test -z "$(ETAGS_ARGS)$$tags$$unique" \
+	  || $(ETAGS) $(ETAGSFLAGS) $(AM_ETAGSFLAGS) $(ETAGS_ARGS) \
+	     $$tags $$unique
+ctags: CTAGS
+CTAGS:  $(HEADERS) $(SOURCES)  $(TAGS_DEPENDENCIES) \
+		$(TAGS_FILES) $(LISP)
+	tags=; \
+	here=`pwd`; \
+	list='$(SOURCES) $(HEADERS)  $(LISP) $(TAGS_FILES)'; \
+	unique=`for i in $$list; do \
+	    if test -f "$$i"; then echo $$i; else echo $(srcdir)/$$i; fi; \
+	  done | \
+	  $(AWK) '    { files[$$0] = 1; } \
+	       END { for (i in files) print i; }'`; \
+	test -z "$(CTAGS_ARGS)$$tags$$unique" \
+	  || $(CTAGS) $(CTAGSFLAGS) $(AM_CTAGSFLAGS) $(CTAGS_ARGS) \
+	     $$tags $$unique
+
+GTAGS:
+	here=`$(am__cd) $(top_builddir) && pwd` \
+	  && cd $(top_srcdir) \
+	  && gtags -i $(GTAGS_ARGS) $$here
+
+distclean-tags:
+	-rm -f TAGS ID GTAGS GRTAGS GSYMS GPATH tags
+
+distdir: $(DISTFILES)
+	@srcdirstrip=`echo "$(srcdir)" | sed 's|.|.|g'`; \
+	topsrcdirstrip=`echo "$(top_srcdir)" | sed 's|.|.|g'`; \
+	list='$(DISTFILES)'; for file in $$list; do \
+	  case $$file in \
+	    $(srcdir)/*) file=`echo "$$file" | sed "s|^$$srcdirstrip/||"`;; \
+	    $(top_srcdir)/*) file=`echo "$$file" | sed "s|^$$topsrcdirstrip/|$(top_builddir)/|"`;; \
+	  esac; \
+	  if test -f $$file || test -d $$file; then d=.; else d=$(srcdir); fi; \
+	  dir=`echo "$$file" | sed -e 's,/[^/]*$$,,'`; \
+	  if test "$$dir" != "$$file" && test "$$dir" != "."; then \
+	    dir="/$$dir"; \
+	    $(mkdir_p) "$(distdir)$$dir"; \
+	  else \
+	    dir=''; \
+	  fi; \
+	  if test -d $$d/$$file; then \
+	    if test -d $(srcdir)/$$file && test $$d != $(srcdir); then \
+	      cp -pR $(srcdir)/$$file $(distdir)$$dir || exit 1; \
+	    fi; \
+	    cp -pR $$d/$$file $(distdir)$$dir || exit 1; \
+	  else \
+	    test -f $(distdir)/$$file \
+	    || cp -p $$d/$$file $(distdir)/$$file \
+	    || exit 1; \
+	  fi; \
+	done
+check-am: all-am
+check: check-am
+all-am: Makefile $(PROGRAMS) $(HEADERS)
+installdirs:
+	for dir in "$(DESTDIR)$(bindir)"; do \
+	  test -z "$$dir" || $(mkdir_p) "$$dir"; \
+	done
+install: install-am
+install-exec: install-exec-am
+install-data: install-data-am
+uninstall: uninstall-am
+
+install-am: all-am
+	@$(MAKE) $(AM_MAKEFLAGS) install-exec-am install-data-am
+
+installcheck: installcheck-am
+install-strip:
+	$(MAKE) $(AM_MAKEFLAGS) INSTALL_PROGRAM="$(INSTALL_STRIP_PROGRAM)" \
+	  install_sh_PROGRAM="$(INSTALL_STRIP_PROGRAM)" INSTALL_STRIP_FLAG=-s \
+	  `test -z '$(STRIP)' || \
+	    echo "INSTALL_PROGRAM_ENV=STRIPPROG='$(STRIP)'"` install
+mostlyclean-generic:
+
+clean-generic:
+
+distclean-generic:
+	-rm -f $(CONFIG_CLEAN_FILES)
+
+maintainer-clean-generic:
+	@echo "This command is intended for maintainers to use"
+	@echo "it deletes files that may require special tools to rebuild."
+clean: clean-am
+
+clean-am: clean-binPROGRAMS clean-generic clean-libtool mostlyclean-am
+
+distclean: distclean-am
+	-rm -rf ./$(DEPDIR)
+	-rm -f Makefile
+distclean-am: clean-am distclean-compile distclean-generic \
+	distclean-libtool distclean-tags
+
+dvi: dvi-am
+
+dvi-am:
+
+html: html-am
+
+info: info-am
+
+info-am:
+
+install-data-am:
+
+install-exec-am: install-binPROGRAMS
+
+install-info: install-info-am
+
+install-man:
+
+installcheck-am:
+
+maintainer-clean: maintainer-clean-am
+	-rm -rf ./$(DEPDIR)
+	-rm -f Makefile
+maintainer-clean-am: distclean-am maintainer-clean-generic
+
+mostlyclean: mostlyclean-am
+
+mostlyclean-am: mostlyclean-compile mostlyclean-generic \
+	mostlyclean-libtool
+
+pdf: pdf-am
+
+pdf-am:
+
+ps: ps-am
+
+ps-am:
+
+uninstall-am: uninstall-binPROGRAMS uninstall-info-am
+
+.PHONY: CTAGS GTAGS all all-am check check-am clean clean-binPROGRAMS \
+	clean-generic clean-libtool ctags distclean distclean-compile \
+	distclean-generic distclean-libtool distclean-tags distdir dvi \
+	dvi-am html html-am info info-am install install-am \
+	install-binPROGRAMS install-data install-data-am install-exec \
+	install-exec-am install-info install-info-am install-man \
+	install-strip installcheck installcheck-am installdirs \
+	maintainer-clean maintainer-clean-generic mostlyclean \
+	mostlyclean-compile mostlyclean-generic mostlyclean-libtool \
+	pdf pdf-am ps ps-am tags uninstall uninstall-am \
+	uninstall-binPROGRAMS uninstall-info-am
+
+# Tell versions [3.59,3.63) of GNU make to not export all variables.
+# Otherwise a system limit (for SysV at least) may be exceeded.
+.NOEXPORT:

sampler/Observer.cpp

@@ -0,0 +1,79 @@
+/*
+ * PSAMP Reference Implementation
+ *
+ * Observer.cpp
+ *
+ * Implementation of the packet capturing thread
+ *
+ * Author: Michael Drueing <michael@drueing.de>
+ *
+ */
+	 
+#include <pcap.h>
+#include <iostream>
+#include <unistd.h>
+
+#include "Observer.h"
+#include "Globals.h"
+#include "Thread.h"
+
+using namespace std;
+
+//
+// void *Observer::observerThread(void *)
+//
+// This is the main observer loop. It graps packets from libpcap and
+// dispatches them to the registered receivers.
+//
+void *Observer::observerThread(void *arg)
+{
+  // first we need to get the instance back from the void *arg
+  Observer *obs = (Observer *)arg;
+  Packet *p;
+  int numReceivers = obs->receivers.size();
+  const unsigned char *rawPacketData;
+  void *myPacketData;
+  struct pcap_pkthdr packetHeader;
+  
+  // start capturing packets
+  LOG("Observer: Capturing started for device %s\n", obs->captureInterface);
+  obs->captureDevice=pcap_open_live(obs->captureInterface, obs->capturelen, 1, 2000, obs->errorBuffer);
+  // check for errors
+  if(!obs->captureDevice)
+  {
+    LOG("Observer: Error initializing pcap interface: %s\n", obs->errorBuffer);
+    return NULL;
+  }
+  
+  while(!obs->exitFlag)
+  {
+    // get next packet (no zero-copy possible *sigh*)
+    rawPacketData = pcap_next(obs->captureDevice, &packetHeader);
+    if(!rawPacketData)
+      continue; // no packet data was available
+    
+    if(!(myPacketData=malloc(packetHeader.caplen))) {
+    	/*
+	 FIXME!
+	 ALARM - no more memory available
+	 1) Start throwing away packets !
+	 2) Notify user !
+	 3) Try to resolve (?)
+	 */
+    }
+    memcpy(myPacketData, rawPacketData, packetHeader.caplen);
+    
+    p = new Packet(myPacketData, packetHeader.caplen, numReceivers);
+    p->timestamp = packetHeader.ts;
+    //LOG("Observer: received packet at %d.%04d, len=%d\n", 
+    //    p->timestamp.tv_sec, p->timestamp.tv_usec / 1000, packetHeader.caplen);
+    
+    // broadcast packet to all receivers
+    for(vector<ConcurrentQueue<Packet> *>::iterator it = obs->receivers.begin();
+      it != obs->receivers.end(); ++it)
+    {
+      (*it)->push(p);
+    }
+  }
+}
+

sampler/Observer.h

@@ -0,0 +1,111 @@
+/*
+ * PSAMP Reference Implementation
+ *
+ * Observer.h
+ *
+ * Declarations for observing process
+ *
+ * Author: Michael Drueing <michael@drueing.de>
+ *
+ */
+
+#ifndef OBSERVER_H
+#define OBSERVER_H
+
+// default pcap packet capture length
+#define CAPTURE_LENGTH 128
+/*
+ maximum physical packet length
+ you may want to adjust this on a special jumbo-framed GBit network
+ */
+#define CAPTURE_PHYSICAL_MAX 1526
+#include <vector>
+#include "Globals.h"
+#include "Thread.h"
+#include "ConcurrentQueue.h"
+#include "Packet.h"
+#include "PacketReceiver.h"
+#include <pcap.h>
+
+class Observer
+{
+public:
+  // ObserverThread constructor
+  Observer() : thread(Observer::observerThread), exitFlag(false), capturelen(CAPTURE_LENGTH)
+  {    
+    // query all available capture devices
+    LOG("Observer: Finding devices\n");
+    pcap_findalldevs(&allDevices, errorBuffer);
+    if (!allDevices)
+    {
+      LOG("Observer: error getting list of interfaces. Reason: %s\n", errorBuffer);
+    }
+    for (pcap_if_t *dev = allDevices; dev != NULL; dev=dev->next )
+    {
+      LOG("  Name=%s, DESC=%s\n", dev->name, dev->description);
+    }
+
+    LOG("Observer: Setting pcap snaplen to %d\n", capturelen);
+
+  };
+
+  ~Observer()
+  {
+    LOG("Observer: freeing devices...\n");
+    
+    if (captureDevice)
+      pcap_close(captureDevice);
+    
+    pcap_freealldevs(allDevices);
+  };
+
+  void startCapture(char *ifdesc)
+  {
+    captureInterface = ifdesc;
+    thread.run(this);
+  };
+
+  void terminateCapture()
+  {
+    exitFlag = true;
+  };
+
+  void addReceiver(PacketReceiver *recv)
+  {
+    receivers.push_back(recv->getQueue());
+  };
+  
+  bool Observer::setCaptureLen(int x)
+  {
+      if(x > CAPTURE_PHYSICAL_MAX) {
+          LOG("Capture length %d exceeds physical MTU %d (with header)\n", x, CAPTURE_PHYSICAL_MAX);
+          return false;
+      }
+      capturelen=x;
+
+      return true;
+  }
+
+  int Observer::getCaptureLen()
+  {
+      return capturelen;
+  }
+
+protected:
+  Thread thread;
+  
+  // pointer to list of pcap-devices
+  pcap_if_t *allDevices;
+  pcap_t *captureDevice;
+  char errorBuffer[PCAP_ERRBUF_SIZE];
+  int capturelen;
+
+  static void *observerThread(void *);
+
+public:
+  bool exitFlag;
+  char *captureInterface;
+  std::vector<ConcurrentQueue<Packet> *> receivers;
+};
+
+#endif

sampler/Packet.cpp

@@ -0,0 +1,17 @@
+/*
+ * PSAMP Reference Implementation
+ *
+ * Packet.cpp
+ *
+ * Encapsulates a captured packet with simple, thread-aware
+ * reference-(usage-) counting.
+ *
+ * Author: Michael Drueing <michael@drueing.de>
+ *
+ */
+
+#include "Packet.h"
+
+// this is the DEFAULT value (Ethernet specific)
+int Packet::IPHeaderOffset = 14;
+

sampler/Packet.h

@@ -0,0 +1,125 @@
+/*
+ * PSAMP Reference Implementation
+ *
+ * Packet.h
+ *
+ * Encapsulates a captured packet with simple, thread-aware
+ * reference-(usage-) counting.
+ *
+ * Author: Michael Drueing <michael@drueing.de>
+ *
+ */
+
+#ifndef PACKET_H
+#define PACKET_H
+
+#include <cstdlib>
+#include <cstdio>
+#include <ctime>
+#include <cstring>
+#include <stdint.h>
+#include <netinet/in.h>
+#include <sys/time.h>
+
+#include "Globals.h"
+#include "Lock.h"
+
+class Packet
+{
+public:
+  /*
+   data: the raw packet data from the wire, including physical header
+   ipHeader: start of the IP header: data + (physical dependent) IP header offset
+   transportHeader: start of the transport layer header (TCP/UDP): ip_header + variable IP header length
+   */
+  void *data;
+  void *ipHeader;
+  void *transportHeader;
+  
+  // The length of the packet in bytes
+  unsigned int length;
+    
+  // when was the packet received?
+  struct timeval timestamp;
+  
+  // construct a new Packet for a specified number of 'users'
+  Packet(void *packetData, unsigned int len, int numUsers = 1) : users(numUsers), refCountLock()
+  {
+    data = packetData;
+    ipHeader = (unsigned char *)data + IPHeaderOffset;
+    transportHeader = (unsigned char *)ipHeader + ipTransportHeaderOffset(ipHeader);
+    length = len;
+
+    /*
+     DO NOT SET TIMESTAMP HERE
+     IS SET IN OBSERVER!
+     */
+  };
+  
+  // Delete the packet and free all data associated with it. Should only be called
+  // if users==0 !
+  ~Packet()
+  {
+    if (users > 0)
+      LOG("Packet: WARNING: freeing in-use packet!\n");
+      
+    free(data);
+  }
+  
+  // call this function after processing the packet, NOT delete()!
+  void release()
+  {
+    int newUsers;
+    
+    refCountLock.lock();
+    --users;
+    newUsers = users;
+    refCountLock.unlock();
+   
+    if (newUsers < 0)
+      LOG("Packet: WARNING: trying to free already freed packet!\n");
+    
+    if (newUsers == 0)
+    {
+      delete this;
+    }	
+  };
+  
+  // read data from the IP header
+  void getPacketData(int offset, void *dest, int size) const
+  {
+    memcpy(dest, (char *)ipHeader + offset, size);
+  }
+  
+  // return a pointer into the packet to IP header offset given
+  void *getPacketData(int offset) const
+  {
+    return (char *)ipHeader + offset;
+  }
+  
+private:
+  // the raw offset at which the IP header starts in the packet
+  // for Ethernet, this is 14 bytes (MAC header size)
+  static int IPHeaderOffset;
+  
+  // Number of concurrent users of this packet. Decremented each time
+  // release() is called. After it reaches zero, the packet is deleted.
+  int users;
+
+  Lock refCountLock;
+  
+  // return the offset the transport header lies; IP knows about variable ip options field
+  static inline unsigned int ipTransportHeaderOffset(void *ipPacket)
+  {
+    /*
+     the header length (incl. options field) is:
+     last 4 bits in the first byte * 4bytes
+    */
+    unsigned char len = *((unsigned char *)ipPacket) & 0x0f;
+
+    return len << 2;
+  }
+  
+};
+
+#endif

sampler/PacketProcessor.h

@@ -0,0 +1,27 @@
+/*
+ * PSAMP Reference Implementation
+ *
+ * PacketProcessor.h
+ *
+ * Base class for a packet processor (filter/sampler)
+ *
+ * Author: Michael Drueing <michael@drueing.de>
+ *
+ */
+
+#ifndef PACKET_PROCESSOR_H
+#define PACKET_PROCESSOR_H
+ 
+#include "Globals.h"
+#include "Packet.h"
+
+class PacketProcessor
+{
+public:
+  // processPacket(Packet *)
+  // processes a packet and returns true if the packet passes the
+  // filter/sampler and false if it should be dropped
+  virtual bool processPacket(const Packet *p) = 0;
+};
+
+#endif

sampler/PacketReceiver.h

@@ -0,0 +1,41 @@
+/*
+ * PSAMP Reference Implementation
+ *
+ * PacketReceiver.h
+ *
+ * Base class for all Packet-receiving classes
+ *
+ * Author: Michael Drueing <michael@drueing.de>
+ *
+ */
+
+#ifndef PACKET_RECEIVER_H
+#define PACKET_RECEIVER_H
+
+#include "ConcurrentQueue.h"
+#include "Packet.h"
+
+class PacketReceiver
+{
+protected:
+  ConcurrentQueue<Packet> *queue;
+  
+public:
+  PacketReceiver()
+  {
+    queue = new ConcurrentQueue<Packet>();
+  }
+
+  virtual ~PacketReceiver()
+  {
+    delete queue;
+  }
+
+  inline virtual ConcurrentQueue<Packet> *getQueue() const
+  {
+    return queue;
+  }
+  
+};
+
+#endif

sampler/PacketSink.cpp

@@ -0,0 +1,31 @@
+/*
+ * PSAMP Reference Implementation
+ *
+ * PacketSink.cpp
+ *
+ * Implementation of a simple packet sink
+ *
+ * Author: Michael Drueing <michael@drueing.de>
+ *
+ */
+
+#include <iostream>
+#include "PacketSink.h"
+
+using namespace std;
+
+void *PacketSink::packetSinkProcess(void *arg)
+{
+  PacketSink *sink = (PacketSink *)arg;
+  ConcurrentQueue<Packet> *queue = sink->getQueue();
+  Packet *p;
+  
+  cerr << "PacketSink started" << endl;
+  while (!sink->exitFlag)
+  {
+    p = queue->pop();
+    p->release();
+    //cerr << "*";
+  }
+}
+

sampler/PacketSink.h

@@ -0,0 +1,48 @@
+/*
+ * PSAMP Reference Implementation
+ *
+ * PacketSink.h
+ *
+ * Implementation of a dummy packet sink
+ *
+ * Author: Michael Drueing <michael@drueing.de>
+ *
+ */
+
+#ifndef PACKET_SINK_H
+#define PACKET_SINK_H
+
+#include "Globals.h"
+#include "Thread.h"
+#include "PacketReceiver.h"
+
+class PacketSink : public PacketReceiver
+{
+public:
+  PacketSink() : thread(PacketSink::packetSinkProcess), exitFlag(false)
+  {
+  };
+
+  ~PacketSink()
+  {
+  };
+
+  inline void runSink()
+  {
+    thread.run(this);
+  };
+
+  inline void terminateSink()
+  {
+    exitFlag = true;
+  };
+
+protected:
+  Thread thread;
+  static void *packetSinkProcess(void *);
+
+public:
+  bool exitFlag;
+};
+
+#endif

sampler/RandomSampler.cpp

@@ -0,0 +1,53 @@
+/*
+ * PSAMP Reference Implementation
+ *
+ * RandomSampler.cpp
+ *
+ * Random n-out-of-N sampling of packets
+ *
+ * Author: Michael Drueing <michael@drueing.de>
+ *
+ */
+ 
+#include <cstdlib>
+#include <ctime>
+
+#include "RandomSampler.h"
+
+RandomSampler::RandomSampler(int n, int N) : samplingSize(N), acceptSize(n), currentPos(0)
+{
+  int pos;
+  
+  if (n > N)
+  {
+    fprintf(stderr, "RandomSampler: %d-out-of%d makes no sense!\n", n, N);
+    exit(3);
+  }
+  
+  sampleMask.clear();
+  sampleMask.insert(sampleMask.begin(), N, false);
+  
+  srand(time(0));
+  
+  // setup sampling bitfield
+  // TODO: There might be a more elegant solution to this...
+  for (int i = 0; i < acceptSize; i++)
+  {
+    // find a free spot in the sampleMask (i.e. a position
+    // with FALSE in it) and set it to TRUE
+    do
+    {
+      pos = rand() % samplingSize;
+    } while (sampleMask[pos]);
+    sampleMask[pos] = true;
+  }
+};
+
+bool RandomSampler::processPacket(const Packet *p)
+{
+  bool accepted = sampleMask[currentPos];
+  
+  currentPos = (currentPos + 1) % samplingSize;
+  
+  return accepted;
+}

sampler/RandomSampler.h

@@ -0,0 +1,35 @@
+/*
+ * PSAMP Reference Implementation
+ *
+ * RandomSampler.h
+ *
+ * Random n-out-of-N sampling of packets
+ *
+ * Author: Michael Drueing <michael@drueing.de>
+ *
+ */
+
+// implementation of the sampler explained in section 3.1.2.1 of PSAMP-Sample-Tech
+
+#ifndef RANDOM_SAMPLER_H
+#define RANDOM_SAMPLER_H
+
+#include "Globals.h"
+#include "PacketProcessor.h"
+#include <vector>
+
+class RandomSampler : public PacketProcessor
+{
+public:
+  RandomSampler(int n, int N);
+  
+  virtual bool processPacket(const Packet *p);
+  
+protected:
+  int samplingSize; // N
+  int acceptSize;   // n
+  int currentPos;
+  std::vector<bool> sampleMask;
+};
+
+#endif

sampler/Semaphore.h

@@ -0,0 +1,52 @@
+/*
+ * PSAMP Reference Implementation
+ *
+ * Semaphore.h
+ *
+ * Class which encapsulates a semaphore
+ *
+ * Author: Michael Drueing <michael@drueing.de>
+ *
+ */
+
+#ifndef SEMAPHORE_H
+#define SEMAPHORE_H
+
+#include <semaphore.h>
+
+class Semaphore
+{
+private:
+  // the system semaphore used (currently using pthread semaphores)
+  sem_t semaphore;
+  
+public:
+  // constructs a new Semaphore
+  inline Semaphore(int initialValue = 0)
+  {
+    sem_init(&semaphore, 0, initialValue);
+  };
+  
+  // deletes a Semaphore.
+  inline ~Semaphore()
+  {
+    sem_destroy(&semaphore);
+  };
+  
+  inline void wait()
+  {
+    sem_wait(&semaphore);
+  };
+
+  inline void post()
+  {
+    sem_post(&semaphore);
+  };
+
+  inline bool try_wait()
+  {
+    return (sem_trywait(&semaphore) == 0 ? true : false);
+  };
+};
+
+#endif

sampler/SystematicSampler.cpp

@@ -0,0 +1,36 @@
+/*
+ * PSAMP Reference Implementation
+ *
+ * SystematicSampler.cpp
+ *
+ * Implementation of systematic sampler
+ *
+ * Author: Michael Drueing <michael@drueing.de>
+ *
+ */
+
+#include "SystematicSampler.h"
+
+bool SystematicSampler::processPacket(const Packet *p)
+{
+  struct timeval elapsed;
+  unsigned long msecs;
+  
+  // calculate time elapsed since start of sampling
+  timersub(&p->timestamp, &this->startTime, &elapsed);
+  
+  // calculate number of milliseconds
+  msecs = (elapsed.tv_sec * 1000) + (elapsed.tv_usec / 1000);
+    
+  packetCount++;
+  
+  if (samplingType == SystematicSampler::TimeBasedSampler)
+  {
+    // normalize to interval borders and check if it's <= onTime
+    return ((msecs % interval) < samplingOnTime);
+  }
+  else
+  {
+    return ((packetCount % interval) < samplingOnTime);
+  }
+}

sampler/SystematicSampler.h

@@ -0,0 +1,57 @@
+/*
+ * PSAMP Reference Implementation
+ *
+ * SystematicSampler.h
+ *
+ * Systematic sampling of packets (count-based/time-based)
+ *
+ * Author: Michael Drueing <michael@drueing.de>
+ *
+ */
+ 
+#ifndef SYSTEMATIC_SAMPLER_H
+#define SYSTEMATIC_SAMPLER_H
+
+#include <sys/time.h>
+#include "Globals.h"
+#include "PacketProcessor.h"
+
+class SystematicSampler : public PacketProcessor
+{
+public:
+  enum SystematicSamplerType {
+    CountBasedSampler = 0x01,
+    TimeBasedSampler = 0x02
+  };
+  
+  // constructs a new systematic sampler (see section 3.1.1 of PSAMP-Sample-Tech)
+  // if type is CountBasedSampler then onTime and offTime specify how long to sample.
+  // for example if onTime = 10 and offTime = 90, we sample 10 packets out of 100
+  // if type is TimeBasedSampler, then onTime and offTime are specified in milliseconds
+  // and specify how long to keep capturing packets.
+  SystematicSampler(SystematicSamplerType type, unsigned long onTime, unsigned long offTime) :
+      samplingType(type), samplingOnTime(onTime), samplingOffTime(offTime), packetCount(0)
+  {
+    gettimeofday(&startTime, 0);
+    
+    interval = samplingOnTime + samplingOffTime;
+  };
+
+  ~SystematicSampler();
+
+  virtual bool processPacket(const Packet *p);
+  
+protected:
+  SystematicSamplerType samplingType;
+  unsigned long samplingOnTime;
+  unsigned long samplingOffTime;
+  unsigned long interval;
+  
+  // for time-based sampler:
+  struct timeval startTime;
+  
+  // for count-based samplers
+  unsigned long packetCount;
+};
+
+#endif

sampler/Template.cpp

@@ -0,0 +1,120 @@
+/*
+ * PSAMP Reference Implementation
+ *
+ * Template.cpp
+ *
+ * A Template definition
+ *
+ * Author: Michael Drueing <michael@drueing.de>
+ *
+ */
+ 
+#include <cstring>
+#include <fstream>
+#include <iostream>
+#include "Template.h"
+#include "Globals.h"
+
+using namespace std;
+
+
+bool Template::addField(uint16_t id, uint16_t len)
+{
+	uint16_t offset;
+
+	switch(id) {
+	
+	case FT_SRCIP4:
+		offset=12;
+		break;
+		
+	case FT_DSTIP4:
+		offset=16;
+		break;
+		
+	case FT_PROTO:
+		offset=9;
+		break;
+		
+	case FT_SRCPORT:
+		offset=20;
+		break;
+	
+	case FT_DSTPORT:
+		offset=22;
+		break;
+	default:
+		LOG("ID %d currently not supported\n", id);
+		return false;	
+	}
+
+	addFieldWithOffset(id, len, offset);
+	return true;
+}
+
+
+void AddFieldFromString(Template *temp, const char *field)
+{
+  if (strncasecmp(field, "SRCIP4", 6) == 0)
+  {
+    temp->addFieldWithOffset(FT_SRCIP4, 4, 12); // source address is as offset 12    
+  }
+  else if (strncasecmp(field, "DSTIP4", 6) == 0)
+  {
+    temp->addFieldWithOffset(FT_DSTIP4, 4, 16); // dest address is at offset 16
+  }
+  else if (strncasecmp(field, "PROTO", 5) == 0)
+  {
+    temp->addFieldWithOffset(FT_PROTO, 2, 9); // protocol is at offset 9
+  }
+  else if (strncasecmp(field, "SRCPORT", 7) == 0)
+  {
+    temp->addFieldWithOffset(FT_SRCPORT, 2, 20);  // source port is as offset 20 (TCP offset 0)
+  }
+  else if (strncasecmp(field, "DSTPORT", 7) == 0)
+  {
+    temp->addFieldWithOffset(FT_DSTPORT, 2, 22); // dest port is at offset 22 (TCP offset 2)
+  }
+}
+
+Template *Template::readFromFile(const char *fileName)
+{
+  char buffer[256];
+  Template *tmp = 0;
+  
+  ifstream f(fileName);
+  
+  // get template id
+  while (!f.eof())
+  {
+    f.getline(buffer, 255);
+    if ((buffer[0] == '#') || (buffer[0] == 0x0d) || (buffer[0] == 0x0a) || (buffer[0] == 0))
+      continue;
+    
+    if (strncasecmp(buffer, "ID ", 3) == 0)
+    {      
+      // assign template id
+      tmp = new Template(strtol(buffer + 3, 0, 10));
+      break;
+    }
+    else
+    {
+      LOG("Expected ID\n");
+      return 0;
+    }    
+  }  
+  
+  // get template fields
+  while (!f.eof())
+  {
+    f.getline(buffer, 255);
+    if ((buffer[0] == '#') || (buffer[0] == 0x0d) || (buffer[0] == 0x0a) || (buffer[0] == 0))
+      continue;
+      
+    AddFieldFromString(tmp, buffer);
+  }
+  
+  f.close();
+  
+  return tmp;
+}

sampler/Template.h

@@ -0,0 +1,84 @@
+/*
+ * PSAMP Reference Implementation
+ *
+ * Template.h
+ *
+ * A Template definition
+ *
+ * Author: Michael Drueing <michael@drueing.de>
+ *
+ */
+
+#ifndef TEMPLATE_H
+#define TEMPLATE_H
+
+#include <stdint.h>
+#include "Globals.h"
+
+#define MAX_TEMPLATE_FIELDS 512
+
+// the different field types used
+#define FT_SRCIP4   8
+#define FT_DSTIP4  12
+#define FT_PROTO    4
+#define FT_SRCPORT  7
+#define FT_DSTPORT 11
+
+class Template
+{
+private:
+  // the template-id for this template
+  unsigned short templateID;
+  // the number of fields in this template
+  unsigned short fieldCount;
+  
+  // the field types for each field
+  unsigned short fieldType[MAX_TEMPLATE_FIELDS];
+  
+  // the length of each field
+  unsigned short fieldLength[MAX_TEMPLATE_FIELDS];
+  
+  // the offset in the TCP/IP packet for each field
+  unsigned short fieldPacketOffset[MAX_TEMPLATE_FIELDS];
+
+public:
+  Template(unsigned short id) : templateID(id), fieldCount(0)
+  {   
+  };
+  
+  ~Template()
+  {
+  };
+  
+  int getFieldCount() const
+  {
+    return fieldCount;
+  };
+  
+  void addFieldWithOffset(unsigned short type, unsigned short length, unsigned short offset)
+  {
+    //LOG("Adding field type %d, length %d, offset %d\n", type, length, offset);
+    fieldType[fieldCount] = type;
+    fieldLength[fieldCount] = length;
+    fieldPacketOffset[fieldCount] = offset;
+    fieldCount++;
+  };
+  
+  void getFieldInfo(int num, unsigned short *type, unsigned short *length, unsigned short *offset) const
+  {
+    *type = fieldType[num];
+    *length = fieldLength[num];
+    *offset = fieldPacketOffset[num];
+  }
+  
+  int getTemplateID() const
+  {
+    return templateID;
+  }
+  
+  static Template *readFromFile(const char *fileName);
+  bool Template::addField(uint16_t id, uint16_t len);
+
+};
+
+#endif

sampler/Test.cpp

@@ -0,0 +1,64 @@
+/*
+ * PSAMP Reference Implementation
+ *
+ * Test.cpp
+ *
+ * Test software
+ *
+ * Author: Michael Drueing <michael@drueing.de>
+ *
+ */
+
+#include "Globals.h"
+#include "Packet.h"
+#include "Observer.h"
+#include "PacketSink.h"
+#include "ExporterSink.h"
+#include "Filter.h"
+#include "Template.h"
+#include "IPHeaderFilter.h"
+
+#include <unistd.h>
+#include <iostream>
+
+using namespace std;
+
+int main(int argc, char *argv[])
+{
+  Observer *obs = new Observer();
+  Filter *filter = new Filter();
+  Template *tmpl = Template::readFromFile("template.cfg");
+  //PacketProcessor *samp = new SystematicSampler(SystematicSampler::CountBasedSampler, 10, 10);
+  PacketProcessor *samp = new RandomSampler(10, 20);
+  IPHeaderFilter *flt1 = new IPHeaderFilter(9, 1, CMP_EQ, 1); // ICMP
+  //PacketSink *sink = new PacketSink();
+  ExporterSink *sink = new ExporterSink(tmpl, 2342);
+  
+  sink->AddCollector("127.0.0.1", 4711, "UDP");
+
+  // connect the output of the observer to the packet sink
+  //obs->addReceiver(sink->getQueue());
+  
+  obs->addReceiver(filter);
+  filter->setReceiver(sink);
+  //filter->addProcessor(samp);
+  filter->addProcessor(flt1);
+  
+  sink->runSink();
+  filter->startFilter();
+  obs->startCapture("lo");
+  
+  sleep(10);
+  
+  cerr << "terminating...";
+  obs->terminateCapture();
+  filter->terminate();
+  sink->terminateSink();
+  cerr << "done." << endl;
+  LOG("%ld of %ld packets passed (%05.2f%%)\n", filter->pktOut, filter->pktIn,
+            (float)(100.0*filter->pktOut)/(float)(filter->pktIn));
+  
+  delete obs;  
+  return 0;
+}
+

sampler/Thread.h

@@ -0,0 +1,59 @@
+/*
+ * PSAMP Reference Implementation
+ *
+ * Thread.h
+ *
+ * Implementation of a base class for a thread
+ *
+ * Author: Michael Drueing <michael@drueing.de>
+ *
+ */
+
+#ifndef THREAD_H
+#define THREAD_H
+
+#include <pthread.h>
+#include "Globals.h"
+
+typedef void* (*ThreadFunc)(void *);
+
+class Thread
+{
+public:
+  Thread(ThreadFunc threadFunction) : exitFlag(false), f(threadFunction)
+  {
+  };
+
+  bool run(void *threadData)
+  {
+    //data = threadData;
+    return pthread_create(&thread, NULL, f, threadData) == 0;
+  };
+
+  void *join()
+  {
+    void *result = NULL;
+    if (pthread_join(thread, &result))
+      LOG("Thread: ERROR: Join failed\n");
+      
+    return result;
+  };
+
+  bool detach()
+  {
+    return (pthread_detach(thread) == 0);
+  }
+  
+  ~Thread()
+  {
+  };
+
+  bool exitFlag;
+  //void *data;
+  
+private:
+  pthread_t thread;
+  ThreadFunc f;
+};
+
+#endif

sampler/TimeoutLock.h

@@ -0,0 +1,121 @@
+/*
+ * PSAMP Reference Implementation
+ *
+ * TimeoutLock.h
+ *
+ * Class which encapsulates a lock with a timeout.
+ * Currently implemented using a Condition Variable.
+ *
+ * Author: Michael Drueing <michael@drueing.de>
+ *
+ */
+ 
+#ifndef TIMEOUT_LOCK_H
+#define TIMEOUT_LOCK_H
+
+#include "Globals.h"
+
+#include <pthread.h>
+#include <time.h>
+#include <sys/time.h>
+#include <errno.h>
+
+class TimeoutLock
+{
+private:
+  pthread_cond_t cond;
+  pthread_mutex_t mutex;
+  bool locked;
+  
+public:
+  // construct a new lock
+  inline TimeoutLock() : locked(false)
+  {
+    pthread_mutex_init(&mutex, 0);
+    pthread_cond_init(&cond, 0);
+  }
+  
+  // destroy the lock
+  // TODO: actually check if the lock is held. If so, releasing
+  //       it causes unpredictable results!
+  inline ~TimeoutLock()
+  {
+    pthread_cond_destroy(&cond);
+    pthread_mutex_destroy(&mutex);
+  }
+  
+  // Acquire the lock or wait until timeout has expired.
+  // if the lock was acquired, return true
+  // if the timeout was reached without the lock being acquired,
+  // return false
+  // a timeout of -1 means infinite (i.e. results are equal to a "normal" Lock
+  inline bool lock(long timeout_ms = -1)
+  {
+    bool result;
+    struct timeval tv;
+    struct timespec ts;
+    
+    pthread_mutex_lock(&mutex);
+    
+    if (locked)
+    {
+      // there's already someone holding the lock
+      if (timeout_ms == -1)
+      {
+        // wait indefinitely
+        pthread_cond_wait(&cond, &mutex);
+        locked = true;
+      }
+      else
+      {
+        // calculate absolute time for timeout
+        gettimeofday(&tv, 0);
+        // add the timeout value to the current time
+        ts.tv_sec = tv.tv_sec;
+        ts.tv_nsec = tv.tv_usec * 1000L + timeout_ms * 1000000L;
+        if (ts.tv_nsec > 999999999L) // overflow in nsec?
+        {
+          ts.tv_sec += (ts.tv_nsec / 1000000000L);
+          ts.tv_nsec %= 1000000000L;
+        }
+        // lock and wait for signal or timeout
+        if (pthread_cond_timedwait(&cond, &mutex, &ts) == ETIMEDOUT)
+        {
+          // timeout occured
+          locked = false;
+        }
+        else
+        {
+          // lock acquired
+          locked = true;
+        }
+      }
+    }
+    else
+    {
+      // the lock is not yet acquired
+      locked = true;
+    }
+    
+    pthread_mutex_unlock(&mutex);
+    return locked;
+  }
+  
+  // unlock the lock
+  inline void unlock()
+  {
+    pthread_mutex_lock(&mutex);
+    
+    locked = false;             // set to "unlocked"
+    pthread_cond_signal(&cond); // wake up next thread waiting to acquire the lock
+    
+    pthread_mutex_unlock(&mutex);
+  }
+  
+  inline bool try_lock()
+  {
+    // not yet implemented...
+  }
+};
+ 
+#endif