/*
 * Copyright 2016-present Facebook, Inc.
 *
 * 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.facebook.buck.util.concurrent;

import com.google.common.base.Preconditions;
import com.google.common.util.concurrent.Futures;
import com.google.common.util.concurrent.ListenableFuture;
import com.google.common.util.concurrent.SettableFuture;

import java.util.Iterator;
import java.util.LinkedList;
import java.util.List;

/**
 * A semaphore using {@link ListenableFuture}s for acquisition of different resource types
 * rather than blocking.
 */
public class ListeningMultiSemaphore {

  private ResourceAmounts usedValues;
  private final ResourceAmounts maximumValues;
  private final List<ListeningSemaphoreArrayPendingItem> pending = new LinkedList<>();

  public ListeningMultiSemaphore(ResourceAmounts availableResources) {
    this.usedValues = ResourceAmounts.ZERO;
    this.maximumValues = availableResources;
  }

  /**
   * Returns the future which will be completed by the moment when resources will be acquired.
   * Future may be returned already completed. You should subscribe to the future and perform your
   * resource requiring job once the future will be completed and not cancelled.
   * When you finish your job, you must release acquired resources by calling release() method
   * below.
   *
   * @param resources Resource amounts that need to be acquired.
   *
   * @return Future that will be completed once resource will be acquired.
   */
  public synchronized ListenableFuture<Void> acquire(ResourceAmounts resources) {
    if (!checkIfResourcesAvailable(resources)) {
      SettableFuture<Void> pendingFuture = SettableFuture.create();
      pending.add(ListeningSemaphoreArrayPendingItem.of(pendingFuture, resources));
      return pendingFuture;
    }

    increaseUsedResources(resources);
    return Futures.immediateFuture(null);
  }

  /**
   * Releases previously acquired resources.
   *
   * @param resources Resource amounts that need to be released. This argument should
   *                  match one you used during resource acquiring.
   */
  public synchronized void release(ResourceAmounts resources) {
    decreaseUsedResources(resources);
    processPendingFutures();
  }

  public synchronized ResourceAmounts getAvailableResources() {
    return maximumValues.subtract(usedValues);
  }

  public synchronized ResourceAmounts getMaximumValues() {
    return maximumValues;
  }

  public synchronized int getQueueLength() {
    return pending.size();
  }

  private synchronized boolean checkIfResourcesAvailable(ResourceAmounts resources) {
    return !usedValues.append(resources).containsValuesGreaterThan(maximumValues);
  }

  private synchronized void increaseUsedResources(ResourceAmounts resources) {
    ResourceAmounts updatedAmounts = usedValues.append(resources);
    Preconditions.checkArgument(
        !updatedAmounts.containsValuesGreaterThan(maximumValues),
        "Cannot decrease available resources by %s. Current: %s, Maximum: %s",
        resources, usedValues, maximumValues);
    usedValues = updatedAmounts;
  }

  private synchronized void decreaseUsedResources(ResourceAmounts resources) {
    ResourceAmounts updatedAmounts = usedValues.subtract(resources);
    Preconditions.checkArgument(
        !updatedAmounts.containsValuesLessThan(ResourceAmounts.ZERO),
        "Cannot increase available resources by %s. Current: %s, Maximum: %s",
        resources, usedValues, maximumValues);
    usedValues = updatedAmounts;
  }

  private synchronized void processPendingFutures() {
    Iterator<ListeningSemaphoreArrayPendingItem> iterator = pending.iterator();

    while (iterator.hasNext()) {
      ListeningSemaphoreArrayPendingItem item = iterator.next();
      if (checkIfResourcesAvailable(item.getResources())) {
        iterator.remove();
        increaseUsedResources(item.getResources());
        if (!item.getFuture().set(null)) {
          decreaseUsedResources(item.getResources());
        }
      }
    }
  }

}

@@ -688,16 +688,64 @@
 itions;%0A
+import com.google.common.collect.ImmutableList;%0A
 import c
@@ -2657,29 +2657,16 @@
  public 
-synchronized 
 void rel
@@ -2763,16 +2763,710 @@
 Futures(
+getPendingItemsThatCanBeProcessed());%0A  %7D%0A%0A  private synchronized ImmutableList%3CListeningSemaphoreArrayPendingItem%3E%0A  getPendingItemsThatCanBeProcessed() %7B%0A    ImmutableList.Builder%3CListeningSemaphoreArrayPendingItem%3E builder = ImmutableList.builder();%0A%0A    Iterator%3CListeningSemaphoreArrayPendingItem%3E iterator = pending.iterator();%0A    while (!getAvailableResources().equals(ResourceAmounts.ZERO) && iterator.hasNext()) %7B%0A      ListeningSemaphoreArrayPendingItem item = iterator.next();%0A      if (checkIfResourcesAvailable(item.getResources())) %7B%0A        builder.add(item);%0A        increaseUsedResources(item.getResources());%0A        iterator.remove();%0A      %7D%0A    %7D%0A    return builder.build(
 );%0A  %7D%0A%0A
@@ -4752,37 +4752,24 @@
 %0A%0A  private 
-synchronized 
 void process
@@ -4787,24 +4787,21 @@
 res(
-) %7B%0A    Iterator
+ImmutableList
 %3CLis
@@ -4840,76 +4840,82 @@
  ite
-rator = pending.iterator();%0A%0A    while (iterator.hasNext()) %7B%0A      
+ms) %7B%0A    ResourceAmounts failedAmounts = ResourceAmounts.ZERO;%0A%0A    for (
 List
@@ -4954,280 +4954,226 @@
 tem 
-=
+:
  ite
-rator.next();%0A      if (checkIfResourcesAvailable(item.getResources())) %7B%0A        iterator.remove();%0A        increaseUsedResources(item.getResources());%0A        if (!item.getFuture().set(null)) %7B%0A          decreaseUsedResources(item.getResources());%0A        %7D%0A      %7D
+ms) %7B%0A      if (!item.getFuture().set(null)) %7B%0A        failedAmounts = failedAmounts.append(item.getResources());%0A      %7D%0A    %7D%0A%0A    if (!failedAmounts.equals(ResourceAmounts.ZERO)) %7B%0A      release(failedAmounts);
 %0A   

Fix performance in processing pending semaphore items