ResourceWorkManager.java

/*
 * JOnAS: Java(TM) Open Application Server
 * Copyright (C) 1999 Bull S.A.
 * Contact: jonas-team@objectweb.org
 *
 * This library is free software; you can redistribute it and/or
 * modify it under the terms of the GNU Lesser General Public
 * License as published by the Free Software Foundation; either
 * version 2.1 of the License, or any later version.
 *
 * This library is distributed in the hope that it will be useful,
 * but WITHOUT ANY WARRANTY; without even the implied warranty of
 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
 * Lesser General Public License for more details.
 *
 * You should have received a copy of the GNU Lesser General Public
 * License along with this library; if not, write to the Free Software
 * Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA  02111-1307
 * USA
 *
 * Initial developer(s): Eric Hardesty
 * Contributor(s): ______________________________________.
 *
 * --------------------------------------------------------------------------
 * $Id: ResourceWorkManager.java,v 1.10 2005/02/13 23:51:02 ehardesty Exp $
 * --------------------------------------------------------------------------
 */
package org.objectweb.jonas.resource;

import java.util.Collections;
import java.util.HashMap;
import java.util.LinkedList;
import java.util.Map;
import java.util.Vector;

import javax.resource.spi.work.ExecutionContext;
import javax.resource.spi.work.Work;
import javax.resource.spi.work.WorkCompletedException;
import javax.resource.spi.work.WorkEvent;
import javax.resource.spi.work.WorkException;
import javax.resource.spi.work.WorkListener;
import javax.resource.spi.work.WorkManager;
import javax.resource.spi.work.WorkRejectedException;
import javax.transaction.NotSupportedException;
import javax.transaction.SystemException;
import javax.transaction.xa.Xid;

import org.objectweb.jonas_ejb.container.TraceEjb;
import org.objectweb.jotm.Current;
import org.objectweb.transaction.jta.TransactionManager;
import org.objectweb.util.monolog.api.BasicLevel;
import org.objectweb.util.monolog.api.Logger;
public class ResourceWorkManager implements WorkManager {

    private boolean      inited;
    private int          minThreads;
    private int          maxThreads;
    private long         waitTimeout;
    private long         execTimeout;
    private Vector       vThreads;

    private LinkedList   workObjs;
    private int freeThreads;  // free threads ready to work

    private final static int WO_REJECTED = 1;
    private final static int WO_ACCEPTED = 2;
    private final static int WO_STARTED = 3;
    private final static int WO_COMPLETED = 4;

    // Loggers used in the ResourceService
    private static Logger logger = null;

    private TransactionManager tm = null ;

    // Static hashtable: Xid ---> work Object
    private static Map wrkXids = Collections.synchronizedMap(new HashMap());

    public ResourceWorkManager(TransactionManager tmgr, int minThreads, int maxThreads, 
                                int waitTimeout, int execTimeout, Logger mlogger) {
        // This was set to false to only allow Work objects once the RAR has been deployed, but
        // I think that is too restrictive, so I'm setting it to true which allows things to start
        // up right away.  I'm not sure if this is correct, but necessary until the spec is fixed
        // to allow for a ResourceAdapter prepare and then start later.
        inited      = true;
        this.minThreads  = minThreads;
        this.maxThreads  = maxThreads;
        this.waitTimeout = waitTimeout * 1000;
        this.execTimeout = execTimeout * 1000;
        vThreads    = new Vector();
        workObjs    = new LinkedList();
        tm          = tmgr;
        logger      = mlogger;

        // Start the specified number of threads
        for (int i = 0; i < minThreads; i++) {
            WorkerPool wp = new WorkerPool();
            vThreads.add(wp);
            wp.start();
        }

        // Need the Monitor thread to check on timeouts
        (new RWManagerMonitor()).start();
    }


    public void doWork(Work work) throws WorkException {
        doWork(work, INDEFINITE, null, null);
    }

    public void doWork(Work work, long timeout, ExecutionContext eCtx,
                        WorkListener wListen) throws WorkException {

        logger.log(BasicLevel.DEBUG, "work=" + work + " timeout=" + timeout
                   + " execCtx=" + eCtx + " wrkListen=" + wListen);
        int rStat = WorkEvent.WORK_COMPLETED;
        WorkObject wo = new WorkObject(work, timeout, eCtx, wListen, rStat);
        synchronized (wo.lock) {
            add(wo, rStat);
            try {
                wo.lock.wait();
            } catch (InterruptedException ie) {
            }
        }
        if (wo.wException != null) {
            logger.log(BasicLevel.ERROR, "" + wo.wException);
            throw wo.wException;
        }
    }

    public long startWork(Work work) throws WorkException {
        return startWork(work, INDEFINITE, null, null);
    }

    public long startWork(Work work, long timeout, ExecutionContext eCtx,
                           WorkListener wListen) throws WorkException {

        logger.log(BasicLevel.DEBUG, "work=" + work + " timeout=" + timeout
                   + " execCtx=" + eCtx + " wrkListen=" + wListen);
        int rStat = WorkEvent.WORK_STARTED;
        WorkObject wo = new WorkObject(work, timeout, eCtx, wListen, rStat);
        synchronized (wo.lock) {
            add(wo, rStat);
            try {
                wo.lock.wait();
            } catch (InterruptedException ie) {
            }
        }

        if (wo.wException != null) {
            throw wo.wException;
        }
        return (wo.startTime - wo.acceptTime);
    }

    public void scheduleWork(Work work) throws WorkException {
        scheduleWork(work, INDEFINITE, null, null);
    }

    public void scheduleWork(Work work, long timeout, 
                              ExecutionContext eCtx,
                              WorkListener wListen) throws WorkException {

        logger.log(BasicLevel.DEBUG, "work=" + work + " timeout=" + timeout
                   + " execCtx=" + eCtx + " wrkListen=" + wListen);
        int rStat = WorkEvent.WORK_ACCEPTED;
        WorkObject wo = new WorkObject(work, timeout, eCtx, wListen, rStat);
        add(wo, rStat);
    }

    public void setInited() {
        if (!inited) {
            synchronized (workObjs) {
                inited = true;
                workObjs.notifyAll();
            }
        }
    }

    public void releasePool() {
        for (int i = 0; i < vThreads.size(); i++) {
            WorkerPool wp = (WorkerPool) vThreads.get(i);
            wp.shutdown = true;
        }
        synchronized (workObjs) {
            workObjs.notifyAll();
        }
    }

    private void add(WorkObject wo, int rStat) throws WorkException {
        // Only accept scheduleWork objects before deployment is complete
        if (!inited) {
            if (rStat != WorkEvent.WORK_ACCEPTED) {
                WorkRejectedException wex = new WorkRejectedException(
                      "Cannot call doWork or startWork before RA is deployed",
                      WorkException.INTERNAL);
                WorkEvent we = new WorkEvent(wo.work, WorkEvent.WORK_REJECTED, wo.work, wex);
                if (wo.wListen != null) {
                    wo.wListen.workRejected(we);
                }
                throw wex;
            }
        }

        logger.log(BasicLevel.DEBUG, "workobj=" + wo + " rStat=" + rStat);

        // Add last to list, set accepted time and send the event if necessary
        synchronized (workObjs) {
            // If an EnvironmentContext and an Xid exists, then check Xid/Work hash
            if (wo.eCtx != null && wo.eCtx.getXid() != null) {
                if (wrkXids.containsKey(wo.eCtx.getXid())) {
                    WorkRejectedException wex = new WorkRejectedException(
                          "Xid is already being processed by another work object",
                          WorkException.TX_CONCURRENT_WORK_DISALLOWED);
                    WorkEvent we = new WorkEvent(wo.work, WorkEvent.WORK_REJECTED, wo.work, wex);
                    if (wo.wListen != null) {
                        wo.wListen.workRejected(we);
                    }
                    logger.log(BasicLevel.ERROR, "" + wex);
                    throw wex;
                }
                wrkXids.put(wo.eCtx.getXid(), wo.work);
            }
            workObjs.addLast(wo);
            wo.acceptTime = System.currentTimeMillis();
            if (wo.wListen != null) {
                WorkEvent we = new WorkEvent(wo.work, WorkEvent.WORK_ACCEPTED, wo.work, null);
                wo.wListen.workAccepted(we);
            }
            if (freeThreads <= 0) {
                WorkerPool wp = new WorkerPool();
                vThreads.add(wp);
                wp.start();
            }
            workObjs.notify();
        }
    }


    // Inner classes for ResourceWorkManager
    /* Wrapper class for the Work object, keeps state and time information
     */
    private class WorkObject {
        // Work object information
        Work work;
        ExecutionContext eCtx;
        WorkListener wListen;
        WorkException wException;

        // Time values
        long acceptTime;
        long executionTimeout;
        long startTime;
        long timeout;

        // State information
        Object lock;
        int rState;
        int cState;
        boolean rejected;
        
        public WorkObject(Work work, long timeout,
                           ExecutionContext eCtx,
                           WorkListener wListen,
                           int rState) {

            this.eCtx = eCtx;
            this.work = work;
            wException = null;
            this.wListen = wListen;
            
            executionTimeout = 0;
            acceptTime = 0;
            startTime = 0;
            long curTime = System.currentTimeMillis();
            if (timeout == INDEFINITE) {
                this.timeout = timeout;
            } else {
                this.timeout = curTime + timeout;
            }
            if (execTimeout > 0) {
                executionTimeout = curTime + execTimeout;
            }

            lock = new Object();
            this.rState = rState;
            cState = -1;
            rejected = false;
        }
    }


    // Thread pool runs the work objects
    private class WorkerPool extends Thread {
        WorkObject wo = null;
        boolean shutdown = false;

        public WorkerPool() {
            super();
            setDaemon(true);
            logger.log(BasicLevel.DEBUG, getName() + " constructor");
        }

        public void run() {

            while (true) {
                synchronized (workObjs) {
                    boolean waited = false;
                    while (workObjs.isEmpty()) {
                        // If a thread has waited and not gotten any work then there 
                        // should only be minThreads waiting, so if freeThreads is equal
                        // to minThreads then this thread should be stopped
                        if ((waited && freeThreads >= minThreads) || shutdown) {
                            logger.log(BasicLevel.DEBUG, getName() + " ending");
                            return;
                        }
                        try {
                            freeThreads++;
                            workObjs.wait(waitTimeout);
                            freeThreads--;
                            waited = true;
                        } catch (InterruptedException ignored) {
                            freeThreads--;
                            return;
                        }
                    }
                    if (inited) {
                        wo = (WorkObject) workObjs.removeFirst();
                    }
                }

                if (inited && wo != null) {
                    try {
                        checkTime();
                        if (wo.rState == WorkEvent.WORK_STARTED) {
                            logger.log(BasicLevel.DEBUG, "Start notification for " + wo);
                            synchronized (wo.lock) {
                                wo.lock.notify();  // If waiting for execution start
                            }
                        }
                        if (wo.wListen != null) {
                            WorkEvent we = new WorkEvent(wo.work, WorkEvent.WORK_STARTED,
                                                         wo.work, null, wo.startTime);
                            wo.wListen.workStarted(we);
                        }
                        // Need to setup up ExecutionContext if necessary
                        if (wo.eCtx != null) {
                             long timeout = wo.eCtx.getTransactionTimeout();
                             Xid xid = wo.eCtx.getXid();
                             if (xid != null) {
                                 try {
                                     if (timeout != WorkManager.UNKNOWN) {
                                         ((Current) tm).begin(xid, timeout);
                                     } else {
                                         ((Current) tm).begin(xid);
                                     }
                                 } catch (NotSupportedException e) {
                                     logger.log(BasicLevel.ERROR, "cannot start a transaction: NotSupportedException");
                                     WorkException wex = new WorkException(
                                           "Error starting a new transaction", e);
                                     throw wex;
                                 } catch (SystemException e) {
                                     logger.log(BasicLevel.ERROR, "cannot start a transaction:\n", e);
                                     WorkException wex = new WorkException(
                                           "Error starting a new transaction", e);
                                     throw wex;
                                 }
                             }
                        }
                        logger.log(BasicLevel.DEBUG, "Thread: " + getName() + " Running " + wo.work);
                        wo.work.run();
                        logger.log(BasicLevel.DEBUG, "Completed " + wo.work);
                        wo.wException = null;
                    } catch (WorkException wex) {
                        wo.wException = wex;
                    } catch (Throwable th) {
                        // Need to report issue and wrap with WorkException
                        WorkCompletedException wex = new WorkCompletedException(
                              "Error from work object", th);
                        wo.wException = wex;
                        logger.log(BasicLevel.ERROR, "" + wex);
                    } finally {
                        if (wo.eCtx != null && wo.eCtx.getXid() != null) {
                            wrkXids.remove(wo.eCtx.getXid());
                        }
                        if (wo.wListen != null && !wo.rejected) {
                            WorkEvent we = new WorkEvent(wo.work, WorkEvent.WORK_COMPLETED,
                                                         wo.work, wo.wException, wo.startTime);
                            wo.wListen.workCompleted(we);
                        }
                        if (wo.rState == WorkEvent.WORK_COMPLETED) {
                            synchronized (wo.lock) {
                                wo.lock.notify();  // If waiting for execution completion
                            }
                        }
                        // Make sure that our ThreadLocal storage from JOTM is cleared
                        // Another thread could have completed the transaction for us.
                        ((Current) tm).clearThreadTx();
                        wo.work = null;
                    }
                }
            }
        }

        private void checkTime() throws WorkException{

            long curTime = System.currentTimeMillis();

            if (curTime > wo.timeout) {
                WorkException wex = new WorkException(
                      "Error starting work in specified time", WorkException.START_TIMED_OUT);
                if (wo.wListen != null) {
                    WorkEvent we = new WorkEvent(wo.work, WorkEvent.WORK_REJECTED,
                                                 wo.work, wex);
                    wo.wListen.workRejected(we);
                }
                wo.rejected = true;
                logger.log(BasicLevel.ERROR, "" + wex);
                throw wex;
            }
            wo.startTime = curTime;
        }
    }


    // ResourceWorkManager Monitor thread to check for timeouts of work objects
    private class RWManagerMonitor extends Thread {

        public RWManagerMonitor() {
            super("RWManagerMonitor");
            setDaemon(true);
        }

        public void run() {
            int  cnt = 0;
            long curTime;
            while (true) {
                try {
                    sleep(100); // 100 ms
                    cnt++;
                    if (!workObjs.isEmpty()) {
                        for (int i = 0; i < workObjs.size(); i++) {
                            WorkObject wo = (WorkObject) workObjs.get(i);
                            curTime = System.currentTimeMillis();
                            if (curTime > wo.timeout) {
                                synchronized (workObjs) {
                                    if (workObjs.remove(wo)) {
                                        WorkException wex = new WorkException(
                                              "Error starting work in specified time",
                                              WorkException.START_TIMED_OUT);
                                        if (wo.wListen != null) {
                                            WorkEvent we = new WorkEvent(wo.work, WorkEvent.WORK_REJECTED,
                                                                         wo.work, wex);
                                            wo.wListen.workRejected(we);
                                        }
                                        wo.wException = wex;
                                        throw wex;
                                    }
                                }
                            }
                        }
                    }

                    // Check threads for execution timeout or thread issue every second
                    if (cnt >= 10) {
                        cnt = 0;
                        curTime = System.currentTimeMillis();
                        for (int i = 0; i < vThreads.size(); i++) {
                            WorkerPool wp = (WorkerPool) vThreads.get(i);
                            if (!wp.isAlive() && !wp.shutdown) {
                                vThreads.remove(i);
                                if (vThreads.size() < minThreads) {
                                    WorkerPool nwp = new WorkerPool();
                                    vThreads.add(nwp);
                                    nwp.start();
                                }
                            } else if (wp.wo != null && wp.wo.executionTimeout > 0
                                       && curTime > wp.wo.executionTimeout) {
                                if (wp.wo.work != null) {
                                    wp.wo.work.release();
                                }
                            }
                        }
                    }
                } catch (Exception ex) {
                    logger.log(BasicLevel.ERROR, ex.toString());
                }
            }
        }
    }

}

---