/*
    * 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.utils;
  
  import org.apache.hadoop.util.Progressable;
  import org.apache.log4j.Logger;
  
  import com.google.common.collect.Sets;
  
  import java.util.HashSet;
  import java.util.Set;
  
  /**
   * This barrier is used when we don't know how many events are we waiting on
   * from the start. Instead we have a set of task ids, and each of those will,
   * at some point of time, give the information about how many events from it
   * should we expect. Barrier will be waiting for all the tasks to notify it
   * about that number of events, and than it will also wait for all the events
   * to happen.
   *
   * requirePermits() corresponds to task notifying us how many events from it
   * to expect, and releasePermits() notifies us about events happening.
   *
   * This class is currently used during preparation of aggregators.
   *
   * User must follow this protocol for concurrent access:
   *
   * (1) an object instance is constructed
   * (2) arbitrarily many times
   *     (2a) concurrent calls to requirePermits(), releasePermits() and
   *          waitForRequiredPermits() are issued
   *     (2b) waitForRequiredPermits() returns
   *
   * Note that the next cycle of calls to requirePermits() or releasePermits()
   * cannot start until the previous call to waitForRequiredPermits()
   * has returned.
   *
   * Methods of this class are thread-safe.
   */
  public class TaskIdsPermitsBarrier {
    /** Class logger */
    private static final Logger LOG =
        Logger.getLogger(TaskIdsPermitsBarrier.class);
    /** Msecs to refresh the progress meter */
    private static final int MSEC_PERIOD = 10000;
    /** Maximum number of task ids to list in the log */
    private static final int MAX_TASK_IDS_TO_LOG = 10;
    /** Progressable for reporting progress */
    private final Progressable progressable;
    /** Number of permits we are currently waiting for */
    private long waitingOnPermits = 0;
    /** Set of task ids which required permits already */
    private final Set<Integer> arrivedTaskIds = new HashSet<Integer>();
    /** Logger */
    private final TimedLogger logger;
  
    /**
     * Constructor
     *
     * @param progressable Progressable for reporting progress
     */
    public TaskIdsPermitsBarrier(Progressable progressable) {
      this.progressable = progressable;
      logger = new TimedLogger(MSEC_PERIOD, LOG);
    }
  
    /**
     * Wait until permits have been required desired number of times,
     * and all required permits are available
     *
     * @param expectedTaskIds List of task ids which we are waiting permits from
     */
    public synchronized void waitForRequiredPermits(
        Set<Integer> expectedTaskIds) {
      while (arrivedTaskIds.size() < expectedTaskIds.size() ||
          waitingOnPermits > 0) {
        try {
          wait(MSEC_PERIOD);
        } catch (InterruptedException e) {
          throw new IllegalStateException("waitForRequiredPermits: " +
              "InterruptedException occurred");
        }
        progressable.progress();
       if (LOG.isInfoEnabled()) {
         if (arrivedTaskIds.size() < expectedTaskIds.size()) {
           String logSuffix = "";
           if (expectedTaskIds.size() - arrivedTaskIds.size() <=
               MAX_TASK_IDS_TO_LOG) {
             Sets.SetView<Integer> difference =
                 Sets.difference(expectedTaskIds, arrivedTaskIds);
             logSuffix = ", task ids: " + difference;
           }
           logger.info("waitForRequiredPermits: " +
               "Waiting for " +
               (expectedTaskIds.size() - arrivedTaskIds.size()) +
               " more tasks to send their aggregator data" +
               logSuffix);
         } else {
           logger.info("waitForRequiredPermits: " +
               "Waiting for " + waitingOnPermits + " more aggregator requests");
         }
       }
     }
 
     // Reset for the next time to use
     arrivedTaskIds.clear();
     waitingOnPermits = 0;
   }
 
   /**
    * Require more permits. This will increase the number of times permits
    * were required. Doesn't wait for permits to become available.
    *
    * @param permits Number of permits to require
    * @param taskId Task id which required permits
    */
   public synchronized void requirePermits(long permits, int taskId) {
     arrivedTaskIds.add(taskId);
     waitingOnPermits += permits;
     notifyAll();
   }
 
   /**
    * Release one permit.
    */
   public synchronized void releaseOnePermit() {
     releasePermits(1);
   }
 
   /**
    * Release some permits.
    *
    * @param permits Number of permits to release
    */
   public synchronized void releasePermits(long permits) {
     waitingOnPermits -= permits;
     notifyAll();
   }
 }