/**
 * JOnAS: Java(TM) Open Application Server
 * Copyright (C) 1999-2007 Bull S.A.S
 * Contact: jonas-team@ow2.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
 *
 * --------------------------------------------------------------------------
 * $Id: JWorkManager.java 21566 2011-08-08 12:28:12Z cazauxj $
 * --------------------------------------------------------------------------
 */


package org.ow2.jonas.workmanager.internal;

import java.util.LinkedList;
import java.util.List;

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.util.monolog.api.BasicLevel;
import org.objectweb.util.monolog.api.Logger;
import org.ow2.jonas.lib.management.javaee.ManagedObject;
import org.ow2.jonas.lib.management.reconfig.PropertiesConfigurationData;
import org.ow2.jonas.lib.util.Log;
import org.ow2.jonas.tm.TransactionService;



/**
 * Jonas Implementation of the Resource Work Manager
 * @author durieuxp
 */
public class JWorkManager extends ManagedObject implements WorkManager {

    /**
     *  Name as used to label configuration properties.
     */
    public static final String SERVICE_NAME = "workmanager";

    // The service configuration properties
    /**
     * name of the 'minworkthreads' configuration parameter
     */
    static final String MINPOOLSIZE = "jonas.service.wm.minworkthreads";
    /**
     * name of the 'maxworkthreads' configuration parameter
     */
    static final String MAXPOOLSIZE = "jonas.service.wm.maxworkthreads";

    protected List workList = new LinkedList();

    protected static int poolnumber = 0;
    protected static int threadnumber = 0;

    protected int maxpoolsz;
    protected int minpoolsz;
    protected int poolsz; // current size of thread pool
    protected int freeThreads;  // free threads ready to work
    protected long waitingTime; // in millisec

    protected boolean valid = true; // set to false when WorkManager is removed.

    protected static final long FEW_MORE_SECONDS = 3000;

    private static Logger logger = null;

    private TransactionService transactionService;

    /**
     * Value used as sequence number by reconfiguration notifications.
     */
    private long sequenceNumber;

    /**
     * Constructor
     * @param threadwait max time in seconds a thread will wait
     */
    public JWorkManager(final int minsz, final int maxsz, final TransactionService transactionService, final long threadwait) {
        minpoolsz = minsz;
        maxpoolsz = maxsz;
        waitingTime = threadwait * 1000L;
        this.transactionService = transactionService;
        poolnumber++;
        logger = Log.getLogger(Log.JONAS_WORK_MGR_PREFIX);
        if (logger.isLoggable(BasicLevel.DEBUG)) {
            logger.log(BasicLevel.DEBUG, "thread pool #" + poolnumber);
            logger.log(BasicLevel.DEBUG, "minpoolsz = " + minsz + " maxpoolsz = " + maxsz);
        }
        for (poolsz = 0; poolsz < minsz; poolsz++) {
            WorkThread st = new WorkThread(this, threadnumber++, poolnumber);
            st.start();
        }
        sequenceNumber = 0;
    }

    // --------------------------------------------------------------------------
    // Management
    // --------------------------------------------------------------------------

    /**
     * @return current pool size
     */
    public int getCurrentPoolSize() {
        return poolsz;
    }

    /**
     * @return min pool size
     */
    public int getMinPoolSize() {
        return minpoolsz;
    }

    /**
     * @return max pool size
     */
    public int getMaxPoolSize() {
        return maxpoolsz;
    }

    /**
     * Set the min pool size
     * @param minsz
     */
    public void setMinPoolSize(final int minsz) {
        minpoolsz = minsz;
        // Send a notification containing the new value of this property to the
        // listener MBean
        String propName = MINPOOLSIZE;
        String propValue = new Integer(minsz).toString();
        sendReconfigNotification(getSequenceNumber(), SERVICE_NAME, new PropertiesConfigurationData(propName, propValue));
    }

    /**
     * Set the max pool size
     * @param maxsz
     */
    public void setMaxPoolSize(final int maxsz) {
        maxpoolsz = maxsz;
        // Send a notification containing the new value of this property to the
        // listener MBean
        String propName = MAXPOOLSIZE;
        String propValue = new Integer(maxsz).toString();
        sendReconfigNotification(getSequenceNumber(), SERVICE_NAME, new PropertiesConfigurationData(propName, propValue));
    }

    /**
     * Save updated configuration.
     */
    public void saveConfig() {
        // Send save reconfig notification
        sendSaveNotification(getSequenceNumber(), SERVICE_NAME);
    }

    /**
     * Return a sequence number and increase this number.
     * @return a sequence number
     */
    protected long getSequenceNumber() {
        return ++sequenceNumber;
    }
    // --------------------------------------------------------------------------
    // WorkManager implementation
    // --------------------------------------------------------------------------

    /**
     * Accepts a Work instance for processing.
     * This call blocks until the Work instance completes execution.
     * There is no guarantee on when the accepted Work instance would start execution ie.,
     * there is no time constraint to start execution.
     * @param work The unit of work to be done. Could be long or short-lived.
     * @throws WorkRejectedException a Work instance has been rejected from further processing.
     * @throws WorkCompletedException a Work instance has completed execution with an exception.
     * @throws WorkException
     */
    public void doWork(final Work work) throws WorkException {
        doMyWork(work, INDEFINITE, null, null, 0);
    }

    /**
     * Accepts a Work instance for processing. This call blocks until the Work
     * instance completes execution.
     * @param work The unit of work to be done. Could be long or short-lived.
     * @param timeout a time duration (in milliseconds) within which the
     *        execution of the Work instance must start. Otherwise, the Work
     *        instance is rejected with a WorkRejectedException set to an
     *        appropriate error code (WorkRejectedException.TIMED_OUT).
     * @param ectx an object containing the execution context with which the
     *        submitted Work instance must be executed.
     * @param listener an object which would be notified when the various Work
     *        processing events (work accepted, work rejected, work started,
     *        work completed) occur.
     * @throws WorkRejectedException a Work instance has been rejected from
     *         further processing.
     * @throws WorkCompletedException a Work instance has completed execution
     *         with an exception.
     * @throws WorkException
     */
    public void doWork(final Work work, final long timeout, final ExecutionContext ectx, final WorkListener listener) throws WorkException {
        if (logger.isLoggable(BasicLevel.DEBUG)) {
            logger.log(BasicLevel.DEBUG, "");
        }
        if (listener != null) {
            listener.workAccepted(new WorkEvent(this, WorkEvent.WORK_ACCEPTED, work, null));
        }
        doMyWork(work, timeout, ectx, listener, System.currentTimeMillis());
    }

    /**
     * Accepts a Work instance for processing. This call blocks until the Work
     * instance starts execution but not until its completion. There is no
     * guarantee on when the accepted Work instance would start execution ie.,
     * there is no time constraint to start execution.
     * @param work The unit of work to be done. Could be long or short-lived.
     * @return the time elapsed (in milliseconds) from Work acceptance until
     *         start of execution. Note, this does not offer real-time
     *         guarantees. It is valid to return -1, if the actual start delay
     *         duration is unknown.
     * @throws WorkRejectedException a Work instance has been rejected from
     *         further processing.
     * @throws WorkException
     */
    public long startWork(final Work work) throws WorkException {
        return startWork(work, INDEFINITE, null, null);
    }

    /**
     * Accepts a Work instance for processing. This call blocks until the Work
     * instance starts execution but not until its completion. There is no
     * guarantee on when the accepted Work instance would start execution ie.,
     * there is no time constraint to start execution.
     * @param work The unit of work to be done. Could be long or short-lived.
     * @param timeout a time duration (in milliseconds) within which the
     *        execution of the Work instance must start. Otherwise, the Work
     *        instance is rejected with a WorkRejectedException set to an
     *        appropriate error code (WorkRejectedException.TIMED_OUT).
     * @param ectx an object containing the execution context with which the
     *        submitted Work instance must be executed.
     * @param listener an object which would be notified when the various Work
     *        processing events (work accepted, work rejected, work started,
     *        work completed) occur.
     * @return the time elapsed (in milliseconds) from Work acceptance until
     *         start of execution. Note, this does not offer real-time
     *         guarantees. It is valid to return -1, if the actual start delay
     *         duration is unknown.
     * @throws WorkRejectedException a Work instance has been rejected from
     *         further processing.
     * @throws WorkException
     */
    public long startWork(final Work work, final long timeout, final ExecutionContext ectx, final WorkListener listener) throws WorkException {
        if (logger.isLoggable(BasicLevel.DEBUG)) {
            logger.log(BasicLevel.DEBUG, "");
        }
        JWork mywork = new JWork(work, timeout, ectx, listener);
        if (listener != null) {
            listener.workAccepted(new WorkEvent(this, WorkEvent.WORK_ACCEPTED, work, null));
        }
        long starttime = System.currentTimeMillis();
        long duration = 0;
        synchronized (workList) {
            workList.add(mywork);
            if (poolsz < maxpoolsz && workList.size() > freeThreads) {
                // We need one more thread.
                poolsz++;
                WorkThread st = new WorkThread(this, threadnumber++, poolnumber);
                st.start();
            } else {
                workList.notify();
            }
        }
        // Wait until my work is started.
        boolean started = false;
        synchronized (mywork) {
            if (!mywork.isStarted()) {
                try {
                    // No need to wait after timeout is elapsed
                    long waittime = waitingTime;
                    if (timeout < waittime) {
                        waittime = timeout + FEW_MORE_SECONDS;
                    }
                    mywork.wait(waittime);
                } catch (InterruptedException e) {
                    throw new WorkRejectedException("Interrupted");
                }
            }
            started = mywork.isStarted();
        }
        duration = System.currentTimeMillis() - starttime;
        if (!started) {
            synchronized (workList) {
                // Remove the work in the list
                if (! workList.remove(mywork)) {
                    if (logger.isLoggable(BasicLevel.DEBUG)) {
                        logger.log(BasicLevel.DEBUG, "cannot remove work");
                    }
                }
                throw new WorkRejectedException(WorkException.START_TIMED_OUT);
            }
        }
        return duration;
    }

    /**
     * Accepts a Work instance for processing. This call does not block and
     * returns immediately once a Work instance has been accepted for
     * processing. There is no guarantee on when the submitted Work instance
     * would start execution ie., there is no time constraint to start
     * execution.
     * @param work The unit of work to be done. Could be long or short-lived.
     * @param timeout a time duration (in milliseconds) within which the
     *        execution of the Work instance must start. Otherwise, the Work
     *        instance is rejected with a WorkRejectedException set to an
     *        appropriate error code (WorkRejectedException.TIMED_OUT).
     * @param ectx an object containing the execution context with which the
     *        submitted Work instance must be executed.
     * @param listener an object which would be notified when the various Work
     *        processing events (work accepted, work rejected, work started,
     *        work completed) occur.
     * @throws WorkRejectedException a Work instance has been rejected from
     *         further processing.
     * @throws WorkException
     */
    public void scheduleWork(final Work work) throws WorkException {
        scheduleWork(work, INDEFINITE, null, null);
    }

    /**
     * Accepts a Work instance for processing. This call does not block and
     * returns immediately once a Work instance has been accepted for
     * processing. There is no guarantee on when the submitted Work instance
     * would start execution ie., there is no time constraint to start
     * execution.
     * @param work The unit of work to be done. Could be long or short-lived.
     * @param timeout a time duration (in milliseconds) within which the
     *        execution of the Work instance must start. Otherwise, the Work
     *        instance is rejected with a WorkRejectedException set to an
     *        appropriate error code (WorkRejectedException.TIMED_OUT).
     * @param ectx an object containing the execution context with which the
     *        submitted Work instance must be executed.
     * @param listener an object which would be notified when the various Work
     *        processing events (work accepted, work rejected, work started,
     *        work completed) occur.
     * @throws WorkRejectedException a Work instance has been rejected from
     *         further processing.
     * @throws WorkException
     */
    public void scheduleWork(final Work work, final long timeout, final ExecutionContext ectx, final WorkListener listener) throws WorkException {
        if (logger.isLoggable(BasicLevel.DEBUG)) {
            logger.log(BasicLevel.DEBUG, "");
        }
        JWork mywork = new JWork(work, timeout, ectx, listener);
        if (listener != null) {
            listener.workAccepted(new WorkEvent(this, WorkEvent.WORK_ACCEPTED, work, null));
        }
         synchronized (workList) {
            workList.add(mywork);
            if (poolsz < maxpoolsz && workList.size() > freeThreads) {
                // We need one more thread.
                poolsz++;
                WorkThread st = new WorkThread(this, threadnumber++, poolnumber);
                st.start();
            } else {
                // Just wake up a thread waiting for work.
                workList.notify();
            }
        }
    }

    /**
     * Internal method doing the work.
     */
    private void doMyWork(final Work work, final long timeout, final ExecutionContext ectx, final WorkListener listener, final long creationTime) throws WorkException {
        if (logger.isLoggable(BasicLevel.DEBUG)) {
            logger.log(BasicLevel.DEBUG, "timeout=" + timeout);
        }
        // Notify the listener that the work is started or rejected by timeout.
        if (listener != null) {
            long duration = System.currentTimeMillis() - creationTime;
            if (duration > timeout) {
                // This can occur only in case of scheduleWork
                logger.log(BasicLevel.WARN, "REJECTED: duration=" + duration);
               listener.workRejected(new WorkEvent(this, WorkEvent.WORK_REJECTED, work, null));
                return;
            }
            listener.workStarted(new WorkEvent(this, WorkEvent.WORK_STARTED, work, null));
        }

        // Setup ExecutionContext
        // TODO: Check this if doWork (same thread)
        Xid xid = null;
        if (ectx != null) {
             xid = ectx.getXid();
             if (xid != null) {
                 long txtimeout = ectx.getTransactionTimeout();
                 try {
                     transactionService.attachTransaction(xid, txtimeout);
                 } catch (NotSupportedException e) {
                     throw new WorkException("Error starting a new transaction", e);
                 } catch (SystemException e) {
                     throw new WorkException("Error starting a new transaction", e);
                 }
             }
        }

        ClassLoader oldCL = Thread.currentThread().getContextClassLoader();
        try {
            Thread.currentThread().setContextClassLoader(work.getClass().getClassLoader());
            work.run();
            // Notify the listener that the work is completed.
            if (listener != null) {
                listener.workCompleted(new WorkEvent(this, WorkEvent.WORK_COMPLETED, work, null));
            }
        } catch (Exception e) {
            if (listener != null) {
                listener.workCompleted(new WorkEvent(this, WorkEvent.WORK_COMPLETED, work, null));
            }
            throw new WorkCompletedException(e);
        } finally {
            Thread.currentThread().setContextClassLoader(oldCL);
            if (xid != null) {
                transactionService.detachTransaction();
            }
        }
    }

    /**
     * Get the next JWork object to be run.
     * @return next JWork object to be run, or null if thread must end.
     */
    public void nextWork() throws WorkException, InterruptedException {
        JWork run = null;
        boolean haswait = false;
        synchronized (workList) {
            while (workList.isEmpty()) {
                if ((haswait && freeThreads > minpoolsz) || !valid) {
                    poolsz--;
                    throw new InterruptedException("Thread ending");
                }
                try {
                    freeThreads++;
                    if (logger.isLoggable(BasicLevel.DEBUG)) {
                        logger.log(BasicLevel.DEBUG, "waiting");
                    }
                    workList.wait(waitingTime);
                    if (logger.isLoggable(BasicLevel.DEBUG)) {
                        logger.log(BasicLevel.DEBUG, "notified");
                    }
                    freeThreads--;
                    haswait = true;
                } catch (InterruptedException e) {
                    freeThreads--;
                    poolsz--;
                    throw e;
                }
            }
            run = (JWork) workList.remove(0);
            // In case startWork() was called
            synchronized(run) {
                if (logger.isLoggable(BasicLevel.DEBUG)) {
                    logger.log(BasicLevel.DEBUG, "start new work");
                }
                run.setStarted();
                run.notify();
            }
        }
        doMyWork(run.getWork(),
                 run.getTimeout(),
                 run.getExecutionContext(),
                 run.getWorkListener(),
                 run.getCreationTime());
    }

    /**
     * Remove this WorkManager : Stop all threads
     */
    public synchronized void stopThreads() {
        if (logger.isLoggable(BasicLevel.DEBUG)) {
            logger.log(BasicLevel.DEBUG, "");
        }
        valid = false;
        notifyAll();
        poolnumber--;
    }

    class JWork {
        private Work work;
        private long timeout;
        private ExecutionContext ectx;
        private WorkListener listener;
        private long creationTime;
        private boolean started = false;

        public JWork(final Work work, final long timeout, final ExecutionContext ectx, final WorkListener listener) {
            this.work = work;
            this.timeout = timeout;
            this.ectx = ectx;
            this.listener = listener;
            creationTime = System.currentTimeMillis();
            if (logger.isLoggable(BasicLevel.DEBUG)) {
                logger.log(BasicLevel.DEBUG, "timeout=" + timeout);
            }
        }

        public Work getWork() {
            return work;
        }

        public long getTimeout() {
            return timeout;
        }

        public ExecutionContext getExecutionContext() {
            return ectx;
        }

        public WorkListener getWorkListener() {
            return listener;
        }

        public long getCreationTime() {
            return creationTime;
        }

        public boolean isStarted() {
            return started;
        }

        public void setStarted() {
            if (logger.isLoggable(BasicLevel.DEBUG)) {
                logger.log(BasicLevel.DEBUG, "");
            }
            started = true;
        }
    }

    /**
     * Thread executing works for the work manager.
     */
    class WorkThread extends Thread {

        private JWorkManager mgr;
        private int number;

        /**
         * Constructor
         * @param m The WorkManager
         * @param num thread number
         * @param wm workManager number
         */
        WorkThread(final JWorkManager m, final int num, final int wm) {
            super("WorkThread-" + wm + "/" + num);
            mgr = m;
            number = num;
        }

        @Override
        public void run() {
            if (logger.isLoggable(BasicLevel.DEBUG)) {
                logger.log(BasicLevel.DEBUG, "running");
            }
            while (true) {
                try {
                    mgr.nextWork();
                } catch (InterruptedException e) {
                    if (logger.isLoggable(BasicLevel.DEBUG)) {
                        logger.log(BasicLevel.DEBUG, "Exiting: ", e);
                    }
                    return;
                } catch (WorkException e) {
                    logger.log(BasicLevel.ERROR, "Exception during work run: ", e);
                }
            }
        }

    }

}

