diff --git a/include/villas/hook.h b/include/villas/hook.h index c1b120b42..3db1fbd38 100644 --- a/include/villas/hook.h +++ b/include/villas/hook.h @@ -75,7 +75,6 @@ struct hook_type { struct hook { enum state state; - struct sample *prev, *last; struct path *path; struct hook_type *_vt; /**< C++ like Vtable pointer. */ diff --git a/lib/hooks/drop.c b/lib/hooks/drop.c index d6ab88012..4420d1002 100644 --- a/lib/hooks/drop.c +++ b/lib/hooks/drop.c @@ -29,45 +29,71 @@ #include "stats.h" #include "path.h" +struct drop { + struct sample *prev; +}; + +static int drop_start(struct hook *h) +{ + struct drop *d = h->_vd; + + d->prev = NULL; + + return 0; +} + +static int drop_stop(struct hook *h) +{ + struct drop *d = h->_vd; + + if (d->prev) + sample_put(d->prev); + + return 0; +} + static int drop_read(struct hook *h, struct sample *smps[], size_t *cnt) { int i, ok, dist; + + struct drop *d = h->_vd; + struct sample *prev, *cur = NULL; - for (i = 0, ok = 0; i < *cnt; i++) { - h->last = smps[i]; + for (i = 0, ok = 0, prev = d->prev; i < *cnt; i++, prev = cur) { + cur = smps[i]; - if (h->prev) { - dist = h->last->sequence - (int32_t) h->prev->sequence; + if (prev) { + dist = cur->sequence - (int32_t) prev->sequence; if (dist <= 0) { - warn("Dropped sample: sequence=%u, dist=%d, i=%d", h->last->sequence, dist, i); + debug(10, "Reordered sample: sequence=%u, distance=%d", cur->sequence, dist); if (h->path && h->path->stats) stats_update(h->path->stats->delta, STATS_REORDERED, dist); } - else { - struct sample *tmp; - - tmp = smps[i]; - smps[i] = smps[ok]; - smps[ok++] = tmp; - } - - /* To discard the first X samples in 'smps[]' we must - * shift them to the end of the 'smps[]' array. - * In case the hook returns a number 'ok' which is smaller than 'cnt', - * only the first 'ok' samples in 'smps[]' are accepted and further processed. - */ + else + goto ok; } - else { - struct sample *tmp; + else + goto ok; - tmp = smps[i]; - smps[i] = smps[ok]; - smps[ok++] = tmp; - } + continue; - h->prev = h->last; +ok: /* To discard the first X samples in 'smps[]' we must + * shift them to the end of the 'smps[]' array. + * In case the hook returns a number 'ok' which is smaller than 'cnt', + * only the first 'ok' samples in 'smps[]' are accepted and further processed. + */ + + smps[i] = smps[ok]; + smps[ok++] = cur; } + if (cur) + sample_get(cur); + if (d->prev) + sample_put(d->prev); + + d->prev = cur; + *cnt = ok; return 0; @@ -80,7 +106,10 @@ static struct plugin p = { .hook = { .priority = 3, .builtin = true, - .read = drop_read + .read = drop_read, + .start = drop_start, + .stop = drop_stop, + .size = sizeof(struct drop) } }; diff --git a/lib/hooks/restart.c b/lib/hooks/restart.c index 5f314df2f..2ae249611 100644 --- a/lib/hooks/restart.c +++ b/lib/hooks/restart.c @@ -28,18 +28,44 @@ #include "plugin.h" #include "path.h" +struct restart { + struct sample *prev; +}; + +static int restart_start(struct hook *h) +{ + struct restart *r = h->_vd; + + r->prev = NULL; + + return 0; +} + +static int restart_stop(struct hook *h) +{ + struct restart *r = h->_vd; + + if (r->prev) + sample_put(r->prev); + + return 0; +} + static int restart_read(struct hook *h, struct sample *smps[], size_t *cnt) { + int i; + struct restart *r = h->_vd; + struct sample *prev, *cur = NULL; + assert(h->path); - for (int i = 0; i < *cnt; i++) { - h->last = smps[i]; + for (i = 0, prev = r->prev; i < *cnt; i++, prev = cur) { + cur = smps[i]; - if (h->prev) { - if (h->last->sequence == 0 && - h->prev->sequence <= UINT32_MAX - 32) { - warn("Simulation for path %s restarted (prev->seq=%u, current->seq=%u)", - path_name(h->path), h->prev->sequence, h->last->sequence); + if (prev) { + if (cur->sequence == 0 && prev->sequence <= UINT32_MAX - 32) { + warn("Simulation for path %s restarted (previous->seq=%u, current->seq=%u)", + path_name(h->path), prev->sequence, cur->sequence); /* Run restart hooks */ for (size_t i = 0; i < list_length(&h->path->hooks); i++) { @@ -49,9 +75,14 @@ static int restart_read(struct hook *h, struct sample *smps[], size_t *cnt) } } } - - h->prev = h->last; } + + if (cur) + sample_get(cur); + if (r->prev) + sample_put(r->prev); + + r->prev = cur; return 0; } @@ -63,7 +94,10 @@ static struct plugin p = { .hook = { .priority = 1, .builtin = true, - .read = restart_read + .read = restart_read, + .start = restart_start, + .stop = restart_stop, + .size = sizeof(struct restart) } };