Changeset 5447

Timestamp:

Oct 19, 2010, 1:02:17 PM (12 years ago)

Author:

Nicklas Nordborg

Message:

References #1528: Make slot assignment pluggable in job agents

This is now implemented. Select among three different slot managers:

• InternalSlotManager (the default)
• MasterSlotManager
• RemoteSlotManager

Documentation will be added to docbook.

Location:

trunk

Files:

• config/dist/jobagent.properties (modified) (2 diffs)
• src/clients/jobagent/net/sf/basedb/clients/jobagent/Agent.java (modified) (24 diffs)
• src/clients/jobagent/net/sf/basedb/clients/jobagent/JobRunner.java (modified) (3 diffs)
• src/clients/jobagent/net/sf/basedb/clients/jobagent/executors/DummyJobExecutor.java (modified) (4 diffs)
• src/clients/jobagent/net/sf/basedb/clients/jobagent/handlers/MultiProtocolRequestHandler.java (modified) (2 diffs)
• src/clients/jobagent/net/sf/basedb/clients/jobagent/slotmanager (added)
• src/clients/jobagent/net/sf/basedb/clients/jobagent/slotmanager/InternalSlotManager.java (added)
• src/clients/jobagent/net/sf/basedb/clients/jobagent/slotmanager/MasterSlotManager.java (added)
• src/clients/jobagent/net/sf/basedb/clients/jobagent/slotmanager/RemoteSlotManager.java (added)
• src/clients/jobagent/net/sf/basedb/clients/jobagent/slotmanager/Slot.java (added)
• src/clients/jobagent/net/sf/basedb/clients/jobagent/slotmanager/SlotManager.java (added)
• src/core/net/sf/basedb/util/jobagent/JobAgentConnection.java (modified) (2 diffs)
• src/core/net/sf/basedb/util/jobagent/JobAgentInfo.java (modified) (4 diffs)

trunk/config/dist/jobagent.properties

@@ -67 +67 @@
 
 
-# =======================
+# ============================
 # Job agent execution settings
-# =======================
+# ============================
 
-# The name of the executor class that is responsible for starting the job
+## The name of the executor class that is responsible for starting the job
 ## The default is ProcessJobExecutor which starts job in a separate process
 ## The class must implement the net.sf.basedb.clients.JobExecutor interface
@@ -103 +103 @@
 agent.checkinterval=30
 
+
+# ============================
+# Slot manager settings
+# ============================
+
+## The name of the slot manager class that is responsible for assigning a slot
+## to the job. The default is the InternalSlotManager which assign slots based
+## on the estimated execution time.
+agent.slotmanager.class=net.sf.basedb.clients.jobagent.slotmanager.InternalSlotManager
+
+# -------------------------------------------------------------------
+# The master slot manager is like the internal slot manager but also
+# accepts slot assignment on behalf of other job agents. The other
+# job agents should use the ExternalSlotManager and connect to the
+# remote control port of this job agent.
+# -------------------------------------------------------------------
+# agent.slotmanager.class=net.sf.basedb.clients.jobagent.slotmanager.MasterSlotManager
+
+# -------------------------------------------------------------------
+# The remote slot manager uses another jobagent to assign slots.
+# The other job agent should use the MasterSlotManager
+# Options
+#  server=The name/ip of the other job agent
+#  port=The remote control port of the job agent
+# -------------------------------------------------------------------
+# agent.slotmanager.class=net.sf.basedb.clients.jobagent.slotmanager.RemoteSlotManager
+# agent.slotmanager.remote.server=
+# agent.slotmanager.remote.port=
+
 ## Note! A quick job may use a slot from any of the pools reserved for
 ## slower jobs if there are unused slots. Priority values should be between

trunk/src/clients/jobagent/net/sf/basedb/clients/jobagent/Agent.java

@@ -25 +25 @@
 import java.net.InetAddress;
 import java.net.UnknownHostException;
+import java.util.Arrays;
 import java.util.Collections;
 import java.util.HashMap;
@@ -38 +39 @@
 import net.sf.basedb.clients.jobagent.handlers.MultiProtocolRequestHandler;
 import net.sf.basedb.clients.jobagent.handlers.SignalRequestHandler;
+import net.sf.basedb.clients.jobagent.slotmanager.InternalSlotManager;
+import net.sf.basedb.clients.jobagent.slotmanager.Slot;
+import net.sf.basedb.clients.jobagent.slotmanager.SlotManager;
 import net.sf.basedb.core.Application;
 import net.sf.basedb.core.DbControl;
@@ -222 +226 @@
   
   /**
+    The default slot manager to use if none has been specified
+    in the configuration file.
+  */
+  public static final Class<? extends SlotManager> DEFAULT_SLOT_MANAGER = 
+    InternalSlotManager.class;
+  
+  /**
     The default check interval in seconds.
   */
@@ -259 +270 @@
   private final Class<? extends JobExecutor> executorClass;
   
+  // Slot manager
+  private final Class<? extends SlotManager> slotManagerClass;
+  
   private InetAddress serverAddress;
   private JobAgentServerConnection server;
@@ -264 +278 @@
   private AgentSignalReceiver signalReceiver;
   private JobExecutor jobExecutor;
+  private SlotManager slotManager;
 
   private TimerTask jobQueueChecker;
@@ -270 +285 @@
   private boolean isRunning;
   private SessionControl sc;
-  
-  /**
-    The current number of threads executing in each slot.
-  */
-  private final Map<Job.ExecutionTime, Integer> usedSlots;
-  
-  /**
-    The maximum number of threads that are allowed in each slot.
-  */
-  private final Map<Job.ExecutionTime, Integer> maxSlots;
   
   /**
@@ -346 +351 @@
     this.executorClass = getJobExecutorClass(properties.getProperty("agent.executor.class"));
     
+    // Slot manager
+    this.slotManagerClass = getSlotManagerClass(properties.getProperty("agent.slotmanager.class"));
+    
     // Slots and priorities
-    this.usedSlots = new HashMap<Job.ExecutionTime, Integer>();
-    this.maxSlots = new HashMap<Job.ExecutionTime, Integer>();
     this.priorities = new HashMap<Job.ExecutionTime, Integer>();
     for (Job.ExecutionTime et : Job.ExecutionTime.values())
     {
-      usedSlots.put(et, 0);
       String configName = "agent."+et.name().toLowerCase();
-      int configuredSlots = Values.getInt(properties.getProperty(configName+".slots"), et.getDefaultSlots());
-      maxSlots.put(et, configuredSlots);
       int priority = Values.getInt(properties.getProperty(configName+".priority"), et.getDefaultPriority());
       priorities.put(et, priority);
@@ -488 +491 @@
   private Map<String, Class<? extends CustomRequestHandler>> getCustomRequestHandlerClasses()
   {
-    Map<String, Class<? extends CustomRequestHandler>> handlers = new HashMap<String, Class<? extends CustomRequestHandler>>();
+    log.debug("Loading custom request handler classes");
+    Map<String, Class<? extends CustomRequestHandler>> handlers = 
+      new HashMap<String, Class<? extends CustomRequestHandler>>();
+    
     for (String property : properties.stringPropertyNames())
     {
@@ -500 +506 @@
           if (!CustomRequestHandler.class.isAssignableFrom(clazz))
           {
-            log.warn("Class '" + className + "' doesn't implement the CustomRequestHandler interface, ignored!");
+            log.warn("Class '" + className + "' doesn't implement the CustomRequestHandler interface, ignored.");
           }
           else
@@ -510 +516 @@
         catch (Throwable t)
         {
-          log.warn("Class " + className + " not found, ignored!");
+          log.warn("Custom request handler class '" + className + "' not found, ignored.");
         }
       }
     }
+    log.debug("Custom request handler classes lodaed: " + handlers.size());
     return handlers;
   }
@@ -559 +566 @@
     try
     {
-      executor = className == null ? 
-        DEFAULT_JOB_EXECUTOR : (Class<JobExecutor>)Class.forName(className);
-      if (!JobExecutor.class.isAssignableFrom(executor))
+      if (className != null)
       {
-        log.warn("Class " + className + " doesn't implement the JobExecutor interface, using " + 
-          DEFAULT_JOB_EXECUTOR.getName() + " instead");
-        executor =  DEFAULT_JOB_EXECUTOR;
+        executor = (Class<JobExecutor>)Class.forName(className);
+        if (!JobExecutor.class.isAssignableFrom(executor))
+        {
+          log.warn("Class '" + className + "' doesn't implement the JobExecutor interface, using '" + 
+            DEFAULT_JOB_EXECUTOR.getName() + "' instead");
+          executor =  DEFAULT_JOB_EXECUTOR;
+        }
       }
     }
     catch (Throwable t)
     {
-      log.warn("Class " + className + " not found, using " + DEFAULT_JOB_EXECUTOR.getName() + 
-        " instead", t);
-    }
+      log.warn("Job executor class '" + className + "' not found, using '" + DEFAULT_JOB_EXECUTOR.getName() + 
+        "' instead", t);
+    }
+    log.info("Job executor class '" + executor.getName() + "' loaded successfully");
     return executor;
   }
+  
+  /**
+    Get the class object for the configured slot manager. If the 
+    specified class can't be found or doesn't implement the 
+    {@link SlotManager} interface a warning message is logged and 
+    the {@link #DEFAULT_SLOT_MANAGER} is used instead.
+    
+    @param className The name of the slot manager class
+    @return The class object for that class or the default job executor
+  */
+  @SuppressWarnings("unchecked")
+  private Class<? extends SlotManager> getSlotManagerClass(String className)
+  {
+    Class<? extends SlotManager> manager = DEFAULT_SLOT_MANAGER;
+    try
+    {
+      if (className != null)
+      {
+        manager = (Class<SlotManager>)Class.forName(className); 
+      }
+      if (!SlotManager.class.isAssignableFrom(manager))
+      {
+        log.warn("Class '" + className + "' doesn't implement the SlotManager interface, using '" + 
+          DEFAULT_SLOT_MANAGER.getName() + "' instead");
+        manager =  DEFAULT_SLOT_MANAGER;
+      }
+    }
+    catch (Throwable t)
+    {
+      log.warn("Slot manager class '" + className + "' not found, using '" + DEFAULT_SLOT_MANAGER.getName() + 
+        "' instead", t);
+    }
+    log.info("Slot manager class '" + manager.getName() + "' loaded successfully");
+    return manager;
+  }
+
   
   /**
@@ -648 +694 @@
       getSessionControl();;
       // Create executor and job queue checker
+      slotManager = createSlotManager();
       jobExecutor = createJobExecutor();
       jobQueueChecker = createJobQueueChecker();
@@ -680 +727 @@
     maybeStopRunningJobs();
     closeJobExecutor();
+    closeSlotManager();
     closeServer();
     if (sc != null) sc.logout();
@@ -790 +838 @@
     if (this.requestHandler == null) return;
     if (handler == null) throw new NullPointerException("handler");
+    if (log.isDebugEnabled())
+    {
+      log.debug("Registering protocol handler " + handler + " for protocols: " + Arrays.asList(protocols));
+    }
     requestHandler.registerProtocols(handler, protocols);
   }
@@ -805 +857 @@
   {
     if (this.requestHandler == null) return;
+    if (log.isDebugEnabled())
+    {
+      log.debug("Unregistering protocol handler for protocols: " + Arrays.asList(protocols));
+    }
     requestHandler.unregisterProtocols(protocols);
   }
@@ -903 +959 @@
     <p>
     Note! This method reserves the slot for the job. It is important that
-    the {@link #jobDone(Job, Job.ExecutionTime)} method is called once
+    the {@link #jobDone(Job, Slot)} method is called once
     the job has completed to return the slot to the pool. Failure to do
     so may result in that the agent thinks that all slots are
@@ -911 +967 @@
     @return The assigned slot or null if no slot is available
   */
-  synchronized Job.ExecutionTime getSlot(Job job)
+  synchronized Slot getSlot(Job job)
   {
     log.debug("Requesting slot for job: " + job);
-    Job.ExecutionTime requested = job.getEstimatedExecutionTime();
-    Job.ExecutionTime slotToUse = null;
-    Job.ExecutionTime[] slots =  Job.ExecutionTime.values();
-
-    // Check all slots from the requested execution time and longer execution times
-    for (int i = requested.ordinal(); i < slots.length; ++i)
-    {
-      if (usedSlots.get(slots[i]) < maxSlots.get(slots[i]))
-      {
-        // This slot has free jobs
-        slotToUse = slots[i];
-        usedSlots.put(slotToUse, usedSlots.get(slotToUse) + 1);
-        activeJobs.add(new JobInfo(job, slotToUse));
-        log.debug("Slot: " + slotToUse + "; used: " + usedSlots.get(slotToUse) + "; max: " + maxSlots.get(slotToUse));
-        break;
-      }
-    }
+    Slot slotToUse = slotManager.getSlot(job);
+
+    if (slotToUse != null)
+    {
+      log.debug("Got slot: " + slotToUse);
+      activeJobs.add(new JobInfo(job, slotToUse.getEstimatedExecutionTime()));
+    }
+    else
+    {
+      log.debug("No available slot");
+    }
+
     // If null we couldn't find a free slot
     return slotToUse;
@@ -960 +1011 @@
     @param usedSlot The slot that was used
   */
-  synchronized void jobDone(Job job, Job.ExecutionTime usedSlot)
-  {
-    usedSlots.put(usedSlot, usedSlots.get(usedSlot) - 1);
-    activeJobs.remove(new JobInfo(job, usedSlot));
+  synchronized void jobDone(Job job, Slot usedSlot)
+  {
+    if (slotManager != null)
+    {
+      slotManager.releaseSlot(usedSlot);
+      activeJobs.remove(new JobInfo(job, usedSlot.getEstimatedExecutionTime()));
+    }
   }
   
@@ -969 +1023 @@
     Close the service listener.
   */
-  private void closeServer()
+  private synchronized void closeServer()
   {
     log.info("Closing service listener: " + server);
@@ -1001 +1055 @@
     Close the job queue checker. 
   */
-  private void closeJobQueueChecker()
+  private synchronized void closeJobQueueChecker()
   {
     log.info("Closing job queue checker: " + jobQueueChecker);
@@ -1026 +1080 @@
     catch (RuntimeException t)
     {
-      log.error("Could not create job executor instance: " + executorClass.getName(), t);
+      log.error("Could not create job executor: " + executorClass.getName(), t);
       executor = null;
       throw t;
@@ -1032 +1086 @@
     catch (Exception e)
     {
-      log.error("Could not create job executor instance: " + executorClass.getName(), e);
+      log.error("Could not create job executor: " + executorClass.getName(), e);
       executor = null;
       throw new RuntimeException(e);
@@ -1039 +1093 @@
   }
   
+
   /**
     Close the job executor.
   */
-  private void closeJobExecutor()
+  private synchronized void closeJobExecutor()
   {
     log.info("Closing job executor: " + jobExecutor);
@@ -1052 +1107 @@
   }
   
+  
+  /**
+    Create a slot manager and initialize it.
+    @return A SlotManager instance
+    @since 2.16
+  */
+  private SlotManager createSlotManager()
+  {
+    log.info("Creating slot manager: " + slotManagerClass.getName());
+    SlotManager manager = null;
+    try
+    {
+      manager = slotManagerClass.newInstance();
+      manager.init(this);
+    }
+    catch (RuntimeException t)
+    {
+      log.error("Could not create slot manager: " + slotManagerClass.getName(), t);
+      manager = null;
+      throw t;
+    }
+    catch (Exception e)
+    {
+      log.error("Could not create slot manager: " + slotManagerClass.getName(), e);
+      manager = null;
+      throw new RuntimeException(e);
+    }
+    return manager;
+  }
+
+  /**
+    Close the slot manager.
+  */
+  private synchronized void closeSlotManager()
+  {
+    log.info("Closing slot manager: " + slotManager);
+    if (slotManager != null)
+    {
+      slotManager.close();
+      slotManager = null;
+    }
+  }
+
   /**
     Try to stop running jobs by interrupting the threads they are executing in.
   */
-  private void maybeStopRunningJobs()
+  private synchronized void maybeStopRunningJobs()
   {
     log.info("Stopping running jobs. " + activeJobs.size() + " job(s) still active.");

trunk/src/clients/jobagent/net/sf/basedb/clients/jobagent/JobRunner.java

@@ -24 +24 @@
 import java.util.Arrays;
 
+import net.sf.basedb.clients.jobagent.slotmanager.Slot;
 import net.sf.basedb.core.Application;
 import net.sf.basedb.core.DbControl;
@@ -83 +84 @@
   public void run()
   {
-    Job.ExecutionTime slotToUse = agent.getSlot(job);
+    Slot slotToUse = agent.getSlot(job);
     if (slotToUse == null)
     {
@@ -122 +123 @@
       try
       {
-        jobExecutor.executeJob(sc, agent, j, settings, slotToUse);
+        jobExecutor.executeJob(sc, agent, j, settings, slotToUse.getEstimatedExecutionTime());
       }
       catch (Throwable t)

trunk/src/clients/jobagent/net/sf/basedb/clients/jobagent/executors/DummyJobExecutor.java

@@ -30 +30 @@
 import net.sf.basedb.core.JobAgentSettings;
 import net.sf.basedb.core.SessionControl;
+import net.sf.basedb.core.signal.SignalHandler;
 import net.sf.basedb.core.signal.SignalReceiver;
 import net.sf.basedb.core.signal.ThreadSignalHandler;
@@ -93 +94 @@
     boolean aborted = false;
     Throwable error = null;
+    SignalReceiver signalReceiver = agent.getSignalReceiver();
+    SignalHandler signalHandler = null;
     try
     {
@@ -101 +104 @@
       if (wait > 0)
       {
-        SignalReceiver signalReceiver = agent.getSignalReceiver();
+        signalHandler = new ThreadSignalHandler();
         job.setSignalTransporter(signalReceiver.getSignalTransporterClass(), 
-          signalReceiver.registerSignalHandler(new ThreadSignalHandler()));
+          signalReceiver.registerSignalHandler(signalHandler));
         job.setProgress(50, "Halfway; waiting " + wait + " seconds");
         dc.commit();
@@ -142 +145 @@
     finally
     {
+      if (signalHandler != null) signalReceiver.unregisterSignalHandler(signalHandler);
       if (dc != null) dc.close();
     }

trunk/src/clients/jobagent/net/sf/basedb/clients/jobagent/handlers/MultiProtocolRequestHandler.java

@@ -29 +29 @@
 
 import net.sf.basedb.clients.jobagent.Agent;
+import net.sf.basedb.util.Values;
 import net.sf.basedb.util.jobagent.RequestHandler;
 
@@ -127 +128 @@
     if (log.isInfoEnabled())
     {
-      log.info("Returning '" + answer.replaceAll("\n", "\\\\n") + "' for command '" + cmd +"' to " + remote);
+      log.info("Returning '" + Values.getString(answer).replaceAll("\n", "\\\\n") + "' for command '" + cmd +"' to " + remote);
     }
     return answer;

trunk/src/core/net/sf/basedb/util/jobagent/JobAgentConnection.java

@@ -26 +26 @@
 import java.net.InetSocketAddress;
 import java.net.Socket;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.regex.Matcher;
+import java.util.regex.Pattern;
 
 import net.sf.basedb.util.SocketUtil;
@@ -42 +48 @@
 {
 
+  /**
+    Regexp used to parse a typical answer.
+  */
+  private static final Pattern HEADER_REGEXP = Pattern.compile("(.*):(.*)");
+  
+  /**
+    Utility method for parsing a 'typical' answer from a job agent.
+    The typical answer is a string that contains a key-value pair on each
+    line, separated by a colon. This method allows multiple entries for
+    the same key.
+    
+    @param answer The answer
+    @return A map with the key as index and a list with all values
+    @since 2.16
+  */
+  public static Map<String, List<String>> parseAnswer(String answer)
+  {
+    Map<String, List<String>> headers = new HashMap<String, List<String>>();
+    Matcher m = HEADER_REGEXP.matcher(answer);
+    while (m.find())
+    {
+      String header = m.group(1);
+      String value = m.group(2);
+      List<String> values = headers.get(header);
+      if (values == null)
+      {
+        values = new ArrayList<String>();
+        headers.put(header, values);
+      }
+      values.add(value);
+    }
+    return headers;
+  }
+
+  /**
+    Utility method for parsing a 'typical' answer from a job agent.
+    The typical answer is a string that contains a key-value pair on each
+    line, separated by a colon. This method allows only a single entry for
+    the same key.
+    
+    @param answer The answer
+    @return A map with the key as index to the value
+    @since 2.16
+  */
+  public static Map<String, String> parseSimpleAnswer(String answer)
+  {
+    Map<String, String> headers = new HashMap<String, String>();
+    Matcher m = HEADER_REGEXP.matcher(answer);
+    while (m.find())
+    {
+      String header = m.group(1);
+      String value = m.group(2);
+      headers.put(header, value);
+    }
+    return headers;
+  }
+  
+  
   private final String server;
   private final int port;

trunk/src/core/net/sf/basedb/util/jobagent/JobAgentInfo.java

@@ -22 +22 @@
 package net.sf.basedb.util.jobagent;
 
-import java.util.ArrayList;
-import java.util.HashMap;
 import java.util.HashSet;
 import java.util.List;
 import java.util.Map;
 import java.util.Set;
-import java.util.regex.Matcher;
-import java.util.regex.Pattern;
 
 import net.sf.basedb.util.Values;
@@ -44 +40 @@
 {
 
-  private static final Pattern HEADER_REGEXP = Pattern.compile("(.*):(.*)");
-  
   private final long created;
   private final Boolean paused;
@@ -91 +85 @@
   public JobAgentInfo(String answer)
   {
-    Map<String, List<String>> headers = getHeaders(answer);
+    Map<String, List<String>> headers = JobAgentConnection.parseAnswer(answer);
     this.created = System.currentTimeMillis();
     String tempPaused = getValue(headers, "Status");
@@ -201 +195 @@
     return sb.toString();
   }
-
-  private Map<String, List<String>> getHeaders(String answer)
-  {
-    Map<String, List<String>> headers = new HashMap<String, List<String>>();
-    Matcher m = HEADER_REGEXP.matcher(answer);
-    while (m.find())
-    {
-      String header = m.group(1);
-      String value = m.group(2);
-      List<String> values = headers.get(header);
-      if (values == null)
-      {
-        values = new ArrayList<String>();
-        headers.put(header, values);
-      }
-      values.add(value);
-    }
-    return headers;
-  }
   
   private String getValue(Map<String, List<String>> headers, String header)