Merge branch 'armo' of github.com:RWTH-ACS/S2SS into armo

server/include/config.h

@@ -52,11 +52,18 @@
 	  { "/etc/machine-id", "0d8399d0216314f083b9ed2053a354a8" }, \
 	  { "/dev/sda2", "\x53\xf6\xb5\xeb\x8b\x16\x46\xdc\x8d\x8f\x5b\x70\xb8\xc9\x1a\x2a", 0x468 } }
 
-/* Hook function configuration */
-#define HOOK_FIR_INDEX		 1 /**< The first value of message should be filtered. */
-#define HOOK_TS_INDEX		-1 /**< The last value of message should be overwritten by a timestamp. */
-#define HOOK_DECIMATE_RATIO 	30 /**< Only forward every 30th message to the destination nodes. */
+/* Hard coded configuration of hook functions */
+#define HOOK_FIR_INDEX		0	/**< Which value inside a message should be filtered? */
+#define HOOK_FIR_COEFFS		{ -0.003658148158728, -0.008882653268281, 0.008001024183003,	\
+				  0.08090485991761,    0.2035239551043,   0.3040703593515,	\
+				  0.3040703593515,     0.2035239551043,   0.08090485991761,	\
+				  0.008001024183003,  -0.008882653268281,-0.003658148158728 }
+	
+#define HOOK_TS_INDEX		-1	/**< The last value of message should be overwritten by a timestamp. */
+#define HOOK_DECIMATE_RATIO 	30	/**< Only forward every 30th message to the destination nodes. */
 
+#define HOOK_DEDUP_TYPE		HOOK_ASYNC
+#define HOOK_DEDUP_TRESH	1e-3	/**< Do not send messages when difference of values to last message is smaller than this threshold */
 /** Global configuration */
 struct settings {
 	/** Process priority (lower is better) */

server/include/hooks.h

@@ -76,6 +76,9 @@ struct hook {
  * @{
  */
 
+/** Example hook: Drop messages whose values are similiar to the previous ones */
+int hook_deduplicate(struct path *);
+
 /** Example hook: Print the message. */
 int hook_print(struct path *p);
 

server/include/msg_format.h

@@ -53,7 +53,7 @@
 }
 
 /** Initialize a message */
-#define MSG_INIT(i) (struct msg) {	\
+#define MSG_INIT(i) {	\
 	.version = MSG_VERSION,		\
 	.type = MSG_TYPE_DATA,		\
 	.endian = MSG_ENDIAN_HOST,	\

server/src/hooks.c

@@ -13,6 +13,7 @@
  *********************************************************************************/
 
 #include <string.h>
+#include <math.h>
 
 #include "timing.h"
 #include "config.h"
@@ -25,6 +26,33 @@
 
 struct list hooks;
 
+REGISTER_HOOK("deduplicate", 99, hook_deduplicate, HOOK_DEDUP_TYPE)
+int hook_deduplicate(struct path *p)
+{
+	int ret = 0;
+#if HOOK_DEDUP_TYPE == HOOK_ASYNC
+	/** Thread local storage (TLS) is used to maintain a copy of the last run of the hook */
+	static __thread struct msg previous = MSG_INIT(0);
+	struct msg *prev = &previous;
+#else
+	struct msg *prev = p->previous;
+#endif
+	struct msg *cur = p->current;
+	
+	for (int i = 0; i < MIN(cur->length, prev->length); i++) {
+		if (fabs(cur->data[i].f - prev->data[i].f) > HOOK_DEDUP_TRESH)
+			goto out;
+	}
+	
+	ret = -1; /* no appreciable change in values, we will drop the packet */
+
+out:
+#if HOOK_DEDUP_TYPE == HOOK_ASYNC
+	memcpy(prev, cur, sizeof(struct msg)); /* save current message for next run */
+#endif
+	return ret;
+}
+
 REGISTER_HOOK("print", 99, hook_print, HOOK_MSG)
 int hook_print(struct path *p)
 {
@@ -59,31 +87,39 @@ int hook_ts(struct path *p)
 	return 0;
 }
 
-REGISTER_HOOK("fir", 99, hook_fir, HOOK_POST)
+REGISTER_HOOK("fir", 99, hook_fir, HOOK_MSG)
 int hook_fir(struct path *p)
 {
-	/** Simple FIR-LP: F_s = 1kHz, F_pass = 100 Hz, F_block = 300
+	/** Coefficients for simple FIR-LowPass:
+	 *   F_s = 1kHz, F_pass = 100 Hz, F_block = 300
+	 *
 	 * Tip: Use MATLAB's filter design tool and export coefficients
-	 *      with the integrated C-Header export */
-	static const double coeffs[] = {
-		-0.003658148158728, -0.008882653268281, 0.008001024183003,
-		0.08090485991761,    0.2035239551043,   0.3040703593515,
-		0.3040703593515,     0.2035239551043,   0.08090485991761,
-		0.008001024183003,  -0.008882653268281,-0.003658148158728 };
+	 *      with the integrated C-Header export
+	 */
+	static const double coeffs[] = HOOK_FIR_COEFFS;
+	
+	/** Per path thread local storage for unfiltered sample values.
+	 * The message ringbuffer (p->pool & p->current) will contain filtered data!
+	 */
+	static __thread double *past = NULL;
+	
+	/** @todo Avoid dynamic allocation at runtime */
+	if (!past)
+		alloc(p->poolsize * sizeof(double));
+	
+	
+	/* Current value of interest */
+	float *cur = &p->current->data[HOOK_FIR_INDEX].f;
+	
+	/* Save last sample, unfiltered */
+	past[p->received % p->poolsize] = *cur;
 
-	/* Accumulator */
-	double sum = 0;
-
-	/** Trim FIR length to length of history buffer */
-	int len = MIN(ARRAY_LEN(coeffs), p->poolsize);
-
-	for (int i = 0; i < len; i++) {
-		struct msg *old = &p->pool[(p->poolsize+p->received-i) % p->poolsize];
-
-		sum += coeffs[i] * old->data[HOOK_FIR_INDEX].f;
-	}
-
-	p->current->data[HOOK_FIR_INDEX].f = sum;
+	/* Reset accumulator */
+	*cur = 0;
+	
+	/* FIR loop */
+	for (int i = 0; i < MIN(ARRAY_LEN(coeffs), p->poolsize); i++)
+		*cur += coeffs[i] * past[p->received+p->poolsize-i];
 
 	return 0;
 }