[riftsaw-commits] riftsaw SVN: r618 - in branches/DBIssue/runtime/engine/src/main/java/org/jboss/soa/bpel/runtime/engine/ode: scheduler and 1 other directory.

Thursday, 8 April 2010


Author: objectiser
Date: 2010-04-08 11:41:12 -0400 (Thu, 08 Apr 2010)
New Revision: 618

Added:
  
branches/DBIssue/runtime/engine/src/main/java/org/jboss/soa/bpel/runtime/engine/ode/scheduler/
  
branches/DBIssue/runtime/engine/src/main/java/org/jboss/soa/bpel/runtime/engine/ode/scheduler/Job.java
  
branches/DBIssue/runtime/engine/src/main/java/org/jboss/soa/bpel/runtime/engine/ode/scheduler/JobComparatorByDate.java
  
branches/DBIssue/runtime/engine/src/main/java/org/jboss/soa/bpel/runtime/engine/ode/scheduler/SchedulerThread.java
  
branches/DBIssue/runtime/engine/src/main/java/org/jboss/soa/bpel/runtime/engine/ode/scheduler/Task.java
  
branches/DBIssue/runtime/engine/src/main/java/org/jboss/soa/bpel/runtime/engine/ode/scheduler/TaskRunner.java
  
branches/DBIssue/runtime/engine/src/main/java/org/jboss/soa/bpel/runtime/engine/ode/scheduler/TransientScheduler.java
Modified:
  
branches/DBIssue/runtime/engine/src/main/java/org/jboss/soa/bpel/runtime/engine/ode/BPELEngineImpl.java
Log:
Transient job scheduler, to avoid additional db usage - to help isolate db concurrency
issues.

Modified:
branches/DBIssue/runtime/engine/src/main/java/org/jboss/soa/bpel/runtime/engine/ode/BPELEngineImpl.java
===================================================================
---
branches/DBIssue/runtime/engine/src/main/java/org/jboss/soa/bpel/runtime/engine/ode/BPELEngineImpl.java	2010-04-08
14:23:33 UTC (rev 617)
+++
branches/DBIssue/runtime/engine/src/main/java/org/jboss/soa/bpel/runtime/engine/ode/BPELEngineImpl.java	2010-04-08
15:41:12 UTC (rev 618)
@@ -40,6 +40,7 @@
 import org.apache.ode.utils.Properties;
 import org.jboss.soa.bpel.runtime.engine.BPELEngine;
 import org.jboss.soa.bpel.runtime.engine.InvocationAdapter;
+import org.jboss.soa.bpel.runtime.engine.ode.scheduler.TransientScheduler;
 import org.w3c.dom.Element;
 
 import javax.sql.DataSource;
@@ -440,7 +441,8 @@
   }
 
   protected Scheduler createScheduler() {
-    SimpleScheduler scheduler = new SimpleScheduler(new GUID().toString(),new
JdbcDelegate(_db.getDataSource()), _odeConfig.getProperties());
+	//SimpleScheduler scheduler = new SimpleScheduler(new GUID().toString(),new
JdbcDelegate(_db.getDataSource()), _odeConfig.getProperties());
+	TransientScheduler scheduler = new TransientScheduler(new GUID().toString(),
_odeConfig.getProperties());
     scheduler.setExecutorService(_executorService);
     scheduler.setTransactionManager(_txMgr);
 

Added:
branches/DBIssue/runtime/engine/src/main/java/org/jboss/soa/bpel/runtime/engine/ode/scheduler/Job.java
===================================================================
---
branches/DBIssue/runtime/engine/src/main/java/org/jboss/soa/bpel/runtime/engine/ode/scheduler/Job.java	
                       (rev 0)
+++
branches/DBIssue/runtime/engine/src/main/java/org/jboss/soa/bpel/runtime/engine/ode/scheduler/Job.java	2010-04-08
15:41:12 UTC (rev 618)
@@ -0,0 +1,62 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.jboss.soa.bpel.runtime.engine.ode.scheduler;
+
+import java.util.Map;
+
+import org.apache.ode.utils.GUID;
+
+/**
+ * Like a task, but a little bit better.
+ * 
+ * @author Maciej Szefler ( m s z e f l e r @ g m a i l . c o m )
+ */
+class Job extends Task {
+    String jobId;
+    boolean transacted;
+    Map<String,Object> detail;
+    boolean persisted = true;
+
+    public Job(long when, boolean transacted, Map<String, Object> jobDetail) {
+        this(when, new GUID().toString(),transacted,jobDetail);
+    }
+    
+    public Job(long when, String jobId, boolean transacted,Map<String, Object>
jobDetail) {
+        super(when);
+        this.jobId = jobId;
+        this.detail = jobDetail;
+        this.transacted = transacted;
+    }
+
+    @Override
+    public int hashCode() {
+        return jobId.hashCode();
+    }
+
+    @Override
+    public boolean equals(Object obj) {
+        return obj instanceof Job && jobId.equals(((Job) obj).jobId);
+    }
+    
+    @Override
+    public String toString() {
+        return "Job "+jobId+" transacted: "+transacted+"
persisted: "+persisted+" details: "+detail;
+    }
+}

Added:
branches/DBIssue/runtime/engine/src/main/java/org/jboss/soa/bpel/runtime/engine/ode/scheduler/JobComparatorByDate.java
===================================================================
---
branches/DBIssue/runtime/engine/src/main/java/org/jboss/soa/bpel/runtime/engine/ode/scheduler/JobComparatorByDate.java	
                       (rev 0)
+++
branches/DBIssue/runtime/engine/src/main/java/org/jboss/soa/bpel/runtime/engine/ode/scheduler/JobComparatorByDate.java	2010-04-08
15:41:12 UTC (rev 618)
@@ -0,0 +1,38 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.jboss.soa.bpel.runtime.engine.ode.scheduler;
+
+import java.util.Comparator;
+
+/**
+ * Compare jobs, using scheduled date as sort criteria.
+ * 
+ * @author Maciej Szefler ( m s z e f l e r @ g m a i l . c o m )
+ */
+class JobComparatorByDate implements Comparator<Task> {
+
+    public int compare(Task o1, Task o2) {
+        long diff = o1.schedDate - o2.schedDate;
+        if (diff < 0) return -1;
+        if (diff > 0) return 1;
+        return 0;
+    }
+
+}

Added:
branches/DBIssue/runtime/engine/src/main/java/org/jboss/soa/bpel/runtime/engine/ode/scheduler/SchedulerThread.java
===================================================================
---
branches/DBIssue/runtime/engine/src/main/java/org/jboss/soa/bpel/runtime/engine/ode/scheduler/SchedulerThread.java	
                       (rev 0)
+++
branches/DBIssue/runtime/engine/src/main/java/org/jboss/soa/bpel/runtime/engine/ode/scheduler/SchedulerThread.java	2010-04-08
15:41:12 UTC (rev 618)
@@ -0,0 +1,197 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.jboss.soa.bpel.runtime.engine.ode.scheduler;
+
+import java.util.concurrent.PriorityBlockingQueue;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.locks.Condition;
+import java.util.concurrent.locks.ReentrantLock;
+
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import org.apache.ode.utils.stl.CollectionsX;
+import org.apache.ode.utils.stl.MemberOfFunction;
+
+/**
+ * Implements the "todo" queue and prioritized scheduling mechanism. 
+ * 
+ * @author mszefler
+ * @author Maciej Szefler ( m s z e f l e r @ g m a i l . c o m )
+ * 
+ */
+class SchedulerThread implements Runnable {
+
+    private static final Log __log = LogFactory.getLog(SchedulerThread.class);
+
+    private static final int TODO_QUEUE_INITIAL_CAPACITY = 200;
+
+    /** Jobs ready for immediate execution. */
+    private PriorityBlockingQueue<Task> _todo;
+
+    /** Lock for managing the queue */
+    private ReentrantLock _lock = new ReentrantLock();
+
+    private Condition _activity = _lock.newCondition();
+
+    private volatile boolean _done;
+
+    private TaskRunner _taskrunner;
+
+    private Thread _thread;
+
+    SchedulerThread(TaskRunner runner) {
+        _todo = new PriorityBlockingQueue<Task>(TODO_QUEUE_INITIAL_CAPACITY,
+                new JobComparatorByDate());
+        _taskrunner = runner;
+    }
+
+    void start() {
+        if (_thread != null)
+            return;
+
+        _done = false;
+        _thread = new Thread(this, "OdeScheduler");
+        _thread.start();
+    }
+
+    /**
+     * Shutdown the thread.
+     */
+    void stop() {
+        if (_thread == null)
+            return;
+
+        _done = true;
+        _lock.lock();
+        try {
+            _activity.signal();
+        } finally {
+            _lock.unlock();
+
+        }
+
+        while (_thread != null)
+            try {
+                _thread.join();
+                _thread = null;
+            } catch (InterruptedException e) {
+                ;
+            }
+
+    }
+
+    /**
+     * Add a job to the todo queue.
+     * 
+     * @param job
+     */
+    void enqueue(Task task) {
+        _lock.lock();
+        try {
+            _todo.add(task);
+            _activity.signal();
+        } finally {
+            _lock.unlock();
+        }
+    }
+
+    /**
+     * Remove a job to the todo queue.
+     *
+     * @param job
+     */
+    void dequeue(Task task) {
+        _lock.lock();
+        try {
+            _todo.remove(task);
+            _activity.signal();
+        } finally {
+            _lock.unlock();
+        }
+    }
+
+    /**
+     * Get the size of the todo queue.
+     * 
+     * @return
+     */
+    public int size() {
+        return _todo.size();
+    }
+
+    /**
+     * Pop items off the todo queue, and send them to the task runner for processing.
+     */
+    public void run() {
+        while (!_done) {
+            _lock.lock();
+            try {
+                long nextjob;
+                while ((nextjob = nextJobTime()) > 0 && !_done)
+                    _activity.await(nextjob, TimeUnit.MILLISECONDS);
+
+                if (!_done && nextjob == 0) {
+                    Task task = _todo.take();
+                    _taskrunner.runTask(task);
+
+                }
+            } catch (InterruptedException ex) {
+                ; // ignore
+            } finally {
+                _lock.unlock();
+            }
+        }
+    }
+
+    /**
+     * Calculate the time until the next available job.
+     * 
+     * @return time until next job, 0 if one is one is scheduled to go, and some
+     *         really large number if there are no jobs to speak of
+     */
+    private long nextJobTime() {
+        assert _lock.isLocked();
+
+        Task job = _todo.peek();
+        if (job == null)
+            return Long.MAX_VALUE;
+
+        return Math.max(0, job.schedDate - System.currentTimeMillis());
+    }
+
+    /**
+     * Remove the tasks of a given type from the list. 
+     * @param tasktype type of task
+     */
+    public void clearTasks(final Class<? extends Task> tasktype) {
+        _lock.lock();
+        try {
+            CollectionsX.remove_if(_todo, new MemberOfFunction<Task>() {
+                @Override
+                public boolean isMember(Task o) {
+                    return tasktype.isAssignableFrom(o.getClass());
+                }
+
+            });
+        } finally {
+            _lock.unlock();
+        }
+    }
+}

Added:
branches/DBIssue/runtime/engine/src/main/java/org/jboss/soa/bpel/runtime/engine/ode/scheduler/Task.java
===================================================================
---
branches/DBIssue/runtime/engine/src/main/java/org/jboss/soa/bpel/runtime/engine/ode/scheduler/Task.java	
                       (rev 0)
+++
branches/DBIssue/runtime/engine/src/main/java/org/jboss/soa/bpel/runtime/engine/ode/scheduler/Task.java	2010-04-08
15:41:12 UTC (rev 618)
@@ -0,0 +1,35 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.jboss.soa.bpel.runtime.engine.ode.scheduler;
+
+/**
+ * The thing that we schedule.
+ * 
+ * @author Maciej Szefler ( m s z e f l e r @ g m a i l . c o m )
+ *
+ */
+class Task {
+    /** Scheduled date/time. */
+    public long schedDate;
+
+    Task(long schedDate) {
+        this.schedDate = schedDate;
+    }
+}

Added:
branches/DBIssue/runtime/engine/src/main/java/org/jboss/soa/bpel/runtime/engine/ode/scheduler/TaskRunner.java
===================================================================
---
branches/DBIssue/runtime/engine/src/main/java/org/jboss/soa/bpel/runtime/engine/ode/scheduler/TaskRunner.java	
                       (rev 0)
+++
branches/DBIssue/runtime/engine/src/main/java/org/jboss/soa/bpel/runtime/engine/ode/scheduler/TaskRunner.java	2010-04-08
15:41:12 UTC (rev 618)
@@ -0,0 +1,32 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.jboss.soa.bpel.runtime.engine.ode.scheduler;
+
+/**
+ * The thing that runs the scheduled tasks.
+ * 
+ * @author Maciej Szefler ( m s z e f l e r @ g m a i l . c o m )
+ *
+ */
+interface TaskRunner {
+
+    
+    public void runTask(Task task);
+}

Added:
branches/DBIssue/runtime/engine/src/main/java/org/jboss/soa/bpel/runtime/engine/ode/scheduler/TransientScheduler.java
===================================================================
---
branches/DBIssue/runtime/engine/src/main/java/org/jboss/soa/bpel/runtime/engine/ode/scheduler/TransientScheduler.java	
                       (rev 0)
+++
branches/DBIssue/runtime/engine/src/main/java/org/jboss/soa/bpel/runtime/engine/ode/scheduler/TransientScheduler.java	2010-04-08
15:41:12 UTC (rev 618)
@@ -0,0 +1,757 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.jboss.soa.bpel.runtime.engine.ode.scheduler;
+
+import java.util.*;
+import java.util.concurrent.Callable;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.CopyOnWriteArraySet;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
+import java.util.concurrent.atomic.AtomicLong;
+
+import javax.transaction.Status;
+import javax.transaction.Synchronization;
+import javax.transaction.SystemException;
+import javax.transaction.Transaction;
+import javax.transaction.TransactionManager;
+
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import org.apache.ode.bpel.iapi.ContextException;
+import org.apache.ode.bpel.iapi.Scheduler;
+
+/**
+ * A reliable and relatively simple scheduler that uses a database to persist information
about 
+ * scheduled tasks.
+ * 
+ * The challenge is to achieve high performance in a small memory footprint without loss
of reliability
+ * while supporting distributed/clustered configurations.
+ * 
+ * The design is based around three time horizons: "immediate", "near
future", and "everything else". 
+ * Immediate jobs (i.e. jobs that are about to be up) are written to the database and
kept in
+ * an in-memory priority queue. When they execute, they are removed from the database.
Near future
+ * jobs are placed in the database and assigned to the current node, however they are not
stored in
+ * memory. Periodically jobs are "upgraded" from near-future to immediate
status, at which point they
+ * get loaded into memory. Jobs that are further out in time, are placed in the database
without a 
+ * node identifer; when they are ready to be "upgraded" to near-future jobs
they are assigned to one
+ * of the known live nodes. Recovery is rather straighforward, with stale node
identifiers being 
+ * reassigned to known good nodes.       
+ * 
+ * @author Maciej Szefler ( m s z e f l e r @ g m a i l . c o m )
+ *
+ */
+public class TransientScheduler implements Scheduler, TaskRunner {
+    private static final Log __log = LogFactory.getLog(TransientScheduler.class);
+
+    /**
+     * Jobs scheduled with a time that is between [now, now+immediateInterval] will be
assigned to the current node, and placed
+     * directly on the todo queue.
+     */
+    long _immediateInterval = 30000;
+
+    /**
+     * Jobs scheduled with a time that is between
(now+immediateInterval,now+nearFutureInterval) will be assigned to the current
+     * node, but will not be placed on the todo queue (the promoter will pick them up).
+     */
+    long _nearFutureInterval = 10 * 60 * 1000;
+
+    /** 10s of no communication and you are deemed dead. */
+    long _staleInterval = 10000;
+
+    /**
+     * Estimated sustained transaction per second capacity of the system.
+     * e.g. 100 means the system can process 100 jobs per seconds, on average
+     * This number is used to determine how many jobs to load from the database at once.
+     */
+    int _tps = 100;
+
+    TransactionManager _txm;
+
+    ExecutorService _exec;
+
+    String _nodeId;
+
+    /** Maximum number of jobs in the "near future" / todo queue. */
+    int _todoLimit = 10000;
+
+    /** The object that actually handles the jobs. */
+    volatile JobProcessor _jobProcessor;
+
+    volatile JobProcessor _polledRunnableProcessor;
+    
+    private SchedulerThread _todo;
+
+    /** All the nodes we know about */
+    private CopyOnWriteArraySet<String> _knownNodes = new
CopyOnWriteArraySet<String>();
+
+    /** When we last heard from our nodes. */
+    private ConcurrentHashMap<String, Long> _lastHeartBeat = new
ConcurrentHashMap<String, Long>();
+
+    private boolean _running;
+
+    /** Time for next upgrade. */
+    private AtomicLong _nextUpgrade = new AtomicLong();
+
+    private Random _random = new Random();
+
+    private long _pollIntervalForPolledRunnable =
Long.getLong("org.apache.ode.polledRunnable.pollInterval", 10 * 60 * 1000);
+    
+    public TransientScheduler(String nodeId, Properties conf) {
+        _nodeId = nodeId;
+        _todoLimit = getIntProperty(conf, "ode.scheduler.queueLength",
_todoLimit);
+        _immediateInterval = getLongProperty(conf,
"ode.scheduler.immediateInterval", _immediateInterval);
+        _nearFutureInterval = getLongProperty(conf,
"ode.scheduler.nearFutureInterval", _nearFutureInterval);
+        _staleInterval = getLongProperty(conf, "ode.scheduler.staleInterval",
_staleInterval);
+        _tps = getIntProperty(conf, "ode.scheduler.transactionsPerSecond",
_tps);
+        _todo = new SchedulerThread(this);
+    }
+    
+    public void setPollIntervalForPolledRunnable(long pollIntervalForPolledRunnable) {
+        _pollIntervalForPolledRunnable = pollIntervalForPolledRunnable;
+    }
+
+    private int getIntProperty(Properties props, String propName, int defaultValue) {
+        String s = props.getProperty(propName);
+        if (s != null) return Integer.parseInt(s);
+        else return defaultValue;
+    }
+
+    private long getLongProperty(Properties props, String propName, long defaultValue) {
+        String s = props.getProperty(propName);
+        if (s != null) return Long.parseLong(s);
+        else return defaultValue;
+    }
+        
+    public void setNodeId(String nodeId) {
+        _nodeId = nodeId;
+    }
+
+    public void setStaleInterval(long staleInterval) {
+        _staleInterval = staleInterval;
+    }
+
+    public void setImmediateInterval(long immediateInterval) {
+        _immediateInterval = immediateInterval;
+    }
+
+    public void setNearFutureInterval(long nearFutureInterval) {
+        _nearFutureInterval = nearFutureInterval;
+    }
+
+    public void setTransactionsPerSecond(int tps) {
+        _tps = tps;
+    }
+
+    public void setTransactionManager(TransactionManager txm) {
+        _txm = txm;
+    }
+
+    public void setExecutorService(ExecutorService executorService) {
+        _exec = executorService;
+    }
+
+    public void setPolledRunnableProcesser(JobProcessor polledRunnableProcessor) {
+        _polledRunnableProcessor = polledRunnableProcessor;
+    }
+
+    public void cancelJob(String jobId) throws ContextException {
+        _todo.dequeue(new Job(0, jobId, false, null));
+    }
+
+    public <T> Future<T> execIsolatedTransaction(final Callable<T>
transaction) throws Exception, ContextException {
+        return _exec.submit(new Callable<T>() {
+            public T call() throws Exception {
+                try {
+                    return execTransaction(transaction);
+                } catch (Exception e) {
+                    __log.error("An exception occured while executing an isolated
transaction, " +
+                            "the transaction is going to be abandoned.", e);
+                    return null;
+                }
+            }
+        });
+    }
+
+    public <T> T execTransaction(Callable<T> transaction) throws Exception,
ContextException {
+        try {
+            if (__log.isDebugEnabled()) __log.debug("Beginning a new
transaction");
+            _txm.begin();
+        } catch (Exception ex) {
+            String errmsg = "Internal Error, could not begin transaction.";
+            throw new ContextException(errmsg, ex);
+        }
+        
+        boolean success = false;
+        try {
+            T retval = transaction.call();
+            success = true;
+            return retval;
+        } catch (Exception ex) {
+            throw ex;
+        } finally {
+            if (success) {
+                if (__log.isDebugEnabled()) __log.debug("Commiting on " + _txm
+ "...");
+                _txm.commit();
+            } else {
+                if (__log.isDebugEnabled()) __log.debug("Rollbacking on " +
_txm + "...");
+                _txm.rollback();
+            }
+        }
+    }
+
+    public void setRollbackOnly() throws Exception {
+        _txm.setRollbackOnly();
+    }
+
+    public void registerSynchronizer(final Synchronizer synch) throws ContextException {
+        try {
+            _txm.getTransaction().registerSynchronization(new Synchronization() {
+
+                public void beforeCompletion() {
+                    synch.beforeCompletion();
+                }
+
+                public void afterCompletion(int status) {
+                    synch.afterCompletion(status == Status.STATUS_COMMITTED);
+                }
+
+            });
+        } catch (Exception e) {
+            throw new ContextException("Unable to register synchronizer.", e);
+        }
+    }
+
+    public String schedulePersistedJob(final Map<String, Object> jobDetail, Date
when) throws ContextException {
+        long ctime = System.currentTimeMillis();
+        if (when == null)
+            when = new Date(ctime);
+
+        if (__log.isDebugEnabled())
+            __log.debug("scheduling " + jobDetail + " for " + when);
+
+        return schedulePersistedJob(new Job(when.getTime(), true, jobDetail), when,
ctime);
+    }
+
+    public String scheduleMapSerializableRunnable(MapSerializableRunnable runnable, Date
when) throws ContextException {
+        long ctime = System.currentTimeMillis();
+        if (when == null)
+            when = new Date(ctime);
+
+        Map<String, Object> jobDetails = new HashMap<String, Object>();
+        jobDetails.put("runnable", runnable);
+        runnable.storeToDetailsMap(jobDetails);
+        
+        if (__log.isDebugEnabled())
+            __log.debug("scheduling " + jobDetails + " for " +
when);
+
+        return schedulePersistedJob(new Job(when.getTime(), true, jobDetails), when,
ctime);
+    }
+
+    private String schedulePersistedJob(Job job, Date when, long ctime) throws
ContextException {
+    	/*
+        boolean immediate = when.getTime() <= ctime + _immediateInterval;
+        boolean nearfuture = !immediate && when.getTime() <= ctime +
_nearFutureInterval;
+        try {
+            if (immediate) {
+                // Immediate scheduling means we put it in the DB for safe keeping
+                _db.insertJob(job, _nodeId, true);
+                
+                // And add it to our todo list .
+                if (_todo.size() < _todoLimit) {
+                */
+                    addTodoOnCommit(job);
+                   /*
+                }
+                __log.debug("scheduled immediate job: " + job.jobId);
+            } else if (nearfuture) {
+                // Near future, assign the job to ourselves (why? -- this makes it very
unlikely that we
+                // would get two nodes trying to process the same instance, which causes
unsightly rollbacks).
+                _db.insertJob(job, _nodeId, false);
+                __log.debug("scheduled near-future job: " + job.jobId);
+            } else {
+                // Not the near future, we don't assign a node-id, we'll assign
it later.
+                _db.insertJob(job, null, false);
+                __log.debug("scheduled far-future job: " + job.jobId);
+            }
+        } catch (DatabaseException dbe) {
+            __log.error("Database error.", dbe);
+            throw new ContextException("Database error.", dbe);
+        }
+        */
+        return job.jobId;
+
+    }
+
+    public String scheduleVolatileJob(boolean transacted, Map<String, Object>
jobDetail) throws ContextException {
+        Job job = new Job(System.currentTimeMillis(), transacted, jobDetail);
+        job.persisted = false;
+        addTodoOnCommit(job);
+        return job.toString();
+    }
+
+    public void setJobProcessor(JobProcessor processor) throws ContextException {
+        _jobProcessor = processor;
+    }
+
+    public void shutdown() {
+        stop();
+        _jobProcessor = null;
+        _txm = null;
+        _todo = null;
+    }
+
+    public synchronized void start() {
+        if (_running)
+            return;
+
+        if (_exec == null)
+            _exec = Executors.newCachedThreadPool();
+
+        _todo.clearTasks(UpgradeJobsTask.class);
+        _todo.clearTasks(LoadImmediateTask.class);
+        _todo.clearTasks(CheckStaleNodes.class);
+
+        _knownNodes.clear();
+
+        /*
+        try {
+            execTransaction(new Callable<Void>() {
+
+                public Void call() throws Exception {
+                    _knownNodes.addAll(_db.getNodeIds());
+                    return null;
+                }
+
+            });
+        } catch (Exception ex) {
+            __log.error("Error retrieving node list.", ex);
+            throw new ContextException("Error retrieving node list.", ex);
+        }
+        */
+
+        long now = System.currentTimeMillis();
+        
+        // Pretend we got a heartbeat...
+        for (String s : _knownNodes) _lastHeartBeat.put(s, now);
+
+        // schedule immediate job loading for now!
+        _todo.enqueue(new LoadImmediateTask(now));
+
+        // schedule check for stale nodes, make it random so that the nodes don't
overlap.
+        _todo.enqueue(new CheckStaleNodes(now + randomMean(_staleInterval)));
+
+        // do the upgrade sometime (random) in the immediate interval.
+        _todo.enqueue(new UpgradeJobsTask(now + randomMean(_immediateInterval)));
+
+        _todo.start();
+        _running = true;
+    }
+    
+    private long randomMean(long mean) {
+        return (long) _random.nextDouble() * mean + (mean/2);
+    }
+        
+    public synchronized void stop() {
+        if (!_running)
+            return;
+
+        _todo.stop();
+        _todo.clearTasks(UpgradeJobsTask.class);
+        _todo.clearTasks(LoadImmediateTask.class);
+        _todo.clearTasks(CheckStaleNodes.class);
+        _running = false;
+    }
+
+    /**
+     * Run a job in the current thread.
+     *
+     * @param job job to run.
+     */
+    protected void runJob(final Job job) {
+        final Scheduler.JobInfo jobInfo = new Scheduler.JobInfo(job.jobId, job.detail,
+                (Integer)(job.detail.get("retry") != null ?
job.detail.get("retry") : 0));
+
+        _exec.submit(new Callable<Void>() {
+            public Void call() throws Exception {
+                if (job.transacted) {
+                    try {
+                        execTransaction(new Callable<Void>() {
+                            public Void call() throws Exception {
+                            	/*
+                                if (job.persisted)
+                                    if (!_db.deleteJob(job.jobId, _nodeId))
+                                        throw new
JobNoLongerInDbException(job.jobId,_nodeId);
+                                  */
+                            	
+                                try {
+                                    _jobProcessor.onScheduledJob(jobInfo);
+                                } catch (JobProcessorException jpe) {
+                                    if (jpe.retry) {
+                                        int retry = job.detail.get("retry") !=
null ? (((Integer)job.detail.get("retry")) + 1) : 0;
+                                        if (retry <= 10) {
+                                            //long delay = doRetry(job);
+                                            //__log.error("Error while processing
transaction, retrying in " + delay + "s");
+                                        } else {
+                                            __log.error("Error while processing
transaction after 10 retries, no more retries:"+job);
+                                        }
+                                    } else {
+                                        __log.error("Error while processing
transaction, no retry.", jpe);
+                                    }
+                                    // Let execTransaction know that shit happened.
+                                    throw jpe;
+                                }
+                                return null;
+                            }
+                        });
+                   //} catch (JobNoLongerInDbException jde) {
+                        // This may happen if two node try to do the same job... we try
to avoid
+                        // it the synchronization is a best-effort but not perfect.
+                    //    __log.debug("job no longer in db forced rollback.");
+                    } catch (Exception ex) {
+                        __log.error("Error while executing transaction", ex);
+                    }
+                } else {
+                    _jobProcessor.onScheduledJob(jobInfo);
+                }
+                return null;
+            }
+        });
+    }
+
+    /**
+     * Run a job from a polled runnable thread. The runnable is not persistent,
+     * however, the poller is persistent and wakes up every given interval to
+     * check the status of the runnable.
+     * <ul>
+     * <li>1. The runnable is being scheduled; the poller persistent job
dispatches
+     * the runnable to a runnable delegate thread and schedules itself to a later
time.</li>
+     * <li>2. The runnable is running; the poller job re-schedules itself every
time it
+     * sees the runnable is not completed.</li>
+     * <li>3. The runnable failed; the poller job passes the exception thrown on
the runnable
+     * down, and the standard scheduler retries happen.</li>
+     * <li>4. The runnable completes; the poller persistent does not re-schedule
itself.</li>
+     * <li>5. System powered off and restarts; the poller job does not know what
the status
+     * of the runnable. This is handled just like the case #1.</li>
+     * </ul>
+     * 
+     * There is at least one re-scheduling of the poller job. Since, the runnable's
state is
+     * not persisted, and the same runnable may be tried again after system failure,
+     * the runnable that's used with this polling should be repeatable.
+     *
+     * @param job job to run.
+     */
+    protected void runPolledRunnable(final Job job) {
+        final Scheduler.JobInfo jobInfo = new Scheduler.JobInfo(job.jobId, job.detail,
+                (Integer)(job.detail.get("retry") != null ?
job.detail.get("retry") : 0));
+
+        _exec.submit(new Callable<Void>() {
+            public Void call() throws Exception {
+                try {
+                    execTransaction(new Callable<Void>() {
+                        public Void call() throws Exception {
+                            //if (!_db.deleteJob(job.jobId, _nodeId))
+                            //    throw new JobNoLongerInDbException(job.jobId,_nodeId);
+                            
+                            try {
+                                _polledRunnableProcessor.onScheduledJob(jobInfo);
+                                if(
!"COMPLETED".equals(String.valueOf(jobInfo.jobDetail.get("runnable_status")))
) {
+                                    // the runnable is still in progress, schedule
checker to 10 mins later
+                                    if( _pollIntervalForPolledRunnable < 0 ) {
+                                        if(__log.isWarnEnabled()) __log.warn("The
poll interval for polled runnables is negative; setting it to 1000ms");
+                                        _pollIntervalForPolledRunnable = 1000;
+                                    }
+                                    job.schedDate = System.currentTimeMillis() +
_pollIntervalForPolledRunnable;
+                                    //_db.insertJob(job, _nodeId, false);
+                                    addTodoOnCommit(job); // Added
+                                    System.out.println("GPB: WAS PERSISTENT JOB
INSERT HERE !!!");
+                                }
+                            } catch (JobProcessorException jpe) {
+                                if (jpe.retry) {
+                                    int retry = job.detail.get("retry") != null
? (((Integer)job.detail.get("retry")) + 1) : 0;
+                                    if (retry <= 10) {
+                                       // long delay = doRetry(job);
+                                        //__log.error("Error while processing
transaction, retrying in " + delay + "s");
+                                    } else {
+                                        __log.error("Error while processing
transaction after 10 retries, no more retries:"+job);
+                                    }
+                                } else {
+                                    __log.error("Error while processing transaction,
no retry.", jpe);
+                                }
+                                // Let execTransaction know that shit happened.
+                                throw jpe;
+                            }
+                            return null;
+                        }
+                    });
+                //} catch (JobNoLongerInDbException jde) {
+                    // This may happen if two node try to do the same job... we try to
avoid
+                    // it the synchronization is a best-effort but not perfect.
+                //    __log.debug("job no longer in db forced rollback.");
+                } catch (Exception ex) {
+                    __log.error("Error while executing transaction", ex);
+                }
+                return null;
+            }
+        });
+    }
+    
+    private void addTodoOnCommit(final Job job) {
+        registerSynchronizer(new Synchronizer() {
+            public void afterCompletion(boolean success) {
+                if (success) {
+                    _todo.enqueue(job);
+                }
+            }
+
+            public void beforeCompletion() {
+            }
+        });
+    }
+
+    public boolean isTransacted() {
+        try {
+            Transaction tx = _txm.getTransaction();
+            return (tx != null && tx.getStatus() !=
Status.STATUS_NO_TRANSACTION);
+        } catch (SystemException e) {
+            throw new ContextException("Internal Error: Could not obtain transaction
status.");
+        }
+    }
+
+    public void runTask(Task task) {
+        if (task instanceof Job) {
+            Job job = (Job)task;
+            if( job.detail.get("runnable") != null ) {
+                runPolledRunnable(job);
+            } else {
+                runJob((Job) task);
+            }
+        } else if (task instanceof SchedulerTask)
+            ((SchedulerTask) task).run();
+    }
+
+    public void updateHeartBeat(String nodeId) {
+        if (nodeId == null)
+            return;
+
+        if (_nodeId.equals(nodeId))
+            return;
+
+        _lastHeartBeat.put(nodeId, System.currentTimeMillis());
+        _knownNodes.add(nodeId);
+    }
+
+    boolean doLoadImmediate() {
+        __log.debug("LOAD IMMEDIATE started");
+        
+        /*
+        // don't load anything if we're already half-full;  we've got plenty
to do already
+        if (_todo.size() > _todoLimit/2) return true;
+        
+        List<Job> jobs;
+        try {
+            final int batch = (int) (_immediateInterval * _tps / 1000);
+            jobs = execTransaction(new Callable<List<Job>>() {
+                public List<Job> call() throws Exception {
+                    return _db.dequeueImmediate(_nodeId, System.currentTimeMillis() +
_immediateInterval, batch);
+                }
+            });
+            for (Job j : jobs) {
+                if (__log.isDebugEnabled())
+                    __log.debug("todo.enqueue job from db: " + j.jobId + "
for " + j.schedDate);
+
+                if (_todo.size() < _todoLimit) 
+                    _todo.enqueue(j);
+            }
+            return true;
+        } catch (Exception ex) {
+            __log.error("Error loading immediate jobs from database.", ex);
+            return false;
+        } finally {
+            __log.debug("LOAD IMMEDIATE complete");
+        }
+        */
+        return(false);
+    }
+
+    boolean doUpgrade() {
+        __log.debug("UPGRADE started");
+        /*
+        final ArrayList<String> knownNodes = new
ArrayList<String>(_knownNodes);
+        // Don't forget about self.
+        knownNodes.add(_nodeId);
+        Collections.sort(knownNodes);
+
+        // We're going to try to upgrade near future jobs using the db only.
+        // We assume that the distribution of the trailing digits in the
+        // scheduled time are uniformly distributed, and use modular division
+        // of the time by the number of nodes to create the node assignment.
+        // This can be done in a single update statement.
+        final long maxtime = System.currentTimeMillis() + _nearFutureInterval;
+        try {
+            return execTransaction(new Callable<Boolean>() {
+
+                public Boolean call() throws Exception {
+                    int numNodes = knownNodes.size();
+                    for (int i = 0; i < numNodes; ++i) {
+                        String node = knownNodes.get(i);
+                        _db.updateAssignToNode(node, i, numNodes, maxtime);
+                    }
+                    return true;
+                }
+
+            });
+
+        } catch (Exception ex) {
+            __log.error("Database error upgrading jobs.", ex);
+            return false;
+        } finally {
+            __log.debug("UPGRADE complete");
+        }
+*/
+        return(false);
+    }
+
+    /**
+     * Re-assign stale node's jobs to self.
+     * @param nodeId
+     */
+    void recoverStaleNode(final String nodeId) {
+        __log.debug("recovering stale node " + nodeId);
+        /*
+        try {
+            int numrows = execTransaction(new Callable<Integer>() {
+                public Integer call() throws Exception {
+                    return _db.updateReassign(nodeId, _nodeId);
+                }
+            });
+
+            __log.debug("reassigned " + numrows + " jobs to self.
");
+
+            // We can now forget about this node, if we see it again, it will be
+            // "new to us"
+            _knownNodes.remove(nodeId);
+            _lastHeartBeat.remove(nodeId);
+
+            // Force a load-immediate to catch anything new from the recovered node.
+            doLoadImmediate();
+
+        } catch (Exception ex) {
+            __log.error("Database error reassigning node.", ex);
+        } finally {
+            __log.debug("node recovery complete");
+        }
+*/
+    }
+
+    /*
+    private long doRetry(Job job) throws DatabaseException {
+        int retry = job.detail.get("retry") != null ?
(((Integer)job.detail.get("retry")) + 1) : 0;
+        job.detail.put("retry", retry);
+        long delay = (long)(Math.pow(5, retry));
+        Job jobRetry = new Job(System.currentTimeMillis() + delay*1000, true,
job.detail);
+        _db.insertJob(jobRetry, _nodeId, false);
+        return delay;
+    }
+    */
+
+    private abstract class SchedulerTask extends Task implements Runnable {
+        SchedulerTask(long schedDate) {
+            super(schedDate);
+        }
+    }
+
+    private class LoadImmediateTask extends SchedulerTask {
+        LoadImmediateTask(long schedDate) {
+            super(schedDate);
+        }
+
+        public void run() {
+            boolean success = false;
+            try {
+                success = doLoadImmediate();
+            } finally {
+                if (success)
+                    _todo.enqueue(new LoadImmediateTask(System.currentTimeMillis() +
(long) (_immediateInterval * .90)));
+                else
+                    _todo.enqueue(new LoadImmediateTask(System.currentTimeMillis() +
1000));
+            }
+        }
+
+    }
+
+    /**
+     * Upgrade jobs from far future to immediate future (basically, assign them to a
node).
+     * @author mszefler
+     *
+     */
+    private class UpgradeJobsTask extends SchedulerTask {
+        UpgradeJobsTask(long schedDate) {
+            super(schedDate);
+        }
+
+        public void run() {
+            long ctime = System.currentTimeMillis();
+            long ntime = _nextUpgrade.get();
+            __log.debug("UPGRADE task for " + schedDate + " fired at
" + ctime);
+
+            // We could be too early, this can happen if upgrade gets delayed due to
another
+            // node
+            if (_nextUpgrade.get() > System.currentTimeMillis()) {
+                __log.debug("UPGRADE skipped -- wait another " + (ntime -
ctime) + "ms");
+                _todo.enqueue(new UpgradeJobsTask(ntime));
+                return;
+            }
+
+            boolean success = false;
+            try {
+                success = doUpgrade();
+            } finally {
+                long future = System.currentTimeMillis() + (success ? (long)
(_nearFutureInterval * .50) : 1000);
+                _nextUpgrade.set(future);
+                _todo.enqueue(new UpgradeJobsTask(future));
+                __log.debug("UPGRADE completed, success = " + success + ";
next time in " + (future - ctime) + "ms");
+            }
+        }
+    }
+
+    /**
+     * Check if any of the nodes in our cluster are stale.
+     */
+    private class CheckStaleNodes extends SchedulerTask {
+        CheckStaleNodes(long schedDate) {
+            super(schedDate);
+        }
+
+        public void run() {
+            _todo.enqueue(new CheckStaleNodes(System.currentTimeMillis() +
_staleInterval));
+            __log.debug("CHECK STALE NODES started");
+            for (String nodeId : _knownNodes) {
+                Long lastSeen = _lastHeartBeat.get(nodeId);
+                if ((lastSeen == null || (System.currentTimeMillis() - lastSeen) >
_staleInterval)
+                    && !_nodeId.equals(nodeId))
+                {
+                    recoverStaleNode(nodeId);
+                }
+            }
+        }
+    }
+}


    

2024

2023

2022

2021

2020

2019

2018

2017

2016

2015

2014

2013

2012

2011

2010

2009

[riftsaw-commits] riftsaw SVN: r618 - in branches/DBIssue/runtime/engine/src/main/java/org/jboss/soa/bpel/runtime/engine/ode: scheduler and 1 other directory.