/**
 * 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
 * <p>
 * http://www.apache.org/licenses/LICENSE-2.0
 * <p>
 * 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.hadoop.yarn.server.resourcemanager.reservation;

import java.util.ArrayList;
import java.util.Collections;
import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.SortedMap;
import java.util.TreeMap;
import java.util.TreeSet;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantReadWriteLock;

import org.apache.hadoop.yarn.api.records.ReservationId;
import org.apache.hadoop.yarn.api.records.Resource;
import org.apache.hadoop.yarn.conf.YarnConfiguration;
import org.apache.hadoop.yarn.server.resourcemanager.RMContext;
import org.apache.hadoop.yarn.server.resourcemanager.recovery.RMStateStore;
import org.apache.hadoop.yarn.server.resourcemanager.reservation.RLESparseResourceAllocation.RLEOperator;
import org.apache.hadoop.yarn.server.resourcemanager.reservation.exceptions.PlanningException;
import org.apache.hadoop.yarn.server.resourcemanager.reservation.planning.Planner;
import org.apache.hadoop.yarn.server.resourcemanager.reservation.planning.ReservationAgent;
import org.apache.hadoop.yarn.server.resourcemanager.scheduler.QueueMetrics;
import org.apache.hadoop.yarn.util.Clock;
import org.apache.hadoop.yarn.util.UTCClock;
import org.apache.hadoop.yarn.util.resource.ResourceCalculator;
import org.apache.hadoop.yarn.util.resource.Resources;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * This class represents an in memory representation of the state of our
 * reservation system, and provides accelerated access to both individual
 * reservations and aggregate utilization of resources over time.
 */
public class InMemoryPlan implements Plan {

  private static final Logger LOG = LoggerFactory.getLogger(InMemoryPlan.class);

  private static final Resource ZERO_RESOURCE = Resource.newInstance(0, 0);
  private final RMStateStore rmStateStore;

  private TreeMap<ReservationInterval, Set<InMemoryReservationAllocation>> currentReservations =
      new TreeMap<ReservationInterval, Set<InMemoryReservationAllocation>>();

  private RLESparseResourceAllocation rleSparseVector;

  private PeriodicRLESparseResourceAllocation periodicRle;

  private Map<String, RLESparseResourceAllocation> userResourceAlloc =
      new HashMap<String, RLESparseResourceAllocation>();

  private Map<String, RLESparseResourceAllocation> userPeriodicResourceAlloc =
      new HashMap<String, RLESparseResourceAllocation>();

  private Map<String, RLESparseResourceAllocation> userActiveReservationCount =
      new HashMap<String, RLESparseResourceAllocation>();

  private Map<ReservationId, InMemoryReservationAllocation> reservationTable =
      new HashMap<ReservationId, InMemoryReservationAllocation>();

  private final ReentrantReadWriteLock readWriteLock =
      new ReentrantReadWriteLock();
  private final Lock readLock = readWriteLock.readLock();
  private final Lock writeLock = readWriteLock.writeLock();
  private final SharingPolicy policy;
  private final ReservationAgent agent;
  private final long step;
  private final ResourceCalculator resCalc;
  private final Resource minAlloc, maxAlloc;
  private final String queueName;
  private final QueueMetrics queueMetrics;
  private final Planner replanner;
  private final boolean getMoveOnExpiry;
  private final Clock clock;
  private final long maxPeriodicity;

  private Resource totalCapacity;

  public InMemoryPlan(QueueMetrics queueMetrics, SharingPolicy policy,
      ReservationAgent agent, Resource totalCapacity, long step,
      ResourceCalculator resCalc, Resource minAlloc, Resource maxAlloc,
      String queueName, Planner replanner, boolean getMoveOnExpiry,
      RMContext rmContext) {
    this(queueMetrics, policy, agent, totalCapacity, step, resCalc, minAlloc,
        maxAlloc, queueName, replanner, getMoveOnExpiry,
        YarnConfiguration.DEFAULT_RM_RESERVATION_SYSTEM_MAX_PERIODICITY,
        rmContext);
  }

  public InMemoryPlan(QueueMetrics queueMetrics, SharingPolicy policy,
      ReservationAgent agent, Resource totalCapacity, long step,
      ResourceCalculator resCalc, Resource minAlloc, Resource maxAlloc,
      String queueName, Planner replanner, boolean getMoveOnExpiry,
      long maxPeriodicity, RMContext rmContext) {
    this(queueMetrics, policy, agent, totalCapacity, step, resCalc, minAlloc,
        maxAlloc, queueName, replanner, getMoveOnExpiry, maxPeriodicity,
        rmContext, new UTCClock());
  }

  @SuppressWarnings("checkstyle:parameternumber")
  public InMemoryPlan(QueueMetrics queueMetrics, SharingPolicy policy,
      ReservationAgent agent, Resource totalCapacity, long step,
      ResourceCalculator resCalc, Resource minAlloc, Resource maxAlloc,
      String queueName, Planner replanner, boolean getMoveOnExpiry,
      long maxPeriodicty, RMContext rmContext, Clock clock) {
    this.queueMetrics = queueMetrics;
    this.policy = policy;
    this.agent = agent;
    this.step = step;
    this.totalCapacity = totalCapacity;
    this.resCalc = resCalc;
    this.minAlloc = minAlloc;
    this.maxAlloc = maxAlloc;
    this.rleSparseVector = new RLESparseResourceAllocation(resCalc);
    this.maxPeriodicity = maxPeriodicty;
    this.periodicRle =
        new PeriodicRLESparseResourceAllocation(resCalc, this.maxPeriodicity);
    this.queueName = queueName;
    this.replanner = replanner;
    this.getMoveOnExpiry = getMoveOnExpiry;
    this.clock = clock;
    this.rmStateStore = rmContext.getStateStore();
  }

  @Override
  public QueueMetrics getQueueMetrics() {
    return queueMetrics;
  }

  private RLESparseResourceAllocation getUserRLEResourceAllocation(String user,
      long period) {
    RLESparseResourceAllocation resAlloc = null;
    if (period > 0) {
      if (userPeriodicResourceAlloc.containsKey(user)) {
        resAlloc = userPeriodicResourceAlloc.get(user);
      } else {
        resAlloc = new PeriodicRLESparseResourceAllocation(resCalc,
            periodicRle.getTimePeriod());
        userPeriodicResourceAlloc.put(user, resAlloc);
      }
    } else {
      if (userResourceAlloc.containsKey(user)) {
        resAlloc = userResourceAlloc.get(user);
      } else {
        resAlloc = new RLESparseResourceAllocation(resCalc);
        userResourceAlloc.put(user, resAlloc);
      }
    }
    return resAlloc;
  }

  private void gcUserRLEResourceAllocation(String user, long period) {
    if (period > 0) {
      if (userPeriodicResourceAlloc.get(user).isEmpty()) {
        userPeriodicResourceAlloc.remove(user);
      }
    } else {
      if (userResourceAlloc.get(user).isEmpty()) {
        userResourceAlloc.remove(user);
      }
    }
  }

  private void incrementAllocation(ReservationAllocation reservation) {
    assert (readWriteLock.isWriteLockedByCurrentThread());
    Map<ReservationInterval, Resource> allocationRequests =
        reservation.getAllocationRequests();
    // check if we have encountered the user earlier and if not add an entry
    String user = reservation.getUser();
    long period = reservation.getPeriodicity();
    RLESparseResourceAllocation resAlloc =
        getUserRLEResourceAllocation(user, period);

    RLESparseResourceAllocation resCount = userActiveReservationCount.get(user);
    if (resCount == null) {
      resCount = new RLESparseResourceAllocation(resCalc);
      userActiveReservationCount.put(user, resCount);
    }

    long earliestActive = Long.MAX_VALUE;
    long latestActive = Long.MIN_VALUE;

    for (Map.Entry<ReservationInterval, Resource> r : allocationRequests
        .entrySet()) {

      if (period > 0L) {
        for (int i = 0; i < periodicRle.getTimePeriod() / period; i++) {

          long rStart = r.getKey().getStartTime() + i * period;
          long rEnd = r.getKey().getEndTime() + i * period;

          // handle wrap-around
          if (rEnd > periodicRle.getTimePeriod()) {
            long diff = rEnd - periodicRle.getTimePeriod();
            rEnd = periodicRle.getTimePeriod();
            ReservationInterval newInterval = new ReservationInterval(0, diff);
            periodicRle.addInterval(newInterval, r.getValue());
            resAlloc.addInterval(newInterval, r.getValue());
          }

          ReservationInterval newInterval =
              new ReservationInterval(rStart, rEnd);
          periodicRle.addInterval(newInterval, r.getValue());
          resAlloc.addInterval(newInterval, r.getValue());
        }

      } else {
        rleSparseVector.addInterval(r.getKey(), r.getValue());
        resAlloc.addInterval(r.getKey(), r.getValue());
        if (Resources.greaterThan(resCalc, totalCapacity, r.getValue(),
            ZERO_RESOURCE)) {
          earliestActive = Math.min(earliestActive, r.getKey().getStartTime());
          latestActive = Math.max(latestActive, r.getKey().getEndTime());
        }
      }
    }
    // periodic reservations are active from start time and good till cancelled
    if (period > 0L) {
      earliestActive = reservation.getStartTime();
      latestActive = Long.MAX_VALUE;
    }
    resCount.addInterval(new ReservationInterval(earliestActive, latestActive),
        Resource.newInstance(1, 1));
  }

  private void decrementAllocation(ReservationAllocation reservation) {
    assert (readWriteLock.isWriteLockedByCurrentThread());
    Map<ReservationInterval, Resource> allocationRequests =
        reservation.getAllocationRequests();
    String user = reservation.getUser();
    long period = reservation.getPeriodicity();
    RLESparseResourceAllocation resAlloc =
        getUserRLEResourceAllocation(user, period);

    long earliestActive = Long.MAX_VALUE;
    long latestActive = Long.MIN_VALUE;
    for (Map.Entry<ReservationInterval, Resource> r : allocationRequests
        .entrySet()) {
      if (period > 0L) {
        for (int i = 0; i < periodicRle.getTimePeriod() / period; i++) {

          long rStart = r.getKey().getStartTime() + i * period;
          long rEnd = r.getKey().getEndTime() + i * period;

          // handle wrap-around
          if (rEnd > periodicRle.getTimePeriod()) {
            long diff = rEnd - periodicRle.getTimePeriod();
            rEnd = periodicRle.getTimePeriod();
            ReservationInterval newInterval = new ReservationInterval(0, diff);
            periodicRle.removeInterval(newInterval, r.getValue());
            resAlloc.removeInterval(newInterval, r.getValue());
          }

          ReservationInterval newInterval =
              new ReservationInterval(rStart, rEnd);
          periodicRle.removeInterval(newInterval, r.getValue());
          resAlloc.removeInterval(newInterval, r.getValue());
        }
      } else {
        rleSparseVector.removeInterval(r.getKey(), r.getValue());
        resAlloc.removeInterval(r.getKey(), r.getValue());
        if (Resources.greaterThan(resCalc, totalCapacity, r.getValue(),
            ZERO_RESOURCE)) {
          earliestActive = Math.min(earliestActive, r.getKey().getStartTime());
          latestActive = Math.max(latestActive, r.getKey().getEndTime());
        }
      }
    }
    gcUserRLEResourceAllocation(user, period);

    RLESparseResourceAllocation resCount = userActiveReservationCount.get(user);
    // periodic reservations are active from start time and good till cancelled
    if (period > 0L) {
      earliestActive = reservation.getStartTime();
      latestActive = Long.MAX_VALUE;
    }
    resCount.removeInterval(
        new ReservationInterval(earliestActive, latestActive),
        Resource.newInstance(1, 1));
    if (resCount.isEmpty()) {
      userActiveReservationCount.remove(user);
    }
  }

  public Set<ReservationAllocation> getAllReservations() {
    readLock.lock();
    try {
      if (currentReservations != null) {
        Set<ReservationAllocation> flattenedReservations =
            new TreeSet<ReservationAllocation>();
        for (Set<InMemoryReservationAllocation> res : currentReservations
            .values()) {
          flattenedReservations.addAll(res);
        }
        return flattenedReservations;
      } else {
        return null;
      }
    } finally {
      readLock.unlock();
    }
  }

  @Override
  public boolean addReservation(ReservationAllocation reservation,
      boolean isRecovering) throws PlanningException {
    // Verify the allocation is memory based otherwise it is not supported
    InMemoryReservationAllocation inMemReservation =
        (InMemoryReservationAllocation) reservation;
    if (inMemReservation.getUser() == null) {
      String errMsg = "The specified Reservation with ID "
          + inMemReservation.getReservationId() + " is not mapped to any user";
      LOG.error(errMsg);
      throw new IllegalArgumentException(errMsg);
    }
    writeLock.lock();
    try {
      if (reservationTable.containsKey(inMemReservation.getReservationId())) {
        String errMsg = "The specified Reservation with ID "
            + inMemReservation.getReservationId() + " already exists";
        LOG.error(errMsg);
        throw new IllegalArgumentException(errMsg);
      }
      // Validate if we can accept this reservation, throws exception if
      // validation fails
      if (!isRecovering) {
        policy.validate(this, inMemReservation);
        // we record here the time in which the allocation has been accepted
        reservation.setAcceptanceTimestamp(clock.getTime());
        if (rmStateStore != null) {
          rmStateStore.storeNewReservation(
              ReservationSystemUtil.buildStateProto(inMemReservation),
              getQueueName(), inMemReservation.getReservationId().toString());
        }
      }
      ReservationInterval searchInterval = new ReservationInterval(
          inMemReservation.getStartTime(), inMemReservation.getEndTime());
      Set<InMemoryReservationAllocation> reservations =
          currentReservations.get(searchInterval);
      if (reservations == null) {
        reservations = new HashSet<InMemoryReservationAllocation>();
      }
      if (!reservations.add(inMemReservation)) {
        LOG.error("Unable to add reservation: {} to plan.",
            inMemReservation.getReservationId());
        return false;
      }
      currentReservations.put(searchInterval, reservations);
      reservationTable.put(inMemReservation.getReservationId(),
          inMemReservation);
      incrementAllocation(inMemReservation);
      LOG.info("Successfully added reservation: {} to plan.",
          inMemReservation.getReservationId());
      return true;
    } finally {
      writeLock.unlock();
    }
  }

  @Override
  public boolean updateReservation(ReservationAllocation reservation)
      throws PlanningException {
    writeLock.lock();
    boolean result = false;
    try {
      ReservationId resId = reservation.getReservationId();
      ReservationAllocation currReservation = getReservationById(resId);
      if (currReservation == null) {
        String errMsg = "The specified Reservation with ID " + resId
            + " does not exist in the plan";
        LOG.error(errMsg);
        throw new IllegalArgumentException(errMsg);
      }
      // validate if we can accept this reservation, throws exception if
      // validation fails
      policy.validate(this, reservation);
      if (!removeReservation(currReservation)) {
        LOG.error("Unable to replace reservation: {} from plan.",
            reservation.getReservationId());
        return result;
      }
      try {
        result = addReservation(reservation, false);
      } catch (PlanningException e) {
        LOG.error("Unable to update reservation: {} from plan due to {}.",
            reservation.getReservationId(), e.getMessage());
      }
      if (result) {
        LOG.info("Successfully updated reservation: {} in plan.",
            reservation.getReservationId());
        return result;
      } else {
        // rollback delete
        addReservation(currReservation, false);
        LOG.info("Rollbacked update reservation: {} from plan.",
            reservation.getReservationId());
        return result;
      }
    } finally {
      writeLock.unlock();
    }
  }

  private boolean removeReservation(ReservationAllocation reservation) {
    assert (readWriteLock.isWriteLockedByCurrentThread());
    ReservationInterval searchInterval = new ReservationInterval(
        reservation.getStartTime(), reservation.getEndTime());
    Set<InMemoryReservationAllocation> reservations =
        currentReservations.get(searchInterval);
    if (reservations != null) {
      if (rmStateStore != null) {
        rmStateStore.removeReservation(getQueueName(),
            reservation.getReservationId().toString());
      }
      if (!reservations.remove(reservation)) {
        LOG.error("Unable to remove reservation: {} from plan.",
            reservation.getReservationId());
        return false;
      }
      if (reservations.isEmpty()) {
        currentReservations.remove(searchInterval);
      }
    } else {
      String errMsg = "The specified Reservation with ID "
          + reservation.getReservationId() + " does not exist in the plan";
      LOG.error(errMsg);
      throw new IllegalArgumentException(errMsg);
    }
    reservationTable.remove(reservation.getReservationId());
    decrementAllocation(reservation);
    LOG.info("Sucessfully deleted reservation: {} in plan.",
        reservation.getReservationId());
    return true;
  }

  @Override
  public boolean deleteReservation(ReservationId reservationID) {
    writeLock.lock();
    try {
      ReservationAllocation reservation = getReservationById(reservationID);
      if (reservation == null) {
        String errMsg = "The specified Reservation with ID " + reservationID
            + " does not exist in the plan";
        LOG.error(errMsg);
        throw new IllegalArgumentException(errMsg);
      }
      return removeReservation(reservation);
    } finally {
      writeLock.unlock();
    }
  }

  @Override
  public void archiveCompletedReservations(long tick) {
    // Since we are looking for old reservations, read lock is optimal
    LOG.debug("Running archival at time: {}", tick);
    List<InMemoryReservationAllocation> expiredReservations =
        new ArrayList<InMemoryReservationAllocation>();
    readLock.lock();
    // archive reservations and delete the ones which are beyond
    // the reservation policy "window"
    try {
      long archivalTime = tick - policy.getValidWindow();
      ReservationInterval searchInterval =
          new ReservationInterval(archivalTime, archivalTime);
      SortedMap<ReservationInterval, Set<InMemoryReservationAllocation>> reservations =
          currentReservations.headMap(searchInterval, true);
      if (!reservations.isEmpty()) {
        for (Set<InMemoryReservationAllocation> reservationEntries : reservations
            .values()) {
          for (InMemoryReservationAllocation reservation : reservationEntries) {
            if (reservation.getEndTime() <= archivalTime) {
              expiredReservations.add(reservation);
            }
          }
        }
      }
    } finally {
      readLock.unlock();
    }
    if (expiredReservations.isEmpty()) {
      return;
    }
    // Need write lock only if there are any reservations to be deleted
    writeLock.lock();
    try {
      for (InMemoryReservationAllocation expiredReservation : expiredReservations) {
        removeReservation(expiredReservation);
      }
    } finally {
      writeLock.unlock();
    }
  }

  @Override
  public Set<ReservationAllocation> getReservationsAtTime(long tick) {
    return getReservations(null, new ReservationInterval(tick, tick), "");
  }

  @Override
  public long getStep() {
    return step;
  }

  @Override
  public SharingPolicy getSharingPolicy() {
    return policy;
  }

  @Override
  public ReservationAgent getReservationAgent() {
    return agent;
  }

  @Override
  public RLESparseResourceAllocation getReservationCountForUserOverTime(
      String user, long start, long end) {
    readLock.lock();
    try {
      RLESparseResourceAllocation userResAlloc =
          userActiveReservationCount.get(user);

      if (userResAlloc != null) {
        return userResAlloc.getRangeOverlapping(start, end);
      } else {
        return new RLESparseResourceAllocation(resCalc);
      }
    } finally {
      readLock.unlock();
    }
  }

  @Override
  public RLESparseResourceAllocation getConsumptionForUserOverTime(String user,
      long start, long end) {
    readLock.lock();
    try {
      // merge periodic and non-periodic allocations
      RLESparseResourceAllocation userResAlloc = userResourceAlloc.get(user);
      RLESparseResourceAllocation userPeriodicResAlloc =
          userPeriodicResourceAlloc.get(user);

      if (userResAlloc != null && userPeriodicResAlloc != null) {
        return RLESparseResourceAllocation.merge(resCalc, totalCapacity,
            userResAlloc, userPeriodicResAlloc, RLEOperator.add, start, end);
      }
      if (userResAlloc != null) {
        return userResAlloc.getRangeOverlapping(start, end);
      }
      if (userPeriodicResAlloc != null) {
        return userPeriodicResAlloc.getRangeOverlapping(start, end);
      }
    } catch (PlanningException e) {
      LOG.warn("Exception while trying to merge periodic"
          + " and non-periodic user allocations: {}", e.getMessage(), e);
    } finally {
      readLock.unlock();
    }
    return new RLESparseResourceAllocation(resCalc);
  }

  @Override
  public Resource getTotalCommittedResources(long t) {
    readLock.lock();
    try {
      return Resources.add(rleSparseVector.getCapacityAtTime(t),
          periodicRle.getCapacityAtTime(t));
    } finally {
      readLock.unlock();
    }
  }

  @Override
  public Set<ReservationAllocation> getReservations(ReservationId reservationID,
      ReservationInterval interval) {
    return getReservations(reservationID, interval, null);
  }

  @Override
  public Set<ReservationAllocation> getReservations(ReservationId reservationID,
      ReservationInterval interval, String user) {
    if (reservationID != null) {
      ReservationAllocation allocation = getReservationById(reservationID);
      if (allocation == null) {
        return Collections.emptySet();
      }
      return Collections.singleton(allocation);
    }

    long startTime = interval == null ? 0 : interval.getStartTime();
    long endTime = interval == null ? Long.MAX_VALUE : interval.getEndTime();

    ReservationInterval searchInterval =
        new ReservationInterval(endTime, Long.MAX_VALUE);
    readLock.lock();
    try {
      SortedMap<ReservationInterval, Set<InMemoryReservationAllocation>> res =
          currentReservations.headMap(searchInterval, true);
      if (!res.isEmpty()) {
        Set<ReservationAllocation> flattenedReservations = new HashSet<>();
        for (Set<InMemoryReservationAllocation> resEntries : res.values()) {
          for (InMemoryReservationAllocation reservation : resEntries) {
            // validate user
            if (user != null && !user.isEmpty()
                && !reservation.getUser().equals(user)) {
              continue;
            }
            // handle periodic reservations
            long period = reservation.getPeriodicity();
            if (period > 0) {
              // The shift is used to remove the wrap around for the
              // reservation interval. The wrap around will still
              // exist for the search interval.
              long shift = reservation.getStartTime() % period;
              // This is the duration of the reservation since
              // duration < period.
              long periodicReservationEnd =
                  (reservation.getEndTime() -shift) % period;
              long periodicSearchStart = (startTime - shift) % period;
              long periodicSearchEnd = (endTime - shift) % period;
              long searchDuration = endTime - startTime;

              // 1. If the searchDuration is greater than the period, then
              // the reservation is within the interval. This will allow
              // us to ignore cases where search end > search start >
              // reservation end.
              // 2/3. If the search end is less than the reservation end, or if
              // the search start is less than the reservation end, then the
              // reservation will be in the reservation since
              // periodic reservation start is always zero. Note that neither
              // of those values will ever be negative.
              // 4. If the search end is less than the search start, then
              // there is a wrap around, and both values are implicitly
              // greater than the reservation end because of condition 2/3,
              // so the reservation is within the search interval.
              if (searchDuration > period
                  || periodicSearchEnd < periodicReservationEnd
                  || periodicSearchStart < periodicReservationEnd
                  || periodicSearchStart > periodicSearchEnd) {
                flattenedReservations.add(reservation);
              }
            } else {
              // check for non-periodic reservations
              if (reservation.getEndTime() > startTime) {
                flattenedReservations.add(reservation);
              }
            }
          }
        }
        return Collections.unmodifiableSet(flattenedReservations);
      } else {
        return Collections.emptySet();
      }
    } finally {
      readLock.unlock();
    }
  }

  @Override
  public ReservationAllocation getReservationById(ReservationId reservationID) {
    if (reservationID == null) {
      return null;
    }
    readLock.lock();
    try {
      return reservationTable.get(reservationID);
    } finally {
      readLock.unlock();
    }
  }

  @Override
  public Resource getTotalCapacity() {
    readLock.lock();
    try {
      return Resources.clone(totalCapacity);
    } finally {
      readLock.unlock();
    }
  }

  @Override
  public RLESparseResourceAllocation getAvailableResourceOverTime(String user,
      ReservationId oldId, long start, long end, long period)
      throws PlanningException {
    readLock.lock();
    try {

      // for non-periodic return simple available resources
      if (period == 0) {

        // create RLE of totCapacity
        TreeMap<Long, Resource> totAvailable = new TreeMap<Long, Resource>();
        totAvailable.put(start, Resources.clone(totalCapacity));
        RLESparseResourceAllocation totRLEAvail =
            new RLESparseResourceAllocation(totAvailable, resCalc);

        // subtract used from available
        RLESparseResourceAllocation netAvailable;

        netAvailable = RLESparseResourceAllocation.merge(resCalc,
            Resources.clone(totalCapacity), totRLEAvail, rleSparseVector,
            RLEOperator.subtractTestNonNegative, start, end);

        // remove periodic component
        netAvailable = RLESparseResourceAllocation.merge(resCalc,
            Resources.clone(totalCapacity), netAvailable, periodicRle,
            RLEOperator.subtractTestNonNegative, start, end);

        // add back in old reservation used resources if any
        ReservationAllocation old = reservationTable.get(oldId);
        if (old != null) {

          RLESparseResourceAllocation addBackPrevious =
              old.getResourcesOverTime(start, end);
          netAvailable = RLESparseResourceAllocation.merge(resCalc,
              Resources.clone(totalCapacity), netAvailable, addBackPrevious,
              RLEOperator.add, start, end);
        }
        // lower it if this is needed by the sharing policy
        netAvailable = getSharingPolicy().availableResources(netAvailable, this,
            user, oldId, start, end);
        return netAvailable;
      } else {

        if (periodicRle.getTimePeriod() % period != 0) {
          throw new PlanningException("The reservation periodicity (" + period
              + ") must be" + " an exact divider of the system maxPeriod ("
              + periodicRle.getTimePeriod() + ")");
        }

        if (period < (end - start)) {
          throw new PlanningException(
              "Invalid input: (end - start) = (" + end + " - " + start + ") = "
                  + (end - start) + " > period = " + period);
        }

        // find the minimum resources available among all the instances that fit
        // in the LCM
        long numInstInLCM = periodicRle.getTimePeriod() / period;

        RLESparseResourceAllocation minOverLCM =
            getAvailableResourceOverTime(user, oldId, start, end, 0);
        for (int i = 1; i < numInstInLCM; i++) {

          long rStart = start + i * period;
          long rEnd = end + i * period;

          // recursive invocation of non-periodic range (to pick raw-info)
          RLESparseResourceAllocation snapShot =
              getAvailableResourceOverTime(user, oldId, rStart, rEnd, 0);

          // time-align on start
          snapShot.shift(-(i * period));

          // pick the minimum amount of resources in each time interval
          minOverLCM =
              RLESparseResourceAllocation.merge(resCalc, getTotalCapacity(),
                  minOverLCM, snapShot, RLEOperator.min, start, end);

        }

        return minOverLCM;

      }
    } finally {
      readLock.unlock();
    }
  }

  @Override
  public Resource getMinimumAllocation() {
    return Resources.clone(minAlloc);
  }

  @Override
  public void setTotalCapacity(Resource cap) {
    writeLock.lock();
    try {
      totalCapacity = Resources.clone(cap);
    } finally {
      writeLock.unlock();
    }
  }

  public long getEarliestStartTime() {
    readLock.lock();
    try {
      return rleSparseVector.getEarliestStartTime();
    } finally {
      readLock.unlock();
    }
  }

  @Override
  public long getLastEndTime() {
    readLock.lock();
    try {
      return rleSparseVector.getLatestNonNullTime();
    } finally {
      readLock.unlock();
    }
  }

  @Override
  public ResourceCalculator getResourceCalculator() {
    return resCalc;
  }

  @Override
  public String getQueueName() {
    return queueName;
  }

  @Override
  public Resource getMaximumAllocation() {
    return Resources.clone(maxAlloc);
  }

  @Override
  public long getMaximumPeriodicity() {
    return this.maxPeriodicity;
  }

  public String toCumulativeString() {
    readLock.lock();
    try {
      return rleSparseVector.toString() + "\n" + periodicRle.toString();
    } finally {
      readLock.unlock();
    }
  }

  @Override
  public Planner getReplanner() {
    return replanner;
  }

  @Override
  public boolean getMoveOnExpiry() {
    return getMoveOnExpiry;
  }

  @Override
  public String toString() {
    readLock.lock();
    try {
      StringBuilder planStr = new StringBuilder("In-memory Plan: ");
      planStr.append("Parent Queue: ").append(queueName)
          .append(" Total Capacity: ").append(totalCapacity).append(" Step: ")
          .append(step);
      for (ReservationAllocation reservation : getAllReservations()) {
        planStr.append(reservation);
      }
      return planStr.toString();
    } finally {
      readLock.unlock();
    }
  }

  @Override
  public Set<ReservationAllocation> getReservationByUserAtTime(String user,
      long t) {
    readLock.lock();
    try {
      Set<ReservationAllocation> resSet = new HashSet<ReservationAllocation>();
      for (ReservationAllocation ra : getReservationsAtTime(t)) {
        String resUser = ra.getUser();
        if (resUser != null && resUser.equals(user)) {
          resSet.add(ra);
        }
      }
      return resSet;
    } finally {
      readLock.unlock();
    }
  }

  @Override
  public RLESparseResourceAllocation getCumulativeLoadOverTime(long start,
      long end) throws PlanningException {
    readLock.lock();
    try {

      RLESparseResourceAllocation ret =
          rleSparseVector.getRangeOverlapping(start, end);
      ret = RLESparseResourceAllocation.merge(resCalc, totalCapacity, ret,
          periodicRle.getRangeOverlapping(start, end), RLEOperator.add, start,
          end);

      return ret;
    } finally {
      readLock.unlock();
    }
  }
}