OgreWorkQueue.h
Go to the documentation of this file.
/*
-----------------------------------------------------------------------------
This source file is part of OGRE
(Object-oriented Graphics Rendering Engine)
For the latest info, see http://www.ogre3d.org/
Copyright (c) 2000-2013 Torus Knot Software Ltd
Permission is hereby granted, free of charge, to any person obtaining a copy
of this software and associated documentation files (the "Software"), to deal
in the Software without restriction, including without limitation the rights
to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
copies of the Software, and to permit persons to whom the Software is
furnished to do so, subject to the following conditions:
The above copyright notice and this permission notice shall be included in
all copies or substantial portions of the Software.
THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
THE SOFTWARE.
-----------------------------------------------------------------------------
*/
#ifndef __OgreWorkQueue_H__
#define __OgreWorkQueue_H__
#include "OgrePrerequisites.h"
#include "OgreAny.h"
#include "OgreSharedPtr.h"
#include "Threading/OgreThreadHeaders.h"
#include "OgreHeaderPrefix.h"
namespace Ogre
{
class _OgreExport WorkQueue : public UtilityAlloc
{
protected:
typedef map<String, uint16>::type ChannelMap;
ChannelMap mChannelMap;
uint16 mNextChannel;
OGRE_MUTEX(mChannelMapMutex);
public:
typedef unsigned long long int RequestID;
class _OgreExport Request : public UtilityAlloc
{
friend class WorkQueue;
protected:
uint16 mChannel;
uint16 mType;
Any mData;
uint8 mRetryCount;
RequestID mID;
mutable bool mAborted;
public:
Request(uint16 channel, uint16 rtype, const Any& rData, uint8 retry, RequestID rid);
~Request();
void abortRequest() const { mAborted = true; }
uint16 getChannel() const { return mChannel; }
uint16 getType() const { return mType; }
const Any& getData() const { return mData; }
uint8 getRetryCount() const { return mRetryCount; }
RequestID getID() const { return mID; }
bool getAborted() const { return mAborted; }
};
struct _OgreExport Response : public UtilityAlloc
{
const Request* mRequest;
bool mSuccess;
String mMessages;
Any mData;
public:
Response(const Request* rq, bool success, const Any& data, const String& msg = StringUtil::BLANK);
~Response();
const Request* getRequest() const { return mRequest; }
bool succeeded() const { return mSuccess; }
const String& getMessages() const { return mMessages; }
const Any& getData() const { return mData; }
void abortRequest() { mRequest->abortRequest(); mData.destroy(); }
};
class _OgreExport RequestHandler
{
public:
RequestHandler() {}
virtual ~RequestHandler() {}
virtual bool canHandleRequest(const Request* req, const WorkQueue* srcQ)
{ (void)srcQ; return !req->getAborted(); }
virtual Response* handleRequest(const Request* req, const WorkQueue* srcQ) = 0;
};
class _OgreExport ResponseHandler
{
public:
ResponseHandler() {}
virtual ~ResponseHandler() {}
virtual bool canHandleResponse(const Response* res, const WorkQueue* srcQ)
{ (void)srcQ; return !res->getRequest()->getAborted(); }
virtual void handleResponse(const Response* res, const WorkQueue* srcQ) = 0;
};
WorkQueue() : mNextChannel(0) {}
virtual ~WorkQueue() {}
virtual void startup(bool forceRestart = true) = 0;
virtual void addRequestHandler(uint16 channel, RequestHandler* rh) = 0;
virtual void removeRequestHandler(uint16 channel, RequestHandler* rh) = 0;
virtual void addResponseHandler(uint16 channel, ResponseHandler* rh) = 0;
virtual void removeResponseHandler(uint16 channel, ResponseHandler* rh) = 0;
virtual RequestID addRequest(uint16 channel, uint16 requestType, const Any& rData, uint8 retryCount = 0,
bool forceSynchronous = false, bool idleThread = false) = 0;
virtual void abortRequest(RequestID id) = 0;
virtual void abortRequestsByChannel(uint16 channel) = 0;
virtual void abortPendingRequestsByChannel(uint16 channel) = 0;
virtual void abortAllRequests() = 0;
virtual void setPaused(bool pause) = 0;
virtual bool isPaused() const = 0;
virtual void setRequestsAccepted(bool accept) = 0;
virtual bool getRequestsAccepted() const = 0;
virtual void processResponses() = 0;
virtual unsigned long getResponseProcessingTimeLimit() const = 0;
virtual void setResponseProcessingTimeLimit(unsigned long ms) = 0;
virtual void shutdown() = 0;
virtual uint16 getChannel(const String& channelName);
};
class _OgreExport DefaultWorkQueueBase : public WorkQueue
{
public:
DefaultWorkQueueBase(const String& name = StringUtil::BLANK);
virtual ~DefaultWorkQueueBase();
const String& getName() const;
virtual size_t getWorkerThreadCount() const;
virtual void setWorkerThreadCount(size_t c);
virtual bool getWorkersCanAccessRenderSystem() const;
virtual void setWorkersCanAccessRenderSystem(bool access);
virtual void _processNextRequest();
virtual void _threadMain() = 0;
virtual bool isShuttingDown() const { return mShuttingDown; }
virtual void addRequestHandler(uint16 channel, RequestHandler* rh);
virtual void removeRequestHandler(uint16 channel, RequestHandler* rh);
virtual void addResponseHandler(uint16 channel, ResponseHandler* rh);
virtual void removeResponseHandler(uint16 channel, ResponseHandler* rh);
virtual RequestID addRequest(uint16 channel, uint16 requestType, const Any& rData, uint8 retryCount = 0,
bool forceSynchronous = false, bool idleThread = false);
virtual void abortRequest(RequestID id);
virtual void abortRequestsByChannel(uint16 channel);
virtual void abortPendingRequestsByChannel(uint16 channel);
virtual void abortAllRequests();
virtual void setPaused(bool pause);
virtual bool isPaused() const;
virtual void setRequestsAccepted(bool accept);
virtual bool getRequestsAccepted() const;
virtual void processResponses();
virtual unsigned long getResponseProcessingTimeLimit() const { return mResposeTimeLimitMS; }
virtual void setResponseProcessingTimeLimit(unsigned long ms) { mResposeTimeLimitMS = ms; }
protected:
String mName;
size_t mWorkerThreadCount;
bool mWorkerRenderSystemAccess;
bool mIsRunning;
unsigned long mResposeTimeLimitMS;
typedef deque<Request*>::type RequestQueue;
typedef deque<Response*>::type ResponseQueue;
RequestQueue mRequestQueue; // Guarded by mRequestMutex
RequestQueue mProcessQueue; // Guarded by mProcessMutex
ResponseQueue mResponseQueue; // Guarded by mResponseMutex
struct _OgreExport WorkerFunc OGRE_THREAD_WORKER_INHERIT
{
DefaultWorkQueueBase* mQueue;
WorkerFunc(DefaultWorkQueueBase* q)
: mQueue(q) {}
void operator()();
void operator()() const;
void run();
};
WorkerFunc* mWorkerFunc;
class _OgreExport RequestHandlerHolder : public UtilityAlloc
{
protected:
OGRE_RW_MUTEX(mRWMutex);
RequestHandler* mHandler;
public:
RequestHandlerHolder(RequestHandler* handler)
: mHandler(handler) {}
// Disconnect the handler to allow it to be destroyed
void disconnectHandler()
{
// write lock - must wait for all requests to finish
OGRE_LOCK_RW_MUTEX_WRITE(mRWMutex);
mHandler = 0;
}
RequestHandler* getHandler() { return mHandler; }
Response* handleRequest(const Request* req, const WorkQueue* srcQ)
{
// Read mutex so that multiple requests can be processed by the
// same handler in parallel if required
OGRE_LOCK_RW_MUTEX_READ(mRWMutex);
Response* response = 0;
if (mHandler)
{
if (mHandler->canHandleRequest(req, srcQ))
{
response = mHandler->handleRequest(req, srcQ);
}
}
return response;
}
};
// Hold these by shared pointer so they can be copied keeping same instance
typedef SharedPtr<RequestHandlerHolder> RequestHandlerHolderPtr;
typedef list<RequestHandlerHolderPtr>::type RequestHandlerList;
typedef list<ResponseHandler*>::type ResponseHandlerList;
typedef map<uint16, RequestHandlerList>::type RequestHandlerListByChannel;
typedef map<uint16, ResponseHandlerList>::type ResponseHandlerListByChannel;
RequestHandlerListByChannel mRequestHandlers;
ResponseHandlerListByChannel mResponseHandlers;
RequestID mRequestCount; // Guarded by mRequestMutex
bool mPaused;
bool mAcceptRequests;
bool mShuttingDown;
//NOTE: If you lock multiple mutexes at the same time, the order is important!
// For example if threadA locks mIdleMutex first then tries to lock mProcessMutex,
// and threadB locks mProcessMutex first, then mIdleMutex. In this case you can get livelock and the system is dead!
//RULE: Lock mProcessMutex before other mutex, to prevent livelocks
OGRE_MUTEX(mIdleMutex);
OGRE_MUTEX(mRequestMutex);
OGRE_MUTEX(mProcessMutex);
OGRE_MUTEX(mResponseMutex);
OGRE_RW_MUTEX(mRequestHandlerMutex);
void processRequestResponse(Request* r, bool synchronous);
Response* processRequest(Request* r);
void processResponse(Response* r);
virtual void notifyWorkers() = 0;
void addRequestWithRID(RequestID rid, uint16 channel, uint16 requestType, const Any& rData, uint8 retryCount);
RequestQueue mIdleRequestQueue; // Guarded by mIdleMutex
bool mIdleThreadRunning; // Guarded by mIdleMutex
Request* mIdleProcessed; // Guarded by mProcessMutex
bool processIdleRequests();
};
}
#include "OgreHeaderSuffix.h"
#endif
Ogre::Any::destroy
void destroy()
Definition: OgreAny.h:122
Ogre::WorkQueue::mChannelMap
ChannelMap mChannelMap
Definition: OgreWorkQueue.h:74
Ogre::WorkQueue::Request::abortRequest
void abortRequest() const
Set the abort flag.
Definition: OgreWorkQueue.h:105
Ogre::WorkQueue::RequestID
unsigned long long int RequestID
Numeric identifier for a request.
Definition: OgreWorkQueue.h:79
Ogre::DefaultWorkQueueBase::RequestHandlerList
list< RequestHandlerHolderPtr >::type RequestHandlerList
Definition: OgreWorkQueue.h:536
Ogre::DefaultWorkQueueBase::RequestHandlerHolder::getHandler
RequestHandler * getHandler()
Get handler pointer - note, only use this for == comparison or similar, do not attempt to call it as ...
Definition: OgreWorkQueue.h:511
Ogre::WorkQueue::Request::mID
RequestID mID
Identifier (assigned by the system)
Definition: OgreWorkQueue.h:96
Ogre::WorkQueue::Response::mRequest
const Request * mRequest
Pointer to the request that this response is in relation to.
Definition: OgreWorkQueue.h:125
Ogre::DefaultWorkQueueBase::RequestHandlerHolder::disconnectHandler
void disconnectHandler()
Definition: OgreWorkQueue.h:501
Ogre::DefaultWorkQueueBase::RequestHandlerHolderPtr
SharedPtr< RequestHandlerHolder > RequestHandlerHolderPtr
Definition: OgreWorkQueue.h:534
Ogre::DefaultWorkQueueBase::RequestHandlerListByChannel
map< uint16, RequestHandlerList >::type RequestHandlerListByChannel
Definition: OgreWorkQueue.h:538
_OgreExport
#define _OgreExport
Definition: OgrePlatform.h:257
Ogre::WorkQueue::ResponseHandler::canHandleResponse
virtual bool canHandleResponse(const Response *res, const WorkQueue *srcQ)
Return whether this handler can process a given response.
Definition: OgreWorkQueue.h:208
Ogre::DefaultWorkQueueBase::RequestHandlerHolder::RequestHandlerHolder
RequestHandlerHolder(RequestHandler *handler)
Definition: OgreWorkQueue.h:497
Ogre::DefaultWorkQueueBase::mIdleRequestQueue
RequestQueue mIdleRequestQueue
Definition: OgreWorkQueue.h:567
OgreSharedPtr.h
Ogre::WorkQueue
Interface to a general purpose request / response style background work queue.
Definition: OgreWorkQueue.h:70
Ogre::DefaultWorkQueueBase::mRequestCount
RequestID mRequestCount
Definition: OgreWorkQueue.h:543
Ogre::DefaultWorkQueueBase::mIdleProcessed
Request * mIdleProcessed
Definition: OgreWorkQueue.h:569
Ogre::WorkQueue::ChannelMap
map< String, uint16 >::type ChannelMap
Definition: OgreWorkQueue.h:73
Ogre::WorkQueue::RequestHandler::canHandleRequest
virtual bool canHandleRequest(const Request *req, const WorkQueue *srcQ)
Return whether this handler can process a given request.
Definition: OgreWorkQueue.h:173
Ogre::DefaultWorkQueueBase::ResponseQueue
deque< Response * >::type ResponseQueue
Definition: OgreWorkQueue.h:466
Ogre::DefaultWorkQueueBase::OGRE_THREAD_WORKER_INHERIT::WorkerFunc
WorkerFunc(DefaultWorkQueueBase *q)
Definition: OgreWorkQueue.h:476
Ogre::WorkQueue::Request::getAborted
bool getAborted() const
Get the abort flag.
Definition: OgreWorkQueue.h:117
Ogre::DefaultWorkQueueBase::mWorkerThreadCount
size_t mWorkerThreadCount
Definition: OgreWorkQueue.h:460
Ogre::DefaultWorkQueueBase::mName
String mName
Definition: OgreWorkQueue.h:459
Ogre::DefaultWorkQueueBase::getResponseProcessingTimeLimit
virtual unsigned long getResponseProcessingTimeLimit() const
Get the time limit imposed on the processing of responses in a single frame, in milliseconds (0 indic...
Definition: OgreWorkQueue.h:455
Ogre::DefaultWorkQueueBase::setResponseProcessingTimeLimit
virtual void setResponseProcessingTimeLimit(unsigned long ms)
Set the time limit imposed on the processing of responses in a single frame, in milliseconds (0 indic...
Definition: OgreWorkQueue.h:457
Ogre::WorkQueue::Request::getID
RequestID getID() const
Get the identifier of this request.
Definition: OgreWorkQueue.h:115
OGRE_MUTEX
#define OGRE_MUTEX(name)
Definition: OgreThreadDefinesBoost.h:47
Ogre::DefaultWorkQueueBase::mIdleThreadRunning
bool mIdleThreadRunning
Definition: OgreWorkQueue.h:568
Ogre::WorkQueue::Response::abortRequest
void abortRequest()
Abort the request.
Definition: OgreWorkQueue.h:145
Ogre::Any
Variant type that can hold Any other type.
Definition: OgreAny.h:56
Ogre::WorkQueue::ResponseHandler::ResponseHandler
ResponseHandler()
Definition: OgreWorkQueue.h:199
Ogre::WorkQueue::Request::mRetryCount
uint8 mRetryCount
Retry count - set this to non-zero to have the request try again on failure.
Definition: OgreWorkQueue.h:94
OGRE_LOCK_RW_MUTEX_WRITE
#define OGRE_LOCK_RW_MUTEX_WRITE(name)
Definition: OgreThreadDefinesBoost.h:64
Ogre::WorkQueue::~WorkQueue
virtual ~WorkQueue()
Definition: OgreWorkQueue.h:222
Ogre::WorkQueue::Response::mMessages
String mMessages
Any diagnostic messages.
Definition: OgreWorkQueue.h:129
Ogre::DefaultWorkQueueBase::mRequestQueue
RequestQueue mRequestQueue
Definition: OgreWorkQueue.h:467
Ogre::DefaultWorkQueueBase::isShuttingDown
virtual bool isShuttingDown() const
Returns whether the queue is trying to shut down.
Definition: OgreWorkQueue.h:422
Ogre::WorkQueue::Request
General purpose request structure.
Definition: OgreWorkQueue.h:83
OgrePrerequisites.h
Ogre::DefaultWorkQueueBase::mWorkerRenderSystemAccess
bool mWorkerRenderSystemAccess
Definition: OgreWorkQueue.h:461
Ogre::deque
Definition: OgrePrerequisites.h:477
Ogre::WorkQueue::Response::mData
Any mData
Data associated with the result of the process.
Definition: OgreWorkQueue.h:131
Ogre::WorkQueue::Response::succeeded
bool succeeded() const
Return whether this is a successful response.
Definition: OgreWorkQueue.h:139
Ogre::WorkQueue::Response
General purpose response structure.
Definition: OgreWorkQueue.h:122
OGRE_RW_MUTEX
#define OGRE_RW_MUTEX(name)
Definition: OgreThreadDefinesBoost.h:62
Ogre::uint8
unsigned char uint8
Definition: OgrePlatform.h:361
Ogre::DefaultWorkQueueBase::OGRE_THREAD_WORKER_INHERIT::mQueue
DefaultWorkQueueBase * mQueue
Definition: OgreWorkQueue.h:474
Ogre::DefaultWorkQueueBase::mRequestHandlers
RequestHandlerListByChannel mRequestHandlers
Definition: OgreWorkQueue.h:541
Ogre::String
_StringBase String
Definition: OgrePrerequisites.h:439
Ogre::WorkQueue::Request::mAborted
bool mAborted
Abort Flag.
Definition: OgreWorkQueue.h:98
Ogre::WorkQueue::RequestHandler::~RequestHandler
virtual ~RequestHandler()
Definition: OgreWorkQueue.h:165
Ogre::DefaultWorkQueueBase::mWorkerFunc
WorkerFunc * mWorkerFunc
Definition: OgreWorkQueue.h:485
Ogre::WorkQueue::Request::getRetryCount
uint8 getRetryCount() const
Get the remaining retry count.
Definition: OgreWorkQueue.h:113
Ogre::WorkQueue::Request::mType
uint16 mType
The request type, as an integer within the channel (user can define enumerations on this) ...
Definition: OgreWorkQueue.h:90
Ogre::DefaultWorkQueueBase::RequestQueue
deque< Request * >::type RequestQueue
Definition: OgreWorkQueue.h:465
Ogre::DefaultWorkQueueBase::mIsRunning
bool mIsRunning
Definition: OgreWorkQueue.h:462
Ogre::DefaultWorkQueueBase::mProcessQueue
RequestQueue mProcessQueue
Definition: OgreWorkQueue.h:468
Ogre::WorkQueue::Request::getChannel
uint16 getChannel() const
Get the request channel (top level categorisation)
Definition: OgreWorkQueue.h:107
Ogre::DefaultWorkQueueBase::mResponseHandlers
ResponseHandlerListByChannel mResponseHandlers
Definition: OgreWorkQueue.h:542
OgreHeaderSuffix.h
Ogre::DefaultWorkQueueBase::mResposeTimeLimitMS
unsigned long mResposeTimeLimitMS
Definition: OgreWorkQueue.h:463
Ogre::DefaultWorkQueueBase::mShuttingDown
bool mShuttingDown
Definition: OgreWorkQueue.h:546
Ogre::WorkQueue::Request::getType
uint16 getType() const
Get the type of this request within the given channel.
Definition: OgreWorkQueue.h:109
Ogre::DefaultWorkQueueBase
Base for a general purpose request / response style background work queue.
Definition: OgreWorkQueue.h:360
OGRE_LOCK_RW_MUTEX_READ
#define OGRE_LOCK_RW_MUTEX_READ(name)
Definition: OgreThreadDefinesBoost.h:63
Ogre::WorkQueue::Response::getRequest
const Request * getRequest() const
Get the request that this is a response to (NB destruction destroys this)
Definition: OgreWorkQueue.h:137
OgreThreadHeaders.h
Ogre::WorkQueue::mNextChannel
uint16 mNextChannel
Definition: OgreWorkQueue.h:75
OgreAny.h
Ogre::DefaultWorkQueueBase::RequestHandlerHolder::handleRequest
Response * handleRequest(const Request *req, const WorkQueue *srcQ)
Process a request if possible.
Definition: OgreWorkQueue.h:516
Ogre::WorkQueue::Response::getData
const Any & getData() const
Return the response data (user defined, only valid on success)
Definition: OgreWorkQueue.h:143
Ogre::WorkQueue::ResponseHandler
Interface definition for a handler of responses.
Definition: OgreWorkQueue.h:196
Ogre
Definition: OgreAndroidLogListener.h:34
Ogre::WorkQueue::Request::mChannel
uint16 mChannel
The request channel, as an integer.
Definition: OgreWorkQueue.h:88
Ogre::WorkQueue::RequestHandler::RequestHandler
RequestHandler()
Definition: OgreWorkQueue.h:164
Ogre::DefaultWorkQueueBase::ResponseHandlerListByChannel
map< uint16, ResponseHandlerList >::type ResponseHandlerListByChannel
Definition: OgreWorkQueue.h:539
Ogre::SharedPtr
Reference-counted shared pointer, used for objects where implicit destruction is required.
Definition: OgrePrerequisites.h:310
Ogre::DefaultWorkQueueBase::mResponseQueue
ResponseQueue mResponseQueue
Definition: OgreWorkQueue.h:469
OgreHeaderPrefix.h
Ogre::DefaultWorkQueueBase::RequestHandlerHolder::mHandler
RequestHandler * mHandler
Definition: OgreWorkQueue.h:495
Ogre::WorkQueue::Request::mData
Any mData
The details of the request (user defined)
Definition: OgreWorkQueue.h:92
Ogre::map
Definition: OgrePrerequisites.h:533
Ogre::DefaultWorkQueueBase::ResponseHandlerList
list< ResponseHandler * >::type ResponseHandlerList
Definition: OgreWorkQueue.h:537
Ogre::list::type
std::list< T, A > type
Definition: OgrePrerequisites.h:508
Ogre::WorkQueue::Response::getMessages
const String & getMessages() const
Get any diagnostic messages about the process.
Definition: OgreWorkQueue.h:141
Ogre::WorkQueue::ResponseHandler::~ResponseHandler
virtual ~ResponseHandler()
Definition: OgreWorkQueue.h:200
Ogre::WorkQueue::RequestHandler
Interface definition for a handler of requests.
Definition: OgreWorkQueue.h:161
Ogre::DefaultWorkQueueBase::OGRE_THREAD_WORKER_INHERIT
Thread function.
Definition: OgreWorkQueue.h:472
Ogre::DefaultWorkQueueBase::mPaused
bool mPaused
Definition: OgreWorkQueue.h:544
Ogre::WorkQueue::WorkQueue
WorkQueue()
Definition: OgreWorkQueue.h:221
Ogre::StringUtil::BLANK
static const String BLANK
Constant blank string, useful for returning by ref where local does not exist.
Definition: OgreString.h:196
Ogre::DefaultWorkQueueBase::mAcceptRequests
bool mAcceptRequests
Definition: OgreWorkQueue.h:545
Ogre::uint16
unsigned short uint16
Definition: OgrePlatform.h:360
Ogre::AllocatedObject
Superclass for all objects that wish to use custom memory allocators when their new / delete operator...
Definition: OgreMemoryAllocatedObject.h:58
Ogre::WorkQueue::Request::getData
const Any & getData() const
Get the user details of this request.
Definition: OgreWorkQueue.h:111
Ogre::WorkQueue::Response::mSuccess
bool mSuccess
Whether the work item succeeded or not.
Definition: OgreWorkQueue.h:127
Ogre::DefaultWorkQueueBase::RequestHandlerHolder
Intermediate structure to hold a pointer to a request handler which provides insurance against the ha...
Definition: OgreWorkQueue.h:491

Copyright © 2012 Torus Knot Software Ltd
This work is licensed under a Creative Commons Attribution-ShareAlike 3.0 Unported License.