1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19 package org.apache.hadoop.hbase.quotas;
20
21
22 import org.apache.hadoop.conf.Configuration;
23 import org.apache.hadoop.hbase.HBaseConfiguration;
24 import org.apache.hadoop.hbase.classification.InterfaceAudience;
25 import org.apache.hadoop.hbase.classification.InterfaceStability;
26 import org.apache.hadoop.hbase.protobuf.ProtobufUtil;
27 import org.apache.hadoop.hbase.protobuf.generated.QuotaProtos.Throttle;
28 import org.apache.hadoop.hbase.protobuf.generated.QuotaProtos.TimedQuota;
29
30
31
32
33 @InterfaceAudience.Private
34 @InterfaceStability.Evolving
35 public class TimeBasedLimiter implements QuotaLimiter {
36 private static final Configuration conf = HBaseConfiguration.create();
37 private RateLimiter reqsLimiter = null;
38 private RateLimiter reqSizeLimiter = null;
39 private RateLimiter writeReqsLimiter = null;
40 private RateLimiter writeSizeLimiter = null;
41 private RateLimiter readReqsLimiter = null;
42 private RateLimiter readSizeLimiter = null;
43
44 private TimeBasedLimiter() {
45 if (FixedIntervalRateLimiter.class.getName().equals(
46 conf.getClass(RateLimiter.QUOTA_RATE_LIMITER_CONF_KEY, AverageIntervalRateLimiter.class)
47 .getName())) {
48 reqsLimiter = new FixedIntervalRateLimiter();
49 reqSizeLimiter = new FixedIntervalRateLimiter();
50 writeReqsLimiter = new FixedIntervalRateLimiter();
51 writeSizeLimiter = new FixedIntervalRateLimiter();
52 readReqsLimiter = new FixedIntervalRateLimiter();
53 readSizeLimiter = new FixedIntervalRateLimiter();
54 } else {
55 reqsLimiter = new AverageIntervalRateLimiter();
56 reqSizeLimiter = new AverageIntervalRateLimiter();
57 writeReqsLimiter = new AverageIntervalRateLimiter();
58 writeSizeLimiter = new AverageIntervalRateLimiter();
59 readReqsLimiter = new AverageIntervalRateLimiter();
60 readSizeLimiter = new AverageIntervalRateLimiter();
61 }
62 }
63
64 static QuotaLimiter fromThrottle(final Throttle throttle) {
65 TimeBasedLimiter limiter = new TimeBasedLimiter();
66 boolean isBypass = true;
67 if (throttle.hasReqNum()) {
68 setFromTimedQuota(limiter.reqsLimiter, throttle.getReqNum());
69 isBypass = false;
70 }
71
72 if (throttle.hasReqSize()) {
73 setFromTimedQuota(limiter.reqSizeLimiter, throttle.getReqSize());
74 isBypass = false;
75 }
76
77 if (throttle.hasWriteNum()) {
78 setFromTimedQuota(limiter.writeReqsLimiter, throttle.getWriteNum());
79 isBypass = false;
80 }
81
82 if (throttle.hasWriteSize()) {
83 setFromTimedQuota(limiter.writeSizeLimiter, throttle.getWriteSize());
84 isBypass = false;
85 }
86
87 if (throttle.hasReadNum()) {
88 setFromTimedQuota(limiter.readReqsLimiter, throttle.getReadNum());
89 isBypass = false;
90 }
91
92 if (throttle.hasReadSize()) {
93 setFromTimedQuota(limiter.readSizeLimiter, throttle.getReadSize());
94 isBypass = false;
95 }
96 return isBypass ? NoopQuotaLimiter.get() : limiter;
97 }
98
99 public void update(final TimeBasedLimiter other) {
100 reqsLimiter.update(other.reqsLimiter);
101 reqSizeLimiter.update(other.reqSizeLimiter);
102 writeReqsLimiter.update(other.writeReqsLimiter);
103 writeSizeLimiter.update(other.writeSizeLimiter);
104 readReqsLimiter.update(other.readReqsLimiter);
105 readSizeLimiter.update(other.readSizeLimiter);
106 }
107
108 private static void setFromTimedQuota(final RateLimiter limiter, final TimedQuota timedQuota) {
109 limiter.set(timedQuota.getSoftLimit(), ProtobufUtil.toTimeUnit(timedQuota.getTimeUnit()));
110 }
111
112 @Override
113 public void checkQuota(long writeReqs, long estimateWriteSize, long readReqs,
114 long estimateReadSize) throws ThrottlingException {
115 if (!reqsLimiter.canExecute(writeReqs + readReqs)) {
116 ThrottlingException.throwNumRequestsExceeded(reqsLimiter.waitInterval());
117 }
118 if (!reqSizeLimiter.canExecute(estimateWriteSize + estimateReadSize)) {
119 ThrottlingException.throwRequestSizeExceeded(
120 reqSizeLimiter.waitInterval(estimateWriteSize + estimateReadSize));
121 }
122
123 if (estimateWriteSize > 0) {
124 if (!writeReqsLimiter.canExecute(writeReqs)) {
125 ThrottlingException.throwNumWriteRequestsExceeded(writeReqsLimiter.waitInterval());
126 }
127 if (!writeSizeLimiter.canExecute(estimateWriteSize)) {
128 ThrottlingException.throwWriteSizeExceeded(
129 writeSizeLimiter.waitInterval(estimateWriteSize));
130 }
131 }
132
133 if (estimateReadSize > 0) {
134 if (!readReqsLimiter.canExecute(readReqs)) {
135 ThrottlingException.throwNumReadRequestsExceeded(readReqsLimiter.waitInterval());
136 }
137 if (!readSizeLimiter.canExecute(estimateReadSize)) {
138 ThrottlingException.throwReadSizeExceeded(
139 readSizeLimiter.waitInterval(estimateReadSize));
140 }
141 }
142 }
143
144 @Override
145 public void grabQuota(long writeReqs, long writeSize, long readReqs, long readSize) {
146 assert writeSize != 0 || readSize != 0;
147
148 reqsLimiter.consume(writeReqs + readReqs);
149 reqSizeLimiter.consume(writeSize + readSize);
150
151 if (writeSize > 0) {
152 writeReqsLimiter.consume(writeReqs);
153 writeSizeLimiter.consume(writeSize);
154 }
155 if (readSize > 0) {
156 readReqsLimiter.consume(readReqs);
157 readSizeLimiter.consume(readSize);
158 }
159 }
160
161 @Override
162 public void consumeWrite(final long size) {
163 reqSizeLimiter.consume(size);
164 writeSizeLimiter.consume(size);
165 }
166
167 @Override
168 public void consumeRead(final long size) {
169 reqSizeLimiter.consume(size);
170 readSizeLimiter.consume(size);
171 }
172
173 @Override
174 public boolean isBypass() {
175 return false;
176 }
177
178 @Override
179 public long getWriteAvailable() {
180 return writeSizeLimiter.getAvailable();
181 }
182
183 @Override
184 public long getReadAvailable() {
185 return readSizeLimiter.getAvailable();
186 }
187
188 @Override
189 public String toString() {
190 StringBuilder builder = new StringBuilder();
191 builder.append("TimeBasedLimiter(");
192 if (!reqsLimiter.isBypass()) builder.append("reqs=" + reqsLimiter);
193 if (!reqSizeLimiter.isBypass()) builder.append(" resSize=" + reqSizeLimiter);
194 if (!writeReqsLimiter.isBypass()) builder.append(" writeReqs=" + writeReqsLimiter);
195 if (!writeSizeLimiter.isBypass()) builder.append(" writeSize=" + writeSizeLimiter);
196 if (!readReqsLimiter.isBypass()) builder.append(" readReqs=" + readReqsLimiter);
197 if (!readSizeLimiter.isBypass()) builder.append(" readSize=" + readSizeLimiter);
198 builder.append(')');
199 return builder.toString();
200 }
201 }