1  /*
     2   * Licensed to the Apache Software Foundation (ASF) under one or more
     3   *  contributor license agreements.  The ASF licenses this file to You
     4   * under the Apache License, Version 2.0 (the "License"); you may not
     5   * use this file except in compliance with the License.
     6   * You may obtain a copy of the License at
     7   *
     8   *     http://www.apache.org/licenses/LICENSE-2.0
     9   *
    10   * Unless required by applicable law or agreed to in writing, software
    11   * distributed under the License is distributed on an "AS IS" BASIS,
    12   * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
    13   * See the License for the specific language governing permissions and
    14   * limitations under the License.  For additional information regarding
    15   * copyright in this work, please see the NOTICE file in the top level
    16   * directory of this distribution.
    17   */
    18  
    19  package org.apache.roller.weblogger.business.runnable;
    20  
    21  import java.util.ArrayList;
    22  import java.util.Date;
    23  import java.util.List;
    24  import java.util.concurrent.ExecutorService;
    25  import java.util.concurrent.Executors;
    26  import java.util.concurrent.Future;
    27  import org.apache.commons.lang.StringUtils;
    28  import org.apache.commons.logging.Log;
    29  import org.apache.commons.logging.LogFactory;
    30  import org.apache.roller.weblogger.WebloggerException;
    31  import org.apache.roller.weblogger.business.InitializationException;
    32  import org.apache.roller.weblogger.config.WebloggerConfig;
    33  import org.apache.roller.weblogger.pojos.TaskLock;
    34  
    35  
    36  /**
    37   * Manage Roller's thread use.
    38   */
    39  @com.google.inject.Singleton
    40  public abstract class ThreadManagerImpl implements ThreadManager {
    41      
             /* 
    P/P       *  Method: org.apache.roller.weblogger.business.runnable.ThreadManagerImpl__static_init
              * 
              *  Postconditions:
              *    init'ed(log)
              */
    42      private static final Log log = LogFactory.getLog(ThreadManagerImpl.class);
    43      
    44      // our own scheduler thread
    45      private Thread schedulerThread = null;
    46      
    47      // a simple thread executor
    48      private final ExecutorService serviceScheduler;
    49      
    50      
             /* 
    P/P       *  Method: void org.apache.roller.weblogger.business.runnable.ThreadManagerImpl()
              * 
              *  Presumptions:
              *    org.apache.commons.logging.LogFactory:getLog(...)@42 != null
              * 
              *  Postconditions:
              *    this.schedulerThread == null
              *    init'ed(this.serviceScheduler)
              */
    51      public ThreadManagerImpl() {
    52          
    53          log.info("Instantiating Thread Manager");
    54          
    55          serviceScheduler = Executors.newCachedThreadPool();
    56      }
    57      
    58      
    59      public void initialize() throws InitializationException {
    60                      
    61          // initialize tasks, making sure that each task has a tasklock record in the db
                 /* 
    P/P           *  Method: void initialize()
                  * 
                  *  Preconditions:
                  *    init'ed(this.schedulerThread)
                  *    org/apache/roller/weblogger/config/WebloggerConfig.config != null
                  *    org/apache/roller/weblogger/config/WebloggerConfig.log != null
                  * 
                  *  Presumptions:
                  *    arr$.length@64 <= 232-1
                  *    java.lang.Class:forName(...)@72 != null
                  *    org.apache.commons.lang.StringUtils:stripAll(...)@64 != null
                  *    org.apache.commons.logging.LogFactory:getLog(...)@42 != null
                  * 
                  *  Postconditions:
                  *    org/apache/roller/weblogger/business/pings/PingQueueProcessor.theInstance == old org/apache/roller/weblogger/business/pings/PingQueueProcessor.theInstance
                  *    this.schedulerThread == One-of{old this.schedulerThread, &new Thread(initialize#8)}
                  *    this.schedulerThread != null
                  *    new PingQueueProcessor(init#1*) num objects == undefined
                  *    new PingQueueProcessor(init#1*) num objects == 0, if init'ed
                  *    new PingQueueProcessor(init#1*).pingQueueMgr == undefined
                  *    new PingQueueProcessor(init#1*).pingQueueMgr == null
                  *    new Thread(initialize#8) num objects <= 1
                  * 
                  *  Test Vectors:
                  *    this.schedulerThread: Inverse{null}, Addr_Set{null}
                  *    org.apache.roller.weblogger.business.runnable.ThreadManagerImpl:getTaskLockByName(...)@77: Inverse{null}, Addr_Set{null}
                  */
    62          List<RollerTask> webloggerTasks = new ArrayList<RollerTask>();
    63          String tasksStr = WebloggerConfig.getProperty("tasks.enabled");
    64          String[] tasks = StringUtils.stripAll(StringUtils.split(tasksStr, ","));
    65          for ( String taskName : tasks ) {
    66              
    67              String taskClassName = WebloggerConfig.getProperty("tasks."+taskName+".class");
    68              if(taskClassName != null) {
    69                  log.info("Initializing task: "+taskName);
    70                  
    71                  try {
    72                      Class taskClass = Class.forName(taskClassName);
    73                      RollerTask task = (RollerTask) taskClass.newInstance();
    74                      task.init();
    75                      
    76                      // make sure there is a tasklock record in the db
    77                      TaskLock taskLock = getTaskLockByName(task.getName());
    78                      if (taskLock == null) {
    79                          log.debug("Task record does not exist, inserting empty record to start with");
    80  
    81                          // insert an empty record
    82                          taskLock = new TaskLock();
    83                          taskLock.setName(task.getName());
    84                          taskLock.setLastRun(new Date(0));
    85                          taskLock.setTimeAquired(new Date(0));
    86                          taskLock.setTimeLeased(0);
    87  
    88                          // save it
    89                          this.saveTaskLock(taskLock);
    90                      }
    91                      
    92                      // add it to the list of configured tasks
    93                      webloggerTasks.add(task);
    94                      
    95                  } catch (ClassCastException ex) {
    96                      log.warn("Task does not extend RollerTask class", ex);
    97                  } catch (WebloggerException ex) {
    98                      log.error("Error scheduling task", ex);
    99                  } catch (Exception ex) {
   100                      log.error("Error instantiating task", ex);
   101                  }
   102              }
   103          }
   104          
   105          // create scheduler
   106          TaskScheduler scheduler = new TaskScheduler(webloggerTasks);
   107          
   108          // start scheduler thread, but only if it's not already running
+  109          if (schedulerThread == null && scheduler != null) {
   110              log.debug("Starting scheduler thread");
   111              schedulerThread = new Thread(scheduler, "Roller Weblogger Task Scheduler");
   112              // set thread priority between MAX and NORM so we get slightly preferential treatment
   113              schedulerThread.setPriority((Thread.MAX_PRIORITY + Thread.NORM_PRIORITY)/2);
   114              schedulerThread.start();
   115          }
   116      }
   117      
   118      
   119      public void executeInBackground(Runnable runnable)
   120              throws InterruptedException {
                 /* 
    P/P           *  Method: void executeInBackground(Runnable)
                  * 
                  *  Preconditions:
                  *    this.serviceScheduler != null
                  */
+  121          Future task = serviceScheduler.submit(runnable);
   122      }
   123      
   124      
   125      public void executeInForeground(Runnable runnable)
   126              throws InterruptedException {
                 /* 
    P/P           *  Method: void executeInForeground(Runnable)
                  * 
                  *  Preconditions:
                  *    this.serviceScheduler != null
                  * 
                  *  Presumptions:
                  *    java.util.concurrent.ExecutorService:submit(...)@127 != null
                  * 
                  *  Test Vectors:
                  *    java.util.concurrent.Future:isDone(...)@132: {1}, {0}
                  */
   127          Future task = serviceScheduler.submit(runnable);
   128          
   129          // since this task is really meant to be executed within this calling 
   130          // thread, here we can add a little code here to loop until it realizes 
   131          // the task is done
   132          while(!task.isDone()) {
   133              Thread.sleep(500);
   134          }
   135      }
   136      
   137      
   138      public void shutdown() {
   139          
                 /* 
    P/P           *  Method: void shutdown()
                  * 
                  *  Preconditions:
                  *    init'ed(this.schedulerThread)
                  *    this.serviceScheduler != null
                  * 
                  *  Presumptions:
                  *    org.apache.commons.logging.LogFactory:getLog(...)@42 != null
                  * 
                  *  Test Vectors:
                  *    this.schedulerThread: Addr_Set{null}, Inverse{null}
                  */
   140          log.debug("starting shutdown sequence");
   141          
   142          // trigger an immediate shutdown of any backgrounded tasks
   143          serviceScheduler.shutdownNow();
   144          
   145          // only stop if we are already running
   146          if(schedulerThread != null) {
   147              log.debug("Stopping scheduler");
   148              schedulerThread.interrupt();
   149          }
   150      }
   151      
   152      
   153      public void release() {
   154          // no-op
             /* 
    P/P       *  Method: void release()
              */
   155      }
   156      
   157      
   158      /**
   159       * Default implementation of lease registration, always returns true.
   160       * 
   161       * Subclasses should override this method if they plan to run in an
   162       * environment that supports clustered deployments.
   163       */
   164      public boolean registerLease(RollerTask task) {
                 /* 
    P/P           *  Method: bool registerLease(RollerTask)
                  * 
                  *  Postconditions:
                  *    return_value == 1
                  */
   165          return true;
   166      }
   167      
   168      
   169      /**
   170       * Default implementation of lease unregistration, always returns true.
   171       * 
   172       * Subclasses should override this method if they plan to run in an
   173       * environment that supports clustered deployments.
   174       */
   175      public boolean unregisterLease(RollerTask task) {
                 /* 
    P/P           *  Method: bool unregisterLease(RollerTask)
                  * 
                  *  Postconditions:
                  *    return_value == 1
                  */
   176          return true;
   177      }
   178      
   179  }