1 | package example.globalplugin; |
---|
2 | |
---|
3 | |
---|
4 | import java.util.List; |
---|
5 | import java.util.Properties; |
---|
6 | |
---|
7 | import org.apache.commons.logging.Log; |
---|
8 | import org.apache.commons.logging.LogFactory; |
---|
9 | |
---|
10 | import schedframe.events.scheduling.SchedulingEvent; |
---|
11 | import schedframe.scheduling.SchedulerDescription; |
---|
12 | import schedframe.scheduling.manager.resources.ResourceManager; |
---|
13 | import schedframe.scheduling.manager.tasks.JobRegistry; |
---|
14 | import schedframe.scheduling.plan.SchedulingPlanInterface; |
---|
15 | import schedframe.scheduling.plan.impl.Allocation; |
---|
16 | import schedframe.scheduling.plan.impl.ScheduledTask; |
---|
17 | import schedframe.scheduling.plan.impl.SchedulingPlan; |
---|
18 | import schedframe.scheduling.plugin.grid.Module; |
---|
19 | import schedframe.scheduling.plugin.grid.ModuleList; |
---|
20 | import schedframe.scheduling.plugin.grid.ResourceDiscovery; |
---|
21 | import schedframe.scheduling.queue.QueueDescription; |
---|
22 | import schedframe.scheduling.queue.TaskQueue; |
---|
23 | import schedframe.scheduling.queue.TaskQueueList; |
---|
24 | import schedframe.scheduling.tasks.TaskInterface; |
---|
25 | import schedframe.scheduling.tasks.WorkloadUnit; |
---|
26 | |
---|
27 | public class GridFCFSLoadBalancingPlugin extends BaseGlobalPlugin { |
---|
28 | |
---|
29 | private Log log = LogFactory.getLog(GridFCFSLoadBalancingPlugin.class); |
---|
30 | |
---|
31 | public SchedulingPlanInterface<?> schedule(SchedulingEvent event, |
---|
32 | TaskQueueList queues, |
---|
33 | JobRegistry jobRegistry, |
---|
34 | ResourceManager resourceManager, ModuleList modules) { |
---|
35 | |
---|
36 | ResourceDiscovery resources = null; |
---|
37 | for(int i = 0; i < modules.size(); i++){ |
---|
38 | Module m = modules.get(i); |
---|
39 | switch(m.getType()){ |
---|
40 | case RESOURCE_DISCOVERY: resources = (ResourceDiscovery) m; |
---|
41 | break; |
---|
42 | } |
---|
43 | } |
---|
44 | |
---|
45 | SchedulingPlan plan = new SchedulingPlan(); |
---|
46 | |
---|
47 | TaskQueue q = queues.get(0); |
---|
48 | int size = q.size(); |
---|
49 | |
---|
50 | // order of the resources on this list is not determined |
---|
51 | List<SchedulerDescription> availableResources = resources.getResources(); |
---|
52 | int resourceIdx = -1; |
---|
53 | |
---|
54 | for(int i = 0; i < size; i++) { |
---|
55 | WorkloadUnit job = q.remove(0); |
---|
56 | TaskInterface<?> task = (TaskInterface<?>)job; |
---|
57 | |
---|
58 | resourceIdx = findLeastLoadedResourceIdx(availableResources); |
---|
59 | SchedulerDescription sd = availableResources.get(resourceIdx); |
---|
60 | |
---|
61 | Allocation allocation = new Allocation(); |
---|
62 | allocation.setProcessesCount(1); |
---|
63 | allocation.setProviderName(sd.getProvider().getProviderId()); |
---|
64 | |
---|
65 | ScheduledTask scheduledTask = new ScheduledTask(task); |
---|
66 | scheduledTask.setTaskId(task.getId()); |
---|
67 | scheduledTask.setJobId(task.getJobId()); |
---|
68 | scheduledTask.addAllocation(allocation); |
---|
69 | |
---|
70 | plan.addTask(scheduledTask); |
---|
71 | } |
---|
72 | return plan; |
---|
73 | |
---|
74 | } |
---|
75 | |
---|
76 | public String getName() { |
---|
77 | return getClass().getName(); |
---|
78 | } |
---|
79 | |
---|
80 | public void init(Properties properties) { |
---|
81 | // no extra initialization is expected. |
---|
82 | } |
---|
83 | |
---|
84 | private int findLeastLoadedResourceIdx(List<SchedulerDescription> availableResources){ |
---|
85 | int resourceIdx = -1; |
---|
86 | long minLoad = Long.MAX_VALUE; |
---|
87 | |
---|
88 | for(int i = 0; i < availableResources.size(); i++){ |
---|
89 | SchedulerDescription sd = availableResources.get(i); |
---|
90 | long totalLoad = getQueueLoad(sd.getAvailableQueues()); |
---|
91 | if(totalLoad < minLoad){ |
---|
92 | resourceIdx = i; |
---|
93 | minLoad = totalLoad; |
---|
94 | } |
---|
95 | } |
---|
96 | return resourceIdx; |
---|
97 | } |
---|
98 | |
---|
99 | private long getQueueLoad(List<QueueDescription> queues){ |
---|
100 | long load = 0; |
---|
101 | for(QueueDescription queue : queues){ |
---|
102 | load = load + queue.getLoad(); |
---|
103 | } |
---|
104 | return load; |
---|
105 | } |
---|
106 | |
---|
107 | } |
---|