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