Changeset 4074

Timestamp:

Jan 8, 2008, 1:06:44 PM (15 years ago)

Author:

Nicklas Nordborg

Message:

References #636: Kill a running job

Infrastructure is now in place. Implementation for internal job queue has been tested and working. Still need:

• Implement signal transportation to job agents and via progress reporter
• Implement support in other core plug-ins (only import plug-ins based on FlatFileParser? are working right now)
• Document everything

Location:

trunk

Files:

• config/dist/base.config (modified) (1 diff)
• config/dist/log4j.properties (modified) (1 diff)
• src/core/net/sf/basedb/core/Install.java (modified) (1 diff)
• src/core/net/sf/basedb/core/InternalJobQueue.java (modified) (4 diffs)
• src/core/net/sf/basedb/core/Job.java (modified) (8 diffs)
• src/core/net/sf/basedb/core/PluginExecutionRequest.java (modified) (4 diffs)
• src/core/net/sf/basedb/core/Update.java (modified) (2 diffs)
• src/core/net/sf/basedb/core/data/JobData.java (modified) (1 diff)
• src/core/net/sf/basedb/core/signal/AbstractSignalHandler.java (modified) (1 diff)
• src/core/net/sf/basedb/core/signal/AbstractSignalReceiver.java (modified) (6 diffs)
• src/core/net/sf/basedb/core/signal/AbstractSignalTransporter.java (modified) (7 diffs)
• src/core/net/sf/basedb/core/signal/DelegatingSignalHandler.java (modified) (5 diffs)
• src/core/net/sf/basedb/core/signal/LocalSignalReceiver.java (modified) (3 diffs)
• src/core/net/sf/basedb/core/signal/LocalSignalTransporter.java (modified) (3 diffs)
• src/core/net/sf/basedb/core/signal/Signal.java (modified) (2 diffs)
• src/core/net/sf/basedb/core/signal/SignalHandler.java (modified) (2 diffs)
• src/core/net/sf/basedb/core/signal/SignalReceiver.java (modified) (2 diffs)
• src/core/net/sf/basedb/core/signal/SignalTransporter.java (modified) (1 diff)
• src/core/net/sf/basedb/core/signal/SocketSignalReceiver.java (modified) (7 diffs)
• src/core/net/sf/basedb/core/signal/SocketSignalTransporter.java (modified) (2 diffs)
• src/core/net/sf/basedb/core/signal/ThreadSignalHandler.java (modified) (2 diffs)
• src/core/net/sf/basedb/util/QueryParameters.java (added)
• src/plugins/core/net/sf/basedb/plugins/AbstractFlatFileImporter.java (modified) (3 diffs)
• www/images/abort.png (added)
• www/views/jobs/index.jsp (modified) (2 diffs)
• www/views/jobs/view_job.jsp (modified) (6 diffs)

trunk/config/dist/base.config

@@ -77 +77 @@
 # If the internal job queue should be enabled or not
 jobqueue.internal.enabled            = true
+
+# Signal receiver class for sending signals to running jobs (to abort them)
+jobqueue.internal.signalreceiver.class = net.sf.basedb.core.signal.LocalSignalReceiver
+jobqueue.internal.signalreceiver.init  = localhost:0
 
 # If plugins with useInteralJobQueue = false should be executed or not

trunk/config/dist/log4j.properties

@@ -54 +54 @@
 #log4j.logger.net.sf.basedb.core.storage=debug
 
+### Log Signal handling events
+#log4j.logger.net.sf.basedb.core.signal=debug
+
+
 # -----------------
 # Migration loggers

trunk/src/core/net/sf/basedb/core/Install.java

@@ -107 +107 @@
     method.
   */
-  public static final int NEW_SCHEMA_VERSION = Integer.valueOf(46).intValue();
+  public static final int NEW_SCHEMA_VERSION = Integer.valueOf(47).intValue();
   
   public static synchronized void createTables(boolean update, final ProgressReporter progress)

trunk/src/core/net/sf/basedb/core/InternalJobQueue.java

@@ -26 +26 @@
 
 import net.sf.basedb.core.plugin.Response;
+import net.sf.basedb.core.signal.SignalReceiver;
 import net.sf.basedb.core.data.JobData;
 import net.sf.basedb.core.data.PluginDefinitionData;
@@ -201 +202 @@
   */
   private boolean ignoreUseInternalJobQueueFlag = false;
-  
+
+  /**
+    Use the local signal receiver only.
+  */
+  private static SignalReceiver signalReceiver;
+
   /**
     Create the job queue. The one and only instance is created
@@ -226 +232 @@
     }
     runners = Collections.synchronizedSet(new HashSet<JobRunner>());
+    
+    // Signal receiver
+    String signalReceiverClass = Config.getString("jobqueue.internal.signalreceiver.class");
+    String signalReceiverInit = Config.getString("jobqueue.internal.signalreceiver.init");
+    if (signalReceiverClass == null)
+    {
+      signalReceiverClass = "net.sf.basedb.core.signal.LocalSignalReceiver";
+      if (signalReceiverInit == null) signalReceiverInit = "localhost:0";
+    }
+    try
+    {
+      signalReceiver = (SignalReceiver)Class.forName(signalReceiverClass).newInstance();
+      signalReceiver.init(signalReceiverInit);
+    }
+    catch (Exception ex)
+    {
+      throw new BaseException(ex);
+    }
     
     // Configure thread group
@@ -430 +454 @@
           throw t;
         }
-        //exec.setSignalReceiver(signalReceiver);
+        exec.registerSignalReceiver(signalReceiver, false);
         dc.commit();
         

trunk/src/core/net/sf/basedb/core/Job.java

@@ -39 +39 @@
 import net.sf.basedb.core.plugin.Response;
 import net.sf.basedb.core.plugin.Plugin.MainType;
+import net.sf.basedb.core.signal.SignalTransporter;
 
 import java.util.Collection;
@@ -591 +592 @@
   }
   
+  public void setSignalTransporter(Class<? extends SignalTransporter> clazz, String initParams)
+  {
+    checkPermission(Permission.WRITE);
+    if (clazz == null) throw new InvalidUseOfNullException("clazz");
+    if (initParams == null) initParams = "";
+    getData().setSignalTransporter(clazz.getName() + ":" + initParams);
+  }
+  
+  public boolean hasSignalTransporter()
+  {
+    return getData().getSignalTransporter() != null;
+  }
+  
+  public SignalTransporter getSignalTransporter()
+  {
+    String tmp = getData().getSignalTransporter();
+    SignalTransporter transporter = null;
+    if (tmp != null)
+    {
+      try
+      {
+        String[] tmp2 = tmp.split(":", 2);
+        String transporterClass = tmp2[0];
+        String initParams = tmp2.length > 1 ? tmp2[1] : null;
+        transporter = (SignalTransporter)Class.forName(transporterClass).newInstance();
+        transporter.init(initParams);
+      }
+      catch (Exception ex)
+      {
+        throw new BaseException(ex);
+      }
+    }
+    return transporter;
+  }
+  
   /**
     Get the date and time the job ended.
@@ -705 +741 @@
     data.setEnded(new Date());
     data.setStackTrace(null);
+    data.setSignalTransporter(null);
     if (getSendMessage()) sendMessage();
   }
@@ -726 +763 @@
     data.setStackTrace(null);
     data.setEnded(new Date());
+    data.setSignalTransporter(null);
     if (getSendMessage()) sendMessage();
   }
@@ -747 +785 @@
     data.setPercentComplete(100);
     data.setStatus(Status.ERROR.getValue());
+    data.setSignalTransporter(null);
     if (errors != null)
     {
@@ -1227 +1266 @@
     public void display(int percent, String message)
     {
-      // Do not update unless at least some time has passed
-      if (System.currentTimeMillis() - lastUpdate > UPDATE_INTERVAL)
+      // Do not update unless at least some time has passed, or the plug-in has finished
+      if (System.currentTimeMillis() - lastUpdate > UPDATE_INTERVAL || percent == 100)
       {
         lastUpdate = System.currentTimeMillis();
@@ -1284 +1323 @@
     }
     
-    void setStarted()
+    void setStarted(Class<? extends SignalTransporter> signalTransporter, String signalId)
     {
       DbControl dc = null;
@@ -1291 +1330 @@
         dc = sc.newDbControl();
         Job job = Job.getById(dc, jobId);
-        job.start("Starting...", server);       
+        if (signalTransporter != null && signalId != null)
+        {
+          job.setSignalTransporter(signalTransporter, signalId);
+        }
+        job.start("Starting...", server);
         dc.commit();
       }

trunk/src/core/net/sf/basedb/core/PluginExecutionRequest.java

@@ -31 +31 @@
 import net.sf.basedb.core.signal.SignalReceiver;
 import net.sf.basedb.core.signal.SignalTarget;
+import net.sf.basedb.core.signal.SignalTransporter;
 
 /**
@@ -64 +65 @@
 {
   private Job.ProgressReporterImpl progress;
+  private SignalReceiver signalReceiver = null;
+  private boolean forceSignalReceiver = false;
   
   PluginExecutionRequest(SessionControl sc, Plugin plugin, String command,
@@ -87 +90 @@
     Request request = new RequestImpl(false);
     Response response = pluginResponse.getResponseImpl();
-    if (progress != null) progress.setStarted();
+    SignalHandler signalHandler = null;
     try
     {
-      getPlugin().run(request, response, progress);
+      Plugin plugin = getPlugin();
+      String signalId = null;
+      Class<? extends SignalTransporter> signalTransporter = null;
+      if (plugin instanceof SignalTarget)
+      {
+        // Register a signal handler, receiver and transporter
+        signalHandler = ((SignalTarget)plugin).getSignalHandler();
+        if (!forceSignalReceiver && signalHandler != null)
+        {
+          SignalReceiver override = signalHandler.getSignalReceiver();
+          if (override != null) signalReceiver = override;
+        }
+        if (signalReceiver != null && signalHandler != null)
+        {
+          signalId = signalReceiver.registerSignalHandler(signalHandler);
+          signalTransporter = signalReceiver.getSignalTransporterClass();
+        }
+      }
+      if (progress != null) progress.setStarted(signalTransporter, signalId);
+      plugin.run(request, response, progress);
     }
     catch (Throwable t)
     {
       pluginResponse.setError("Error invoking plugin: " + t.getMessage() , t);
+    }
+    finally
+    {
+      if (signalReceiver != null && signalHandler != null)
+      {
+        signalReceiver.unregisterSignalHandler(signalHandler);
+      }
     }
     done();
@@ -121 +150 @@
   }
   
-  public void setSignalReceiver(SignalReceiver signalReceiver)
+  /**
+    Register a signal receiver that will receive signals for this job.
+    The plug-in must implement the {@link SignalTarget} interface
+    and provide a {@link SignalHandler}. If not, the job will not be
+    registered with the signal receiver and it will not be possible to
+    send signals to the job.
+    
+    @param signalReceiver The signal recevier to use
+    @param force If true, the specified signal recevier will always be used,
+      even if the plug-in wants to use another receiver implementation
+  */
+  public void registerSignalReceiver(SignalReceiver signalReceiver, boolean force)
   {
-    Plugin p = getPlugin();
-    SignalHandler signalHandler = null;
-    if (p instanceof SignalTarget)
-    {
-      signalHandler = ((SignalTarget)p).getSignalHandler();
-    }
-    if (signalHandler != null)
-    {
-      String ID = signalReceiver.registerSignalHandler(signalHandler);
-      Job job = getJob();
-      //job.setSignalReceiverId(ID);
-    }
+    this.signalReceiver = signalReceiver;
+    this.forceSignalReceiver = force;
   }
   

trunk/src/core/net/sf/basedb/core/Update.java

@@ -544 +544 @@
     </td>
   </tr>
+  <tr>
+    <td>47</td>
+    <td>
+      <ul>
+      <li>Added {@link net.sf.basedb.core.data.JobData#getSignalTransporter()}.
+      </ul>
+      No special database update is needed. Only increase the schema version.
+    </td>
+  </tr>
   </table>
 
@@ -774 +783 @@
         if (progress != null) progress.display((int)(45*progress_factor), "--Updating schema version: " + schemaVersion + " -> 46...");
         schemaVersion = setSchemaVersionInTransaction(session, 46);
+      }
+
+      if (schemaVersion < 47)
+      {
+        if (progress != null) progress.display((int)(46*progress_factor), "--Updating schema version: " + schemaVersion + " -> 47...");
+        schemaVersion = setSchemaVersionInTransaction(session, 47);
       }
 

trunk/src/core/net/sf/basedb/core/data/JobData.java

@@ -356 +356 @@
   }
 
+  /**
+    The maximum allowed length of the signal transporter string.
+  */
+  public static final int MAX_SIGNAL_TRANSPORTER_LENGTH = 655536;
+  private String signalTransporter;
+  /**
+    Identifies the signal transporter to use. This is a string with 
+    two parts separated by colon (:). The first part is the class name
+    of the signal transporter class, the second part is the initialisation
+    string for the transporter.
+    @hibernate.property column="`signal_transporter`" type="text" not-null="false"
+    @since 2.6
+  */
+  public String getSignalTransporter()
+  {
+    return signalTransporter;
+  }
+  public void setSignalTransporter(String signalTransporter)
+  {
+    this.signalTransporter = signalTransporter;
+  }
   
   private Map<String, ParameterValueData<?>> parameters;

trunk/src/core/net/sf/basedb/core/signal/AbstractSignalHandler.java

@@ -78 +78 @@
     return supported != null && supported.contains(signal);
   }
+  public SignalReceiver getSignalReceiver()
+  {
+    return null;
+  }
   // -------------------------------------------
 

trunk/src/core/net/sf/basedb/core/signal/AbstractSignalReceiver.java

@@ -24 +24 @@
 package net.sf.basedb.core.signal;
 
+import java.net.URI;
 import java.util.Collection;
 import java.util.Collections;
@@ -34 +35 @@
   return values for the {@link #registerSignalHandler(SignalHandler)} method.
   This class will generate values in the form of {@link java.net.URI}:s.
-  <code>signal://receiverId/handlerId?supportedSignals</code>. 
+  <code>signal://handlerId@receiverId/?supportedSignals</code>. 
   
   <ul>
   <li>The <i>receiverId</i> part is given by the parameter in the 
-    {@link #init(String)} method and must be set by the implementing subclass. 
+    {@link #init(String)} method and must be set by the implementing 
+    subclass. 
     
   <li>The <i>handlerId</i> part is given by calling {@link System#identityHashCode(Object)} 
@@ -51 +53 @@
   as higher-level objects.
   <p>
-  Subclasses should override the {@link #getSignalHandlerId(SignalHandler)} if they 
+  Subclasses should override the {@link #getGlobalSignalId(SignalHandler)} if they 
   want to use a different ID generation scheme.
 
@@ -62 +64 @@
 {
 
+  /**
+    Log signals processing.
+  */
+  private static final org.apache.log4j.Logger logger = 
+    org.apache.log4j.LogManager.getLogger("net.sf.basedb.core.signal.AbstractSignalReceiver");
+  
+
+  /**
+    Maps local signal handler ID to -> signal handler itself.
+  */
   private Map<String, SignalHandler> handlers;
   private String receiverId;
@@ -77 +89 @@
   /**
     Initialise the signal receiver. 
-    @param params The ID of the signal receiver, needed if the default 
+    @param receiverId The ID of the signal receiver, needed if the default 
       ID generation should be used
   */
-  public void init(String params)
+  public void init(String receiverId)
   {
     handlers = Collections.synchronizedMap(new HashMap<String, SignalHandler>());
-    receiverId = params;
-  }
-  
+    this.receiverId = receiverId;
+    logger.info("Initializing signal receiver: id=" + receiverId);
+  }
+  /**
+    Close this signal receiver.
+  */
   public void close()
   {
+    logger.info("Closing signal receiver: receiver id=" + receiverId);
     if (handlers != null) handlers.clear();
     handlers = null;
   }
 
+  /**
+    Register a signal handler with this receiver.
+    @param handler The signal handler to register
+    @return The global ID of the signal handler
+  */
   public String registerSignalHandler(SignalHandler handler)
   {
-    String id = getSignalHandlerId(handler);
-    if (handlers != null) handlers.put(id, handler);
-    return id;
-  }
-
+    String globalId = getGlobalSignalId(handler);
+    String localId = getLocalSignalHandlerId(handler);
+    if (handlers != null) handlers.put(localId, handler);
+    logger.info("Register signal handler: recevier id = " + receiverId + 
+      "; global id=" + globalId + "; local id=" + localId);
+    logger.debug("Current number of registered signal handlers: " + handlers.size());
+    return globalId;
+  }
+
+  /**
+    Unregister a signal handler.
+    @param handler The signal handler to unregister
+  */
   public void unregisterSignalHandler(SignalHandler handler)
   {
-    if (handlers != null) handlers.remove(getSignalHandlerId(handler));
+    String localId = getLocalSignalHandlerId(handler);
+    logger.info("Unregister signal handler: recevier id = " + receiverId + 
+      "; local id=" + localId);
+    if (handlers != null) handlers.remove(localId);
+    logger.debug("Current number of registered signal handlers: " + handlers.size());
   }
   // -------------------------------------------
 
   /**
-    Generate a signal handler ID string. This string is returned by
+    Generate a signal ID string. This string is returned by
     the {@link #registerSignalHandler(SignalHandler)} method and is used
     in {@link SignalTransporter#init(String)} method to initialise
-    a transporter object.
+    a transporter object so that it can send signals to the specified handler.
+    See the class documentation for a description of the format of the
+    generated string.
     
     @param handler The signal handler to generate the ID for
     @return The signal handler ID
   */
-  protected String getSignalHandlerId(SignalHandler handler)
+  protected String getGlobalSignalId(SignalHandler handler)
   {
     StringBuilder sb = new StringBuilder();
-    sb.append("signal://").append(receiverId).append("/");
-    sb.append(System.identityHashCode(handler)).append("?");
+    sb.append("signal://");
+    sb.append(getLocalSignalHandlerId(handler)).append("@");
+    sb.append(receiverId).append("/?");
     Collection<Signal> signals = handler.getSupportedSignals();
     if (signals != null)
@@ -134 +170 @@
   
   /**
+    Get the local signal handler id of the given signal handler.
+    This implementation simply return the system hashcode for the
+    handler.
+    @param handler The handler to get the id for
+    @return The local handler id
+  */
+  protected String getLocalSignalHandlerId(SignalHandler handler)
+  {
+    return String.valueOf(System.identityHashCode(handler));
+  }
+  
+  /**
     Get the signal handler with a given ID.
-    @param handlerId The signal handler ID
+    @param localId The local signal handler ID as 
+      returned by the {@link #getLocalSignalHandlerId(SignalHandler)}
+      method
     @return The signal handler, or null if no handler is found
   */
-  protected SignalHandler getSignalHandler(String handlerId)
-  {
-    return handlers == null ? null : handlers.get(handlerId);
-  }
+  protected SignalHandler getSignalHandler(String localId)
+  {
+    return handlers == null ? null : handlers.get(localId);
+  }
+  
+  /**
+    Process a signal message. If the message can't be understood or
+    if no handler can be found this method does nothing. The signal will
+    be delivered to the signal handler in the current thread.
+    @param message The message to process, the format of the message
+      must be compatible with the message that {@link 
+      AbstractSignalTransporter#generateSignalMessage(Signal)} generates
+  */
+  protected void processSignalMessage(String message)
+  {
+    logger.debug("Processing signal message: " + message);
+    if (message == null) return;
+    try
+    {
+      URI uri = new URI(message);
+      String localId = uri.getUserInfo();
+      Signal signal = Signal.getSignal(uri.getQuery());
+      SignalHandler signalHandler = getSignalHandler(localId);
+      if (signalHandler != null) 
+      {
+        signalHandler.handleSignal(signal);
+      }
+      else
+      {
+        logger.warn("No signal handler found for id: " + localId);
+      }
+    }
+    catch (Exception ex)
+    {
+      // Ignore invalid messages
+      logger.warn("Could not process signal message: " + message, ex);
+    }
+  }
+  
   
 }

trunk/src/core/net/sf/basedb/core/signal/AbstractSignalTransporter.java

@@ -30 +30 @@
 
 /**
+  Abstract base class that is useful when implementing signal transporters.
 
   @author nicklas
@@ -38 +39 @@
   implements SignalTransporter
 {
+  /**
+    Log signals processing.
+  */
+  private static final org.apache.log4j.Logger logger = 
+    org.apache.log4j.LogManager.getLogger("net.sf.basedb.core.signal.AbstractSignalTransporter");
   
-  private String signalHandlerId;
+  private String globalSignalId;
   private URI signalHandlerURI;
   private Collection<Signal> signals;
@@ -52 +58 @@
   /**
     Initialise the transporter. We expect the parameter to be the one
-    returned from {@link AbstractSignalReceiver#getSignalHandlerId(SignalHandler)}.
+    returned from {@link AbstractSignalReceiver#getGlobalSignalId(SignalHandler)}.
     If the receiver subclass overrided that method, the corresponding transporter
     subclass should override this method.
+    @see AbstractSignalReceiver for a description of the format expected
   */
   public void init(String params)
   {
-    signalHandlerId = params;
+    logger.info("Initializing signal transporter: id=" + params);
+    globalSignalId = params;
   }
-  
   
   public Collection<Signal> getSupportedSignals()
@@ -67 +74 @@
     {
       signals = new HashSet<Signal>();
-      URI uri = getSignalHandlerURI();
+      URI uri = getSignalURI();
       String query = uri.getQuery();
       if (query != null)
@@ -83 +90 @@
   // -------------------------------------------
 
-  protected String getSignalHandlerId()
+  /**
+    Generate a signal message string for the given signal. The string
+    will have the following format:
+    <code>signal://handlerId@receiverId/?signal</code>
+    @see AbstractSignalReceiver#processSignalMessage(String)
+  */
+  protected String generateSignalMessage(Signal signal)
   {
-    return signalHandlerId;
+    String message = "signal://" + getHandlerId() + "@" + getReceiverId() + "/?" + signal.getId();
+    return message;
   }
   
-  protected URI getSignalHandlerURI()
+  /**
+    Get the raw ID string that was passed to the {@link #init(String)}
+    method.
+  */
+  protected String getGlobalSignalId()
+  {
+    return globalSignalId;
+  }
+  
+  /**
+    Get the URI representation of the global signal ID.
+    @return An URI
+    @throws SignalException If the signal ID is not a valid URI.
+  */
+  protected URI getSignalURI()
   {
     if (signalHandlerURI == null)
@@ -94 +122 @@
       try
       {
-        signalHandlerURI = new URI(signalHandlerId);
+        signalHandlerURI = new URI(globalSignalId);
       }
       catch (URISyntaxException ex)
       {
-        throw new SignalException(signalHandlerId, ex);
+        throw new SignalException(globalSignalId, ex);
       }
     }
@@ -104 +132 @@
   }
   
+  /**
+    Get the receiverId part of the signal URI.
+    @return The ID of the receiver
+  */
   protected String getReceiverId()
   {
-    return getSignalHandlerURI().getAuthority();
+    URI uri = getSignalURI();
+    String receiverId = uri.getHost();
+    if (uri.getPort() >= 0) receiverId += ":" + uri.getPort();
+    return receiverId;
   }
   
+  /**
+    Get the local handler ID part of the signal URI.
+    @return The local ID of the signal handler
+  */
   protected String getHandlerId()
   {
-    return getSignalHandlerURI().getPath();
+    return getSignalURI().getUserInfo();
   }
   

trunk/src/core/net/sf/basedb/core/signal/DelegatingSignalHandler.java

@@ -46 +46 @@
 {
   /**
+    Log signals processing.
+  */
+  private static final org.apache.log4j.Logger logger = 
+    org.apache.log4j.LogManager.getLogger("net.sf.basedb.core.signal.DelegatingSignalHandler");
+  
+  /**
     Holds the registered signal handlers.
   */
@@ -51 +57 @@
   
   /**
+    If a specific signal receiver must be used.
+  */
+  private SignalReceiver signalReceiver;
+  
+  /**
     Create a new delegating signal handler. Signal handlers to
     delegate to should be registered with 
     {@link #registerSignalHandler(SignalHandler)}.
-   */
+  */
   public DelegatingSignalHandler()
   {
+    this(null);
+  }
+  
+  /**
+    Create a new delegating signal handler using a specified signal
+    receiver. Signal handlers to delegate to should be registered with 
+    {@link #registerSignalHandler(SignalHandler)}.
+    @param signalReceiver The signal receiver that should receive the signals,
+      or null to use the system default signal receiver
+  */
+  public DelegatingSignalHandler(SignalReceiver signalReceiver)
+  {
     this.handlers = new HashMap<Signal, Set<SignalHandler>>();
+    this.signalReceiver = signalReceiver;
   }
   
@@ -90 +114 @@
   public void handleSignal(Signal signal)
   {
+    logger.debug("Got signal: " + signal);
     Set<SignalHandler> all = handlers.get(signal);
-    if (all == null) throw new UnsupportedSignalException(signal);
+    if (all == null) 
+    {
+      logger.debug("Signal not supported " + signal);
+      throw new UnsupportedSignalException(signal);
+    }
     for (SignalHandler handler : all)
     {
+      logger.debug("Sending signal " + signal + " to: " + handler);
       handler.handleSignal(signal);
     }
+  }
+  
+  public SignalReceiver getSignalReceiver()
+  {
+    return signalReceiver;
   }
   // -------------------------------------------
@@ -106 +141 @@
   public void registerSignalHandler(SignalHandler handler)
   {
+    logger.debug("Registering signal handler: " + handler);
     for (Signal signal : handler.getSupportedSignals())
     {
@@ -124 +160 @@
   public void unregisterSignalHandler(SignalHandler handler)
   {
+    logger.debug("Unregistering signal handler: " + handler);
     Iterator<Map.Entry<Signal, Set<SignalHandler>>> it = handlers.entrySet().iterator();
     while (it.hasNext())

trunk/src/core/net/sf/basedb/core/signal/LocalSignalReceiver.java

@@ -40 +40 @@
   extends AbstractSignalReceiver
 {
+  /**
+    Log signals processing.
+  */
+  private static final org.apache.log4j.Logger logger = 
+    org.apache.log4j.LogManager.getLogger("net.sf.basedb.core.signal.LocalSignalReceiver");
 
+  /**
+    Holds all registered signal receivers.
+  */
   private static Map<String, LocalSignalReceiver> receivers =
     Collections.synchronizedMap(new HashMap<String, LocalSignalReceiver>());
-  
+
   /**
     Get a signal receiver with a given ID.
@@ -88 +96 @@
   }
   
+  /**
+    @return {@link LocalSignalTransporter}
+  */
   public Class<? extends SignalTransporter> getSignalTransporterClass()
   {
@@ -94 +105 @@
   // -------------------------------------------
   
+  /**
+    Send the signal to a registered handler. If no handler with the
+    given ID is found this method does nothing. The signal will be
+    processed by the handler in the current thread.
+    @param handlerId The ID of a registered handler
+    @param signal The signal to send
+  */
   public void send(String handlerId, Signal signal)
   {
+    logger.info("Receiving signal " + signal.getId() + " to " + handlerId);
     SignalHandler handler = super.getSignalHandler(handlerId);
     if (handler != null) handler.handleSignal(signal);

trunk/src/core/net/sf/basedb/core/signal/LocalSignalTransporter.java

@@ -25 +25 @@
 
 /**
+  A signal transporter implementation that can transport signals within
+  the local virtual machine only. The corresponding receiver class
+  is {@link LocalSignalReceiver}.
 
   @author nicklas
@@ -33 +36 @@
   extends AbstractSignalTransporter
 {
+  /**
+    Log signals processing.
+  */
+  private static final org.apache.log4j.Logger logger = 
+    org.apache.log4j.LogManager.getLogger("net.sf.basedb.core.signal.LocalSignalTransporter");
 
+  /**
+    Create a new signal transporter instance.
+  */
   public LocalSignalTransporter()
   {}
@@ -41 +52 @@
     -------------------------------------------
   */
+  /**
+    Send the signal to a local signal receiver. This method will call
+    {@link LocalSignalReceiver#getSignalReceiver(String)} to find a registered
+    signal receiver and then call {@link LocalSignalReceiver#send(String, Signal)}.
+    The signal is sent and processed by the signal handler in the current thread.
+  */
   public void send(Signal signal)
   {
+    logger.info("Sending signal " + signal.getId() + " to " + getGlobalSignalId());
     LocalSignalReceiver receiver = LocalSignalReceiver.getSignalReceiver(getReceiverId());
-    if (receiver != null) receiver.send(getSignalHandlerId(), signal);
+    if (receiver != null) receiver.send(getHandlerId(), signal);
   }
   // -------------------------------------------

trunk/src/core/net/sf/basedb/core/signal/Signal.java

@@ -55 +55 @@
     Holds all registered signals.
   */
-  private static Map<String, Signal> signals = new HashMap<String, Signal>();
+  private static Map<String, Signal> signals;
   
   /**
@@ -69 +69 @@
   public static synchronized Signal registerSignal(String id, String name, String description)
   {
-    if (signals.containsKey(id))
+    if (signals == null) signals = new HashMap<String, Signal>();
+    Signal s = signals.get(id);
+    if (s == null)
     {
-      return signals.get(id);
+      s = new Signal(id, name, description);
+      signals.put(id, s);
     }
-    Signal s = new Signal(id, name, description);
-    signals.put(id, s);
     return s;
   }

trunk/src/core/net/sf/basedb/core/signal/SignalHandler.java

@@ -33 +33 @@
   in most cases notify the target about it. How this is done is up to each
   implementation.
+  <p>
+  Signal handler implementations need to be partly thread safe. Once they have 
+  been registered with a {@link SignalReceiver} they may receive multiple signals
+  in different threads at the same time.
 
   @author nicklas
@@ -59 +63 @@
   */
   public boolean supports(Signal signal);
+  
+  /**
+    Return a specific signal receiever that must be used with this
+    signal handler. Null should be returned to let the system select
+    an appropriate signal receiver. If a non-null value is returned, the
+    system should use this signal receiver instead of the system default.
+    @return A signal receiver or null
+  */
+  public SignalReceiver getSignalReceiver();
+  
 }

trunk/src/core/net/sf/basedb/core/signal/SignalReceiver.java

@@ -32 +32 @@
   the {@link SocketSignalReceiver} and {@link SocketSignalTransporter} implementations
   which uses network sockets to transport signals.
+  <p>
+  Signal receivers must be thread safe since a single receiver may be used to 
+  handle signals for multiple jobs at the same time.
   
   @author nicklas
   @version 2.6
+  @see SignalTransporter
+  @see SignalHandler
   @base.modified $Date$
 */
@@ -75 +80 @@
     @return A string that allows a transporter instance locate and send a signal to
       the given signal handler
+    @see AbstractSignalReceiver
   */
   public String registerSignalHandler(SignalHandler handler);

trunk/src/core/net/sf/basedb/core/signal/SignalTransporter.java

@@ -38 +38 @@
   and {@link SocketSignalTransporter} uses a {@link java.net.URI} that contains
   the IP number and port of the receiver and an ID for identifying the signal handler.
+  <p>
+  Signal transporters need not be thread safe since a new instance is created 
+  for each signal that is going to be sent.
 
   @author nicklas

trunk/src/core/net/sf/basedb/core/signal/SocketSignalReceiver.java

@@ -28 +28 @@
 import java.net.InetSocketAddress;
 import java.net.ServerSocket;
+import java.net.Socket;
+import java.net.UnknownHostException;
 import java.nio.channels.ClosedByInterruptException;
 import java.nio.channels.ServerSocketChannel;
 import java.nio.channels.SocketChannel;
-
+import java.util.HashSet;
+import java.util.List;
+import java.util.Set;
+
+import net.sf.basedb.core.Application;
+import net.sf.basedb.util.QueryParameters;
 import net.sf.basedb.util.SocketUtil;
 import net.sf.basedb.util.Values;
@@ -38 +45 @@
   A signal receiver implementation that listens on a socket for incoming signals.
   Accordingly, this receiver supports sending signals between different virtual
-  machines running on different servers. The initialisation parameter
-  for this should be the port number the receiver should listen on. If null, the
-  receiver will open a port number at random.
+  machines running on different servers. The initialisation string for this  
+  class should be of the format:
+  
+  <p>
+  <code>port=xx&allow=ip-address&allow=ip-address...</code>
+
+  <p>
+  where <code>port</code> is the port number the signal receiver will listen on and
+  the <code>allow</code> parts are the ip name or numbers of hosts that are allowed
+  to send signals to the receiver. Except for the special case <code>allow=*</code>,
+  which allows any remote host to send signals an exact match is required. The
+  local host is always allowed to send signals.
+  <p>
+  If no port is given, the signal receiver will randomly choose a free port 
+  <p>
+  If no <code>allow</code> tags are given only allow connections from the
+  local host are allowed.
   
   @author nicklas
@@ -49 +70 @@
   extends AbstractSignalReceiver
 {
+  /**
+    Log signals processing.
+  */
+  private static final org.apache.log4j.Logger logger = 
+    org.apache.log4j.LogManager.getLogger("net.sf.basedb.core.signal.SocketSignalReceiver");
 
   private InetAddress ip;
   private int port;
   private Thread listener;
+  private Set<InetAddress> allow;
+  private boolean allowAll;
   
   /**
     Create a new socket signal receiver. Before it can be used it must
     be initialised with {@link #init(String)}.
-   */
+  */
   public SocketSignalReceiver()
   {}
@@ -69 +97 @@
     receiver should listen on. If the parameter can't be parsed as a numeric value,
     a random free port will be used. This method will open the socket and 
+    start a separate thread that listens for incoming signals.
   */
   public void init(String params)
   {
-    port = Values.getInt(params, 0);
-    
+    QueryParameters qp = QueryParameters.parseQueryString(params);
+    // Get the port number we listen on
+    port = Values.getInt(qp.getValue("port"), 0);
+    
+    // Get allowed hosts
+    allow = new HashSet<InetAddress>();
+    allowAll = false;
+    List<String> allowIp = qp.getValues("allow");
+    if (allowIp != null)
+    {
+      for (String ip : allowIp)
+      {
+        if ("*".equals(ip)) 
+        {
+          logger.debug("Allow all hosts enabled!");
+          allowAll = true;
+        }
+        else
+        {
+          try
+          {
+            logger.debug("Adding host " + ip + " to list of allowed hosts.");
+            allow.add(InetAddress.getByName(ip));
+          }
+          catch (UnknownHostException ex)
+          {
+            logger.warn("Unknown host: " + ip, ex);
+          }
+        }
+      }
+    }
+    
+    logger.info("Starting socket signal receiver on port: " + port);
     try
     {
@@ -81 +141 @@
       port = socket.getLocalPort();
       ip = socket.getInetAddress();
-      listener = new Thread(new ListenerThread(channel), 
-          "ListenerThread."+this.toString());
+      listener = new Thread(new ListenerThread(channel), "ListenerThread."+port);
       listener.start();
+      logger.info("Socket signal receiver is listening on port: " + port);
     }
     catch (IOException ex)
     {
+      logger.error("Could not initialise socket signal receiver", ex);
       throw new SignalException(ex);
     }
-    super.init(ip.getHostAddress() + ":" + port);
-  }
-
-  
+    super.init(Application.getHostName() + ":" + port);
+  }
+
+  /**
+    @return {@link SocketSignalTransporter}
+  */
   public Class<? extends SignalTransporter> getSignalTransporterClass()
   {
@@ -98 +161 @@
   }
 
+  /**
+    Close this receiver and the socket it is listening on.
+  */
   public void close()
   {
+    logger.info("Interrupting socket signal receiver on port: " + port);
     listener.interrupt();
   }
@@ -133 +200 @@
         {
           SocketChannel incoming = socket.accept();
+          Socket inSocket = incoming.socket();
+          InetAddress remoteHost = inSocket.getInetAddress();
           
-          String cmd = SocketUtil.read(incoming.socket(), true);
-          String[] tmp = cmd.split("#");
-          String signalHandlerId = tmp[0];
-          Signal signal = Signal.getSignal(tmp[1]);
-          SignalHandler signalHandler = getSignalHandler(signalHandlerId);
-          signalHandler.handleSignal(signal);
+          logger.debug("Incoming signal: remote host=" + remoteHost);
+          
+          // Check if we are allowed to accept connections from the remote host
+          if (allowAll || SocketUtil.isLocalHost(remoteHost) || allow.contains(remoteHost))
+          {
+            // Read the incoming message
+            String message = SocketUtil.read(inSocket, true);
+            logger.debug("Incoming signal accepted: remote host=" + remoteHost +
+              "; message=" + message);
+            processSignalMessage(message);
+          }
+          else
+          {
+            logger.debug("Incoming signal rejected: remote host=" + remoteHost);
+          }
+          
           interrupted = Thread.interrupted();
         }
         catch (ClosedByInterruptException ex)
         {
+          logger.info("Shutting down socket signal receiver on port: " + port);
           interrupted = true;
         }
-        catch (Throwable t)
-        {
-          interrupted = true;
+        catch (IOException ex)
+        {
+          logger.warn("Error on socket signal receiver on port: " + port, ex);
         }
       }

trunk/src/core/net/sf/basedb/core/signal/SocketSignalTransporter.java

@@ -27 +27 @@
 import java.net.URI;
 
-import net.sf.basedb.core.BaseException;
 import net.sf.basedb.util.SocketUtil;
 
@@ -40 +39 @@
 {
 
+  /**
+    Log signals processing.
+  */
+  private static final org.apache.log4j.Logger logger = 
+    org.apache.log4j.LogManager.getLogger("net.sf.basedb.core.signal.SocketSignalTransporter");
+  
+  /**
+    Create a new socket signal transporter. Before it can be used it must
+    be initialised with {@link #init(String)}.
+  */
+  public SocketSignalTransporter()
+  {}
+  
   /*
     From the SignalTransporter interface
     -------------------------------------------
   */
+  /**
+    Connect to the remote host and send the signal.
+  */
   public void send(Signal signal)
   {
-    URI uri = getSignalHandlerURI();
-    String host = uri.getAuthority();
+    logger.info("Sending signal " + signal.getId() + " to " + getGlobalSignalId());
+    URI uri = getSignalURI();
+    String host = uri.getHost();
     int port = uri.getPort();
-    
+    String message = generateSignalMessage(signal);
+    logger.debug("The message is: " + message);
+    Socket s = null;
     try
     {
-      Socket s = new Socket(host, port);
-      SocketUtil.send(s, getSignalHandlerId() + "#" + signal.getId(), true);
-      SocketUtil.close(s);
+      s = new Socket(host, port);
+      SocketUtil.send(s, message, true);
     }
     catch (Exception ex)
     {
-      throw new BaseException(ex);
+      logger.error("Error sending signal " + signal.getId() + " to " + getGlobalSignalId(), ex);
+      throw new SignalException(ex);
     }
-
+    finally
+    {
+      if (s != null) SocketUtil.close(s);
+    }
   }
   // -------------------------------------------

trunk/src/core/net/sf/basedb/core/signal/ThreadSignalHandler.java

@@ -78 +78 @@
 {
 
+  /**
+    Log signals processing.
+  */
+  private static final org.apache.log4j.Logger logger = 
+    org.apache.log4j.LogManager.getLogger("net.sf.basedb.core.signal.ThreadSignalHandler");
+
+  
   private static final Set<Signal> supported = Collections.singleton(Signal.ABORT);
 
@@ -122 +129 @@
   public void handleSignal(Signal signal)
   {
+    logger.debug("Got signal: " + signal);
     if (!workerThread.isAlive()) return;
     if (Signal.ABORT.equals(signal))
     {
+      logger.debug("Sending signal " + signal + " to thread: " + workerThread);
       workerThread.interrupt();
     }
     else
     {
+      logger.debug("Signal not supported " + signal);
       throw new UnsupportedSignalException(signal);
     }

trunk/src/plugins/core/net/sf/basedb/plugins/AbstractFlatFileImporter.java

@@ -52 +52 @@
 import net.sf.basedb.core.plugin.Response;
 import net.sf.basedb.core.plugin.Plugin;
+import net.sf.basedb.core.signal.SignalException;
 import net.sf.basedb.core.signal.SignalHandler;
 import net.sf.basedb.core.signal.SignalTarget;
@@ -598 +599 @@
             }
             // In case the server is shutting down... throw exception, rollback and quit
-            if (Thread.interrupted()) throw new BaseException("Thread was interrupted.");
+            if (Thread.interrupted()) throw new SignalException("Aborted by user.");
           }
         }
@@ -641 +642 @@
             }
             // In case the server is shutting down... throw exception, rollback and quit
-            if (Thread.interrupted()) throw new BaseException("Thread was interrupted.");
+            if (Thread.interrupted()) throw new SignalException("Aborted by user.");
           }
         }

trunk/www/views/jobs/index.jsp

@@ -39 +39 @@
   import="net.sf.basedb.core.PermissionDeniedException"
   import="net.sf.basedb.core.ItemAlreadyExistsException"
+  import="net.sf.basedb.core.signal.SignalTransporter"
+  import="net.sf.basedb.core.signal.Signal"
   import="net.sf.basedb.util.RemovableUtil"
   import="net.sf.basedb.util.ShareableUtil"
@@ -238 +240 @@
     redirect = viewPage;
   }
+  else if ("AbortJob".equals(cmd))
+  {
+    ItemContext cc = Base.getAndSetCurrentContext(sc, itemType, pageContext, defaultContext);
+    dc = sc.newDbControl();
+    Job job = Job.getById(dc, cc.getId());
+    if (job.getStatus() == Job.Status.WAITING)
+    {
+      job.doneError("Aborted by user");
+    }
+    else
+    {
+      SignalTransporter signalTransporter = job.getSignalTransporter();
+      if (signalTransporter != null) signalTransporter.send(Signal.ABORT);
+    }
+    dc.commit();
+    Thread.sleep(500);
+    redirect = viewPage;
+  }
   else
   {

trunk/www/views/jobs/view_job.jsp

@@ -47 +47 @@
   import="net.sf.basedb.core.plugin.GuiContext"
   import="net.sf.basedb.core.plugin.Plugin"
+  import="net.sf.basedb.core.signal.SignalTransporter"
+  import="net.sf.basedb.core.signal.Signal"
   import="net.sf.basedb.clients.web.Base"
   import="net.sf.basedb.clients.web.util.HTML"
@@ -57 +59 @@
   import="java.util.List"
   import="java.util.Collections"
+  import="java.util.Collection"
 %>
 <%@ taglib prefix="base" uri="/WEB-INF/base.tld" %>
@@ -103 +106 @@
   Formatter<Date> dateFormatter = FormatterFactory.getDateFormatter(sc);
   Formatter<Date> dateTimeFormatter = FormatterFactory.getDateTimeFormatter(sc);
+  
+  // Check if the plug-in supports the "Abort" signal
+  boolean supportsAbort = status == Job.Status.WAITING;
+  if (status == Job.Status.EXECUTING)
+  {
+    try
+    {
+      SignalTransporter signalTransporter = job.getSignalTransporter();
+      Collection<Signal> supportedSignals = signalTransporter != null ? 
+        signalTransporter.getSupportedSignals() : null;
+      supportsAbort = supportedSignals == null || supportedSignals.contains(Signal.ABORT);
+    }
+    catch (Exception ex)
+    {}
+  }
   %>
 
@@ -114 +132 @@
     {
       setTimeout('location.reload()', 10000);
+    }
+  }
+  function abortJob()
+  {
+    if (confirm('Are you sure? This action may not be undone'))
+    {
+      location.href = 'index.jsp?ID=<%=ID%>&cmd=AbortJob&item_id=<%=itemId%>';
     }
   }
@@ -289 +314 @@
       </table>
       </t:tab>
-      
       <%
       if (job.getStackTrace() != null)
@@ -474 +498 @@
       %>
       <%
+      if (supportsAbort)
+      {
+        %>
+        <base:button onclick="abortJob()" title="Abort&hellip;" image="abort.png" />
+        <%
+      }
+      %>
+      <%
       if (job.getStatus() == Job.Status.ERROR && job.getJobType() == Job.Type.RUN_PLUGIN)
       {