content: /*
   * Copyright 1999-2018 Alibaba Group Holding Ltd.
   *
   * Licensed under the Apache License, Version 2.0 (the "License");
   * you may not use this file except in compliance with the License.
   * You may obtain a copy of the License at
   *
   *      http://www.apache.org/licenses/LICENSE-2.0
   *
   * Unless required by applicable law or agreed to in writing, software
   * distributed under the License is distributed on an "AS IS" BASIS,
   * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
   * See the License for the specific language governing permissions and
   * limitations under the License.
   */
  package com.alibaba.csp.sentinel.cluster.flow.statistic.concurrent.expire;
  
  import com.alibaba.csp.sentinel.cluster.flow.rule.ClusterFlowRuleManager;
  import com.alibaba.csp.sentinel.cluster.flow.statistic.concurrent.CurrentConcurrencyManager;
  import com.alibaba.csp.sentinel.cluster.flow.statistic.concurrent.TokenCacheNode;
  import com.alibaba.csp.sentinel.cluster.server.connection.ConnectionManager;
  import com.alibaba.csp.sentinel.concurrent.NamedThreadFactory;
  import com.alibaba.csp.sentinel.log.RecordLog;
  import com.alibaba.csp.sentinel.util.AssertUtil;
  import com.googlecode.concurrentlinkedhashmap.ConcurrentLinkedHashMap;
  
  import java.util.ArrayList;
  import java.util.List;
  import java.util.concurrent.Executors;
  import java.util.concurrent.ScheduledExecutorService;
  import java.util.concurrent.TimeUnit;
  import java.util.concurrent.atomic.AtomicInteger;
  
  /**
   * We need to consider the situation that the token client goes offline
   * or the resource call times out. It can be detected by sourceTimeout
   * and clientTimeout. The resource calls timeout detection is triggered
   * on the token client. If the resource is called over time, the token
   * client will request the token server to release token or refresh the
   * token. The client offline detection is triggered on the token server.
   * If the offline detection time is exceeded, token server will trigger
   * the detection token client’s status. If the token client is offline,
   * token server will delete the corresponding tokenId. If it is not offline,
   * token server will continue to save it.
   *
   * @author yunfeiyanggzq
   **/
  public class RegularExpireStrategy implements ExpireStrategy {
      /**
       * The max number of token deleted each time,
       * the number of expired key-value pairs deleted each time does not exceed this number
       */
      private long executeCount = 1000;
      /**
       * Length of time for task execution
       */
      private long executeDuration = 800;
      /**
       * Frequency of task execution
       */
      private long executeRate = 1000;
      /**
       * the local cache of tokenId
       */
      private ConcurrentLinkedHashMap<Long, TokenCacheNode> localCache;
  
      @SuppressWarnings("PMD.ThreadPoolCreationRule")
      private static ScheduledExecutorService executor = Executors.newScheduledThreadPool(1,
              new NamedThreadFactory("regular clear expired token thread", true));
  
  
      public RegularExpireStrategy(ConcurrentLinkedHashMap<Long, TokenCacheNode> localCache) {
          AssertUtil.isTrue(localCache != null, " local cache can't be null");
          this.localCache = localCache;
      }
  
      @Override
      public void startClearTaskRegularly() {
          executor.scheduleAtFixedRate(new ClearExpiredTokenTask(), 0, executeRate, TimeUnit.MILLISECONDS);
      }
  
      private class ClearExpiredTokenTask implements Runnable {
          @Override
          public void run() {
              try {
                  clearToken();
              } catch (Throwable e) {
                  e.printStackTrace();
                  RecordLog.warn("[RegularExpireStrategy] undefined throwable during clear token: ", e);
              }
          }
      }
  
      private void clearToken() {
          long start = System.currentTimeMillis();
          List<Long> keyList = new ArrayList<>(localCache.keySet());
          for (int i = 0; i < executeCount && i < keyList.size(); i++) {
              // time out execution exit
              if (System.currentTimeMillis() - start > executeDuration) {
                  RecordLog.info("[RegularExpireStrategy] End the process of expired token detection because of execute time is more than executeDuration: {}", executeDuration);
                  break;
              }
              Long key = keyList.get(i);
              TokenCacheNode node = localCache.get(key);
              if (node == null) {
                  continue;
              }
  
              // remove the token whose client is offline and saved for more than clientTimeout
              if (!ConnectionManager.isClientOnline(node.getClientAddress()) && node.getClientTimeout() - System.currentTimeMillis() < 0) {
                  removeToken(key, node);
                  RecordLog.info("[RegularExpireStrategy] Delete the expired token<{}> because of client offline for ruleId<{}>", node.getTokenId(), node.getFlowId());
                  continue;
              }
  
              // If we find that token's save time is more than 2 times of the client's call resource timeout time,
              // the token will be determined to timeout.
              long resourceTimeout = ClusterFlowRuleManager.getFlowRuleById(node.getFlowId()).getClusterConfig().getResourceTimeout();
              if (System.currentTimeMillis() - node.getResourceTimeout() > resourceTimeout) {
                  removeToken(key, node);
                  RecordLog.info("[RegularExpireStrategy] Delete the expired token<{}> because of resource timeout for ruleId<{}>", node.getTokenId(), node.getFlowId());
              }
          }
      }
  
      private void removeToken(long tokenId, TokenCacheNode node) {
          if (localCache.remove(tokenId) == null) {
              RecordLog.info("[RegularExpireStrategy] Token<{}> is already released for ruleId<{}>", tokenId, node.getFlowId());
              return;
          }
          AtomicInteger nowCalls = CurrentConcurrencyManager.get(node.getFlowId());
          if (nowCalls == null) {
              return;
          }
          nowCalls.getAndAdd(node.getAcquireCount() * -1);
      }
  }
  