diff --git a/server/include/config.h b/server/include/config.h index 5dec0bcd8..36fe004a4 100644 --- a/server/include/config.h +++ b/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) */ diff --git a/server/include/hooks.h b/server/include/hooks.h index 8cc8094ae..b54e11f82 100644 --- a/server/include/hooks.h +++ b/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); diff --git a/server/include/msg_format.h b/server/include/msg_format.h index e0cc9dbeb..eb680a57e 100644 --- a/server/include/msg_format.h +++ b/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, \ diff --git a/server/src/hooks.c b/server/src/hooks.c index 733d2ba9b..9811b9bd1 100644 --- a/server/src/hooks.c +++ b/server/src/hooks.c @@ -13,6 +13,7 @@ *********************************************************************************/ #include +#include #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; }