summaryrefslogtreecommitdiffstats
path: root/runtime/wtp.h
diff options
context:
space:
mode:
Diffstat (limited to 'runtime/wtp.h')
-rw-r--r--runtime/wtp.h14
1 files changed, 9 insertions, 5 deletions
diff --git a/runtime/wtp.h b/runtime/wtp.h
index b9cb07c5..e1180177 100644
--- a/runtime/wtp.h
+++ b/runtime/wtp.h
@@ -50,7 +50,7 @@ typedef enum {
/* the worker thread pool (wtp) object */
-typedef struct wtp_s {
+struct wtp_s {
BEGINobjInstance;
int bOptimizeUniProc; /* cache for the equally-named global setting, pulled at time of queue creation */
wtpState_t wtpState;
@@ -67,12 +67,14 @@ typedef struct wtp_s {
int bThrdStateChanged; /* at least one thread state has changed if 1 */
/* end sync variables */
/* user objects */
- void *pUsr; /* pointer to user object */
+ void *pUsr; /* pointer to user object (in this case, the queue the wtp belongs to) */
pthread_mutex_t *pmutUsr;
pthread_cond_t *pcondBusy; /* condition the user will signal "busy again, keep runing" on (awakes worker) */
rsRetVal (*pfChkStopWrkr)(void *pUsr, int);
+ rsRetVal (*pfGetDeqBatchSize)(void *pUsr, int*); /* obtains max dequeue count from queue config */
+ rsRetVal (*pfObjProcessed)(void *pUsr, wti_t *pWti); /* indicate user object is processed */
rsRetVal (*pfRateLimiter)(void *pUsr);
- rsRetVal (*pfIsIdle)(void *pUsr, int);
+ rsRetVal (*pfIsIdle)(void *pUsr, wtp_t *pWtp);
rsRetVal (*pfDoWork)(void *pUsr, void *pWti, int);
rsRetVal (*pfOnIdle)(void *pUsr, int);
rsRetVal (*pfOnWorkerCancel)(void *pUsr, void*pWti);
@@ -80,7 +82,7 @@ typedef struct wtp_s {
rsRetVal (*pfOnWorkerShutdown)(void *pUsr);
/* end user objects */
uchar *pszDbgHdr; /* header string for debug messages */
-} wtp_t;
+};
/* some symbolic constants for easier reference */
@@ -104,8 +106,10 @@ int wtpGetCurNumWrkr(wtp_t *pThis, int bLockMutex);
PROTOTYPEObjClassInit(wtp);
PROTOTYPEpropSetMethFP(wtp, pfChkStopWrkr, rsRetVal(*pVal)(void*, int));
PROTOTYPEpropSetMethFP(wtp, pfRateLimiter, rsRetVal(*pVal)(void*));
-PROTOTYPEpropSetMethFP(wtp, pfIsIdle, rsRetVal(*pVal)(void*, int));
+PROTOTYPEpropSetMethFP(wtp, pfGetDeqBatchSize, rsRetVal(*pVal)(void*, int*));
+PROTOTYPEpropSetMethFP(wtp, pfIsIdle, rsRetVal(*pVal)(void*, wtp_t*));
PROTOTYPEpropSetMethFP(wtp, pfDoWork, rsRetVal(*pVal)(void*, void*, int));
+PROTOTYPEpropSetMethFP(wtp, pfObjProcessed, rsRetVal(*pVal)(void*, wti_t*));
PROTOTYPEpropSetMethFP(wtp, pfOnIdle, rsRetVal(*pVal)(void*, int));
PROTOTYPEpropSetMethFP(wtp, pfOnWorkerCancel, rsRetVal(*pVal)(void*,void*));
PROTOTYPEpropSetMethFP(wtp, pfOnWorkerStartup, rsRetVal(*pVal)(void*));