Changeset 4078

Timestamp:

Jan 14, 2008, 12:21:58 PM (16 years ago)

Author:

Nicklas Nordborg

Message:

References #636: Kill a running job

• Signals are now working with job agents and the progress reporter.
• Lowess and intensity calculator plug-in now support signals
• Documentation has been written.

More things to do:

• Add support in rest of core plug-ins

Location:

trunk

Files:

• config/dist/base.config (modified) (1 diff)
• doc/src/docbook/appendix/base.config.xml (modified) (1 diff)
• doc/src/docbook/developerdoc/api_overview.xml (modified) (1 diff)
• doc/src/docbook/developerdoc/plugin_developer.xml (modified) (1 diff)
• doc/src/docbook/figures/uml/corelayer.signals.png (added)
• doc/src/uml/baseuml.mdzip (modified) (previous)
• src/clients/jobagent/net/sf/basedb/clients/jobagent/Agent.java (modified) (5 diffs)
• src/clients/jobagent/net/sf/basedb/clients/jobagent/AgentSignalReceiver.java (added)
• src/clients/jobagent/net/sf/basedb/clients/jobagent/executors/DummyJobExecutor.java (modified) (5 diffs)
• src/clients/jobagent/net/sf/basedb/clients/jobagent/executors/ProcessJobExecutor.java (modified) (7 diffs)
• src/clients/jobagent/net/sf/basedb/clients/jobagent/executors/ThreadJobExecutor.java (modified) (8 diffs)
• src/clients/jobagent/net/sf/basedb/clients/jobagent/handlers/DefaultRequestHandler.java (modified) (2 diffs)
• src/clients/jobagent/net/sf/basedb/clients/jobagent/handlers/SignalRequestHandler.java (added)
• src/core/net/sf/basedb/core/Application.java (modified) (1 diff)
• src/core/net/sf/basedb/core/InternalJobQueue.java (modified) (3 diffs)
• src/core/net/sf/basedb/core/Job.java (modified) (3 diffs)
• src/core/net/sf/basedb/core/PluginExecutionRequest.java (modified) (3 diffs)
• src/core/net/sf/basedb/core/signal/AbstractSignalHandler.java (modified) (4 diffs)
• src/core/net/sf/basedb/core/signal/AbstractSignalReceiver.java (modified) (9 diffs)
• src/core/net/sf/basedb/core/signal/DelegatingSignalHandler.java (modified) (8 diffs)
• src/core/net/sf/basedb/core/signal/ForwardingSignalHandler.java (added)
• src/core/net/sf/basedb/core/signal/LocalSignalReceiver.java (modified) (1 diff)
• src/core/net/sf/basedb/core/signal/ProgressReporterSignalHandler.java (added)
• src/core/net/sf/basedb/core/signal/SignalHandler.java (modified) (2 diffs)
• src/core/net/sf/basedb/core/signal/SignalReceivedException.java (added)
• src/core/net/sf/basedb/core/signal/SignalReceiver.java (modified) (2 diffs)
• src/core/net/sf/basedb/core/signal/SocketSignalReceiver.java (modified) (10 diffs)
• src/core/net/sf/basedb/core/signal/SocketSignalTransporter.java (modified) (1 diff)
• src/core/net/sf/basedb/core/signal/ThreadSignalHandler.java (modified) (6 diffs)
• src/core/net/sf/basedb/util/QueryParameters.java (modified) (1 diff)
• src/plugins/core/net/sf/basedb/plugins/AbstractFlatFileImporter.java (modified) (3 diffs)
• src/plugins/core/net/sf/basedb/plugins/IntensityCalculatorPlugin.java (modified) (6 diffs)
• src/plugins/core/net/sf/basedb/plugins/LowessNormalization.java (modified) (6 diffs)
• www/views/jobs/view_job.jsp (modified) (3 diffs)

trunk/config/dist/base.config

@@ -80 +80 @@
 # 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
+jobqueue.internal.signalreceiver.init  = jobqueue:0
 
 # If plugins with useInteralJobQueue = false should be executed or not

trunk/doc/src/docbook/appendix/base.config.xml

@@ -379 +379 @@
 
     <varlistentry>
+      <term><property>jobqueue.internal.signalreceiver.class</property></term>
+      <listitem>
+        <para>
+         A class implementing the <interfacename docapi="net.sf.basedb.core.signal"
+         >SignalReceiver</interfacename>
+         interface. The class must have a public no-argument constructor. If
+         no value is specified the default setting is:
+         <classname docapi="net.sf.basedb.core.signal"
+         >net.sf.basedb.core.signal.LocalSignalReceiver</classname>.
+         </para>
+         <para>
+         Change to <classname docapi="net.sf.basedb.core.signal"
+         >net.sf.basedb.core.signal.SocketSignalReceiver</classname>
+         if the internal job queue must be able to receive signals from outside
+         this JVM.
+         </para>
+      </listitem>
+    </varlistentry>
+    
+    <varlistentry>
+      <term><property>jobqueue.internal.signalreceiver.init</property></term>
+      <listitem>
+        <para>
+        Initialisation string sent to <methodname>SignalReceiver.init()</methodname>.
+        The syntax and meaning of the string depends on the actual implementation
+        that is used. Please see the javadoc for more information.
+         </para>
+      </listitem>
+    </varlistentry>
+
+    <varlistentry>
       <term><property>jobqueue.internal.checkinterval</property></term>
       <listitem>

trunk/doc/src/docbook/developerdoc/api_overview.xml

@@ -2806 +2806 @@
       
     </sect2>
+    
+    <sect2 id="core_api.signals">
+      <title>Sending signals (to plug-ins)</title>
+    
+      <para>
+        BASE has a simple system for sending signals between different parts of
+        a system. This signalling system was initially developed to be able to
+        kill plug-ins that a user for some reason wanted to abort. The signalling
+        system as such is not limited to this and it can be used for other purposes 
+        as well. Signals can of course be handled internally in a single JVM but
+        also sent externally to other JVM:s running on the same or a different 
+        computer. The transport mechanism for signals is decoupled from the actual
+        handling of them. If you want to, you could implement a signal transporter
+        that sends signal as emails and the target plug-in would never know.
+      </para>
+      
+      <para>
+        The remainder of this section will focus mainly on the sending and
+        transportation of signals. For more information about handling signals
+        on the receiving end, see <xref linkend="plugin_developer.signals" />.
+      </para>
+      
+      <sect3 id="core_api.signals.diagram">
+        <title>Diagram of classes and methods</title>
+        <figure id="core_api.figures.signals">
+          <title>The signalling system</title>
+          <screenshot>
+            <mediaobject>
+              <imageobject>
+                <imagedata 
+                  align="center"
+                  scalefit="1" width="100%"
+                  fileref="figures/uml/corelayer.signals.png" format="PNG" />
+              </imageobject>
+            </mediaobject>
+          </screenshot>
+        </figure>
+      
+        <para>
+          The signalling system is rather simple. An object that wish
+          to receieve signals must implement the 
+          <interfacename docapi="net.sf.basedb.core.signal"
+          >SignalTarget</interfacename>. It's only method 
+          is <methodname>getSignalHandler()</methodname>. A 
+          <interfacename docapi="net.sf.basedb.core.signal"
+          >SignalHandler</interfacename> is an object that 
+          knows what to do when a signal is delivered to it. The target object
+          may implement the <interfacename>SignalHandler</interfacename> itself
+          or use one of the existing handlers. 
+        </para>
+        
+        <para>
+          The difficult part here is to be aware that a signal is usually
+          delivered by a separate thread. The target object must be aware
+          of this and know how to handle multiple threads. As an example we
+          can use the <classname docapi="net.sf.basedb.core.signal"
+          >ThreadSignalHandler</classname> which simply
+          calls <code>Thread.interrupt()</code> to deliver a signal. The target 
+          object that uses this signal handler it must know that it should check
+          <code>Thread.interrupted()</code> at regular intervals from the main
+          thread. If that method returns true, it means that the <constant>ABORT</constant>
+          signal has been delivered and the main thread should clean up and exit as
+          soon as possible.
+        </para>
+        
+        <para>
+          Even if a signal handler could be given directly to the party
+          that may be interested in sending a signal to the target this
+          is not recommended. This would only work when sending signals
+          within the same virtual machine. The signalling system includes
+          <interfacename docapi="net.sf.basedb.core.signal"
+          >SignalTransporter</interfacename> and
+          <interfacename docapi="net.sf.basedb.core.signal"
+          >SignalReceiver</interfacename> objects that are used
+          to decouple the sending of signals with the handling of signals. The
+          implementation usually comes in pairs, for example 
+          <classname docapi="net.sf.basedb.core.signal"
+          >SocketSignalTransporters</classname> and <classname 
+          docapi="net.sf.basedb.core.signal">SocketSignalReceiver</classname>.
+        </para>
+        
+        <para>
+          Setting up the transport mechanism is usually a system responsibility.
+          Only the system know what kind of transport that is appropriate for it's current 
+          setup. Ie. should signals be delievered by TCP/IP sockets, only internally, or 
+          should a delivery mechanism based on web services be implemented? 
+          If a system wants to receive signals it must create an appropriate
+          <interfacename>SignalReceiver</interfacename> object. Within BASE the 
+          internal job queue set up it's own signalling system that can be used to
+          send signals (eg. kill) running jobs. The job agents do the same but uses
+          a different implementation. See <xref linkend="appendix.base.config.jobqueue" />
+          for more information about how to configure the internal job queue's 
+          signal receiver. In both cases, there is only one signal receiver instance
+          active in the system.
+        </para>
+        
+        <para>
+          Let's take the internal job queue as an example. Here is how it works:
+        </para>
+        
+        <itemizedlist>
+        <listitem>
+          <para>
+          When the internal job queue is started, it will also create a signal
+          receiver instance according to the settings in <filename>base.config</filename>.
+          The default is to create <classname docapi="net.sf.basedb.core.signal"
+          >LocalSignalReceiver</classname>
+          which can only be used inside the same JVM. If needed, this can
+          be changed to a <classname docapi="net.sf.basedb.core.signal"
+          >SocketSignalReceiver</classname> or any other
+          user-provided implementation.
+          </para>
+        </listitem>
+        
+        <listitem>
+          <para>
+          When the job queue has found a plug-in to execute it will check if
+          it also implements the <interfacename docapi="net.sf.basedb.core.signal"
+          >SignalTarget</interfacename>
+          interface. If it does, a signal handler is created and registered
+          with the signal receiver. This is actually done by the BASE core 
+          by calling <methodname>PluginExecutionRequest.registerSignalReceiver()</methodname>
+          which also makes sure that the the ID returned from the registration is
+          stored in the database together with the job item representing the 
+          plug-in to execute.
+          </para>
+        </listitem>
+        
+        <listitem>
+          <para>
+          Now, when the web client see's a running job which has a non-empty 
+          signal transporter property, the <guilabel>Abort</guilabel>
+          button is activated. If the user clicks this button the BASE core
+          uses the information in the database to create 
+          <interfacename docapi="net.sf.basedb.core.signal"
+          >SignalTransporter</interfacename> object. This
+          is simply done by calling <code>Job.getSignalTransporter()</code>.
+          The created signal transporter knows how to send a signal 
+          to the signal receiver it was first registered with. When the
+          signal arrives at the receiver it will find the handler for it
+          and call <code>SignalHandler.handleSignal()</code>. This will in it's turn
+          trigger some action in the signal target which soon will abort what
+          it is doing and exit.
+          </para>
+        </listitem>
+        </itemizedlist>
+        
+        
+      </sect3>
+    
+    </sect2>
+    
   </sect1>
 

trunk/doc/src/docbook/developerdoc/plugin_developer.xml

@@ -4365 +4365 @@
   </sect1>
   
+  <sect1 id="plugin_developer.signals">
+    <title>Enable support for aborting a running a plug-in</title>
+    
+    <para>
+      BASE includes a simple signalling system that can be used to send
+      signals to plug-ins. The system was primarly developed to allow a user
+      to kill a plug-in when it is executing. Therfore, the focus of this chapter
+      will be how to implement a plug-in to make it possible to kill it
+      during it's execution.
+    </para>
+    
+    <para>
+      Since we don't want to do this by brute force such as destroying the
+      process or stopping thread the plug-in executes in, cooperation is needed
+      by the plug-in. First, the plug-in must implement the 
+      <interfacename docapi="net.sf.basedb.core.signal">SignalTarget</interfacename>
+      interface. From this, a <interfacename 
+      docapi="net.sf.basedb.core.signal">SignalHandler</interfacename> can be created.
+      A plug-in may choose to implement it's own signal handler or use an existing
+      implementation. BASE, for example, provides the <classname docapi="net.sf.basedb.core.signal"
+      >ThreadSignalHandler</classname> implementation that supports the <constant>ABORT</constant> signal.
+      This is a simple implementation that just calls <code>Thread.interrupt()</code>
+      on the plug-in worker thread. This may cause two different effects:
+    </para>
+    
+    <itemizedlist>
+    <listitem>
+      <para>
+      The <code>Thread.interrupted()</code> flag is set. The plug-in must check this
+      at regular intervals and if the flag is set it must cleanup, rollback 
+      open transactions and exit as soon as possible.
+      </para>
+    </listitem>
+    
+    <listitem>
+      <para>
+      If the plug-in is waiting in a blocking call that is interruptable, for
+      example <code>Thread.sleep()</code>, an <classname>InterruptedException</classname>
+      is thrown. This should cause the same actions as if the flag was set to happen.
+      </para>
+      
+      <warning>
+        <title>Not all blocking calls are interruptable</title>
+        <para>
+        For example calling <code>InputStream.read()</code> may leave the
+        plug-in waiting in a non-interruptable state. In this case there
+        is nothing BASE can do to wake it up again.
+        </para>
+      </warning>
+    </listitem>
+    </itemizedlist>
+    
+    <para>
+      Here is a general outline for a plug-in that uses the 
+      <classname>ThreadSignalHandler</classname>.
+    </para>
+    <programlisting language="java">
+<![CDATA[
+private ThreadSignalHandler signalHandler;
+public SignalHandler getSignalHandler()
+{
+   signalHandler = new ThreadSignalHandler();
+   return signalHandler;
+}
+
+public void run(Request request, Response response, ProgressReporter progress)
+{
+   if (signalHandler != null) signalHandler.setWorkerThread(null);
+   beginTransaction();
+   boolean done = false;
+   boolean interrupted = false;
+   while (!done && !interrupted)
+   {
+      try
+      {
+         done = doSomeWork(); // NOTE! This must not take forever!
+         interrupted = Thread.interrupted();
+      }
+      catch (InterruptedException ex)
+      {
+         // NOTE! Try-catch is only needed if thread calls 
+         // a blocking method that is interruptable 
+         interrupted = true;
+      }
+   }
+   if (interrupted)
+   {
+      rollbackTransaction();
+      response.setError("Aborted by user", null);
+   }
+   else
+   {
+      commitTransaction();
+      response.setDone("Done");
+   }
+}
+]]>
+</programlisting>
+    
+    <para>
+      Another signal handler implementation is the
+      <classname docapi="net.sf.basedb.core.signal">ProgressReporterSignalHandler</classname>.
+      See that javadoc for information about how to use it.
+      For more information about the signalling system as a whole,
+      see <xref linkend="core_api.signals" />.
+    </para>
+    
+  </sect1>
+  
   <sect1 id="plugin_developer.classload">
     <title>How BASE load plug-in classes</title>

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

@@ -251 +251 @@
   private JobAgentServerConnection server;
   private RequestHandler requestHandler;
+  private AgentSignalReceiver signalReceiver;
   private JobExecutor jobExecutor;
 
@@ -275 +276 @@
   
   private final Set<JobInfo> activeJobs;
-  
+
+  /**
+    Timeout to wait for jobs to act on the ABORT signal when stopping.
+    Default value is 20 seconds.
+  */
+  private int closeTimeout = 20000;
+
   /**
     The group were all job runners are placed.
@@ -531 +538 @@
     if (server == null)
     {
+      this.signalReceiver = new AgentSignalReceiver(this);
+      this.signalReceiver.init(null);
       this.requestHandler = requestHandler == null ? new DefaultRequestHandler(this) : requestHandler;
       this.server = new JobAgentServerConnection(port, this.requestHandler, logServer);
@@ -685 +694 @@
     }
     return allow;
+  }
+  
+  /**
+    Get the signal receiver that is processing signal messages
+    on behalf of this job agent.
+    @since 2.6
+  */
+  public AgentSignalReceiver getSignalReceiver()
+  {
+    return signalReceiver;
   }
   
@@ -917 +936 @@
   
   /**
-    Try to stop running jobs by interrupting the threads thaey are executing in.
+    Try to stop running jobs by interrupting the threads they are executing in.
   */
   private void maybeStopRunningJobs()
   {
     log.info("Stopping running jobs. " + activeJobs.size() + " job(s) still active.");
-    // Interrupt all threads. Hopefully they will do as we tell them.
-    runnersGroup.interrupt();
+    
+    // Send ABORT to all jobs, that support signals
+    signalReceiver.close(closeTimeout);
   }
   

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

@@ -24 +24 @@
 package net.sf.basedb.clients.jobagent.executors;
 
+import java.util.Collections;
+
 import net.sf.basedb.clients.jobagent.Agent;
 import net.sf.basedb.clients.jobagent.JobExecutor;
@@ -30 +32 @@
 import net.sf.basedb.core.JobAgentSettings;
 import net.sf.basedb.core.SessionControl;
+import net.sf.basedb.core.signal.SignalReceiver;
+import net.sf.basedb.core.signal.ThreadSignalHandler;
 import net.sf.basedb.util.Values;
 
@@ -89 +93 @@
     log.info("Executing job: " + job);
     DbControl dc = null;
+    boolean aborted = false;
+    Throwable error = null;
     try
     {
@@ -96 +102 @@
       if (wait > 0)
       {
+        SignalReceiver signalReceiver = agent.getSignalReceiver();
+        job.setSignalTransporter(signalReceiver.getSignalTransporterClass(), 
+          signalReceiver.registerSignalHandler(new ThreadSignalHandler()));
         job.setProgress(50, "Halfway; waiting " + wait + " seconds");
         dc.commit();
@@ -105 +114 @@
           }
         }
+        catch (InterruptedException ex)
+        {
+          aborted = true;
+        }
         catch (Throwable t)
         {
           log.error(t.getMessage(), t);
+          error = t;
         }
         dc  = sc.newDbControl();
         dc.reattachItem(job);
       }
-      job.doneOk("Not really, but used for testing job agent");
+      if (aborted)
+      {
+        job.doneError("Aborted by user");
+      }
+      else if (error != null)
+      {
+        job.doneError(error.getMessage(), Collections.singleton(error));
+      }
+      else
+      {
+        job.doneOk("Not really, but used for testing job agent");
+      }
       dc.commit();
       log.info("Done executing: " + job);

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

@@ -42 +42 @@
 import net.sf.basedb.core.PluginDefinition;
 import net.sf.basedb.core.SessionControl;
+import net.sf.basedb.core.signal.Signal;
+import net.sf.basedb.core.signal.SignalHandler;
+import net.sf.basedb.core.signal.SignalReceiver;
+import net.sf.basedb.core.signal.SignalTransporter;
+import net.sf.basedb.core.signal.ThreadSignalHandler;
 import net.sf.basedb.util.Values;
 
@@ -101 +106 @@
   private String javaBin;
   private String options;
+  private SignalReceiver signalReceiver;
+  private SignalHandler signalHandler;
   
   public ProcessJobExecutor()
@@ -120 +127 @@
     if (javaBin == null) javaBin = "java";
     options = Values.getString(agent.getProperty("agent.executor.process.options"), "-server");
+    signalReceiver = agent.getSignalReceiver();
   }
   
@@ -126 +134 @@
   {
     log.info("Executing job: " + job);
-    
+
     // Generate command for new process
     List<String> cmd = new ArrayList<String>(10);
@@ -164 +172 @@
     cmd.add("-t"); // Thread priority
     cmd.add(Integer.toString(agent.getThreadPriority(job.getEstimatedExecutionTime())));
+    cmd.add("-x"); // Port the job agent is listening on for remote control
+    cmd.add(Integer.toString(agent.getPort()));
     
     Process process = null;
@@ -201 +211 @@
         new InputStreamReader(process.getInputStream()), result));
       t.start();
-
+      
       try
       {
         log.info("Waiting for process to end");
+        signalHandler = new ThreadSignalHandler();
+        signalReceiver.registerSignalHandler(signalHandler);
         int exitCode = process.waitFor();
         if (exitCode != 0)
@@ -224 +236 @@
       {
         log.info("Job was interrupted: " + job, ex);
-        // Kill the process
-        process.destroy();
+        // First, send ABORT to the job if it supports it
         dc = sc.newDbControl();
         job = Job.getById(dc, job.getId());
-        job.doneError(ex.getMessage(), Arrays.asList(ex));
+        SignalTransporter signalTransporter = job.getSignalTransporter();
+        if (signalTransporter != null)
+        {
+          signalTransporter.send(Signal.ABORT);
+        }
+        else
+        {
+          job.doneError("Aborted by user", Arrays.asList(ex));
+          process.destroy();
+        }
         dc.commit();
+      }
+      finally
+      {
+        signalReceiver.unregisterSignalHandler(signalHandler);          
       }
     }

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

@@ -39 +39 @@
 import net.sf.basedb.core.SessionControl;
 import net.sf.basedb.core.plugin.Response;
+import net.sf.basedb.core.signal.SignalReceiver;
+import net.sf.basedb.core.signal.SocketSignalReceiver;
 import net.sf.basedb.util.SocketUtil;
 import net.sf.basedb.util.Values;
@@ -65 +67 @@
     org.apache.log4j.LogManager.getLogger("net.sf.basedb.clients.jobagent.executors.ThreadJobExecutor");
   
+  private SignalReceiver signalReceiver;
   
   public ThreadJobExecutor()
@@ -74 +77 @@
   */
   public void init(Agent agent)
-  {}
+  {
+    this.signalReceiver = agent == null ? null : agent.getSignalReceiver();
+  }
   
   public void executeJob(SessionControl sc, Agent agent, Job job, JobAgentSettings settings,
@@ -99 +104 @@
         log.error("Error executing job: " + job, t);
       }
+      exec.registerSignalReceiver(signalReceiver);
       dc.commit();
 
@@ -130 +136 @@
   // -------------------------------------------
   
+  private void setSignalReceiver(SignalReceiver signalReceiver)
+  {
+    this.signalReceiver = signalReceiver;
+  }
+  
   public static void main(String[] args)
   {
@@ -138 +149 @@
     String login = cmdLine.getOption("-u");
     String password = cmdLine.getOption("-p");
+    String proxyPort = cmdLine.getOption("-x");
     int threadPriority = Values.getInt(cmdLine.getOption("-t"), Thread.NORM_PRIORITY);
 
@@ -152 +164 @@
       ThreadJobExecutor executor = new ThreadJobExecutor();
       
+      if (proxyPort != null)
+      {
+        SocketSignalReceiver signalReceiver = new SocketSignalReceiver();
+        signalReceiver.init("proxy=" + Application.getHostName() + ":" + proxyPort);
+        executor.setSignalReceiver(signalReceiver);
+        Runtime.getRuntime().addShutdownHook(new Thread(new ShutdownHook(signalReceiver)));
+      }
+      
       sc = Application.newSessionControl("net.sf.basedb.clients.jobagent", 
         SocketUtil.getLocalHost().toString(), null);
@@ -188 +208 @@
   }
   
+  private static class ShutdownHook
+    implements Runnable
+  {
+    private SignalReceiver signalReceiver;
+    
+    private ShutdownHook(SignalReceiver signalReceiver)
+    {
+      this.signalReceiver = signalReceiver;
+    }
+    public void run()
+    {
+      signalReceiver.close(5000);
+    }
+  }
 }

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

@@ -69 +69 @@
     registerHandler(new StopRequestHandler(agent), "stop");
     registerHandler(new PauseRequestHandler(agent), "pause");
+    if (agent.getSignalReceiver() != null)
+    {
+      registerHandler(new SignalRequestHandler(agent), "signal");
+    }
   }
   
@@ -89 +93 @@
     String answer = null;
     RequestHandler handler = commandHandlers.get(cmd);
+    if (handler == null && cmd != null && cmd.startsWith("signal://"))
+    {
+      handler = commandHandlers.get("signal");
+    }
     if (!agent.isAllowedControl(remote, cmd))
     {

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

@@ -568 +568 @@
     // Stop timer and clean the cache of SessionControl objects
     if (timer != null) timer.cancel();
+    if (internalJobQueue != null) internalJobQueue.close();
+    if (secondaryStorageController != null) secondaryStorageController.close();
     timer = null;
     scheduler = null;
     cleanSessionControlCache(true);
     sessionCache = null;
-    if (internalJobQueue != null) internalJobQueue.close();
-    if (secondaryStorageController != null) secondaryStorageController.close();
 
     Keyring.unload();

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

@@ -203 +203 @@
   private boolean ignoreUseInternalJobQueueFlag = false;
 
+  /**
+    Timeout to wait for jobs to act on the ABORT signal when shutting down BASE.
+    Default value is 20 seconds.
+  */
+  private int closeTimeout = 20000;
+  
   /**
     Use the local signal receiver only.
@@ -381 +387 @@
     sc.logout();
 
-    // Interrupt all threads. Hopefully they will do as we tell them.
-    runnersGroup.interrupt();
+    // Send ABORT to all jobs that supports signalling
+    signalReceiver.close(closeTimeout);
   }
   
@@ -454 +460 @@
           throw t;
         }
-        exec.registerSignalReceiver(signalReceiver, false);
+        exec.registerSignalReceiver(signalReceiver);
         dc.commit();
         

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

@@ -592 +592 @@
   }
   
+  /**
+    Set information needed to create a signal transporter object that can be used
+    to send signals to the plug-in that is currently executing this job.
+    @param clazz The signal transporter class
+    @param initParams Initialisation parameters that will be passed
+      to {@link SignalTransporter#init(String)}
+    @throws PermissionDeniedException If the user doesn't have write
+      permission for the job
+    @since 2.6
+  */
   public void setSignalTransporter(Class<? extends SignalTransporter> clazz, String initParams)
   {
@@ -600 +610 @@
   }
   
+  /**
+    Checks if a signal transporter has been registered for this job or not.
+    @return TRUE if a signal transporter is registered, FALSE otherwise
+    @since 2.6
+  */
   public boolean hasSignalTransporter()
   {
@@ -605 +620 @@
   }
   
+  /**
+    Create and initialise a signal transporter object that can be used
+    to send signal to the plug-in running this job. Note! Depending on
+    the transport mechanism used, it is not certain that the created
+    transporter can actually send the signal. For example, some transporters
+    only work if the plug-in is running in the same virtual machine, while others
+    supports sending signals to other JVM:s on the same or a different
+    computer.
+    @return An initialised signal transporter object, or null
+      if no signal transporter has been registered
+    @throws PermissionDeniedException If the user doesn't have write
+      permission for the job
+    @since 2.6
+  */
   public SignalTransporter getSignalTransporter()
   {
+    checkPermission(Permission.WRITE);
     String tmp = getData().getSignalTransporter();
     SignalTransporter transporter = null;

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

@@ -66 +66 @@
   private Job.ProgressReporterImpl progress;
   private SignalReceiver signalReceiver = null;
-  private boolean forceSignalReceiver = false;
   
   PluginExecutionRequest(SessionControl sc, Plugin plugin, String command,
@@ -100 +99 @@
         // 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)
         {
@@ -158 +152 @@
     
     @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)
+  public void registerSignalReceiver(SignalReceiver signalReceiver)
   {
     this.signalReceiver = signalReceiver;
-    this.forceSignalReceiver = force;
   }
   

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

@@ -54 +54 @@
   protected AbstractSignalHandler()
   {
-    this.supported = new HashSet<Signal>();
+    this(null);
   }
-  
+
   /**
     Create a new signal handler that supports the given signals.
-    @param supported A set with the signals that are supported
+    
+    @param supported A collection with the signals that are initially supported.
+      More signals can be added with {@link #addSignal(Signal)}
   */
-  protected AbstractSignalHandler(Set<Signal> supported)
+  protected AbstractSignalHandler(Collection<Signal> supported)
   {
-    this.supported = new HashSet<Signal>(supported);
+    this.supported = new HashSet<Signal>();
+    if (supported != null) this.supported.addAll(supported);
   }
   
@@ -78 +81 @@
     return supported != null && supported.contains(signal);
   }
-  public SignalReceiver getSignalReceiver()
-  {
-    return null;
-  }
   // -------------------------------------------
 
@@ -88 +87 @@
     @param signal The signal to add
   */
-  public void addSignal(Signal signal)
+  protected void addSignal(Signal signal)
   {
     if (signal != null) supported.add(signal);
@@ -96 +95 @@
     @param signal The signal to remove
   */
-  public void removeSignal(Signal signal)
+  protected void removeSignal(Signal signal)
   {
     supported.remove(signal);

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

@@ -25 +25 @@
 
 import java.net.URI;
+import java.util.ArrayList;
 import java.util.Collection;
 import java.util.Collections;
@@ -42 +43 @@
     subclass. 
     
-  <li>The <i>handlerId</i> part is given by calling {@link System#identityHashCode(Object)} 
-    for the signal handler. 
+  <li>The <i>handlerId</i> part is given by calling {@link #getLocalSignalHandlerId(SignalHandler)}.
+    This method may be overridden by subclasses that needs a different method of
+    local ID generation. The important thing is that the ID is unique among all
+    registered signal handlers.
     
   <li>The <i>supportedSignals</i> part is created by joining all signal ID:s 
@@ -76 +79 @@
   private Map<String, SignalHandler> handlers;
   private String receiverId;
+  private Thread notifyThread;
   
   /**
@@ -99 +103 @@
   }
   /**
-    Close this signal receiver.
-  */
-  public void close()
+    Close this signal receiver. If a subclass overrides this method it
+    should call <code>super.close()</code> if it wants to
+    use the default shutdown notification to signal handlers. This
+    implementation will start a separate notification thread that sends
+    the {@link Signal#ABORT} signal to all registered handlers. The current 
+    thread will wait at most the given time. If the number of registered handlers
+    goes down to 0 before the time has ended the current thread will be awakened
+    so it can continue. Note that the notification thread will usually finish in 
+    a short time, but it may take longer for all worker thread of each job to
+    react to the signal.
+    @param wait Number of milliseconds to wait for registered jobs to abort,
+      use a negative value to disable notification, and 0 to disable waiting
+  */
+  public void close(int wait)
   {
     logger.info("Closing signal receiver: receiver id=" + receiverId);
-    if (handlers != null) handlers.clear();
+    if (handlers == null) return;
+    if (wait >= 0 && handlers.size() > 0)
+    {
+      Thread abortAll = new Thread(
+        new Runnable()
+        {
+          public void run()
+          {
+            logger.info("Sending ABORT to all handlers");
+            sendToAll(Signal.ABORT);
+          }
+        }
+      );
+      abortAll.setDaemon(true);
+      abortAll.start();
+      if (wait > 0)
+      {
+        try
+        {
+          logger.info("Sleeping for " + wait + " milliseconds");
+          notifyThread = Thread.currentThread();
+          Thread.sleep(wait);
+          notifyThread = null;
+        }
+        catch (InterruptedException ex)
+        {
+          // Good, all handlers got the signal before timeout
+        }
+      }
+      logger.info("Remaining active handlers: " + handlers.size());
+    }
+    handlers.clear();
     handlers = null;
   }
@@ -115 +161 @@
   public String registerSignalHandler(SignalHandler handler)
   {
+    if (notifyThread != null) throw new SignalException("Signal receiver is shutting down.");
     String globalId = getGlobalSignalId(handler);
     String localId = getLocalSignalHandlerId(handler);
-    if (handlers != null) handlers.put(localId, handler);
+    if (handlers != null) 
+    {
+      if (handlers.containsKey(localId))
+      {
+        throw new SignalException("A handler with ID '" + localId + "' is already registered: "
+          + handlers.get(localId));
+      }
+      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());
+    logger.debug("Current number of registered signal handlers: " + 
+      (handlers == null ? 0 : handlers.size()));
     return globalId;
   }
@@ -133 +190 @@
     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());
+    if (handlers != null) 
+    {
+      handlers.remove(localId);
+      // If called from the close() method() this is waiting for
+      if (notifyThread != null && handlers.size() == 0)
+      {
+        logger.info("Interrupting closedown thread");
+        notifyThread.interrupt();
+      }
+    }
+    logger.debug("Current number of registered signal handlers: " + 
+      (handlers == null ? 0 : handlers.size()));
+  }
+  
+  public void sendToAll(Signal signal)
+  {
+    if (handlers == null) return;
+    logger.info("Sending " + signal + " to " + (handlers.size()) + " handlers");
+    // Copy the signal handlers to a temporary collection,
+    // since sending a signal may cause a handler to get 
+    // unregistered, which may fail or block due to synchronization
+    Collection<SignalHandler> temp = null;
+    synchronized (handlers)
+    {
+      temp = new ArrayList<SignalHandler>(handlers.values());
+    }
+    for (SignalHandler handler : temp)
+    {
+      if (handler.supports(signal)) handler.handleSignal(signal);
+    }
   }
   // -------------------------------------------
 
+  /**
+    Get the receiver ID that was passed to the {@link #init(String)}
+    method.
+  */
+  protected String getReceiverId()
+  {
+    return receiverId;
+  }
+  
   /**
     Generate a signal ID string. This string is returned by
@@ -144 +238 @@
     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.
+    generated string. The string is of the format:
+    <p>
+    <code>signal://handlerId@receiverId/?supportedSignals</code>
+    <p>
+    See class description for detailed information.
     
     @param handler The signal handler to generate the ID for
@@ -154 +252 @@
     sb.append("signal://");
     sb.append(getLocalSignalHandlerId(handler)).append("@");
-    sb.append(receiverId).append("/?");
+    sb.append(getReceiverId()).append("/?");
     Collection<Signal> signals = handler.getSupportedSignals();
     if (signals != null)
@@ -172 +270 @@
     Get the local signal handler id of the given signal handler.
     This implementation simply return the system hashcode for the
-    handler.
+    handler. The returned ID must be unique among the registered
+    signal handlers.
     @param handler The handler to get the id for
     @return The local handler id

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

@@ -26 +26 @@
 import java.util.Collection;
 import java.util.Collections;
-import java.util.HashMap;
 import java.util.HashSet;
 import java.util.Iterator;
@@ -43 +42 @@
 */
 public class DelegatingSignalHandler
-  implements SignalHandler
+  extends AbstractSignalHandler
 {
   /**
@@ -57 +56 @@
   
   /**
-    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 
@@ -68 +62 @@
   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;
+    super();
   }
   
@@ -92 +73 @@
       registered handler
   */
+  @Override
   public Collection<Signal> getSupportedSignals()
   {
@@ -100 +82 @@
     Check if at least one handler has been registered for the given signal.
   */
+  @Override
   public boolean supports(Signal signal)
   {
@@ -126 +109 @@
       handler.handleSignal(signal);
     }
-  }
-  
-  public SignalReceiver getSignalReceiver()
-  {
-    return signalReceiver;
   }
   // -------------------------------------------
@@ -174 +152 @@
     }
   }
-  
 
 }

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

@@ -90 +90 @@
   
   @Override
-  public void close()
+  public void close(int wait)
   {
+    super.close(wait);
     receivers.remove(receiverId);
-    super.close();
   }
   

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

@@ -34 +34 @@
   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.
+  Signal handler implementations need to be implemented in a 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
@@ -63 +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

@@ -57 +57 @@
     Close the receiver. The receiver should close and cleanup any opened
     resources, unregister itself if neccesary and stop listening for signals.
+
+    @param wait If the value is zero or positive, the {@link Signal#ABORT}
+      should be sent to all registered signal handlers. The signal receiver 
+      should then wait at most the specified number of milliseconds for
+      all signal handlers to get unregistered. If the timeout expires before
+      all signal handlers has processed the signal, the signal receiver should
+      continue it's shutdown.
   */
-  public void close();
+  public void close(int wait);
   
   /**
@@ -91 +98 @@
   */
   public void unregisterSignalHandler(SignalHandler handler);
+  
+  /**
+    Send a signal to all registered signal handlers that supports
+    it. The main purpose of this method is to be able to send
+    the {@link Signal#ABORT} when the system is shutting down.
+    @param signal The signal to send
+  */
+  public void sendToAll(Signal signal);
+  
 }

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

@@ -29 +29 @@
 import java.net.ServerSocket;
 import java.net.Socket;
+import java.net.URI;
+import java.net.URISyntaxException;
 import java.net.UnknownHostException;
 import java.nio.channels.ClosedByInterruptException;
@@ -49 +51 @@
   
   <p>
-  <code>port=xx&allow=ip-address&allow=ip-address...</code>
+  <code>port=xx&proxy=host:port&forward=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.
+  where:
+
+  <ul>
+  <li>
+    The <code>port</code> is the port number the signal receiver will listen on.
+    This is optional and if not given a random non-used port will be used.
+  
+  <li>
+    The <code>proxy</code> is the address of a proxy host that will route
+    incoming messages to this signal receiver. The proxy will automatically
+    be added to the list of allowed hosts. This value is optional.
+    
+  <li>
+    The <code>forward</code> attribute is a boolean option (1 or 0).
+    If 1, this signal receiver will act as a proxy and forward signal
+    messages which are of the form:
+    <p>
+    <code><code>signal://handlerID@host:port/?SIGNAL#forward.to.host:port</code>
+    <p>
+    Before the message is forwarded it is re-built like this:
+    <p>
+    <code>signal://handlerId@forward.to.host:port/?SIGNAL</code>
+    
+  <li>
+    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 <code>allow</code> tags are given only allow connections from the
+    local host (and the proxy if given) are allowed.  
+  </ul>
   
   @author nicklas
@@ -77 +101 @@
 
   private InetAddress ip;
+  private Thread listener;
   private int port;
-  private Thread listener;
+  private String proxy;
+  private boolean allowForward;
   private Set<InetAddress> allow;
   private boolean allowAll;
@@ -101 +127 @@
   public void init(String params)
   {
+    logger.info("Initialising: params=" + params);
+    parseInitParameters(params);
+    listenOnSocket();
+    super.init(Application.getHostName() + ":" + port);
+  }
+
+  /**
+    @return {@link SocketSignalTransporter}
+  */
+  public Class<? extends SignalTransporter> getSignalTransporterClass()
+  {
+    return SocketSignalTransporter.class;
+  }
+
+  /**
+    Close this receiver and the socket it is listening on.
+  */
+  public void close(int wait)
+  {
+    if (listener != null) 
+    {
+      logger.info("Interrupting socket signal receiver on port: " + port);
+      listener.interrupt();
+    }
+    super.close(wait);
+  }
+  // -------------------------------------------
+
+  /*
+    From the AbstractSignalReceiver interface
+    -------------------------------------------
+  */
+  /**
+    Override ID generation so we can add proxy information if needed.
+  */
+  @Override
+  protected String getGlobalSignalId(SignalHandler handler)
+  {
+    String globalId = super.getGlobalSignalId(handler);
+    if (proxy != null)
+    {
+      try
+      {
+        URI uri = new URI(globalId);
+        globalId = "signal://" + uri.getUserInfo() + "@" + proxy + 
+          "/?" + uri.getQuery() + "#" + uri.getHost();
+        if (uri.getPort() >= 0) globalId += ":" + uri.getPort();
+      }
+      catch (URISyntaxException ex)
+      {
+        // Should not happen
+      }
+    }
+    return globalId;
+  }
+  /**
+    Process the incoming message. This class will check if the
+    message contains any forwarding information. If so, the 
+    message will be forwarde to the new destination. Otherwise, 
+    the message will be processed as normal.
+  */
+  @Override
+  protected void processSignalMessage(String message)
+  {
+    logger.info("Processing signal message: " + message);
+    boolean forwarded = false;
+    if (allowForward) forwarded = forwardSignalMessageIfNeeded(message);
+    if (!forwarded) super.processSignalMessage(message);
+  }
+  // -------------------------------------------
+
+  /**
+    Parse the initialisation parameters. Called from the {@link #init(String)}
+    method.
+    @param params Parameter string passed to {@link #init(String)} method
+  */
+  protected void parseInitParameters(String params)
+  {
     QueryParameters qp = QueryParameters.parseQueryString(params);
     // Get the port number we listen on
     port = Values.getInt(qp.getValue("port"), 0);
+    logger.info("port=" + port);
+
+    // Are we allowed to forward the messages
+    allowForward = Values.getBoolean(qp.getValue("forward"));
+    logger.info("forward=" + allowForward);
     
     // Get allowed hosts
@@ -113 +222 @@
       for (String ip : allowIp)
       {
+        logger.info("allow=" + ip);
         if ("*".equals(ip)) 
         {
@@ -133 +243 @@
     }
     
+    // Get proxy
+    proxy = qp.getValue("proxy");
+    logger.info("proxy=" + proxy);
+    if (proxy != null)
+    {
+      try
+      {
+        URI proxyURI = new URI("singal://" + proxy);
+        allow.add(InetAddress.getByName(proxyURI.getHost()));
+      }
+      catch (Exception ex)
+      {
+        throw new SignalException("Invalid proxy: " + proxy, ex);
+      }
+    }
+  } 
+  
+  /**
+    Start listening for incoming signals.
+  */
+  protected void listenOnSocket()
+  {
     logger.info("Starting socket signal receiver on port: " + port);
     try
@@ -142 +274 @@
       ip = socket.getInetAddress();
       listener = new Thread(new ListenerThread(channel), "ListenerThread."+port);
+      listener.setDaemon(true);
       listener.start();
       logger.info("Socket signal receiver is listening on port: " + port);
@@ -150 +283 @@
       throw new SignalException(ex);
     }
-    super.init(Application.getHostName() + ":" + port);
-  }
-
-  /**
-    @return {@link SocketSignalTransporter}
-  */
-  public Class<? extends SignalTransporter> getSignalTransporterClass()
-  {
-    return SocketSignalTransporter.class;
-  }
-
-  /**
-    Close this receiver and the socket it is listening on.
-  */
-  public void close()
-  {
-    logger.info("Interrupting socket signal receiver on port: " + port);
-    listener.interrupt();
-  }
-  // -------------------------------------------
-
+  }
+  
+  /**
+    If the message has forwarding information, forward the message to
+    the other host. 
+    @return TRUE if the message was forwarded, FALSE otherwise
+  */
+  protected boolean forwardSignalMessageIfNeeded(String message)
+  {
+    if (message == null) return false;
+    try
+    {
+      URI uri = new URI(message);
+      String fragment = uri.getFragment();
+      if (fragment != null)
+      {
+        Signal signal = Signal.getSignal(uri.getQuery());
+        String init = "signal://" + uri.getUserInfo() + "@" + fragment;
+        logger.info("Forwarding signal message: message=" + message + "; to=" + init);
+        SignalTransporter forwardTo = new SocketSignalTransporter();
+        forwardTo.init(init);
+        forwardTo.send(signal);
+        return true;
+      }
+    }
+    catch (Exception ex)
+    {
+      // Ignore invalid messages
+      logger.warn("Could not process signal message: " + message, ex);
+    }
+    return false;
+  }
+  
+  /**
+    Checks if the specified remote host is allowed to send
+    signals to this receiver.
+    @param remoteHost The remote host that is sending the signal
+    @return TRUE if the host is allowed, FALSE otherwise
+  */
+  protected boolean isAllowedHost(InetAddress remoteHost)
+  {
+    return allowAll || SocketUtil.isLocalHost(remoteHost) || allow.contains(remoteHost);
+  }
+  
   
   /**
@@ -206 +362 @@
           
           // Check if we are allowed to accept connections from the remote host
-          if (allowAll || SocketUtil.isLocalHost(remoteHost) || allow.contains(remoteHost))
+          if (isAllowedHost(remoteHost))
           {
             // Read the incoming message
@@ -235 +391 @@
     // -------------------------------------------
   }
+
   
 }

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

@@ -66 +66 @@
     int port = uri.getPort();
     String message = generateSignalMessage(signal);
+    if (uri.getFragment() != null) message += "#" + uri.getFragment(); // For proxy support
     logger.debug("The message is: " + message);
     Socket s = null;

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

@@ -44 +44 @@
   <pre class="code">
 // ... code in worker thread
+threadSignalHandler.setWorkerThread(null);
 beginTransaction();
 boolean done = false;
@@ -91 +92 @@
     received.
   */
-  private final Thread workerThread;
+  private Thread workerThread;
+  
+  /**
+    If TRUE, call Thread.stop() instead of Thread.interrupt()
+  */
+  private boolean forceStop;
   
   /**
@@ -104 +110 @@
   /**
     Create a new thread signal handler.
-    @param thread The worker thread to interrupt when it receieves the 
+    @param workerThread The worker thread to interrupt when it receieves the 
       {@link Signal#ABORT} signal, or null to interrupt the current thread
   */
-  public ThreadSignalHandler(Thread thread)
+  public ThreadSignalHandler(Thread workerThread)
   {
     super(supported);
-    this.workerThread = thread == null ? Thread.currentThread() : thread;
+    setWorkerThread(workerThread);
   }
   
@@ -127 +133 @@
       {@link Signal#ABORT}
   */
+  @SuppressWarnings("deprecation")
   public void handleSignal(Signal signal)
   {
@@ -134 +141 @@
     {
       logger.debug("Sending signal " + signal + " to thread: " + workerThread);
-      workerThread.interrupt();
+      if (forceStop)
+      {
+        workerThread.stop(new InterruptedException("Received signal: ABORT"));
+      }
+      else
+      {
+        workerThread.interrupt();
+      }
     }
     else
@@ -144 +158 @@
   // -------------------------------------------
 
+  /**
+    Set the worker thread that should be interrupted when a signal is
+    receiver.
+    @param workerThread The worker thread, or null to use the current thread
+  */
+  public void setWorkerThread(Thread workerThread)
+  {
+    this.workerThread = workerThread == null ? Thread.currentThread() : workerThread;
+  }
+  
+  /**
+    Call this method with a true value to make the signal handler
+    use {@link Thread#stop(Throwable)} instead of {@link Thread#interrupt()}.
+    Note that the <code>Thread.stop</code> method is deprecated and it is not
+    recommended that it is used. For some plug-ins though this may be the
+    only option, if they are mainly executing code outside of their control
+    which doesn't check the {@link Thread#interrupted()} status. The exception
+    passed to the thread will be a {@link InterruptedException}.
+    @param forceStop TRUE to stop the thread by force, FALSE to signal to
+      it with a flag
+  */
+  public void setForceStop(boolean forceStop)
+  {
+    this.forceStop = forceStop;
+  }
+  
 }

trunk/src/core/net/sf/basedb/util/QueryParameters.java

@@ -63 +63 @@
     if (query != null)
     {
-      Pattern p = Pattern.compile("(.*)=(.*)&?");
+      Pattern p = Pattern.compile("([^=]+)=([^&]+)&?");
       Matcher m = p.matcher(query);
       while (m.find())

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

@@ -413 +413 @@
   private int skippedLines;
   private ClassMapErrorHandler errorHandler;
+  private ThreadSignalHandler signalHandler;
 
   public AbstractFlatFileImporter()
@@ -449 +450 @@
   public void run(Request request, Response response, ProgressReporter progress)
   {
+    if (signalHandler != null) signalHandler.setWorkerThread(null);
     String command = request.getCommand();
     if (command.equals(Request.COMMAND_EXECUTE))
@@ -678 +680 @@
   public SignalHandler getSignalHandler()
   {
-    return new ThreadSignalHandler();
+    signalHandler = new ThreadSignalHandler();
+    return signalHandler;
   }
   // -------------------------------------------

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

@@ -57 +57 @@
 import net.sf.basedb.core.query.Hql;
 import net.sf.basedb.core.query.Orders;
+import net.sf.basedb.core.signal.ProgressReporterSignalHandler;
+import net.sf.basedb.core.signal.Signal;
+import net.sf.basedb.core.signal.SignalHandler;
+import net.sf.basedb.core.signal.SignalReceivedException;
+import net.sf.basedb.core.signal.SignalTarget;
 import net.sf.basedb.util.IntensityCalculator;
 import net.sf.basedb.util.IntensityCalculatorUtil;
@@ -87 +92 @@
 public class IntensityCalculatorPlugin
   extends AbstractPlugin
-  implements InteractivePlugin 
+  implements InteractivePlugin, SignalTarget
 {
 
@@ -127 +132 @@
 
   private RequestInformation configureJob;
+  private ProgressReporterSignalHandler signalHandler;
   
   /**
@@ -180 +186 @@
   public void run(Request request, Response response, ProgressReporter progress)
   {
+    if (signalHandler != null) 
+    {
+      signalHandler.forwardTo(progress);
+      progress = signalHandler;
+    }
     DbControl dc = sc.newDbControl();
     try
@@ -263 +274 @@
       if (progress != null) progress.display(100, "Done");
       response.setDone("Done");
+    }
+    catch (SignalReceivedException ex)
+    {
+      response.setError("Aborted by user", Arrays.asList(ex));
     }
     catch (Throwable t)
@@ -412 +427 @@
     }
   }
+  // -------------------------------------------
+  /*
+    From the SignalTarget interface
+    -------------------------------------------
+  */
+  public SignalHandler getSignalHandler()
+  {
+    signalHandler = new ProgressReporterSignalHandler(Collections.singleton(Signal.ABORT));
+    return signalHandler;
+  }
+  // -------------------------------------------
 
   

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

@@ -67 +67 @@
 import net.sf.basedb.core.query.SqlResult;
 import net.sf.basedb.core.query.WhenStatement;
+import net.sf.basedb.core.signal.ProgressReporterSignalHandler;
+import net.sf.basedb.core.signal.Signal;
+import net.sf.basedb.core.signal.SignalHandler;
+import net.sf.basedb.core.signal.SignalReceivedException;
+import net.sf.basedb.core.signal.SignalTarget;
 import net.sf.basedb.util.Values;
 
@@ -87 +92 @@
 public class LowessNormalization
   extends AbstractAnalysisPlugin
-  implements InteractivePlugin
+  implements InteractivePlugin, SignalTarget
 {
 
@@ -165 +170 @@
   private RequestInformation configureJob;
 
+  private ProgressReporterSignalHandler signalHandler;
+  
   /*
     From the Plugin interface
@@ -201 +208 @@
     if (command.equals(Request.COMMAND_EXECUTE))
     {
+      if (signalHandler != null) 
+      {
+        signalHandler.forwardTo(progress);
+        progress = signalHandler;
+      }
       DbControl dc = null;
       try
@@ -224 +236 @@
         response.setDone(normalizedSpots + " spots normalized, " + (source.getNumSpots() - normalizedSpots) + " spots removed");
       }
+      catch (SignalReceivedException ex)
+      {
+        response.setError("Aborted by user", Arrays.asList(ex));
+      }
       catch (Throwable ex)
       {
@@ -305 +321 @@
       }
     }
+    catch (SignalReceivedException ex)
+    {
+      response.setError("Aborted by user", Arrays.asList(ex));
+    }
     catch (Throwable ex)
     {
       response.setError(ex.getMessage(), Arrays.asList(ex));
     }
+  }
+  // -------------------------------------------
+
+  /*
+    From the SignalTarget interface
+    -------------------------------------------
+  */
+  public SignalHandler getSignalHandler()
+  {
+    signalHandler = new ProgressReporterSignalHandler(Collections.singleton(Signal.ABORT));
+    return signalHandler;
   }
   // -------------------------------------------

trunk/www/views/jobs/view_job.jsp

@@ -108 +108 @@
   
   // Check if the plug-in supports the "Abort" signal
-  boolean supportsAbort = status == Job.Status.WAITING;
-  if (status == Job.Status.EXECUTING)
+  boolean supportsAbort = status == Job.Status.WAITING && writePermission;
+  if (status == Job.Status.EXECUTING && writePermission)
   {
     try
@@ -116 +116 @@
       Collection<Signal> supportedSignals = signalTransporter != null ? 
         signalTransporter.getSupportedSignals() : null;
-      supportsAbort = supportedSignals == null || supportedSignals.contains(Signal.ABORT);
+      supportsAbort = signalTransporter != null && 
+        (supportedSignals == null || supportedSignals.contains(Signal.ABORT));
     }
     catch (Exception ex)
@@ -172 +173 @@
   </base:head>
   <base:body onload="autoUpdate()">
-    
     <h3 class="docked"><%=title%> <base:help tabcontrol="main" /></h3>
     <t:tabcontrol id="main" active="<%=tab%>" position="bottom" contentstyle="<%="height: "+(int)(scale*320)+"px;"%>">