babeldoc-devel Mailing List for Babeldoc: Universal Document Processor (Page 96)

SourceForge Headquarters 1320 Columbia Street Suite 310 San Diego, CA 92101 +1 (858) 422-6466

Update of /cvsroot/babeldoc/babeldoc/modules/core/src/com/babeldoc/core/pipeline
In directory sc8-pr-cvs1:/tmp/cvs-serv4858/modules/core/src/com/babeldoc/core/pipeline

Modified Files:
	IPipelineStageFactory.java PipelineStageFactory.java 
Log Message:
New pipeline stage factory threading model has been implemented.

Index: IPipelineStageFactory.java
===================================================================
RCS file: /cvsroot/babeldoc/babeldoc/modules/core/src/com/babeldoc/core/pipeline/IPipelineStageFactory.java,v
retrieving revision 1.8
retrieving revision 1.9
diff -C2 -d -r1.8 -r1.9
*** IPipelineStageFactory.java	27 Jun 2003 02:19:58 -0000	1.8
--- IPipelineStageFactory.java	16 Jul 2003 22:45:27 -0000	1.9
***************
*** 70,74 ****
  
  import java.util.Collection;
- import java.util.HashMap;
  import java.util.Map;
  
--- 70,73 ----
***************
*** 90,104 ****
    public String[] getAllPipelineStageNames() throws PipelineException;
  
    /**
!    * Must this pipeline stage be ignored
!    *
!    * @param stageName stage to determine ignorance
!    *
!    * @return true - track the document
     *
!    * @throws com.babeldoc.core.pipeline.PipelineException
     */
! 
!   // public boolean isStageIgnored(String stageName) throws PipelineException;
  
    /**
--- 89,99 ----
    public String[] getAllPipelineStageNames() throws PipelineException;
  
+ 
    /**
!    * Get the resolver.
     *
!    * @return a reference to the pipeline stage resolver
     */
!   public IPipelineStageResolver getResolver();
  
    /**
***************
*** 158,173 ****
    public PipelineStageConnection[] getPipelineStageConnections(
      String sourceStage, String sinkStage) throws PipelineException;
- 
-   /**
-    * Returns the pipeline stage type for this particular pipeline
-    *
-    * @param stageName
-    *
-    * @return
-    *
-    * @throws com.babeldoc.core.pipeline.PipelineException
-    */
-   public PipelineStageType getPipelineStageType(String stageName)
-       throws PipelineException;
  
    /**
--- 153,156 ----

Index: PipelineStageFactory.java
===================================================================
RCS file: /cvsroot/babeldoc/babeldoc/modules/core/src/com/babeldoc/core/pipeline/PipelineStageFactory.java,v
retrieving revision 1.15
retrieving revision 1.16
diff -C2 -d -r1.15 -r1.16
*** PipelineStageFactory.java	16 Jul 2003 06:04:26 -0000	1.15
--- PipelineStageFactory.java	16 Jul 2003 22:45:27 -0000	1.16
***************
*** 69,81 ****
  import com.babeldoc.core.LogService;
  import com.babeldoc.core.Named;
  import com.babeldoc.core.journal.IJournalTicket;
- import com.babeldoc.core.journal.JournalException;
- import com.babeldoc.core.journal.JournalFactory;
  
  import java.util.Collection;
- import java.util.HashMap;
  import java.util.Map;
- import java.util.Collections;
- import java.util.Vector;
  
  
--- 69,79 ----
  import com.babeldoc.core.LogService;
  import com.babeldoc.core.Named;
+ import com.babeldoc.core.GeneralException;
+ import com.babeldoc.core.pipeline.processor.IPipelineStageProcessor;
+ import com.babeldoc.core.pipeline.processor.PipelineStageProcessorFactory;
  import com.babeldoc.core.journal.IJournalTicket;
  
  import java.util.Collection;
  import java.util.Map;
  
  
***************
*** 98,109 ****
    private Map options;
  
!   /** Hold the constructed stages */
!   // this is put in a ThreadLocal, since with Threads the same stage can be called by several Threads at once
!   private ThreadLocal localStages = new ThreadLocal() {
!     protected synchronized Object initialValue() {
!       getLog().logDebug(I18n.get("019005", "new"));
!       return new HashMap();
!     }
!   };
  
    /**
--- 96,101 ----
    private Map options;
  
!   /** reference to the pipeline stage processor */
!   private IPipelineStageProcessor processor;
  
    /**
***************
*** 173,177 ****
--- 165,178 ----
    public void setOptions(Map options)
        throws PipelineException {
+ 
      this.options = options;
+ 
+     // Setup the processor
+     try {
+       this.processor = PipelineStageProcessorFactory.getProcessor(this,
+           options.get(IPipelineStageProcessor.PROCESSOR));
+     } catch (GeneralException e) {
+       throw new PipelineException("", e);
+     }
    }
  
***************
*** 186,229 ****
  
    /**
-    * Get and instantiate the named pipeline stage.
-    *
-    * @param stageName the unique name of the stage
-    *
-    * @return a reference to the pipeline stage
-    *
-    * @throws PipelineException DOCUMENT ME!
-    */
-   public IPipelineStage getPipelineStage(String stageName)
-       throws PipelineException {
-     Map stages = (Map)localStages.get();
- 
-     if (!stages.containsKey(stageName)) {
-       getLog().logDebug(I18n.get("019005", stageName));
- 
-       try {
-         PipelineStageType pipelineStageType = getPipelineStageType(stageName);
-         Class stage = pipelineStageType.getTypeClass();
- 
-         if (stage != null) {
-           IPipelineStage pstage = (IPipelineStage) (stage.newInstance());
-           pstage.setName(stageName);
-           pstage.setResolver(getResolver());
-           pstage.setPipelineName(this.getName());
-           stages.put(stageName, pstage);
-         } else {
-           throw new PipelineException(I18n.get("019010",
-               pipelineStageType.getTypeName()));
-         }
-       } catch (InstantiationException ie) {
-         throw new PipelineException(ie.getMessage(), ie);
-       } catch (IllegalAccessException iae) {
-         throw new PipelineException(iae.getMessage(), iae);
-       }
-     }
- 
-     return (IPipelineStage) stages.get(stageName);
-   }
- 
-   /**
     * Get all the pipieline stage connections between stages
     *
--- 187,190 ----
***************
*** 241,258 ****
  
    /**
-    * Returns the pipeline stage type for this particular pipeline
-    *
-    * @param stageName
-    *
-    * @return
-    *
-    * @throws PipelineException
-    */
-   public PipelineStageType getPipelineStageType(String stageName)
-       throws PipelineException {
-     return this.getResolver().getPipelineStageType(stageName);
-   }
- 
-   /**
     * Set the resolver.
     *
--- 202,205 ----
***************
*** 266,270 ****
     * Get the resolver.
     *
!    * @return DOCUMENT ME!
     */
    public IPipelineStageResolver getResolver() {
--- 213,217 ----
     * Get the resolver.
     *
!    * @return a reference to the pipeline stage resolver
     */
    public IPipelineStageResolver getResolver() {
***************
*** 281,285 ****
    public boolean isStageTracked(String stageName) {
      try {
!       return this.getPipelineStage(stageName).isTracked();
      } catch (Exception e) {
      }
--- 228,232 ----
    public boolean isStageTracked(String stageName) {
      try {
!       return getProcessor().getPipelineStage(stageName).isTracked();
      } catch (Exception e) {
      }
***************
*** 302,471 ****
    public void process(String name, PipelineDocument document,
      IJournalTicket ticket, Collection finalResults) throws PipelineException {
!     if ((name != null) && !name.equals("null") && (document != null)) {
!       getLog().logInfo(I18n.get("019008", name));
! 
!       // Get the pipeline stage for this name
!       IPipelineStage pstage = this.getPipelineStage(name);
! 
!       if (pstage != null) {
!         try {
!           pstage.initialize();
!         } catch (Exception e) {
!           throw new PipelineException("Exception while initializing", e);
!         }
!         pstage.setDocument(document);
!         pstage.setTicket(ticket);
!         trackDocument(pstage);
! 
!         processPipelineStage(pstage, finalResults);
!       } else {
!         getLog().logError(I18n.get("019008", name), null);
!       }
!     } else {
!       if (finalResults != null) {
!         finalResults.add(new PipelineStageResult(null, document, ticket));
!       }
!     }
!   }
! 
!   /**
!    * Process the pipeline stage and then defer to eithe the multithreaded or single threaded
!    * implementation - this needs to be extracted.
!    *
!    * @param pstage The pipeline stage to process
!    * @param finalResults collection of documents after all processing is completed.
!    * @throws PipelineException
!    */
!   protected void processPipelineStage(IPipelineStage pstage, Collection finalResults)
!       throws PipelineException {
! 
!     // Process the pipeline stage, get the results of the processing
!     PipelineStageResult [] psResults = pstage.processStage();
! 
!     if(psResults!=null && psResults.length>0) {
!       if (pstage.getThreaded()) {
!          int maxThreads = pstage.getMaxThreads();
!          if (maxThreads <= 0) {
!             maxThreads = 2; // default to 2 if <= 0
!          }
!          processPipelineStageResultsParallel(psResults, finalResults, maxThreads);
!       }
!       else {
!          processPipelineStageResults(psResults, finalResults);
!       }
!     }
!   }
! 
!   /**
!    * Process the pipeline stage finalResults.  The gets called from process and
!    * handles the finalResults from the processing.
!    *
!    * @param psResults The results of running the current stage on this document
!    * @param finalResults Collection of documents after all processing done (after null)
!    *
!    * @throws PipelineException
!    */
!   private void processPipelineStageResults(PipelineStageResult[] psResults,
!     Collection finalResults) throws PipelineException {
! 
!     int numResults = psResults.length;
!     for (int i = 0; i < numResults; ++i) {
!       processPipelineStageResult(psResults[i], finalResults);
!     }
!   }
! 
!   /**
!    * Process a single PipelineStageResult.
!    *
!    * @param psResult The pipeline stage result to process
!    * @param finalResults the collection of processed documents.
!    * @throws PipelineException
!    */
!   protected void processPipelineStageResult(PipelineStageResult psResult, Collection finalResults)
!       throws PipelineException {
!     String name = psResult.getNamePipelineStage();
!     PipelineDocument document = psResult.getDocument();
!     IJournalTicket ticket = psResult.getTicket();
! 
!     this.process(name, document, ticket, finalResults);
!   }
! 
!   /**
!    * Execute the finalResults using threads in parallel.  This uses the maxThreads argument
!    * to specify the maximum number of threads to spawn.  There is a slight impedance mismatch here.
!    * The processPipelineStageResults methods (parallel and non-parallel) differ only in the
!    * max threads argument.  This indicates the number of threads to spawn to handle the results.
!    * Now, if we could consider this an attribute of this factory (or even of some threadpool) this
!    * we can remove it from the interface - and we can conflate the interface.
!    *
!    * @param psResults results from the current pipeline stage process
!    * @param finalResults result of the current pipeline stage process
!    * @param maxThreads max number of threads to spawn to process the finalResults
!    */
!   private void processPipelineStageResultsParallel(final PipelineStageResult[] psResults,
!                                                      Collection finalResults,
!                                                      final int maxThreads )
!       throws PipelineException {
!     // ensure synchronized access to finalResults Collection if threads are spawned
!     final Collection syncResults = Collections.synchronizedCollection(finalResults);
!     final int numResults = psResults.length;
!     final Vector exceptions = new Vector();
!     Vector threads = new Vector();
!     for (int i = 0; i < maxThreads; i++) {
!        final int index = i;
!        Runnable r = new Runnable() {
!           public void run() {
!              for (int j = 0; j < numResults; j++) {
!                 if ((j % maxThreads) == index) {
!                    try {
!                      processPipelineStageResult(psResults[j], syncResults);
!                    }
!                    catch (PipelineException p) {
!                      exceptions.add(p);
!                    }
!                 }
!              }
!           }
!        };
!        Thread t = new Thread(r);
!        threads.add(t);
!        t.start();
!     }
! 
!     // wait for any threads to finish before returning
!     for (int i = 0; i < threads.size(); i++) {
!        try {
!           Thread t = (Thread)threads.get(i);
!           t.join();
!        }
!        catch (InterruptedException ie) {
!        }
!     }
!     // see if any of the threads threw any exceptions, and if so, re-throw the first one
!     // TODO: throw an exception encapsulating all the exceptions that were thrown?
!     // This might be harder than it appears unless we extend the PipelineException to
!     // Add this capability
!     if (exceptions.size() > 0) {
!        throw (PipelineException)exceptions.get(0);
!     }
!   }
! 
!   /**
!    * Track the document on the pipeline stage
!    *
!    * @param pstage
!    *
!    * @throws PipelineException DOCUMENT ME!
!    */
!   private void trackDocument(IPipelineStage pstage) throws PipelineException {
!     if (pstage.isTracked()) {
!       try {
!         JournalFactory.getJournal().updateDocument(pstage.getTicket(),
!           pstage.getDocument(), getName() + "." + pstage.getName());
!       } catch (JournalException e) {
!         getLog().logError(e);
!         throw new PipelineException(e.getMessage(), e);
!       }
!     }
    }
  
--- 249,253 ----
    public void process(String name, PipelineDocument document,
      IJournalTicket ticket, Collection finalResults) throws PipelineException {
!     getProcessor().process(name, document, ticket, finalResults);
    }
  
***************
*** 481,484 ****
--- 263,270 ----
  
      return log;
+   }
+ 
+   public IPipelineStageProcessor getProcessor() {
+     return processor;
    }
  }

2003	Jan	Feb (14)	Mar (107)	Apr (211)	May (93)	Jun (158)	Jul (159)	Aug (368)	Sep (188)	Oct (151)	Nov (115)	Dec (98)
2004	Jan (25)	Feb	Mar (33)	Apr (28)	May (116)	Jun (2)	Jul (117)	Aug (19)	Sep (9)	Oct (2)	Nov	Dec (4)
2005	Jan	Feb	Mar	Apr	May (1)	Jun	Jul	Aug	Sep	Oct (2)	Nov (9)	Dec
2006	Jan	Feb	Mar (22)	Apr	May	Jun	Jul	Aug	Sep	Oct	Nov	Dec
2007	Jan	Feb	Mar (6)	Apr	May	Jun	Jul	Aug (267)	Sep	Oct	Nov (6)	Dec (512)
2008	Jan (187)	Feb	Mar	Apr	May	Jun (3)	Jul (6)	Aug	Sep	Oct	Nov	Dec
2011	Jan	Feb	Mar	Apr	May	Jun	Jul	Aug	Sep	Oct (2)	Nov	Dec
2012	Jan	Feb	Mar	Apr (1)	May	Jun	Jul	Aug	Sep	Oct	Nov	Dec

babeldoc-devel Mailing List for Babeldoc: Universal Document Processor (Page 96)

babeldoc-devel — Babeldoc developer list