/*
    * Licensed to the Apache Software Foundation (ASF) under one
    * or more contributor license agreements.  See the NOTICE file
    * distributed with this work for additional information
    * regarding copyright ownership.  The ASF licenses this file
    * to you 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 org.apache.giraph.comm.flow_control;
  
  import com.yammer.metrics.core.Counter;
  import org.apache.giraph.comm.netty.NettyClient;
  import org.apache.giraph.comm.netty.handler.AckSignalFlag;
  import org.apache.giraph.comm.requests.WritableRequest;
  import org.apache.giraph.conf.FloatConfOption;
  import org.apache.giraph.conf.ImmutableClassesGiraphConfiguration;
  import org.apache.giraph.conf.IntConfOption;
  import org.apache.giraph.metrics.GiraphMetrics;
  import org.apache.giraph.metrics.MetricNames;
  import org.apache.giraph.metrics.ResetSuperstepMetricsObserver;
  import org.apache.giraph.metrics.SuperstepMetricsRegistry;
  import org.apache.log4j.Logger;
  
  import java.util.concurrent.atomic.AtomicInteger;
  
  import static org.apache.giraph.conf.GiraphConstants.WAITING_REQUEST_MSECS;
  
  /**
   * Representation of a flow control that limits the aggregate number of open
   * requests to all other workers to a constant user-defined value
   */
  public class StaticFlowControl implements
      FlowControl, ResetSuperstepMetricsObserver {
    /** Maximum number of requests without confirmation we should have */
    public static final IntConfOption MAX_NUMBER_OF_OPEN_REQUESTS =
        new IntConfOption("giraph.maxNumberOfOpenRequests", 10000,
            "Maximum number of requests without confirmation we should have");
    /**
     * After pausing a thread due to too large number of open requests,
     * which fraction of these requests need to be closed before we continue
     */
    public static final FloatConfOption
        FRACTION_OF_REQUESTS_TO_CLOSE_BEFORE_PROCEEDING =
        new FloatConfOption("giraph.fractionOfRequestsToCloseBeforeProceeding",
            0.2f, "Fraction of requests to close before proceeding");
    /** Class logger */
    private static final Logger LOG = Logger.getLogger(StaticFlowControl.class);
  
    /** Maximum number of requests without confirmation we can have */
    private final int maxNumberOfOpenRequests;
    /**
     * Maximum number of requests that can be open after the pause in order to
     * proceed
     */
    private final int numberOfRequestsToProceed;
    /** Netty client used for sending requests */
    private final NettyClient nettyClient;
    /** Waiting interval for checking outstanding requests msecs */
    private final int waitingRequestMsecs;
    /** Dummy object to wait on until enough open requests get completed */
    private final Object requestSpotAvailable = new Object();
    /** Counter for time spent waiting on too many open requests */
    private Counter timeWaitingOnOpenRequests;
    /** Number of threads waiting on too many open requests */
    private final AtomicInteger numWaitingThreads = new AtomicInteger(0);
  
    /**
     * Constructor
     *
     * @param conf configuration
     * @param nettyClient netty client
     */
    public StaticFlowControl(ImmutableClassesGiraphConfiguration conf,
                             NettyClient nettyClient) {
      this.nettyClient = nettyClient;
      maxNumberOfOpenRequests = MAX_NUMBER_OF_OPEN_REQUESTS.get(conf);
      numberOfRequestsToProceed = (int) (maxNumberOfOpenRequests *
          (1 - FRACTION_OF_REQUESTS_TO_CLOSE_BEFORE_PROCEEDING.get(conf)));
      if (LOG.isInfoEnabled()) {
        LOG.info("StaticFlowControl: Limit number of open requests to " +
            maxNumberOfOpenRequests + " and proceed when <= " +
            numberOfRequestsToProceed);
      }
      waitingRequestMsecs = WAITING_REQUEST_MSECS.get(conf);
      GiraphMetrics.get().addSuperstepResetObserver(this);
    }
  
    @Override
    public void newSuperstep(SuperstepMetricsRegistry metrics) {
     timeWaitingOnOpenRequests = metrics.getCounter(
         MetricNames.TIME_SPENT_WAITING_ON_TOO_MANY_OPEN_REQUESTS_MS);
   }
 
   @Override
   public void sendRequest(int destTaskId, WritableRequest request) {
     nettyClient.doSend(destTaskId, request);
     if (nettyClient.getNumberOfOpenRequests() > maxNumberOfOpenRequests) {
       long startTime = System.currentTimeMillis();
       waitSomeRequests();
       timeWaitingOnOpenRequests.inc(System.currentTimeMillis() - startTime);
     }
   }
 
   /**
    * Ensure that at most numberOfRequestsToProceed are not complete.
    * Periodically, check the state of every request.  If we find the connection
    * failed, re-establish it and re-send the request.
    */
   private void waitSomeRequests() {
     numWaitingThreads.getAndIncrement();
     while (nettyClient.getNumberOfOpenRequests() > numberOfRequestsToProceed) {
       // Wait for requests to complete for some time
       synchronized (requestSpotAvailable) {
         if (nettyClient.getNumberOfOpenRequests() <=
             numberOfRequestsToProceed) {
           break;
         }
         try {
           requestSpotAvailable.wait(waitingRequestMsecs);
         } catch (InterruptedException e) {
           throw new IllegalStateException("waitSomeRequests: Got unexpected " +
               "InterruptedException", e);
         }
       }
       nettyClient.logAndSanityCheck();
     }
     numWaitingThreads.getAndDecrement();
   }
 
   @Override
   public void messageAckReceived(int taskId, long requestId, int response) {
     synchronized (requestSpotAvailable) {
       requestSpotAvailable.notifyAll();
     }
   }
 
   @Override
   public AckSignalFlag getAckSignalFlag(int response) {
     return AckSignalFlag.values()[response];
   }
 
   @Override
   public int calculateResponse(AckSignalFlag alreadyDone, int clientId) {
     return alreadyDone.ordinal();
   }
 
   @Override
   public void logInfo() {
     if (LOG.isInfoEnabled()) {
       LOG.info("logInfo: " + numWaitingThreads.get() + " threads waiting " +
           "until number of open requests falls below " +
           numberOfRequestsToProceed);
     }
   }
 
   @Override
   public void waitAllRequests() {
     // This flow control policy does not keep any unsent request. All the open
     // requests are in possession of the network client, so the network client
     // will wait for them to complete. Thus, this flow control policy does not
     // need to do anything regarding flushing the remaining requests.
   }
 
   @Override
   public int getNumberOfUnsentRequests() {
     return 0;
   }
 }