| /* |
| * 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.uima.ducc.rm.scheduler; |
| |
| /** |
| * This class represents a job inside the scheduler. |
| */ |
| import java.util.ArrayList; |
| import java.util.Collections; |
| import java.util.Comparator; |
| import java.util.HashMap; |
| import java.util.Iterator; |
| import java.util.List; |
| import java.util.Map; |
| import java.util.Properties; |
| |
| import org.apache.uima.ducc.common.utils.DuccLogger; |
| import org.apache.uima.ducc.common.utils.SystemPropertyResolver; |
| import org.apache.uima.ducc.common.utils.id.DuccId; |
| import org.apache.uima.ducc.transport.event.common.IDuccTypes.DuccType; |
| |
| |
| public class RmJob |
| implements SchedConstants, |
| IRmJob |
| { |
| DuccLogger logger = DuccLogger.getLogger(RmJob.class, COMPONENT_NAME); |
| static final int DEFAULT_NTHREADS = 4; |
| |
| protected DuccId id; // sched-assigned id (maybe delegate to job manager eventually) |
| protected DuccType ducc_type; // for messages so we can tell what kind of job |
| protected String state = "New"; // UIMA-4577 info only, for the db |
| protected boolean arbitrary_process = false; // Is this an AP? |
| protected String name; // user's name for job |
| protected String resource_class_name; // Name of the res class, from incoming job parms |
| protected ResourceClass resource_class; // The actual class, assigned as job is received in scheduler. |
| protected int user_priority; // user "priority", really apportionment |
| |
| // @deprecated |
| // protected int n_machines; // RESERVE: minimum machines to allocate |
| |
| protected int max_shares; // FAIR_SHARE: maximum N shares to allocate |
| protected boolean is_reservation = false; |
| |
| protected int threads; // threads per process |
| |
| protected int memory; // estimated memory usage |
| protected int nquestions; // number of work-items in total |
| protected int nquestions_remaining; // number of uncompleted work items |
| protected double time_per_item = Double.NaN; // from OR - mean time per work item |
| |
| protected int share_order = 0; // How many shares per process this job requires (calculated on submission) |
| |
| protected int share_cap = Integer.MAX_VALUE; // initially; scheduler policy will reset as the job ages |
| protected int job_cap = 0; // current, cached cap on the job, reset at the start of every cycle |
| protected int pure_fair_share = 0; // pure uncapped un-bonused share for this job |
| |
| protected long submit_time; // when job is submitted ( sched or job-manager sets this ) |
| |
| protected String username; |
| protected User user; // user id, enforced by submit and job manager. we just believe it in sched. |
| |
| protected long serviceId = 0; // services only, the SM-assigned service id. UIMA-4712 ref UIMA-4209 |
| // |
| // We keep track of three things related to machines: |
| // 1. All the machines the job is running on. |
| // 2. The machines the job will be exanded to run on but which aren't yet dispatched |
| // 3. The machines to be removed from the job, but which the job is still running on |
| // |
| protected HashMap<Share, Share> assignedShares; // job is dispatched to these |
| protected HashMap<Share, Share> pendingShares ; // job is scheduled for these but not yet confirmed |
| protected HashMap<Share, Share> pendingRemoves; // job is scheduled to remove these but not confirmed |
| protected HashMap<Share, Share> recoveredShares; // recovery after bounce, need to reconnect these |
| |
| protected int total_assigned = 0; // non-preemptable only, total shares ever assigned |
| |
| // track shares by machine, and machines, to help when we have to give stuff away |
| Map<Machine, Map<Share, Share>> sharesByMachine = new HashMap<Machine, Map<Share, Share>>(); |
| Map<Machine, Machine> machineList = new HashMap<Machine, Machine>(); |
| |
| // protected int shares_given; // during scheduling, how many N-shares we get |
| int[] given_by_order; // during scheduling, how many N-shares we get |
| int[] wanted_by_order; // during scheduling, how many N-shares we we want - volatile, changes during countJobsByOrder |
| protected boolean init_wait; // If True, we're waiting for orchestrator to tell us that init is successful. |
| // Until then, we give out only a very small share. |
| protected boolean completed = false; // RmJob can linger a bit after completion for the |
| // defrag code - must mark it complete |
| |
| // For predicting end of job based on current rate of completion |
| protected int orchestrator_epoch; |
| protected int rm_rate; |
| protected int ducc_epoch; |
| |
| protected Properties jobprops; // input that job is constructed from. currently is condensed from Blade logs (simulation only) |
| |
| protected String reason = null; // if refused, deferred, etc, this is why, for the message |
| boolean refused = false; |
| boolean deferred = false; |
| |
| private static Comparator<IEntity> apportionmentSorter = new ApportionmentSorterCl(); |
| |
| protected RmJob() |
| { |
| } |
| |
| public RmJob(DuccId id) |
| { |
| this.id = id; |
| |
| orchestrator_epoch = SystemPropertyResolver.getIntProperty("ducc.orchestrator.state.publish.rate", 10000); |
| rm_rate = SystemPropertyResolver.getIntProperty("ducc.rm.state.publish.ratio", 4); |
| ducc_epoch = orchestrator_epoch * rm_rate; |
| |
| } |
| |
| // public RmJob(DuccId id, Properties properties) |
| // { |
| // this.jobprops = properties; |
| // this.id = id; |
| // } |
| |
| /** |
| * read the props file and set inital values for internal fields |
| */ |
| public void init() |
| { |
| assignedShares = new HashMap<Share, Share>(); |
| pendingShares = new HashMap<Share, Share>(); |
| pendingRemoves = new HashMap<Share, Share>(); |
| recoveredShares = new HashMap<Share, Share>(); |
| |
| if ( max_shares == 0 ) max_shares = Integer.MAX_VALUE; |
| } |
| |
| public DuccId getId() |
| { |
| return id; |
| } |
| |
| public void setId(DuccId id) |
| { |
| this.id = id; |
| } |
| |
| public long getFriendlyId() |
| { |
| return id.getFriendly(); |
| } |
| |
| public String getName() |
| { |
| return id.toString(); |
| } |
| |
| public void setServiceId(long id) |
| { |
| this.serviceId = id; |
| } |
| |
| public long getServiceId() |
| { |
| return serviceId; |
| } |
| |
| public void setJobName(String name) |
| { |
| this.name = name; |
| } |
| |
| public void setState(String state) { this.state = state; } |
| public String getState() { return this.state; } |
| |
| public void setReservation() |
| { |
| this.is_reservation = true; |
| } |
| |
| public boolean isReservation() |
| { |
| return is_reservation; |
| } |
| |
| /** |
| * For preemptable, must remember the job completed so we don't accidentally reexpand it. Can |
| * happen in defrag and maybe various races with OR state. |
| */ |
| public void markComplete() |
| { |
| completed = true; |
| } |
| |
| public boolean isCompleted() |
| { |
| return completed; |
| } |
| |
| /** |
| * The matrix: |
| * IW = am I in initializaton wait |
| * RU = am I runnable |
| * Resched = do we need to reschedule |
| * Reset = set IW to this |
| * x = don't care |
| * |
| * IW RU Resched Reset |
| * T T T F |
| * T F F T |
| * F T F F |
| * F F x x |
| * |
| * So resched = IR & RU |
| * IW = !RU |
| * |
| * We return resched so caller knows to tickle the scheduler |
| */ |
| public boolean setInitWait(boolean is_running) |
| { |
| boolean resched = init_wait & is_running; |
| init_wait = !is_running; |
| return resched; |
| } |
| |
| /** |
| * Save ref to the class we are in, and init class-based structures. |
| */ |
| public void setResourceClass(ResourceClass cl) |
| { |
| this.resource_class = cl; |
| } |
| |
| /** |
| * Number of questions submitted, all must be answered. This is used by job manager |
| * to know when they've all been dealt with. |
| */ |
| public int nQuestions() |
| { |
| return nquestions; |
| } |
| |
| public void setNQuestions(int allq, int remainingq, double time_per_item) |
| { |
| this.nquestions = allq; |
| this.nquestions_remaining = remainingq; |
| this.time_per_item = time_per_item; |
| } |
| |
| /** |
| * Number of questions still to be answered. Used by scheduler to determing current |
| * machine requirement. |
| */ |
| public int nQuestionsRemaining() |
| { |
| return nquestions_remaining; |
| } |
| |
| |
| public Map<Machine, Map<Share, Share>> getSharesByMachine() |
| { |
| return sharesByMachine; |
| } |
| |
| public Map<Machine, Machine> getMachines() |
| { |
| return machineList; |
| } |
| |
| public int queryDemand() |
| { |
| if ( getSchedulingPolicy() == Policy.FAIR_SHARE ) return getJobCap(); |
| return max_shares; |
| } |
| |
| /** |
| * There are a fair number of piddling little methods to manage shares. This high granularity is |
| * needed in order to manage bits and pieces of the bookkeeping from different threads and queues |
| * without blocking. |
| * |
| * TODO: maybe we can consolidate some of this after it's all worked out. |
| */ |
| |
| /** |
| * Before each scheduling epoch, clear the counting from the last time. |
| */ |
| public void clearShares() |
| { |
| // this.shares_given = 0; |
| given_by_order = null; |
| } |
| |
| /** |
| public void addQShares(int s) |
| { |
| this.shares_given += ( s / share_order ) ; // convert to N-shares == processes |
| } |
| */ |
| |
| public void setPureFairShare(int pfs) |
| { |
| this.pure_fair_share = pfs; |
| } |
| |
| public int getPureFairShare() |
| { |
| return pure_fair_share; |
| } |
| |
| public int[] getGivenByOrder() |
| { |
| return given_by_order; |
| } |
| |
| public void setGivenByOrder(int[] gbo) |
| { |
| this.given_by_order = gbo; |
| } |
| |
| public int getShareWeight() |
| { |
| return 1; // all jobs are the same currently |
| } |
| |
| public void initWantedByOrder(ResourceClass unused) |
| { |
| wanted_by_order = unused.makeArray(); |
| wanted_by_order[share_order] = getJobCap(); |
| wanted_by_order[0] = wanted_by_order[share_order]; |
| } |
| |
| public int[] getWantedByOrder() |
| { |
| return wanted_by_order; |
| } |
| |
| // UIMA-4275 simplifies this |
| public int calculateCap() |
| { |
| return Integer.MAX_VALUE; // no cap for jobs |
| } |
| |
| |
| // UIMA-4275 |
| public int countOccupancy() |
| { |
| |
| return (assignedShares.size() + pendingShares.size()) * share_order; |
| |
| // if ( (given_by_order == null) || (given_by_order[share_order] == 0) ) { |
| // // must use current allocation because we haven't been counted yet |
| // return countNShares() * share_order; |
| // } else { |
| // // new allocation may not match current, so we use that one |
| // return given_by_order[share_order] * share_order; |
| //} |
| } |
| |
| public int countNSharesGiven() |
| { |
| if ( given_by_order == null) { return 0; } |
| return given_by_order[share_order]; |
| } |
| |
| public int countQSharesGiven() |
| { |
| return countNSharesGiven() * share_order; |
| } |
| |
| /** |
| * Number of N-shares I'm losing. |
| */ |
| public int countNSharesLost() |
| { |
| return countNShares() - countNSharesGiven(); |
| } |
| |
| /** |
| * Can I use more 1 more share of this size? |
| * UIMA-4065 |
| * |
| * @param order The size of the available share. Must be an exact match because the |
| * offerer has already done all reasonable splitting and will have a better |
| * use for it if I can't take it. |
| * |
| * The decision is based on the wbo/gbo arrays that the offer has been building up |
| * just before asking this question. |
| * |
| * @return True if I can use the share, false otherwise. |
| */ |
| public boolean canUseBonus(int order) // UIMA-4065 |
| { |
| if ( order != share_order) return false; |
| |
| if ( getGivenByOrder()[0] >= getResourceClass().calculateCap() ) return false; // don't exceed cap UIMA-4275 |
| |
| return (getWantedByOrder()[order] > 0); // yep, still want |
| } |
| |
| /** |
| * Officially allocated shares assigned to this job which are known to be in use. |
| */ |
| public HashMap<Share, Share> getAssignedShares() |
| { |
| return assignedShares; |
| } |
| |
| /** |
| * Shares recovered from the OR during job recovery. |
| */ |
| public HashMap<Share, Share> getRecoveredShares() |
| { |
| return recoveredShares; |
| } |
| |
| /** |
| * Newly allocated shares that have not been dispatched. They're unavailable for scheduling but |
| * job manager doesn't know about them yet. When we tell job manager we'll "promote" them to |
| * the assignedShares list. |
| */ |
| public HashMap<Share, Share> getPendingShares() |
| { |
| return pendingShares; |
| } |
| |
| /** |
| * We're dispatching, move machines to active list, and clear pending list. |
| * Tell caller which machines are affected so it can dispatch them. |
| */ |
| public HashMap<Share, Share> promoteShares() |
| { |
| HashMap<Share, Share> answer = new HashMap<Share, Share>(); |
| for ( Share s : pendingShares.values() ) { |
| assignedShares.put(s, s); |
| Machine m = s.getMachine(); |
| machineList.put(m, m); |
| Map<Share, Share> machine_shares = sharesByMachine.get(m); |
| if ( machine_shares == null ) { |
| machine_shares = new HashMap<Share, Share>(); |
| sharesByMachine.put(m, machine_shares); |
| } |
| machine_shares.put(s, s); |
| |
| answer.put(s, s); |
| } |
| pendingShares.clear(); |
| return answer; |
| } |
| |
| /** |
| * This share is being donated to someone more needy than I - see defrag code in NodepoolScheduler |
| */ |
| public void cancelPending(Share s) |
| { |
| pendingShares.remove(s); |
| } |
| |
| // public void assignReservation(Machine m) |
| // { |
| // reservations.add(m); |
| // } |
| // |
| // public ArrayList<Machine> getReservations(Machine m) |
| // { |
| // return reservations; |
| // } |
| |
| // public int countReservations() |
| // { |
| // return reservations.size(); |
| // } |
| // |
| // public boolean reservationComplete() |
| // { |
| // return reservations.size() == machines; |
| // } |
| |
| /** |
| * Scheduler found us a new toy, make it pending until it's given to job manager. |
| */ |
| public void assignShare(Share s) |
| { |
| pendingShares.put(s, s); |
| total_assigned++; |
| } |
| |
| /** |
| * Non-preemptable, need to know total every assigned, in case one of them dies, must be careful |
| * not to reassign it. |
| */ |
| public int countTotalAssignments() |
| { |
| return total_assigned; |
| } |
| |
| /** |
| * Job recovery: OR reports this share as one it already knew about. |
| */ |
| public void recoverShare(Share s) |
| { |
| if ( (! assignedShares.containsKey(s)) && (!pendingShares.containsKey(s)) ) { |
| recoveredShares.put(s, s); |
| } |
| } |
| |
| public boolean isPendingShare(Share s ) |
| { |
| return pendingShares.containsKey(s); |
| } |
| |
| /** |
| * Do I have un-dispatched shares? |
| */ |
| public boolean isExpanded() |
| { |
| return pendingShares.size() > 0; |
| } |
| |
| public boolean isShrunken() |
| { |
| return pendingRemoves.size() > 0; |
| } |
| |
| public boolean isStable() |
| { |
| return ( |
| ( assignedShares.size() > 0 ) && |
| ( pendingShares.size() == 0 ) && |
| ( pendingRemoves.size() == 0 ) |
| ); |
| } |
| |
| public boolean isDormant() |
| { |
| return ( |
| ( assignedShares.size() == 0 ) && |
| ( pendingShares.size() == 0 ) && |
| ( pendingRemoves.size() == 0 ) |
| ); |
| } |
| |
| public void removeAllShares() |
| { |
| String methodName = "removeAllShares"; |
| if ( logger.isTrace() ) { |
| for ( Map<Share, Share> m : sharesByMachine.values() ) { |
| for ( Share s : m.values() ) { |
| logger.trace(methodName, getId(), "Clear share", s); |
| } |
| } |
| } |
| assignedShares.clear(); |
| pendingShares.clear(); |
| pendingRemoves.clear(); |
| machineList.clear(); |
| sharesByMachine.clear(); |
| } |
| |
| /** |
| * I've shrunk or this share has nothing left to do. Remove this specific share. |
| */ |
| public void removeShare(Share share) |
| { |
| String methodName = "removeShare"; |
| |
| if ( assignedShares.containsKey(share) ) { |
| int prev = assignedShares.size(); |
| assignedShares.remove(share); |
| pendingRemoves.remove(share); |
| |
| Machine m = share.getMachine(); |
| Map<Share, Share> machineShares = sharesByMachine.get(m); |
| machineShares.remove(share); |
| if ( machineShares.size() == 0 ) { |
| sharesByMachine.remove(m); |
| machineList.remove(m); |
| } |
| |
| logger.debug(methodName, getId(), "Job removes ", share.toString(), " reduces from ", prev, " to ", assignedShares.size() + "."); |
| } else { |
| logger.warn(methodName, getId(), "****** Job cannot find share " + share.toString() + " to remove. ******"); |
| } |
| } |
| |
| /** |
| * Remove a process from this job |
| */ |
| public void shrinkByOne(Share share) |
| { |
| String methodName = "shrinkByOne"; |
| if ( assignedShares.containsKey(share) ) { |
| logger.debug(methodName, getId(), "Job schedules " + share.toString() + " for removal."); |
| pendingRemoves.put(share, share); |
| share.evict(); |
| } else { |
| logger.warn(methodName, getId(), "****** Job cannot find share " + share.toString() + " to schedule for removal.******"); |
| } |
| |
| } |
| |
| /** |
| * Shrink by 'shares' from machines of largest order starting from 'order' and decreasing. |
| * Investment is not used, this is to shrink-from-largest-machine to MINIMIZE FRAGMENTATIOPN. |
| * |
| * This implementation is simplest, we just vacate. This is called in the sequence of highest |
| * order wanted so we're leaving the largest holes in the largest machines first. |
| * |
| * @param shares - number of N-shares that are wanted |
| * @param order - N - try to free up space for shares of this size. |
| * @param force - When evicting for non-preemptables, we may need to free the requested |
| * shares even if it puts us over our "fair" count. If this happens |
| * we'll end up "sliding" onto other machines (eventually). |
| * @param nodepool - only interested in shares from this nodepool. |
| * |
| * So when this is called, somebody needs (shares*order) shares, given out in chunks of |
| * (order). |
| * |
| * @returns number of Q-shares actually given up |
| */ |
| public int shrinkByOrderByMachine(int shares, int order, boolean force, NodePool nodepool) |
| { |
| String methodName = "shrinkByOrderByMachine"; |
| |
| if ( shares <= 0 ) { |
| throw new SchedulingException(getId(), "Trying to shrink by " + shares + " shares."); |
| } |
| |
| // These are the machines where I have stuff running. |
| ArrayList<Machine> machinesSorted = new ArrayList<Machine>(); |
| for ( Machine m : machineList.values() ) { |
| if ( (m.getNodepool() == nodepool) && ( m.getShareOrder() >= order) ) { |
| machinesSorted.add(m); |
| } |
| } |
| Collections.sort(machinesSorted, new MachineByOrderSorter()); |
| |
| int given = 0; |
| int shares_to_lose = 0; |
| |
| // |
| // How much to lose? If we're not forcing, then only shares that are evicted because of |
| // the 'howMuch' counts. If forcing then everything until we meet the goal or we run |
| // out of stuff to give. |
| // |
| if ( force ) { |
| shares_to_lose = countNShares(); |
| } else { |
| shares_to_lose = Math.max(0, countNShares() - countNSharesGiven()); |
| } |
| if ( shares_to_lose == 0 ) { |
| return 0; |
| } |
| |
| for ( Machine m : machinesSorted ) { |
| |
| logger.debug(methodName, getId(), "Inspecting machine", m.getId()); |
| ArrayList<Share> slist = new ArrayList<Share>(); |
| |
| for ( Share s : sharesByMachine.get(m).values() ) { // get the still-eligible shares |
| if ( ! s.isEvicted() ) { |
| slist.add(s); |
| } |
| } |
| if ( slist.size() == 0 ) { |
| continue; |
| } |
| |
| int to_give = m.countFreedUpShares(); |
| logger.debug(methodName, getId(), "A given:", given, "to_give:", to_give, "order", order, "shares", shares, "shares_to_lose", shares_to_lose); |
| |
| Iterator<Share> iter = slist.iterator(); |
| while ( iter.hasNext() && ( (given + (to_give/order)) < shares ) && (shares_to_lose > 0) ) { |
| Share s = iter.next(); |
| logger.info(methodName, getId(), "Removing share", s.toString()); |
| pendingRemoves.put(s, s); |
| s.evict(); |
| to_give += share_order; |
| shares_to_lose--; |
| } |
| |
| |
| given += (to_give / order); |
| if ( given >= shares ) { |
| break; |
| } |
| } |
| return given; |
| } |
| |
| /** |
| * Shrink by 'shares' from machines of largest order starting from 'order' and decreasing. |
| * Investment is not used, this is a shrink-from-largest-machine to MINIMIZE LOST WORK at the |
| * possibl eexpense of fragmentation. |
| * |
| * @param shares - number of N-shares that are wanted |
| * @param order - N - try to free up space for shares of this size. |
| * @param force - When evicting for non-preemptables, we may need to free the requested |
| * shares even if it puts us over our "fair" count. If this happens |
| * we'll end up "sliding" onto other machines (eventually). |
| * @param nodepool - only interested in shares from this nodepool. |
| * |
| * So when this is called, somebody needs (shares*order) shares, given out in chunks of |
| * (order). |
| * |
| * @returns number of Q-shares actually given up |
| */ |
| public int shrinkByInvestment(int shares, int order, boolean force, NodePool nodepool) |
| { |
| String methodName = "shrinkByInvestment"; |
| |
| if ( shares <= 0 ) { |
| throw new SchedulingException(getId(), "Trying to shrink by " + shares + " shares."); |
| } |
| |
| logger.debug(methodName, getId(), "Enter: shares", shares, "order", order, "force", force, "nodepool", nodepool.getId(), |
| "nAssignedShares", assignedShares.size(), "nPendingShares", pendingShares.size()); |
| |
| ArrayList<Share> sharesSorted = new ArrayList<Share>(); |
| |
| // must pick up only shares in the given nodepool |
| for ( Share s : assignedShares.values() ) { |
| if ( s.getNodepoolId().equals(nodepool.getId()) && ( !s.isEvicted() ) ) { |
| sharesSorted.add(s); |
| } else { |
| if ( logger.isTrace () ) { |
| logger.trace(methodName, getId(), "Skipping", s.getId(), "s.nodepool", s.getNodepoolId(), "incoming.nodepool", nodepool.getId(), "evicted", s.isEvicted()); |
| } |
| } |
| } |
| |
| if ( sharesSorted.size() == 0 ) { |
| return 0; |
| } |
| |
| if ( logger.isTrace() ) { |
| logger.trace(methodName, getId(), "Shares Before Sort - id, isInitialized, investment:"); |
| for ( Share s : sharesSorted ) { |
| logger.trace(methodName, getId(), s.getId(), s.isInitialized(), s.getInvestment()); |
| } |
| } |
| |
| Collections.sort(sharesSorted, new ShareByInvestmentSorter()); |
| |
| if ( logger.isTrace() ) { |
| logger.trace(methodName, getId(), "Shares After Sort - id, isInitialized, investment:"); |
| for ( Share s : sharesSorted ) { |
| logger.trace(methodName, getId(), s.getId(), s.isInitialized(), s.getInvestment()); |
| } |
| } |
| |
| |
| // |
| // How much to lose? If we're not forcing, then only shares that are evicted because of |
| // the 'howMuch' counts. If forcing then everything until we meet the goal or we run |
| // out of stuff to give. |
| // |
| int shares_given = 0; // number of shares of requested order given - NOT necessarily number of my own processes |
| int processes_to_lose = 0; // number of processes I'm able to lose |
| int processes_given = 0; |
| |
| if ( force ) { |
| processes_to_lose = countNShares(); |
| } else { |
| processes_to_lose = Math.max(0, countNShares() - countNSharesGiven()); |
| } |
| processes_to_lose = Math.min(processes_to_lose, sharesSorted.size()); |
| |
| if ( processes_to_lose == 0 ) { |
| return 0; |
| } |
| |
| while ( (shares_given < shares) && (processes_given < processes_to_lose) ) { |
| |
| int currently_given = 0; |
| |
| if ( logger.isTrace() ) { |
| logger.trace(methodName, getId(), "In loop: Shares given", shares_given, "shares wanted", shares, |
| "processes_to_lose", processes_to_lose, "processes_given", processes_given); |
| } |
| |
| Share s = sharesSorted.get(0); |
| Machine m = s.getMachine(); |
| int to_give = m.countFreedUpShares(); |
| logger.debug(methodName, getId(), "Inspecting share", s.getId()); |
| ArrayList<Share> slist = new ArrayList<Share>(); |
| |
| Iterator<Share> iter = sharesSorted.iterator(); |
| while ( (to_give < order) && iter.hasNext() ) { // if we need more shares from this machine to be useful ... |
| // Here we search the share list for enough more shares on the machine to make up enough shares |
| // to satisy exactly one of the requested sizes. |
| Share ss = iter.next(); |
| if ( ss.getMachine() == s.getMachine() ) { |
| slist.add(ss); |
| to_give += ss.getShareOrder(); |
| } |
| } |
| |
| if ( to_give >= order ) { // did we find enough on the machine to make it useful to evict? |
| //slist.add(s); // didn't put on the list earlier, in case we couldn't use it |
| for ( Share ss : slist ) { |
| logger.info(methodName, getId(), "Removing share", ss.toString()); |
| pendingRemoves.put(ss, ss); |
| ss.evict(); |
| |
| sharesSorted.remove(ss); |
| processes_given++; |
| currently_given++; |
| if ( processes_given >= processes_to_lose ) break; // if this is too few to be useful, defrag will fix it (mostly) |
| } |
| shares_given += (to_give / order); |
| } |
| |
| // |
| // If we gave nothing away we didn't change any of the structures and we'll |
| // never exit. So exit stage left asap right now. |
| // We rarely if ever will enter this but it prevents an infinite loop in |
| // varioius corner cases. |
| // |
| if ( currently_given == 0 ) { |
| logger.debug(methodName, getId(), "Gave no shares, breaking loop"); |
| break; |
| } |
| } |
| |
| return shares_given; |
| } |
| |
| /** |
| * This is an investment shrink. We don't have to walk nodepools because we use nodepool depth of each share |
| * to select deeper pools first. |
| * |
| * The selection strategy is this: |
| * 1. Lowest investment always wins. |
| * 2. Deepest nodepool is secondary sort. |
| * 3. Largest machine is tertiary. |
| * |
| * The assumption is that the job must give up the indicated shares unconditionally. We let the |
| * defragmentation routine to any additional cleanup if this isn't sufficient to satisfy pending expansions. |
| * UIMA-4275 |
| */ |
| public void shrinkBy(int howmany) |
| { |
| String methodName = "shrinkBy"; |
| List<Share> sharesSorted = new ArrayList<Share>(assignedShares.values()); |
| Collections.sort(sharesSorted, new ShareByInvestmentSorter()); |
| |
| int tolose = Math.min(howmany, sharesSorted.size()); |
| |
| for ( int i = 0; i < tolose; i++ ) { |
| Share ss = sharesSorted.get(i); |
| logger.info(methodName, getId(), "Removing share", ss.toString()); |
| pendingRemoves.put(ss, ss); |
| ss.evict(); |
| } |
| } |
| |
| /** |
| * Waiting for somebody to deal with my shrinkage? |
| */ |
| public boolean isShrunk() |
| { |
| return pendingRemoves.size() > 0; |
| } |
| |
| /** |
| * Return the reclaimed shares. |
| */ |
| public HashMap<Share, Share> getPendingRemoves() |
| { |
| return pendingRemoves; |
| } |
| |
| /** |
| * And finally, dump the pending shrinkage. |
| */ |
| public void clearPendingRemoves() |
| { |
| pendingRemoves.clear(); |
| } |
| |
| /** |
| * Recovery complete, clear the share map |
| */ |
| public void clearRecoveredShares() |
| { |
| recoveredShares.clear(); |
| } |
| |
| // public Machine removeLastMachine() |
| // { |
| // return assignedMachines.remove(assignedMachines.size() - 1); |
| // } |
| |
| /** |
| * Find number of nShares (virtual shares) this machine has assigned already. |
| * |
| * If things are slow in the cluster the pending removes might be |
| * non-zero. This is an extreme corner case it's best to be safe. |
| */ |
| public int countNShares() |
| { |
| return assignedShares.size() + pendingShares.size() - pendingRemoves.size(); |
| } |
| |
| public void refuse(String refusal) |
| { |
| String methodName = "refuse"; |
| logger.warn(methodName, id, refusal); |
| this.reason = refusal; |
| deferred = true; |
| } |
| |
| public void defer(String reason) |
| { |
| String methodName = "defer"; |
| logger.info(methodName, id, reason); |
| this.reason = reason; |
| deferred = true; |
| } |
| |
| public String getReason() |
| { |
| return this.reason; |
| } |
| |
| public void setReason(String reason) |
| { |
| this.reason = reason; |
| } |
| |
| public void undefer() |
| { |
| deferred = false; |
| reason = null; |
| } |
| |
| public boolean isRefused() |
| { |
| return refused; |
| } |
| |
| public boolean isDeferred() |
| { |
| return deferred; |
| } |
| |
| public String getRefusalReason() |
| { |
| return reason; |
| } |
| |
| public void setShareOrder(int s) |
| { |
| this.share_order = s; |
| } |
| |
| public int getShareOrder() |
| { |
| return share_order; |
| } |
| |
| /** |
| * During the scheduling algorithm we want to track some things by userid. The "share cap" stuff is used |
| * to keep track of max shares that I can actually use or want during scheduling but is generally just |
| * nonsense. |
| */ |
| public void setShareCap(int cap) |
| { |
| this.share_cap = cap; |
| } |
| |
| public int getShareCap() |
| { |
| return share_cap; |
| } |
| |
| /** |
| * We try to project the maximum number of shares that this job can use, based on the current rate |
| * of completion of work items, and the known initialization time. |
| * |
| * Many jobs have very long initialization times, and will complete in their current allocation before |
| * new processes can get started and initialized. We want to avoid growing (evictions) in that case. |
| * |
| * How to use this ... |
| */ |
| private int getProjectedCap() |
| { |
| String methodName = "getPrjCap"; // want this to line up with getJobCap in logs |
| |
| // UIMA-4882 jrc |
| // Must enhance semantics of init_wait to mean "nothing initialized, and have never seen any |
| // execution time for the job." This accounts for the moment after a job initializes, and before it |
| // gets anything running and helps to throttle expansion until a job starts to run. |
| // |
| // After initialization, the time_per_item will be quite small but non-zero, so we'll tend to predict |
| // a future cap as the moral equicalent of "not too many more needed". For installations without |
| // doubling, or where doubling is too fast, this leads to better controlled expansion if the job |
| // actually is going to compete soon. |
| // |
| // The other part of this update includes the OR updating its "time_per_item" to account for |
| // work items in progress as well as work items completed, so we're guarantteed to get a |
| // time_per_item != 0 shortly after first initialization. |
| // |
| // (We update init_wait here because it's used later and needs to be used with the same |
| // semantics as is used here.) |
| init_wait = init_wait || Double.isNaN(time_per_item) || (time_per_item == 0.0); |
| |
| if ( init_wait ) { // no cap if not initialized, or no per-itme time yet |
| logger.info(methodName, getId(), username, "Cannot predict cap: init_wait", init_wait, "|| time_per_item", time_per_item); |
| return Integer.MAX_VALUE; |
| } |
| |
| // Get average init time |
| int count = 0; |
| long initialization_time = 0; // from OR - |
| for ( Share s : assignedShares.values() ) { |
| long t = s.getInitializationTime(); |
| if ( s.isInitialized() && ( t > 0) ) { |
| count++; |
| initialization_time += t; |
| } |
| } |
| if ( initialization_time > 0 ) { |
| initialization_time = initialization_time / count; // (to seconds) |
| } else { |
| logger.warn(methodName, getId(), username, "Initialization time is 0, project cap and investment will be inaccurate."); |
| } |
| |
| // When in the future we want to estimate the amount of remaining work. |
| long target = initialization_time + ducc_epoch + resource_class.getPredictionFudge(); |
| |
| int nprocesses = countNShares(); |
| double rate = ((double) (nprocesses * threads)) / time_per_item; // number of work items per second |
| // with currently assigned resources |
| long projected = Math.round(target * rate); // expected WI we can do after a new |
| // process gets started |
| |
| long future = Math.max(nquestions_remaining - projected, 0); // work still to do after doubling |
| |
| int answer = 0; |
| |
| answer = (int) future / threads; |
| if ( (future % threads ) > 0 ) answer++; |
| |
| // jrc |
| // Second problem |
| // if future cap is 0, then the future cap is the current number of processes |
| if ( answer == 0 ) { |
| answer = countNShares(); |
| } |
| logger.info(methodName, getId(), username, "O", getShareOrder(),"T", target, "NTh", (nprocesses * threads), "TI", initialization_time, |
| "TR", time_per_item, |
| "R", String.format("%.4e", rate), |
| "QR", nquestions_remaining, "P", projected, "F", future, |
| "ST", submit_time, |
| "return", answer); |
| |
| return answer; // number of processes we expect to need |
| // in the future |
| |
| } |
| |
| /** |
| * This returns the largest number that can actually be used, which will be either the |
| * share cap itself, or nProcess / nThreads, in N shares. |
| */ |
| public void initJobCap() |
| { |
| String methodName = "initJobCap"; |
| |
| if ( isRefused() ) { |
| job_cap = 0; |
| return; |
| } |
| |
| if ( isCompleted() ) { |
| // job is finishing up and will relinquish all shares soon, let's avoid complicating the |
| // world and just wait for it to happen. |
| job_cap = countNShares(); |
| return; |
| } |
| |
| // if ( getSchedulingPolicy() != Policy.FAIR_SHARE ) return; |
| |
| int c = nquestions_remaining / threads; |
| |
| if ( ( nquestions_remaining % threads ) > 0 ) { |
| c++; |
| } |
| |
| int currentResources = countNShares(); |
| c = Math.max(c, currentResources); // if job is ending we could be fragmented and have to be |
| // careful not to underestimate, or we end up possibly |
| // evicting something that should be left alone. |
| |
| // |
| // If waiting for initialization, we have to cap as well on the maximum number of shares |
| // we give out, in case the job can't start, to avoid unnecessary preemption. |
| // |
| // Must convert to N-shares, because that is the number of actual processes, which is the |
| // unit that the initialization cap is specified in. |
| // |
| |
| int base_cap = Math.min(getMaxShares(), c); |
| if ( base_cap < 0 ) base_cap = 0; // getMaxShares comes from OR - protect in case |
| // it's messed up |
| |
| int projected_cap = getProjectedCap(); |
| if ( projected_cap == 0 ) { // we know nothing, this is best guess |
| projected_cap = base_cap; |
| } |
| |
| int potential_cap = base_cap; |
| int actual_cap = 0; |
| |
| if ( resource_class.isUsePrediction() ) { |
| if (projected_cap < base_cap ) { // If we project less need, revise the estimate down |
| potential_cap = Math.max(projected_cap, currentResources); |
| } |
| } |
| |
| if ( init_wait && ( resource_class.getInitializationCap() > 0) ) { |
| actual_cap = Math.min(potential_cap, (resource_class.getInitializationCap())); |
| } else { |
| |
| if ( init_wait ) { // ugly, but true, if not using initialization caps |
| actual_cap = potential_cap; |
| } else if ( resource_class.isExpandByDoubling() ) { |
| if ( currentResources == 0 ) { |
| actual_cap = Math.max(1, resource_class.getInitializationCap()); // if we shrink to 0, need to restart from the init cap |
| actual_cap = Math.min(base_cap, actual_cap); // must re-min this in case we have a base cap < class init cap |
| } else { |
| actual_cap = Math.min(potential_cap, currentResources * 2); |
| } |
| } else { |
| actual_cap = potential_cap; |
| } |
| } |
| |
| logger.info(methodName, getId(), username, "O", getShareOrder(), "Base cap:", base_cap, "Expected future cap:", projected_cap, "potential cap", potential_cap, "actual cap", actual_cap); |
| job_cap = actual_cap; |
| } |
| |
| public int getJobCap() |
| { |
| return job_cap; |
| } |
| |
| public int getMaxShares() |
| { |
| // if set to -1, our max is the number already assigned |
| if ( max_shares < 0 ) { |
| return countNShares(); |
| } else { |
| return max_shares; |
| } |
| } |
| |
| public void setMaxShares(int s) |
| { |
| this.max_shares = s; |
| } |
| |
| public boolean isRunning() |
| { |
| return countNShares() > 0 ? true : false; |
| } |
| |
| public String getUserName() |
| { |
| return username; |
| } |
| |
| public void setUserName(String n) |
| { |
| this.username = n; |
| } |
| |
| public User getUser() |
| { |
| return user; |
| } |
| |
| public void setUser(User u) |
| { |
| this.user = u; |
| } |
| |
| public long getTimestamp() |
| { |
| return submit_time; |
| } |
| |
| public void setTimestamp(long t) |
| { |
| this.submit_time = t; |
| } |
| |
| public int getUserPriority() { |
| return user_priority; |
| } |
| |
| public void setUserPriority(int p) { |
| this.user_priority = p; |
| } |
| |
| public String getClassName() { |
| return resource_class_name; |
| } |
| |
| public void setClassName(String class_name) { |
| this.resource_class_name = class_name; |
| } |
| |
| public int getSchedulingPriority() { |
| return resource_class.getPriority(); |
| } |
| |
| public Policy getSchedulingPolicy() { |
| return resource_class.getPolicy(); |
| } |
| |
| public ResourceClass getResourceClass() { |
| return resource_class; |
| } |
| |
| // UIMA-4275 |
| public boolean exceedsFairShareCap() |
| { |
| return getResourceClass().fairShareCapExceeded(this); |
| } |
| |
| // @deprecated |
| //public int countInstances() { |
| // return n_machines; |
| //} |
| |
| // @deprecated |
| // public void setNInstances(int m) |
| // { |
| // this.n_machines = m; |
| // } |
| |
| public int nThreads() { |
| return threads; |
| } |
| public void setThreads(int th) |
| { |
| this.threads = th; |
| } |
| |
| public int getShareQuantum() |
| { |
| return resource_class.getShareQuantum(); |
| } |
| |
| public int getMemory() { |
| return memory; |
| } |
| |
| public void setMemory(int memory) { |
| this.memory = memory; |
| } |
| |
| public void setDuccType(DuccType type) |
| { |
| this.ducc_type = type; |
| } |
| |
| public DuccType getDuccType() |
| { |
| return this.ducc_type; |
| } |
| |
| // UIMA-4142 |
| public void setArbitraryProcess() |
| { |
| this.arbitrary_process = true; |
| } |
| |
| // UIMA-4142 |
| public boolean isArbitraryProcess() |
| { |
| return (ducc_type == DuccType.Service) && this.arbitrary_process; |
| } |
| |
| // UIMA-4142 |
| public boolean isService() |
| { |
| return (ducc_type == DuccType.Service) && !this.arbitrary_process; |
| } |
| |
| /** |
| * Is at least one of my current shares initialized? |
| */ |
| public boolean isInitialized() |
| { |
| for (Share s : assignedShares.values()) { |
| if ( s.isInitialized() ) return true; |
| } |
| return false; |
| } |
| |
| /** |
| * Logging and debugging - nice to know what my assigned shares are. |
| */ |
| public String printShares() |
| { |
| StringBuffer buf = new StringBuffer("AssignedShares: "); |
| if ( assignedShares.size() == 0 ) { |
| buf.append("<none>"); |
| } else { |
| for ( Share s : assignedShares.values()) { |
| buf.append(s.getId()); |
| buf.append(" "); |
| } |
| } |
| |
| buf.append("\nPendingShares: "); |
| if ( pendingShares.size() == 0 ) { |
| buf.append("<none>"); |
| } else { |
| for ( Share s : pendingShares.values()) { |
| buf.append(s.getId()); |
| buf.append(" "); |
| } |
| } |
| |
| buf.append("\nPendingRemoves: "); |
| if ( pendingRemoves.size() == 0 ) { |
| buf.append("<none>"); |
| } else { |
| for ( Share s : pendingRemoves.values()) { |
| buf.append(s.getId()); |
| buf.append(" "); |
| } |
| } |
| |
| return buf.toString(); |
| } |
| |
| public String getShortType() |
| { |
| String st = "?"; |
| switch ( ducc_type ) { |
| case Reservation: |
| st = "R"; |
| break; |
| case Job: |
| st = "J"; |
| break; |
| case Service: |
| st = ( isArbitraryProcess() ? "M" : "S" ); // UIMA-4142 |
| break; |
| } |
| return st; |
| } |
| |
| public static String getHeader() |
| { |
| // 1 2 3 4 5 6 7 8 9 10 11 12 13 |
| return String.format("%11s %30s %10s %10s %6s %5s %7s %3s %6s %6s %8s %8s %9s", |
| "ID", "JobName", "User", "Class", // 1 2 3 4 |
| "Shares", "Order", "QShares", // 5 6 7 |
| "NTh", "Memory", // 8 9 |
| "nQuest", "Ques Rem", "InitWait", // 10 11 12 |
| "Max P/Nst"); // 13 |
| } |
| |
| public String toString() |
| { |
| int shares = assignedShares.size() + pendingShares.size(); |
| //if ( getSchedulingPolicy() != Policy.FAIR_SHARE ) { |
| // shares = countInstances(); |
| //} |
| |
| // 1 2 3 4 5 6 7 8 9 10 11 12 13 |
| String format = "%11s %30.30s %10s %10s %6d %5d %7d %3d %6d %6d %8d %8s %9d"; |
| String jid = String.format("%1s%10s", getShortType(), id.toString()).replace(' ', '_'); |
| if ( isReservation() ) {// 1 2 3 4 5 6 7 8 9 10 11 12 13 |
| return String.format(format, |
| jid, // 1 |
| name.replace(' ', '_'), username, getClassName(), // 2 3 4 |
| shares, share_order, (shares * share_order), // 5 6 7 |
| 0, memory, // 8 9 |
| 0, 0, 0, // 10 11 12 |
| max_shares); // 13 |
| |
| |
| } else { |
| return String.format(format, |
| jid, // 1 |
| name.replace(' ', '_'), username, getClassName(), // 2 3 4 |
| shares, share_order, (shares * share_order), // 5 6 7 |
| threads, memory, // 8 9 |
| nQuestions(), nQuestionsRemaining(), init_wait, // 10 11 12 |
| max_shares); // 13 |
| } |
| } |
| |
| public String toStringWithHeader() |
| { |
| StringBuilder buf = new StringBuilder(getHeader()); |
| buf.append("\n"); |
| buf.append(toString()); |
| return buf.toString(); |
| } |
| |
| // |
| // Order machines by DECREASING order |
| // |
| class MachineByOrderSorter |
| implements Comparator<Machine> |
| { |
| public int compare(Machine m1, Machine m2) |
| { |
| if (m1.equals(m2)) return 0; |
| return (int) (m2.getShareOrder() - m1.getShareOrder()); |
| } |
| } |
| |
| // pull this out of the ShareByInvest sorter so other sorters can use it - see the |
| // ShareByWealth sorter in NodePoolScheduler |
| // UIMA-4275 |
| public static int compareInvestment(Share s1, Share s2) |
| { |
| if ( s1.equals(s2) ) return 0; |
| |
| int mask = 0; |
| if ( s1.isInitialized() ) mask |= 0x02; |
| if ( s2.isInitialized() ) mask |= 0x01; |
| |
| switch ( mask ) { |
| case 0: // neither is initialized |
| return ( (int) (s1.getInitializationTime() - s2.getInitializationTime()) ); |
| case 1: // s1 not initialized, s2 is |
| return -1; |
| case 2: // s2 initialized, s1 not |
| return 1; |
| default: // both initialized, compare investments |
| } |
| |
| long i1 = s1.getInvestment(); |
| long i2 = s2.getInvestment(); |
| if (i1 == i2 ) { |
| // same invesstment, go to depper nodepool first |
| int d1 = s1.getNodepoolDepth(); |
| int d2 = s1.getNodepoolDepth(); |
| |
| if ( d1 == d2 ) { |
| // same nodepool depth, go for largest machine |
| long m1 = s1.getHostMemory(); |
| long m2 = s2.getHostMemory(); |
| |
| if ( m1 == m2 ) { |
| // last resort, sort on youngest first |
| return (int) (s2.getId().getFriendly() - s1.getId().getFriendly()); |
| } |
| return (int) (m2 - m1); // largest machine |
| } |
| |
| return (d2 - d1); // greatest depth |
| |
| } else { |
| return (int) (i1 - i2); // least investment |
| } |
| } |
| |
| // |
| // Order shares by INCREASING investment |
| // |
| public static class ShareByInvestmentSorter |
| implements Comparator<Share> |
| { |
| public int compare(Share s1, Share s2) |
| { |
| return compareInvestment(s1, s2); // UIMA-4275 |
| } |
| } |
| |
| @Override |
| public int hashCode() |
| { |
| return id.hashCode(); |
| } |
| |
| @Override |
| public boolean equals(Object obj) |
| { |
| |
| if (this == obj) return true; |
| |
| if (obj == null) return false; |
| |
| |
| if (getClass() != obj.getClass()) return false; |
| |
| IRmJob other = (IRmJob) obj; |
| |
| |
| // can't get null id |
| if ( !id.equals(other.getId()) ) return false; |
| |
| //can't get null shares.. normal compare should finish it off. |
| return assignedShares.equals(other.getAssignedShares()); |
| } |
| |
| public Comparator<IEntity> getApportionmentSorter() |
| { |
| return apportionmentSorter; |
| } |
| |
| static private class ApportionmentSorterCl |
| implements Comparator<IEntity> |
| { |
| public int compare(IEntity e1, IEntity e2) |
| { |
| // Order by smallest first. The counter will round up for really |
| // small jobs so they don't get buried in the round-off errors. |
| // |
| // Note that getJobCap() is (must be) pre-computed before this sorter is called. |
| if ( e1.equals(e2) ) return 0; |
| return (int) (e1.getTimestamp() - e2.getTimestamp()); |
| |
| // return (int) (((RmJob)e1).getJobCap() - ((RmJob)e2).getJobCap()); |
| } |
| } |
| |
| } |
