001/**
002 * Licensed to the Apache Software Foundation (ASF) under one
003 * or more contributor license agreements.  See the NOTICE file
004 * distributed with this work for additional information
005 * regarding copyright ownership.  The ASF licenses this file
006 * to you under the Apache License, Version 2.0 (the
007 * "License"); you may not use this file except in compliance
008 * with the License.  You may obtain a copy of the License at
009 *
010 *     http://www.apache.org/licenses/LICENSE-2.0
011 *
012 * Unless required by applicable law or agreed to in writing, software
013 * distributed under the License is distributed on an "AS IS" BASIS,
014 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
015 * See the License for the specific language governing permissions and
016 * limitations under the License.
017 */
018
019package org.apache.hadoop.hbase.quotas;
020
021
022import org.apache.hadoop.conf.Configuration;
023import org.apache.hadoop.hbase.HBaseConfiguration;
024import org.apache.yetus.audience.InterfaceAudience;
025import org.apache.yetus.audience.InterfaceStability;
026import org.apache.hadoop.hbase.shaded.protobuf.ProtobufUtil;
027import org.apache.hadoop.hbase.shaded.protobuf.generated.QuotaProtos.Throttle;
028import org.apache.hadoop.hbase.shaded.protobuf.generated.QuotaProtos.TimedQuota;
029
030/**
031 * Simple time based limiter that checks the quota Throttle
032 */
033@InterfaceAudience.Private
034@InterfaceStability.Evolving
035public class TimeBasedLimiter implements QuotaLimiter {
036  private static final Configuration conf = HBaseConfiguration.create();
037  private RateLimiter reqsLimiter = null;
038  private RateLimiter reqSizeLimiter = null;
039  private RateLimiter writeReqsLimiter = null;
040  private RateLimiter writeSizeLimiter = null;
041  private RateLimiter readReqsLimiter = null;
042  private RateLimiter readSizeLimiter = null;
043  private RateLimiter reqCapacityUnitLimiter = null;
044  private RateLimiter writeCapacityUnitLimiter = null;
045  private RateLimiter readCapacityUnitLimiter = null;
046
047  private TimeBasedLimiter() {
048    if (FixedIntervalRateLimiter.class.getName().equals(
049      conf.getClass(RateLimiter.QUOTA_RATE_LIMITER_CONF_KEY, AverageIntervalRateLimiter.class)
050          .getName())) {
051      reqsLimiter = new FixedIntervalRateLimiter();
052      reqSizeLimiter = new FixedIntervalRateLimiter();
053      writeReqsLimiter = new FixedIntervalRateLimiter();
054      writeSizeLimiter = new FixedIntervalRateLimiter();
055      readReqsLimiter = new FixedIntervalRateLimiter();
056      readSizeLimiter = new FixedIntervalRateLimiter();
057      reqCapacityUnitLimiter = new FixedIntervalRateLimiter();
058      writeCapacityUnitLimiter = new FixedIntervalRateLimiter();
059      readCapacityUnitLimiter = new FixedIntervalRateLimiter();
060    } else {
061      reqsLimiter = new AverageIntervalRateLimiter();
062      reqSizeLimiter = new AverageIntervalRateLimiter();
063      writeReqsLimiter = new AverageIntervalRateLimiter();
064      writeSizeLimiter = new AverageIntervalRateLimiter();
065      readReqsLimiter = new AverageIntervalRateLimiter();
066      readSizeLimiter = new AverageIntervalRateLimiter();
067      reqCapacityUnitLimiter = new AverageIntervalRateLimiter();
068      writeCapacityUnitLimiter = new AverageIntervalRateLimiter();
069      readCapacityUnitLimiter = new AverageIntervalRateLimiter();
070    }
071  }
072
073  static QuotaLimiter fromThrottle(final Throttle throttle) {
074    TimeBasedLimiter limiter = new TimeBasedLimiter();
075    boolean isBypass = true;
076    if (throttle.hasReqNum()) {
077      setFromTimedQuota(limiter.reqsLimiter, throttle.getReqNum());
078      isBypass = false;
079    }
080
081    if (throttle.hasReqSize()) {
082      setFromTimedQuota(limiter.reqSizeLimiter, throttle.getReqSize());
083      isBypass = false;
084    }
085
086    if (throttle.hasWriteNum()) {
087      setFromTimedQuota(limiter.writeReqsLimiter, throttle.getWriteNum());
088      isBypass = false;
089    }
090
091    if (throttle.hasWriteSize()) {
092      setFromTimedQuota(limiter.writeSizeLimiter, throttle.getWriteSize());
093      isBypass = false;
094    }
095
096    if (throttle.hasReadNum()) {
097      setFromTimedQuota(limiter.readReqsLimiter, throttle.getReadNum());
098      isBypass = false;
099    }
100
101    if (throttle.hasReadSize()) {
102      setFromTimedQuota(limiter.readSizeLimiter, throttle.getReadSize());
103      isBypass = false;
104    }
105
106    if (throttle.hasReqCapacityUnit()) {
107      setFromTimedQuota(limiter.reqCapacityUnitLimiter, throttle.getReqCapacityUnit());
108      isBypass = false;
109    }
110
111    if (throttle.hasWriteCapacityUnit()) {
112      setFromTimedQuota(limiter.writeCapacityUnitLimiter, throttle.getWriteCapacityUnit());
113      isBypass = false;
114    }
115
116    if (throttle.hasReadCapacityUnit()) {
117      setFromTimedQuota(limiter.readCapacityUnitLimiter, throttle.getReadCapacityUnit());
118      isBypass = false;
119    }
120    return isBypass ? NoopQuotaLimiter.get() : limiter;
121  }
122
123  public void update(final TimeBasedLimiter other) {
124    reqsLimiter.update(other.reqsLimiter);
125    reqSizeLimiter.update(other.reqSizeLimiter);
126    writeReqsLimiter.update(other.writeReqsLimiter);
127    writeSizeLimiter.update(other.writeSizeLimiter);
128    readReqsLimiter.update(other.readReqsLimiter);
129    readSizeLimiter.update(other.readSizeLimiter);
130    reqCapacityUnitLimiter.update(other.reqCapacityUnitLimiter);
131    writeCapacityUnitLimiter.update(other.writeCapacityUnitLimiter);
132    readCapacityUnitLimiter.update(other.readCapacityUnitLimiter);
133  }
134
135  private static void setFromTimedQuota(final RateLimiter limiter, final TimedQuota timedQuota) {
136    limiter.set(timedQuota.getSoftLimit(), ProtobufUtil.toTimeUnit(timedQuota.getTimeUnit()));
137  }
138
139  @Override
140  public void checkQuota(long writeReqs, long estimateWriteSize, long readReqs,
141      long estimateReadSize, long estimateWriteCapacityUnit, long estimateReadCapacityUnit)
142      throws RpcThrottlingException {
143    if (!reqsLimiter.canExecute(writeReqs + readReqs)) {
144      RpcThrottlingException.throwNumRequestsExceeded(reqsLimiter.waitInterval());
145    }
146    if (!reqSizeLimiter.canExecute(estimateWriteSize + estimateReadSize)) {
147      RpcThrottlingException.throwRequestSizeExceeded(
148          reqSizeLimiter.waitInterval(estimateWriteSize + estimateReadSize));
149    }
150    if (!reqCapacityUnitLimiter.canExecute(estimateWriteCapacityUnit + estimateReadCapacityUnit)) {
151      RpcThrottlingException.throwRequestCapacityUnitExceeded(
152        reqCapacityUnitLimiter.waitInterval(estimateWriteCapacityUnit + estimateReadCapacityUnit));
153    }
154
155    if (estimateWriteSize > 0) {
156      if (!writeReqsLimiter.canExecute(writeReqs)) {
157        RpcThrottlingException.throwNumWriteRequestsExceeded(writeReqsLimiter.waitInterval());
158      }
159      if (!writeSizeLimiter.canExecute(estimateWriteSize)) {
160        RpcThrottlingException.throwWriteSizeExceeded(
161            writeSizeLimiter.waitInterval(estimateWriteSize));
162      }
163      if (!writeCapacityUnitLimiter.canExecute(estimateWriteCapacityUnit)) {
164        RpcThrottlingException.throwWriteCapacityUnitExceeded(
165          writeCapacityUnitLimiter.waitInterval(estimateWriteCapacityUnit));
166      }
167    }
168
169    if (estimateReadSize > 0) {
170      if (!readReqsLimiter.canExecute(readReqs)) {
171        RpcThrottlingException.throwNumReadRequestsExceeded(readReqsLimiter.waitInterval());
172      }
173      if (!readSizeLimiter.canExecute(estimateReadSize)) {
174        RpcThrottlingException.throwReadSizeExceeded(
175            readSizeLimiter.waitInterval(estimateReadSize));
176      }
177      if (!readCapacityUnitLimiter.canExecute(estimateReadCapacityUnit)) {
178        RpcThrottlingException.throwReadCapacityUnitExceeded(
179          readCapacityUnitLimiter.waitInterval(estimateReadCapacityUnit));
180      }
181    }
182  }
183
184  @Override
185  public void grabQuota(long writeReqs, long writeSize, long readReqs, long readSize,
186      long writeCapacityUnit, long readCapacityUnit) {
187    assert writeSize != 0 || readSize != 0;
188
189    reqsLimiter.consume(writeReqs + readReqs);
190    reqSizeLimiter.consume(writeSize + readSize);
191
192    if (writeSize > 0) {
193      writeReqsLimiter.consume(writeReqs);
194      writeSizeLimiter.consume(writeSize);
195    }
196    if (readSize > 0) {
197      readReqsLimiter.consume(readReqs);
198      readSizeLimiter.consume(readSize);
199    }
200    if (writeCapacityUnit > 0) {
201      reqCapacityUnitLimiter.consume(writeCapacityUnit);
202      writeCapacityUnitLimiter.consume(writeCapacityUnit);
203    }
204    if (readCapacityUnit > 0) {
205      reqCapacityUnitLimiter.consume(readCapacityUnit);
206      readCapacityUnitLimiter.consume(readCapacityUnit);
207    }
208  }
209
210  @Override
211  public void consumeWrite(final long size, long capacityUnit) {
212    reqSizeLimiter.consume(size);
213    writeSizeLimiter.consume(size);
214    reqCapacityUnitLimiter.consume(capacityUnit);
215    writeCapacityUnitLimiter.consume(capacityUnit);
216  }
217
218  @Override
219  public void consumeRead(final long size, long capacityUnit) {
220    reqSizeLimiter.consume(size);
221    readSizeLimiter.consume(size);
222    reqCapacityUnitLimiter.consume(capacityUnit);
223    readCapacityUnitLimiter.consume(capacityUnit);
224  }
225
226  @Override
227  public boolean isBypass() {
228    return false;
229  }
230
231  @Override
232  public long getWriteAvailable() {
233    return writeSizeLimiter.getAvailable();
234  }
235
236  @Override
237  public long getReadAvailable() {
238    return readSizeLimiter.getAvailable();
239  }
240
241  @Override
242  public String toString() {
243    StringBuilder builder = new StringBuilder();
244    builder.append("TimeBasedLimiter(");
245    if (!reqsLimiter.isBypass()) {
246      builder.append("reqs=" + reqsLimiter);
247    }
248    if (!reqSizeLimiter.isBypass()) {
249      builder.append(" resSize=" + reqSizeLimiter);
250    }
251    if (!writeReqsLimiter.isBypass()) {
252      builder.append(" writeReqs=" + writeReqsLimiter);
253    }
254    if (!writeSizeLimiter.isBypass()) {
255      builder.append(" writeSize=" + writeSizeLimiter);
256    }
257    if (!readReqsLimiter.isBypass()) {
258      builder.append(" readReqs=" + readReqsLimiter);
259    }
260    if (!readSizeLimiter.isBypass()) {
261      builder.append(" readSize=" + readSizeLimiter);
262    }
263    if (!reqCapacityUnitLimiter.isBypass()) {
264      builder.append(" reqCapacityUnit=" + reqCapacityUnitLimiter);
265    }
266    if (!writeCapacityUnitLimiter.isBypass()) {
267      builder.append(" writeCapacityUnit=" + writeCapacityUnitLimiter);
268    }
269    if (!readCapacityUnitLimiter.isBypass()) {
270      builder.append(" readCapacityUnit=" + readCapacityUnitLimiter);
271    }
272    builder.append(')');
273    return builder.toString();
274  }
275}