ParallelExecStreamScheduler Class Reference

ParallelExecStreamScheduler is a parallel implementation of the ExecStreamScheduler interface. More...

#include <ParallelExecStreamScheduler.h>

Inheritance diagram for ParallelExecStreamScheduler:

List of all members.

Public Member Functions
	ParallelExecStreamScheduler (SharedTraceTarget pTraceTarget, std::string name, ThreadTracker &threadTracker, uint degreeOfParallelism)
	Constructs a new scheduler.
virtual	~ParallelExecStreamScheduler ()
virtual void	addGraph (SharedExecStreamGraph pGraph)
	Adds a graph to be scheduled.
virtual void	removeGraph (SharedExecStreamGraph pGraph)
	Removes a graph currently being scheduled.
virtual void	start ()
	Starts this scheduler, preparing it to execute streams.
virtual void	setRunnable (ExecStream &stream, bool)
	Sets whether that a specific stream should be considered for execution.
virtual void	makeRunnable (ExecStream &stream)
	Requests that a specific stream be considered for execution.
virtual void	abort (ExecStreamGraph &graph)
	Asynchronously aborts execution of any scheduled streams contained by a particular graph and prevents further scheduling.
virtual void	checkAbort () const
	Checks whether there is an abort request for this scheduler, and if so, throws an AbortExcn.
virtual void	stop ()
	Shuts down this scheduler, preventing any further streams from being scheduled.
virtual ExecStreamBufAccessor &	readStream (ExecStream &stream)
	Reads data from a stream, first performing any scheduling necessary to make output available.
virtual void	createBufferProvisionAdapter (ExecStreamEmbryo &embryo)
	Creates a new adapter stream capable of buffering the output of a stream with BUFPROV_CONSUMER for use as input to a stream with BUFPROV_PRODUCER.
virtual uint	getDegreeOfParallelism ()
	Returns: the degree of parallelism implemented by this scheduler, or 1 for a non-parallel scheduler
virtual void	traceStreamBufferContents (ExecStream &stream, ExecStreamBufAccessor &bufAccessor, TraceLevel traceLevel)
	Traces the contents of a stream buffer.
virtual SharedExecStreamBufAccessor	newBufAccessor ()
	Creates a new ExecStreamBufAccessor suitable for use with this scheduler.
virtual void	createCopyProvisionAdapter (ExecStreamEmbryo &embryo)
	Creates a new adapter stream capable of copying the output of a stream with BUFPROV_PRODUCER into the input of a stream with BUFPROV_CONSUMER.
virtual void	initTraceSource (SharedTraceTarget pTraceTarget, std::string name)
	For use when initialization has to be deferred until after construction.
void	trace (TraceLevel level, std::string message) const
	Records a trace message.
bool	isTracing () const
	Returns: true iff tracing is enabled for this source
bool	isTracingLevel (TraceLevel level) const
	Determines whether a particular level is being traced.
TraceTarget &	getTraceTarget () const
	Returns: the TraceTarget for this source
SharedTraceTarget	getSharedTraceTarget () const
	Returns: the SharedTraceTarget for this source
std::string	getTraceSourceName () const
	Gets the name of this source.
void	setTraceSourceName (std::string const &n)
	Sets the name of this source.
TraceLevel	getMinimumTraceLevel () const
void	disableTracing ()
Protected Member Functions
ExecStreamResult	executeStream (ExecStream &stream, ExecStreamQuantum const &quantum)
	Executes one stream, performing tracing if enabled.
virtual void	tracePreExecution (ExecStream &stream, ExecStreamQuantum const &quantum)
	Traces before execution of a stream.
virtual void	tracePostExecution (ExecStream &stream, ExecStreamResult rc)
	Traces after execution of a stream.
virtual void	traceStreamBuffers (ExecStream &stream, TraceLevel inputTupleTraceLevel, TraceLevel outputTupleTraceLevel)
	Traces the states of the input and output buffers adjacent to a stream.
Protected Attributes
bool	tracingFine
StrictMutex	mutex
LocalCondition	condition
Private Types
enum	StreamState { SS_SLEEPING, SS_RUNNING, SS_INHIBITED }
enum	ManagerState { MGR_RUNNING, MGR_STOPPING, MGR_STOPPED }
typedef std::hash_map< ExecStreamId, StreamStateMapEntry >	StreamStateMap
typedef std::deque< ExecStreamId >	InhibitedQueue
Private Member Functions
void	tryExecuteManager ()
void	executeManager ()
void	tryExecuteTask (ExecStream &)
void	executeTask (ExecStream &)
bool	addToQueue (ExecStreamId streamId)
void	signalSentinel (ExecStreamId sentinelId)
void	retryInhibitedQueue ()
void	processCompletedTask (ParallelExecResult const &task)
bool	isInhibited (ExecStreamId streamId)
void	alterNeighborInhibition (ExecStreamId streamId, int delta)
Private Attributes
SharedExecStreamGraph	pGraph
ThreadPool< ParallelExecTask >	threadPool
std::deque< ParallelExecResult >	completedQueue
ThreadTracker &	threadTracker
StreamStateMap	streamStateMap
ManagerState	mgrState
InhibitedQueue	inhibitedQueue
InhibitedQueue	transitQueue
LocalCondition	sentinelCondition
uint	degreeOfParallelism
boost::scoped_ptr< FennelExcn >	pPendingExcn
Friends
class	ParallelExecTask
Classes
struct	StreamStateMapEntry

---

Detailed Description

ParallelExecStreamScheduler is a parallel implementation of the ExecStreamScheduler interface.

For more information, see the design doc in Eigenpedia.

Author:

John Sichi

Version:

Id

//open/dev/fennel/exec/ParallelExecStreamScheduler.h#12

Definition at line 98 of file ParallelExecStreamScheduler.h.

---

Member Typedef Documentation

typedef std::hash_map<ExecStreamId, StreamStateMapEntry> ParallelExecStreamScheduler::StreamStateMap [private]

Definition at line 121 of file ParallelExecStreamScheduler.h.

typedef std::deque<ExecStreamId> ParallelExecStreamScheduler::InhibitedQueue [private]

Definition at line 122 of file ParallelExecStreamScheduler.h.

---

Member Enumeration Documentation

enum ParallelExecStreamScheduler::StreamState [private]

Enumerator:

SS_SLEEPING
SS_RUNNING
SS_INHIBITED

Definition at line 101 of file ParallelExecStreamScheduler.h.

    {
        SS_SLEEPING,
        SS_RUNNING,
        SS_INHIBITED
    };

enum ParallelExecStreamScheduler::ManagerState [private]

Enumerator:

MGR_RUNNING
MGR_STOPPING
MGR_STOPPED

Definition at line 114 of file ParallelExecStreamScheduler.h.

                      {
        MGR_RUNNING,
        MGR_STOPPING,
        MGR_STOPPED
    };

---

Constructor & Destructor Documentation

ParallelExecStreamScheduler::ParallelExecStreamScheduler	(	SharedTraceTarget	pTraceTarget,
		std::string	name,
		ThreadTracker &	threadTracker,
		uint	degreeOfParallelism
	)			[explicit]

Constructs a new scheduler.

Parameters:

	pTraceTarget	the TraceTarget to which messages will be sent, or NULL to disable tracing entirely
	name	the name to use for tracing this scheduler
	threadTracker	tracker to use for threads created during parallelization
	degreeOfParallelism	number of threads to run

Definition at line 37 of file ParallelExecStreamScheduler.cpp.

References degreeOfParallelism, MGR_STOPPED, mgrState, ThreadPoolBase::setThreadTracker(), threadPool, and threadTracker.

    : TraceSource(pTraceTarget, name),
      ExecStreamScheduler(pTraceTarget, name),
      threadTracker(threadTrackerInit)
{
    degreeOfParallelism = degreeOfParallelismInit;
    assert(degreeOfParallelism > 0);
    threadPool.setThreadTracker(threadTracker);
    mgrState = MGR_STOPPED;
}

ParallelExecStreamScheduler::~ParallelExecStreamScheduler

(

)

[virtual]

Definition at line 52 of file ParallelExecStreamScheduler.cpp.

{
}

---

Member Function Documentation

void ParallelExecStreamScheduler::tryExecuteManager

(

)

[private]

Definition at line 228 of file ParallelExecStreamScheduler.cpp.

References completedQueue, SynchMonitoredObject::condition, MGR_RUNNING, mgrState, SynchMonitoredObject::mutex, pPendingExcn, processCompletedTask(), and TRACE_FINE.

Referenced by executeManager().

{
    FENNEL_TRACE(TRACE_FINE,"manager task starting");
    for (;;) {
        StrictMutexGuard mutexGuard(mutex);
        while (completedQueue.empty() && (mgrState == MGR_RUNNING)
            && !pPendingExcn)
        {
            condition.wait(mutexGuard);
        }
        if (pPendingExcn) {
            return;
        }
        if (mgrState != MGR_RUNNING) {
            return;
        }
        while (!completedQueue.empty()) {
            ParallelExecResult result = completedQueue.front();
            completedQueue.pop_front();
            // don't hold lock while doing expensive state maintenance
            mutexGuard.unlock();
            processCompletedTask(result);
            if (pPendingExcn) {
                return;
            }
            mutexGuard.lock();
        }
    }
}

void ParallelExecStreamScheduler::executeManager

(

)

[private]

Definition at line 212 of file ParallelExecStreamScheduler.cpp.

References MGR_STOPPED, mgrState, SynchMonitoredObject::mutex, sentinelCondition, and tryExecuteManager().

Referenced by ParallelExecTask::execute().

{
    // TODO jvs 16-Aug-2008:  RAII
    try {
        tryExecuteManager();
    } catch (...) {
        StrictMutexGuard mutexGuard(mutex);
        mgrState = MGR_STOPPED;
        sentinelCondition.notify_all();
        throw;
    }
    StrictMutexGuard mutexGuard(mutex);
    mgrState = MGR_STOPPED;
    sentinelCondition.notify_all();
}

void ParallelExecStreamScheduler::tryExecuteTask

(

ExecStream &

)

[private]

Definition at line 399 of file ParallelExecStreamScheduler.cpp.

References completedQueue, SynchMonitoredObject::condition, ExecStreamScheduler::executeStream(), ExecStream::getStreamId(), and SynchMonitoredObject::mutex.

Referenced by executeTask().

{
    ExecStreamQuantum quantum;
    ExecStreamResult rc = executeStream(stream, quantum);
    ParallelExecResult result(stream.getStreamId(), rc);

    StrictMutexGuard mutexGuard(mutex);
    completedQueue.push_back(result);
    condition.notify_one();
}

void ParallelExecStreamScheduler::executeTask

(

ExecStream &

)

[private]

Definition at line 379 of file ParallelExecStreamScheduler.cpp.

References ThreadTracker::cloneExcn(), SynchMonitoredObject::condition, SynchMonitoredObject::mutex, pPendingExcn, threadTracker, and tryExecuteTask().

Referenced by ParallelExecTask::execute().

{
    try {
        tryExecuteTask(stream);
    } catch (std::exception &ex) {
        StrictMutexGuard mutexGuard(mutex);
        if (!pPendingExcn) {
            pPendingExcn.reset(threadTracker.cloneExcn(ex));
        }
        condition.notify_one();
    } catch (...) {
        // REVIEW jvs 22-Jul-2008:  panic instead?
        StrictMutexGuard mutexGuard(mutex);
        if (!pPendingExcn) {
            pPendingExcn.reset(new FennelExcn("Unknown error"));
        }
        condition.notify_one();
    }
}

bool ParallelExecStreamScheduler::addToQueue

(

ExecStreamId

streamId

)

[private]

Definition at line 424 of file ParallelExecStreamScheduler.cpp.

References alterNeighborInhibition(), ExecStreamGraphImpl::getStreamFromVertex(), inhibitedQueue, isInhibited(), pPendingExcn, SS_INHIBITED, SS_RUNNING, SS_SLEEPING, streamStateMap, ThreadPool< Task >::submitTask(), and threadPool.

Referenced by processCompletedTask(), and retryInhibitedQueue().

{
    if (pPendingExcn) {
        return false;
    }
    switch (streamStateMap[streamId].state) {
    case SS_SLEEPING:
        {
            ExecStreamGraphImpl &graphImpl =
                dynamic_cast<ExecStreamGraphImpl&>(*pGraph);
            SharedExecStream pStream = graphImpl.getStreamFromVertex(streamId);
            if (!pStream) {
                // sentinel node
                return true;
            }
            if (isInhibited(streamId)) {
                streamStateMap[streamId].state = SS_INHIBITED;
                inhibitedQueue.push_back(streamId);
            } else {
                streamStateMap[streamId].state = SS_RUNNING;
                alterNeighborInhibition(streamId, + 1);
                ParallelExecTask task(*this, pStream.get());
                threadPool.submitTask(task);
            }
        }
        break;
    case SS_INHIBITED:
    case SS_RUNNING:
        // ignore request
        break;
    default:
        permAssert(false);
    }
    return false;
}

void ParallelExecStreamScheduler::signalSentinel

(

ExecStreamId

sentinelId

)

[private]

Definition at line 370 of file ParallelExecStreamScheduler.cpp.

References alterNeighborInhibition(), SynchMonitoredObject::mutex, sentinelCondition, SS_RUNNING, and streamStateMap.

Referenced by processCompletedTask().

{
    alterNeighborInhibition(sentinelId, + 1);

    StrictMutexGuard mutexGuard(mutex);
    streamStateMap[sentinelId].state = SS_RUNNING;
    sentinelCondition.notify_all();
}

void ParallelExecStreamScheduler::retryInhibitedQueue

(

)

[private]

Definition at line 410 of file ParallelExecStreamScheduler.cpp.

References addToQueue(), ExecStreamId, inhibitedQueue, SS_SLEEPING, streamStateMap, and transitQueue.

Referenced by processCompletedTask().

{
    // addToQueue may bounce some back to inhibitedQueue,
    // so process via transitQueue
    transitQueue = inhibitedQueue;
    inhibitedQueue.clear();
    while (!transitQueue.empty()) {
        ExecStreamId inhibitedStreamId = transitQueue.front();
        transitQueue.pop_front();
        streamStateMap[inhibitedStreamId].state = SS_SLEEPING;
        addToQueue(inhibitedStreamId);
    }
}

void ParallelExecStreamScheduler::processCompletedTask

(

ParallelExecResult const &

task

)

[private]

Definition at line 303 of file ParallelExecStreamScheduler.cpp.

References addToQueue(), alterNeighborInhibition(), EXECBUF_EMPTY, EXECBUF_UNDERFLOW, EXECRC_BUF_OVERFLOW, EXECRC_BUF_UNDERFLOW, EXECRC_EOS, EXECRC_QUANTUM_EXPIRED, ExecStreamId, ExecStreamGraphImpl::getBufAccessorFromEdge(), ExecStreamGraphImpl::getGraphRep(), ParallelExecResult::getResultCode(), ExecStreamBufAccessor::getState(), ParallelExecResult::getStreamId(), retryInhibitedQueue(), signalSentinel(), SS_SLEEPING, and streamStateMap.

Referenced by tryExecuteManager().

{
    ExecStreamId current = result.getStreamId();
    ExecStreamGraphImpl &graphImpl =
        dynamic_cast<ExecStreamGraphImpl&>(*pGraph);
    ExecStreamGraphImpl::GraphRep const &graphRep = graphImpl.getGraphRep();

    streamStateMap[current].state = SS_SLEEPING;
    alterNeighborInhibition(current, -1);

    switch (result.getResultCode()) {
    case EXECRC_EOS:
    case EXECRC_BUF_OVERFLOW:
    case EXECRC_BUF_UNDERFLOW:
        {
            ExecStreamGraphImpl::OutEdgeIterPair outEdges =
                boost::out_edges(current, graphRep);
            for (; outEdges.first != outEdges.second; ++(outEdges.first)) {
                ExecStreamGraphImpl::Edge edge = *(outEdges.first);
                ExecStreamBufAccessor &bufAccessor =
                    graphImpl.getBufAccessorFromEdge(edge);
                if (bufAccessor.getState() != EXECBUF_UNDERFLOW) {
                    ExecStreamId consumer = boost::target(edge, graphRep);
                    bool sentinel = addToQueue(consumer);
                    if (sentinel) {
                        if (bufAccessor.getState() != EXECBUF_EMPTY) {
                            signalSentinel(consumer);
                        }
                    }
                }
            }
            ExecStreamGraphImpl::InEdgeIterPair inEdges =
                boost::in_edges(current, graphRep);
            bool sawUnderflow = false;
            for (; inEdges.first != inEdges.second; ++(inEdges.first)) {
                ExecStreamGraphImpl::Edge edge = *(inEdges.first);
                ExecStreamBufAccessor &bufAccessor =
                    graphImpl.getBufAccessorFromEdge(edge);
                if (bufAccessor.getState() == EXECBUF_UNDERFLOW) {
                    ExecStreamId producer = boost::source(edge, graphRep);
                    addToQueue(producer);
                    sawUnderflow = true;
                }
            }
            if (!sawUnderflow &&
                (result.getResultCode() == EXECRC_BUF_UNDERFLOW))
            {
                // sometimes, a stream may return underflow, even though none
                // of its inputs are in underflow state; instead, some are in
                // EOS; we interpret this as a yield request to help keep
                // the stream's state machine logic simpler
                addToQueue(current);
            }
        }
        break;
    case EXECRC_QUANTUM_EXPIRED:
        addToQueue(current);
        break;
    default:
        permAssert(false);
    }

    // REVIEW jvs 4-Jul-2008:  only reschedule inhibited neighbors?
    retryInhibitedQueue();
}

bool ParallelExecStreamScheduler::isInhibited

(

ExecStreamId

streamId

)

[inline, private]

Definition at line 80 of file ParallelExecStreamScheduler.cpp.

References streamStateMap.

Referenced by addToQueue().

{
    return streamStateMap[streamId].inhibitionCount > 0;
}

void ParallelExecStreamScheduler::alterNeighborInhibition	(	ExecStreamId	streamId,
		int	delta
	)			[inline, private]

Definition at line 56 of file ParallelExecStreamScheduler.cpp.

References ExecStreamId, ExecStreamGraphImpl::getGraphRep(), and streamStateMap.

Referenced by addToQueue(), processCompletedTask(), signalSentinel(), and start().

{
    ExecStreamGraphImpl &graphImpl =
        dynamic_cast<ExecStreamGraphImpl&>(*pGraph);
    ExecStreamGraphImpl::GraphRep const &graphRep = graphImpl.getGraphRep();
    ExecStreamGraphImpl::OutEdgeIterPair outEdges =
        boost::out_edges(streamId, graphRep);
    for (; outEdges.first != outEdges.second; ++(outEdges.first)) {
        ExecStreamGraphImpl::Edge edge = *(outEdges.first);
        ExecStreamId consumer = boost::target(edge, graphRep);
        streamStateMap[consumer].inhibitionCount += delta;
        assert(streamStateMap[consumer].inhibitionCount >= 0);
    }
    ExecStreamGraphImpl::InEdgeIterPair inEdges =
        boost::in_edges(streamId, graphRep);
    for (; inEdges.first != inEdges.second; ++(inEdges.first)) {
        ExecStreamGraphImpl::Edge edge = *(inEdges.first);
        ExecStreamId producer = boost::source(edge, graphRep);
        streamStateMap[producer].inhibitionCount += delta;
        assert(streamStateMap[producer].inhibitionCount >= 0);
    }
}

void ParallelExecStreamScheduler::addGraph

(

SharedExecStreamGraph

pGraph

)

[virtual]

Adds a graph to be scheduled.

Some implementations may require all graphs to be added before scheduler is started; others may allow graphs to be added at any time.

Parameters:

pGraph

the graph to be scheduled

Reimplemented from ExecStreamScheduler.

Definition at line 85 of file ParallelExecStreamScheduler.cpp.

References ExecStreamScheduler::addGraph(), and pGraph.

{
    assert(!pGraph);

    ExecStreamScheduler::addGraph(pGraphInit);
    pGraph = pGraphInit;
}

void ParallelExecStreamScheduler::removeGraph

(

SharedExecStreamGraph

pGraph

)

[virtual]

Removes a graph currently being scheduled.

Some implementations may disallow graph removal except when scheduler is stopped; others may disallow graph removal altogether.

Parameters:

pGraph

the graph currently being scheduled

Reimplemented from ExecStreamScheduler.

Definition at line 94 of file ParallelExecStreamScheduler.cpp.

References pGraph, and ExecStreamScheduler::removeGraph().

{
    assert(pGraph == pGraphInit);

    pGraph.reset();
    ExecStreamScheduler::removeGraph(pGraphInit);
}

void ParallelExecStreamScheduler::start

(

)

[virtual]

Starts this scheduler, preparing it to execute streams.

Implements ExecStreamScheduler.

Definition at line 103 of file ParallelExecStreamScheduler.cpp.

References alterNeighborInhibition(), degreeOfParallelism, ExecStreamId, ExecStreamGraphImpl::getBufAccessorFromEdge(), ExecStreamGraphImpl::getGraphRep(), ExecStreamGraphImpl::getStreamFromVertex(), MGR_RUNNING, mgrState, pGraph, pPendingExcn, ExecStreamBufAccessor::requestProduction(), SS_SLEEPING, ThreadPoolBase::start(), streamStateMap, ThreadPool< Task >::submitTask(), threadPool, and TRACE_FINE.

{
    FENNEL_TRACE(TRACE_FINE,"start");
    assert(pGraph->isAcyclic());
    pPendingExcn.reset();

    ExecStreamGraphImpl &graphImpl =
        dynamic_cast<ExecStreamGraphImpl&>(*pGraph);
    ExecStreamGraphImpl::GraphRep const &graphRep = graphImpl.getGraphRep();
    ExecStreamGraphImpl::VertexIterPair vertices = boost::vertices(graphRep);
    while (vertices.first != vertices.second) {
        ExecStreamId streamId = *vertices.first;
        streamStateMap[streamId].state = SS_SLEEPING;
        streamStateMap[streamId].inhibitionCount = 0;
        ++vertices.first;
    }

    vertices = boost::vertices(graphRep);
    while (vertices.first != vertices.second) {
        ExecStreamId streamId = *vertices.first;
        if (!graphImpl.getStreamFromVertex(streamId)) {
            // initially inhibit producers until first call to readStream
            alterNeighborInhibition(streamId, + 1);
            ExecStreamGraphImpl::InEdgeIterPair inEdges =
                boost::in_edges(streamId, graphRep);
            for (; inEdges.first != inEdges.second; ++(inEdges.first)) {
                ExecStreamGraphImpl::Edge edge = *(inEdges.first);
                ExecStreamBufAccessor &bufAccessor =
                    graphImpl.getBufAccessorFromEdge(edge);
                bufAccessor.requestProduction();
            }
        }
        ++vertices.first;
    }

    // +1 for the manager task, which will tie up one thread
    threadPool.start(degreeOfParallelism + 1);

    // kick off the manager task
    ParallelExecTask managerTask(*this, NULL);
    mgrState = MGR_RUNNING;
    threadPool.submitTask(managerTask);
}

void ParallelExecStreamScheduler::setRunnable	(	ExecStream &	stream,
		bool
	)			[virtual]

Sets whether that a specific stream should be considered for execution.

Parameters:

stream

the stream to make runnable

Implements ExecStreamScheduler.

Definition at line 147 of file ParallelExecStreamScheduler.cpp.

{
    permAssert(false);
}

void ParallelExecStreamScheduler::makeRunnable

(

ExecStream &

stream

)

[virtual]

Requests that a specific stream be considered for execution.

Parameters:

stream

the stream to make runnable

Deprecated:

use setRunnable

Reimplemented from ExecStreamScheduler.

Definition at line 152 of file ParallelExecStreamScheduler.cpp.

{
    permAssert(false);
}

void ParallelExecStreamScheduler::abort

(

ExecStreamGraph &

graph

)

[virtual]

Asynchronously aborts execution of any scheduled streams contained by a particular graph and prevents further scheduling.

Returns immediately, not waiting for abort request to be fully processed.

Parameters:

graph

graph to abort; must be one of the graphs associated with this scheduler

Implements ExecStreamScheduler.

Definition at line 157 of file ParallelExecStreamScheduler.cpp.

References SynchMonitoredObject::condition, SynchMonitoredObject::mutex, pPendingExcn, and TRACE_FINE.

{
    StrictMutexGuard mutexGuard(mutex);
    FENNEL_TRACE(TRACE_FINE,"abort requested");

    if (!pPendingExcn) {
        pPendingExcn.reset(new AbortExcn());
    }
    condition.notify_one();
}

void ParallelExecStreamScheduler::checkAbort

(

)

const [virtual]

Checks whether there is an abort request for this scheduler, and if so, throws an AbortExcn.

Reimplemented from ExecStreamScheduler.

Definition at line 168 of file ParallelExecStreamScheduler.cpp.

References pPendingExcn.

{
    if (pPendingExcn) {
        throw AbortExcn();
    }
}

void ParallelExecStreamScheduler::stop

(

)

[virtual]

Shuts down this scheduler, preventing any further streams from being scheduled.

Implements ExecStreamScheduler.

Definition at line 175 of file ParallelExecStreamScheduler.cpp.

References completedQueue, SynchMonitoredObject::condition, inhibitedQueue, MGR_STOPPED, MGR_STOPPING, mgrState, SynchMonitoredObject::mutex, pPendingExcn, sentinelCondition, ThreadPoolBase::stop(), threadPool, and TRACE_FINE.

{
    FENNEL_TRACE(TRACE_FINE,"stop");

    StrictMutexGuard mutexGuard(mutex);
    if (mgrState != MGR_STOPPED) {
        mgrState = MGR_STOPPING;
        condition.notify_one();
        while (mgrState != MGR_STOPPED) {
            sentinelCondition.wait(mutexGuard);
        }
    }
    mutexGuard.unlock();

    threadPool.stop();

    // NOTE jvs 10-Aug-2008:  This is how we keep the cloned excn
    // from becoming a memory leak.  It assumes that the caller
    // doesn't invoke pScheduler->stop() until *after* the exception
    // has been completely handled and is no longer referenced.
    pPendingExcn.reset();

    completedQueue.clear();
    inhibitedQueue.clear();
}

ExecStreamBufAccessor & ParallelExecStreamScheduler::readStream

(

ExecStream &

stream

)

[virtual]

Reads data from a stream, first performing any scheduling necessary to make output available.

Parameters:

stream

the stream from which to read

Returns:

accessor for output data buffer

Implements ExecStreamScheduler.

Definition at line 258 of file ParallelExecStreamScheduler.cpp.

References completedQueue, SynchMonitoredObject::condition, EXECBUF_EMPTY, EXECBUF_UNDERFLOW, EXECRC_BUF_UNDERFLOW, ExecStreamId, ExecStreamGraphImpl::getBufAccessorFromEdge(), ExecStreamGraphImpl::getGraphRep(), ExecStream::getName(), ExecStreamBufAccessor::getState(), ExecStreamGraphImpl::getStreamFromVertex(), ExecStream::getStreamId(), SynchMonitoredObject::mutex, pPendingExcn, ExecStreamBufAccessor::requestProduction(), sentinelCondition, SS_SLEEPING, streamStateMap, and TRACE_FINE.

{
    FENNEL_TRACE(
        TRACE_FINE,
        "entering readStream " << stream.getName());

    ExecStreamId current = stream.getStreamId();
    ExecStreamGraphImpl &graphImpl =
        dynamic_cast<ExecStreamGraphImpl&>(*pGraph);
    ExecStreamGraphImpl::GraphRep const &graphRep = graphImpl.getGraphRep();

    // assert that we're reading from a designated output stream
    assert(boost::out_degree(current,graphRep) == 1);
    ExecStreamGraphImpl::Edge edge =
        *(boost::out_edges(current,graphRep).first);
    ExecStreamBufAccessor &bufAccessor = graphImpl.getBufAccessorFromEdge(edge);
    current = boost::target(edge, graphRep);
    assert(!graphImpl.getStreamFromVertex(current));

    if (bufAccessor.getState() == EXECBUF_EMPTY) {
        bufAccessor.requestProduction();
    } else if (bufAccessor.getState() != EXECBUF_UNDERFLOW) {
        // data or EOS already available
        return bufAccessor;
    }

    // please sir, I'd like some more
    ParallelExecResult result(current, EXECRC_BUF_UNDERFLOW);
    StrictMutexGuard mutexGuard(mutex);
    streamStateMap[current].state = SS_SLEEPING;
    completedQueue.push_back(result);
    condition.notify_one();

    while ((streamStateMap[current].state == SS_SLEEPING) && !pPendingExcn) {
        sentinelCondition.wait(mutexGuard);
    }

    if (pPendingExcn) {
        pPendingExcn->throwSelf();
    }

    return bufAccessor;
}

void ParallelExecStreamScheduler::createBufferProvisionAdapter

(

ExecStreamEmbryo &

embryo

)

[virtual]

Creates a new adapter stream capable of buffering the output of a stream with BUFPROV_CONSUMER for use as input to a stream with BUFPROV_PRODUCER.

Default implementation is ScratchBufferExecStream. Caller is responsible for filling in generic ExecStreamParams after return.

Parameters:

embryo

receives new adapter stream

Reimplemented from ExecStreamScheduler.

Definition at line 201 of file ParallelExecStreamScheduler.cpp.

References ExecStreamEmbryo::init().

{
    // use double buffering so that producers and consumers can run
    // in parallel
    DoubleBufferExecStreamParams adapterParams;
    embryo.init(
        new DoubleBufferExecStream(),
        adapterParams);
}

uint ParallelExecStreamScheduler::getDegreeOfParallelism

(

)

[virtual]

Returns:

the degree of parallelism implemented by this scheduler, or 1 for a non-parallel scheduler

Reimplemented from ExecStreamScheduler.

Definition at line 485 of file ParallelExecStreamScheduler.cpp.

References degreeOfParallelism.

{
    return degreeOfParallelism;
}

ExecStreamResult ExecStreamScheduler::executeStream	(	ExecStream &	stream,
		ExecStreamQuantum const &	quantum
	)			[inline, protected, inherited]

Executes one stream, performing tracing if enabled.

Parameters:

	stream	stream to execute
	quantum	quantum controlling stream execution

Returns:

result of executing stream

Definition at line 243 of file ExecStreamScheduler.h.

References ExecStream::execute(), ExecStreamScheduler::tracePostExecution(), ExecStreamScheduler::tracePreExecution(), and ExecStreamScheduler::tracingFine.

Referenced by DfsTreeExecStreamScheduler::readStream(), and tryExecuteTask().

{
    if (tracingFine) {
        tracePreExecution(stream, quantum);
        ExecStreamResult rc = stream.execute(quantum);
        tracePostExecution(stream, rc);
        return rc;
    } else {
        return stream.execute(quantum);
    }
}

void ExecStreamScheduler::tracePreExecution	(	ExecStream &	stream,
		ExecStreamQuantum const &	quantum
	)			[protected, virtual, inherited]

Traces before execution of a stream.

Parameters:

	stream	stream about to be executed
	quantum	quantum controlling stream execution

Definition at line 112 of file ExecStreamScheduler.cpp.

References ExecStream::getName(), ExecStream::getStreamId(), isMAXU(), ExecStreamQuantum::nTuplesMax, TRACE_FINE, TRACE_FINEST, and ExecStreamScheduler::traceStreamBuffers().

Referenced by ExecStreamScheduler::executeStream().

{
    FENNEL_TRACE(
        TRACE_FINE,
        "executing " << stream.getStreamId() << ' ' << stream.getName());
    if (!isMAXU(quantum.nTuplesMax)) {
        FENNEL_TRACE(
            TRACE_FINE,
            "nTuplesMax = " << quantum.nTuplesMax);
    }

    traceStreamBuffers(stream, TRACE_FINEST, TRACE_FINEST);
}

void ExecStreamScheduler::tracePostExecution	(	ExecStream &	stream,
		ExecStreamResult	rc
	)			[protected, virtual, inherited]

Traces after execution of a stream.

Parameters:

	stream	stream which was just executed
	rc	result code returned by stream

Definition at line 128 of file ExecStreamScheduler.cpp.

References ExecStreamResult_names, ExecStream::getName(), ExecStream::getStreamId(), TRACE_FINE, TRACE_FINER, TRACE_FINEST, and ExecStreamScheduler::traceStreamBuffers().

Referenced by ExecStreamScheduler::executeStream().

{
    FENNEL_TRACE(
        TRACE_FINE,
        "executed " << stream.getStreamId() << ' ' << stream.getName()
        << " with result " << ExecStreamResult_names[rc]);

    traceStreamBuffers(stream, TRACE_FINEST, TRACE_FINER);
}

void ExecStreamScheduler::traceStreamBuffers	(	ExecStream &	stream,
		TraceLevel	inputTupleTraceLevel,
		TraceLevel	outputTupleTraceLevel
	)			[protected, virtual, inherited]

Traces the states of the input and output buffers adjacent to a stream.

Parameters:

	stream	stream whose buffers are to be traced
	inputTupleTraceLevel	trace level at which tuple contents of input buffers are to be traced
	outputTupleTraceLevel	trace level at which tuple contents of output buffers are to be traced

Definition at line 140 of file ExecStreamScheduler.cpp.

References ExecStreamBufState_names, ExecStreamGraphImpl::getBufAccessorFromEdge(), ExecStreamBufAccessor::getConsumptionAvailable(), ExecStream::getGraph(), ExecStreamGraphImpl::getGraphRep(), ExecStreamBufAccessor::getProductionAvailable(), ExecStreamBufAccessor::getState(), ExecStream::getStreamId(), ExecStreamBufAccessor::hasPendingEOS(), TraceSource::isTracingLevel(), TRACE_FINER, and ExecStreamScheduler::traceStreamBufferContents().

Referenced by ExecStreamScheduler::tracePostExecution(), and ExecStreamScheduler::tracePreExecution().

{
    ExecStreamGraphImpl &graphImpl =
        dynamic_cast<ExecStreamGraphImpl&>(stream.getGraph());
    ExecStreamGraphImpl::GraphRep const &graphRep = graphImpl.getGraphRep();

    ExecStreamGraphImpl::InEdgeIterPair inEdges =
        boost::in_edges(stream.getStreamId(),graphRep);
    for (uint i = 0; inEdges.first != inEdges.second;
         ++(inEdges.first),  ++i)
    {
        ExecStreamGraphImpl::Edge edge = *(inEdges.first);
        ExecStreamBufAccessor &bufAccessor =
            graphImpl.getBufAccessorFromEdge(edge);
        FENNEL_TRACE(
            TRACE_FINER,
            "input buffer " << i << ":  "
            << ExecStreamBufState_names[bufAccessor.getState()]
            << (bufAccessor.hasPendingEOS() ? ", EOS pending" : "")
            << ",  consumption available = "
            << bufAccessor.getConsumptionAvailable());
        if (stream.isTracingLevel(inputTupleTraceLevel)) {
            traceStreamBufferContents(
                stream, bufAccessor, inputTupleTraceLevel);
        }
    }

    ExecStreamGraphImpl::OutEdgeIterPair outEdges =
        boost::out_edges(stream.getStreamId(),graphRep);
    for (uint i = 0; outEdges.first != outEdges.second;
         ++(outEdges.first),  ++i) {
        ExecStreamGraphImpl::Edge edge = *(outEdges.first);
        ExecStreamBufAccessor &bufAccessor =
            graphImpl.getBufAccessorFromEdge(edge);
        FENNEL_TRACE(
            TRACE_FINER,
            "output buffer " << i << ":  "
            << ExecStreamBufState_names[bufAccessor.getState()]
            << (bufAccessor.hasPendingEOS() ? ", EOS pending" : "")
            << ",  consumption available = "
            << bufAccessor.getConsumptionAvailable()
            << ",  production available = "
            << bufAccessor.getProductionAvailable());
        if (stream.isTracingLevel(outputTupleTraceLevel)) {
            traceStreamBufferContents(
                stream, bufAccessor, outputTupleTraceLevel);
        }
    }
}

void ExecStreamScheduler::traceStreamBufferContents	(	ExecStream &	stream,
		ExecStreamBufAccessor &	bufAccessor,
		TraceLevel	traceLevel
	)			[virtual, inherited]

Traces the contents of a stream buffer.

Parameters:

	stream	stream whose buffer is being traced
	bufAccessor	accessor for stream buffer
	traceLevel	level at which contents should be traced

Definition at line 193 of file ExecStreamScheduler.cpp.

References ExecStreamBufAccessor::getConsumptionEnd(), ExecStreamBufAccessor::getConsumptionStart(), TupleAccessor::getCurrentByteCount(), ExecStreamBufAccessor::getScratchTupleAccessor(), ExecStreamBufAccessor::getTupleDesc(), TuplePrinter::print(), TupleAccessor::setCurrentTupleBuf(), TraceSource::trace(), and TupleAccessor::unmarshal().

Referenced by ExecStreamScheduler::traceStreamBuffers().

{
    TupleDescriptor const &tupleDesc = bufAccessor.getTupleDesc();
    TupleData tupleData(tupleDesc);
    TupleAccessor &tupleAccessor = bufAccessor.getScratchTupleAccessor();

    for (PConstBuffer pTuple = bufAccessor.getConsumptionStart();
         pTuple != bufAccessor.getConsumptionEnd();
         pTuple += tupleAccessor.getCurrentByteCount())
    {
        tupleAccessor.setCurrentTupleBuf(pTuple);
        // while we're here, we might as well sanity-check the content
        assert(pTuple + tupleAccessor.getCurrentByteCount()
            <= bufAccessor.getConsumptionEnd());
        tupleAccessor.unmarshal(tupleData);
        // TODO:  sanity-check individual data values?
        std::ostringstream oss;
        TuplePrinter tuplePrinter;
        tuplePrinter.print(oss,tupleDesc,tupleData);
        stream.trace(traceLevel,oss.str());
    }
}

SharedExecStreamBufAccessor ExecStreamScheduler::newBufAccessor

(

)

[virtual, inherited]

Creates a new ExecStreamBufAccessor suitable for use with this scheduler.

Returns:

new buffer accessor

Definition at line 78 of file ExecStreamScheduler.cpp.

Referenced by ExecStreamGraphImpl::prepare().

{
    return SharedExecStreamBufAccessor(new ExecStreamBufAccessor());
}

void ExecStreamScheduler::createCopyProvisionAdapter

(

ExecStreamEmbryo &

embryo

)

[virtual, inherited]

Creates a new adapter stream capable of copying the output of a stream with BUFPROV_PRODUCER into the input of a stream with BUFPROV_CONSUMER.

Default implementation is CopyExecStream. Caller is responsible for filling in generic ExecStreamParams after return.

Parameters:

embryo

receives new adapter stream

Definition at line 92 of file ExecStreamScheduler.cpp.

References ExecStreamEmbryo::init().

{
    CopyExecStreamParams adapterParams;
    embryo.init(
        new CopyExecStream(),
        adapterParams);
}

void TraceSource::initTraceSource	(	SharedTraceTarget	pTraceTarget,
		std::string	name
	)			[virtual, inherited]

For use when initialization has to be deferred until after construction.

Parameters:

	pTraceTarget	the TraceTarget to which messages will be sent
	name	the name of this source

Definition at line 46 of file TraceSource.cpp.

References TraceSource::isTracing(), TraceSource::minimumLevel, TraceSource::name, TraceSource::pTraceTarget, and TRACE_OFF.

Referenced by TestBase::beforeTestCase(), TestBase::TestBase(), and TraceSource::TraceSource().

{
    assert(!pTraceTarget.get());

    pTraceTarget = pTraceTargetInit;
    name = nameInit;
    if (isTracing()) {
        minimumLevel = pTraceTarget->getSourceTraceLevel(name);
    } else {
        minimumLevel = TRACE_OFF;
    }
}

void TraceSource::trace	(	TraceLevel	level,
		std::string	message
	)			const [inherited]

Records a trace message.

Normally only called via FENNEL_TRACE.

Parameters:

	level	severity level of event being trace
	message	the text of the message

Definition at line 61 of file TraceSource.cpp.

References TraceSource::getTraceTarget(), TraceSource::isTracing(), TraceSource::name, and TraceTarget::notifyTrace().

Referenced by Calculator::exec(), and ExecStreamScheduler::traceStreamBufferContents().

{
    if (isTracing()) {
        getTraceTarget().notifyTrace(name,level,message);
    }
}

bool TraceSource::isTracing

(

)

const [inline, inherited]

Returns:

true iff tracing is enabled for this source

Definition at line 88 of file TraceSource.h.

Referenced by TraceSource::initTraceSource(), CalcExecStream::prepare(), and TraceSource::trace().

    {
        return pTraceTarget.get() ? true : false;
    }

bool TraceSource::isTracingLevel

(

TraceLevel

level

)

const [inline, inherited]

Determines whether a particular level is being traced.

Parameters:

level

trace level to test

Returns:

true iff tracing is enabled for the given level

Definition at line 100 of file TraceSource.h.

Referenced by ExecStreamScheduler::addGraph(), SimpleExecStreamGovernor::assignCachePages(), SimpleExecStreamGovernor::distributeCachePages(), Calculator::exec(), ExecStreamScheduler::ExecStreamScheduler(), LcsClusterNodeWriter::getLastClusterPageForWrite(), LcsClusterNodeWriter::moveFromTempToIndex(), JavaSinkExecStream::stuffByteBuffer(), and ExecStreamScheduler::traceStreamBuffers().

    {
        return level >= minimumLevel;
    }

TraceTarget& TraceSource::getTraceTarget

(

)

const [inline, inherited]

Returns:

the TraceTarget for this source

Definition at line 108 of file TraceSource.h.

Referenced by TraceSource::trace().

    {
        assert(isTracing());
        return *(pTraceTarget.get());
    }

SharedTraceTarget TraceSource::getSharedTraceTarget

(

)

const [inline, inherited]

Returns:

the SharedTraceTarget for this source

Definition at line 117 of file TraceSource.h.

Referenced by Database::init(), LcsClusterAppendExecStream::initLoad(), and CalcExecStream::prepare().

    {
        return pTraceTarget;
    }

std::string TraceSource::getTraceSourceName

(

)

const [inline, inherited]

Gets the name of this source.

Useful to construct nested names for subcomponents that are also TraceSources.

Returns:

the name

Definition at line 127 of file TraceSource.h.

Referenced by LcsClusterAppendExecStream::initLoad().

    {
        return name;
    }

void TraceSource::setTraceSourceName

(

std::string const &

n

)

[inline, inherited]

Sets the name of this source.

Useful to construct dynamic names for fine-grained filtering.

Definition at line 136 of file TraceSource.h.

    {
        name = n;
    }

TraceLevel TraceSource::getMinimumTraceLevel

(

)

const [inline, inherited]

Definition at line 141 of file TraceSource.h.

    {
        return minimumLevel;
    }

void TraceSource::disableTracing

(

)

[inherited]

Definition at line 68 of file TraceSource.cpp.

References TraceSource::minimumLevel, TraceSource::pTraceTarget, and TRACE_OFF.

Referenced by TestBase::afterTestCase().

{
    pTraceTarget.reset();
    minimumLevel = TRACE_OFF;
}

---

Friends And Related Function Documentation

friend class ParallelExecTask [friend]

Definition at line 124 of file ParallelExecStreamScheduler.h.

---

Member Data Documentation

SharedExecStreamGraph ParallelExecStreamScheduler::pGraph [private]

Definition at line 126 of file ParallelExecStreamScheduler.h.

Referenced by addGraph(), removeGraph(), and start().

ThreadPool<ParallelExecTask> ParallelExecStreamScheduler::threadPool [private]

Definition at line 128 of file ParallelExecStreamScheduler.h.

Referenced by addToQueue(), ParallelExecStreamScheduler(), start(), and stop().

std::deque<ParallelExecResult> ParallelExecStreamScheduler::completedQueue [private]

Definition at line 129 of file ParallelExecStreamScheduler.h.

Referenced by readStream(), stop(), tryExecuteManager(), and tryExecuteTask().

ThreadTracker& ParallelExecStreamScheduler::threadTracker [private]

Definition at line 131 of file ParallelExecStreamScheduler.h.

Referenced by executeTask(), and ParallelExecStreamScheduler().

StreamStateMap ParallelExecStreamScheduler::streamStateMap [private]

Definition at line 133 of file ParallelExecStreamScheduler.h.

Referenced by addToQueue(), alterNeighborInhibition(), isInhibited(), processCompletedTask(), readStream(), retryInhibitedQueue(), signalSentinel(), and start().

ManagerState ParallelExecStreamScheduler::mgrState [private]

Definition at line 134 of file ParallelExecStreamScheduler.h.

Referenced by executeManager(), ParallelExecStreamScheduler(), start(), stop(), and tryExecuteManager().

InhibitedQueue ParallelExecStreamScheduler::inhibitedQueue [private]

Definition at line 136 of file ParallelExecStreamScheduler.h.

Referenced by addToQueue(), retryInhibitedQueue(), and stop().

InhibitedQueue ParallelExecStreamScheduler::transitQueue [private]

Definition at line 137 of file ParallelExecStreamScheduler.h.

Referenced by retryInhibitedQueue().

LocalCondition ParallelExecStreamScheduler::sentinelCondition [private]

Definition at line 138 of file ParallelExecStreamScheduler.h.

Referenced by executeManager(), readStream(), signalSentinel(), and stop().

uint ParallelExecStreamScheduler::degreeOfParallelism [private]

Definition at line 140 of file ParallelExecStreamScheduler.h.

Referenced by getDegreeOfParallelism(), ParallelExecStreamScheduler(), and start().

boost::scoped_ptr<FennelExcn> ParallelExecStreamScheduler::pPendingExcn [private]

Definition at line 142 of file ParallelExecStreamScheduler.h.

Referenced by abort(), addToQueue(), checkAbort(), executeTask(), readStream(), start(), stop(), and tryExecuteManager().

bool ExecStreamScheduler::tracingFine [protected, inherited]

Definition at line 47 of file ExecStreamScheduler.h.

Referenced by ExecStreamScheduler::addGraph(), ExecStreamScheduler::ExecStreamScheduler(), and ExecStreamScheduler::executeStream().

StrictMutex SynchMonitoredObject::mutex [protected, inherited]

Definition at line 38 of file SynchMonitoredObject.h.

Referenced by abort(), LogicalTxnLog::checkpoint(), Database::checkpointImpl(), CheckpointThread::closeImpl(), LogicalTxnLog::commitTxn(), executeManager(), executeTask(), LogicalTxnLog::getOldestActiveTxnId(), LogicalTxnLog::newLogicalTxn(), readStream(), SXMutex::release(), GroupLock::release(), Database::requestCheckpoint(), CheckpointThread::requestCheckpoint(), LogicalTxnLog::rollbackTxn(), TimerThread::run(), CheckpointThread::run(), ThreadPoolBase::runPooledThread(), SXMutex::setSchedulingPolicy(), TimerThread::signalImmediate(), signalSentinel(), ThreadPoolBase::start(), TimerThread::stop(), ThreadPoolBase::stop(), stop(), ThreadPool< RandomAccessRequest >::submitTask(), tryExecuteManager(), tryExecuteTask(), SXMutex::tryUpgrade(), SXMutex::waitFor(), GroupLock::waitFor(), and Database::writeStats().

LocalCondition SynchMonitoredObject::condition [protected, inherited]

Definition at line 39 of file SynchMonitoredObject.h.

Referenced by abort(), Database::checkpointImpl(), CheckpointThread::closeImpl(), LogicalTxnLog::commitTxnWithGroup(), executeTask(), readStream(), SXMutex::release(), GroupLock::release(), Database::requestCheckpoint(), CheckpointThread::requestCheckpoint(), TimerThread::run(), CheckpointThread::run(), ThreadPoolBase::runPooledThread(), TimerThread::signalImmediate(), TimerThread::stop(), ThreadPoolBase::stop(), stop(), ThreadPool< RandomAccessRequest >::submitTask(), tryExecuteManager(), tryExecuteTask(), SXMutex::waitFor(), and GroupLock::waitFor().

---

The documentation for this class was generated from the following files:

• /home/pub/open/dev/fennel/exec/ParallelExecStreamScheduler.h
• /home/pub/open/dev/fennel/exec/ParallelExecStreamScheduler.cpp

---