staxparalleliterateaction.java

Software Testing Automation Framework (STAF)的开发代码

/*****************************************************************************/
/* Software Testing Automation Framework (STAF)                              */
/* (C) Copyright IBM Corp. 2002                                              */
/*                                                                           */
/* This software is licensed under the Common Public License (CPL) V1.0.     */
/*****************************************************************************/

package com.ibm.staf.service.stax;
import java.util.List;
import java.util.HashMap;
import java.util.Iterator;

public class STAXParallelIterateAction implements STAXAction,
                                                  STAXThreadCompleteListener
{
    static final int INIT = 0;
    static final int WAIT_THREADS = 1;
    static final int THREADS_COMPLETE = 2;
    static final int COMPLETE = 3;

    static final String INIT_STRING = "INIT";
    static final String WAIT_THREADS_STRING = "WAIT_THREADS";
    static final String THREADS_COMPLETE_STRING = "THREADS_COMPLETE";
    static final String COMPLETE_STRING = "COMPLETE";
    static final String STATE_UNKNOWN_STRING = "UNKNOWN";

    private STAXParallelIterateAction()
    { /* Do Nothing */ }

    public STAXParallelIterateAction(String itemvar, String in, 
                                     String indexvar, STAXAction action)
    {
        fItemvar = itemvar;
        fIn = in;
        fIndexvar = indexvar; 
        fAction = action;
    }

    public String getStateAsString()
    {
        switch (fState)
        {
            case INIT:
                return INIT_STRING;
            case WAIT_THREADS:
                return WAIT_THREADS_STRING;
            case THREADS_COMPLETE:
                return THREADS_COMPLETE_STRING;
            case COMPLETE:
                return COMPLETE_STRING;
            default:
                return STATE_UNKNOWN_STRING;
        }
    }

    public String getXMLInfo()
    {
        if (fIndexvar.equals(""))
            return "<paralleliterate var=\"" + fItemvar + "\" in=\"" + 
                   fIn + "\">";
        else
            return "<paralleliterate var=\"" + fItemvar + "\" in=\"" + 
                   fIn + "\" indexvar=\"" + fIndexvar + "\">";
    }

    public String getInfo()
    {
        return fListSize + "/" + getStateAsString();
    }

    public String getDetails()
    {
        return "ListSize:" + fListSize + 
               ";State:" + getStateAsString() +
               ";Itemvar:" + fItemvar +
               ";In:" + fIn +
               ";Indexvar:" + fIndexvar +
               ";Action:" + fAction +
               ";HoldThreadCondition:" + fHoldCondition +
               ";HardHoldThreadCondition:" + fHardHoldCondition + 
               ";ThreadMap:" + fThreadMap +
               ";List:" + fList;
    }


    public void execute(STAXThread thread)
    {
        try
        {
            if (fState == INIT)
            {
                // Create a Python list/tuple from fIn and then extract it
                // into a Java List.

                fList = thread.pyListEval(fIn);
                fListSize = fList.size();

                if (fListSize == 0)
                {   // Nothing in list, no threads to start; raise a signal.
                    fState = THREADS_COMPLETE;
                    thread.popAction();
                    thread.setSignalMsgVar("STAXEmptyListMsg", getXMLInfo());
                    thread.raiseSignal("STAXEmptyList");

                    return;
                }
                
                synchronized (fThreadMap)
                {
                    for (int i = 0; i < fListSize; i++)
                    {
                        // Setup up a new thread for each iteration
                        STAXThread childThread = thread.createChildThread();
  
                        // Assign indexvar Jython variable, if provided
                        if (fIndexvar != null)
                            childThread.pySetVar(fIndexvar, new Integer(i));

                        // Assign itemvar Jython variable
                        Object itemValue = (Object)fList.get(i);

                        childThread.pySetVar(fItemvar, itemValue);
                        fThreadMap.put(new Integer(
                            childThread.getThreadNumber()), childThread);
                        childThread.addCompletionNotifiee(this);
                        childThread.pushAction(fAction.cloneAction());
                        childThread.schedule();
                    }
                        
                    fState = WAIT_THREADS;
                    thread.addCondition(fHoldCondition);
                }
          
                return;
            }
            else if (fState == THREADS_COMPLETE)
            {
                fState = COMPLETE;
                thread.popAction();
            }   
        }
        catch (STAXPythonEvaluationException e)
        {
            fState = COMPLETE;
            thread.popAction();
            thread.setSignalMsgVar("STAXPythonEvalMsg", getXMLInfo(), e);
            thread.raiseSignal("STAXPythonEvaluationError");
        }
    }

    public void handleCondition(STAXThread thread, STAXCondition cond)
    {
        synchronized (fThreadMap)
        {
            if (!fThreadMap.isEmpty())
            {
                // Add a hard hold, so we can wait until our children terminate.
                // This is removed by threadComplete().

                thread.addCondition(fHardHoldCondition);

                // Now, iterate of our children and tell them to terminate

                Iterator iter = fThreadMap.values().iterator();

                while (iter.hasNext())
                {
                    STAXThread childThread = (STAXThread)iter.next();
                    childThread.terminate(
                        STAXThread.THREAD_END_STOPPED_BY_PARENT);
                }
            }
        }

        fState = COMPLETE;  
        thread.popAction();
    }

    public STAXAction cloneAction()
    {
        STAXParallelIterateAction clone = new STAXParallelIterateAction();

        clone.fItemvar = fItemvar;
        clone.fIn = fIn;
        clone.fIndexvar = fIndexvar;
        clone.fAction = fAction;

        return clone;
    }

    // STAXThreadCompleteListener method

    public void threadComplete(STAXThread thread, int endCode)
    {
        synchronized (fThreadMap)
        {
            fThreadMap.remove(new Integer(thread.getThreadNumber()));

            if (fThreadMap.isEmpty())
            {
                fState = THREADS_COMPLETE;

                // thread.getParentThread() should be the thread we are
                // running on.

                thread.getParentThread().removeCondition(fHoldCondition);
                thread.getParentThread().removeCondition(fHardHoldCondition);
                thread.getParentThread().schedule();
            }
        }
    }

    private STAXHoldThreadCondition fHoldCondition = 
        new STAXHoldThreadCondition("ParallelIterate");
    private STAXHardHoldThreadCondition fHardHoldCondition =
        new STAXHardHoldThreadCondition("ParallelIterate");
    private int fState = INIT;
    private HashMap fThreadMap = new HashMap();
    private List fList = null;
    private int fListSize = 0;

    private String fItemvar;
    private String fIn;
    private String fIndexvar;
    private STAXAction fAction = null;
}