1/*
2 * librdkafka - Apache Kafka C library
3 *
4 * Copyright (c) 2018 Magnus Edenhill
5 * All rights reserved.
6 *
7 * Redistribution and use in source and binary forms, with or without
8 * modification, are permitted provided that the following conditions are met:
9 *
10 * 1. Redistributions of source code must retain the above copyright notice,
11 * this list of conditions and the following disclaimer.
12 * 2. Redistributions in binary form must reproduce the above copyright notice,
13 * this list of conditions and the following disclaimer in the documentation
14 * and/or other materials provided with the distribution.
15 *
16 * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS"
17 * AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE
18 * IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE
19 * ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT OWNER OR CONTRIBUTORS BE
20 * LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR
21 * CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF
22 * SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS
23 * INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN
24 * CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE)
25 * ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE
26 * POSSIBILITY OF SUCH DAMAGE.
27 */
28
29#ifndef _RDINTERVAL_H_
30#define _RDINTERVAL_H_
31
32#include "rd.h"
33
34typedef struct rd_interval_s {
35 rd_ts_t ri_ts_last; /* last interval timestamp */
36 rd_ts_t ri_fixed; /* fixed interval if provided interval is 0 */
37 int ri_backoff; /* back off the next interval by this much */
38} rd_interval_t;
39
40
41static RD_INLINE RD_UNUSED void rd_interval_init (rd_interval_t *ri) {
42 memset(ri, 0, sizeof(*ri));
43}
44
45
46
47/**
48 * Returns the number of microseconds the interval has been over-shot.
49 * If the return value is >0 (i.e., time for next intervalled something) then
50 * the time interval is updated for the next inteval.
51 *
52 * A current time can be provided in 'now', if set to 0 the time will be
53 * gathered automatically.
54 *
55 * If 'interval_us' is set to 0 the fixed interval will be used, see
56 * 'rd_interval_fixed()'.
57 *
58 * If this is the first time rd_interval() is called after an _init() or
59 * _reset() and the \p immediate parameter is true, then a positive value
60 * will be returned immediately even though the initial interval has not passed.
61 */
62#define rd_interval(ri,interval_us,now) rd_interval0(ri,interval_us,now,0)
63#define rd_interval_immediate(ri,interval_us,now) \
64 rd_interval0(ri,interval_us,now,1)
65static RD_INLINE RD_UNUSED rd_ts_t rd_interval0 (rd_interval_t *ri,
66 rd_ts_t interval_us,
67 rd_ts_t now,
68 int immediate) {
69 rd_ts_t diff;
70
71 if (!now)
72 now = rd_clock();
73 if (!interval_us)
74 interval_us = ri->ri_fixed;
75
76 if (ri->ri_ts_last || !immediate) {
77 diff = now - (ri->ri_ts_last + interval_us + ri->ri_backoff);
78 } else
79 diff = 1;
80 if (unlikely(diff > 0)) {
81 ri->ri_ts_last = now;
82 ri->ri_backoff = 0;
83 }
84
85 return diff;
86}
87
88
89/**
90 * Reset the interval to zero, i.e., the next call to rd_interval()
91 * will be immediate.
92 */
93static RD_INLINE RD_UNUSED void rd_interval_reset (rd_interval_t *ri) {
94 ri->ri_ts_last = 0;
95 ri->ri_backoff = 0;
96}
97
98/**
99 * Back off the next interval by `backoff_us` microseconds.
100 */
101static RD_INLINE RD_UNUSED void rd_interval_backoff (rd_interval_t *ri,
102 int backoff_us) {
103 ri->ri_backoff = backoff_us;
104}
105
106/**
107 * Expedite (speed up) the next interval by `expedite_us` microseconds.
108 * If `expedite_us` is 0 the interval will be set to trigger
109 * immedately on the next rd_interval() call.
110 */
111static RD_INLINE RD_UNUSED void rd_interval_expedite (rd_interval_t *ri,
112 int expedite_us) {
113 if (!expedite_us)
114 ri->ri_ts_last = 0;
115 else
116 ri->ri_backoff = -expedite_us;
117}
118
119/**
120 * Specifies a fixed interval to use if rd_interval() is called with
121 * `interval_us` set to 0.
122 */
123static RD_INLINE RD_UNUSED void rd_interval_fixed (rd_interval_t *ri,
124 rd_ts_t fixed_us) {
125 ri->ri_fixed = fixed_us;
126}
127
128/**
129 * Disables the interval (until rd_interval_init()/reset() is called).
130 * A disabled interval will never return a positive value from
131 * rd_interval().
132 */
133static RD_INLINE RD_UNUSED void rd_interval_disable (rd_interval_t *ri) {
134 /* Set last beat to a large value a long time in the future. */
135 ri->ri_ts_last = 6000000000000000000LL; /* in about 190000 years */
136}
137
138/**
139 * Returns true if the interval is disabled.
140 */
141static RD_INLINE RD_UNUSED int rd_interval_disabled (const rd_interval_t *ri) {
142 return ri->ri_ts_last == 6000000000000000000LL;
143}
144
145#endif /* _RDINTERVAL_H_ */
146