diff options
| -rw-r--r-- | action.c | 4 | ||||
| -rw-r--r-- | action.h | 1 | ||||
| -rw-r--r-- | runtime/atomic.h | 15 |
3 files changed, 19 insertions, 1 deletions
@@ -230,6 +230,7 @@ rsRetVal actionConstruct(action_t **ppThis) pThis->iResumeRetryCount = glbliActionResumeRetryCount; pThis->tLastOccur = datetime.GetTime(NULL); /* done once per action on startup only */ pthread_mutex_init(&pThis->mutActExec, NULL); + INIT_ATOMIC_HELPER_MUT(pThis->mutCAS); SYNC_OBJ_TOOL_INIT(pThis); /* indicate we have a new action */ @@ -1386,7 +1387,8 @@ doSubmitToActionQNotAllMarkBatch(action_t *pAction, batch_t *pBatch) } else { bProcessMarkMsgs = 1; } - } while(ATOMIC_CAS(&pAction->f_time, lastAct, ((msg_t*)(pBatch->pElem[i].pUsrp))->ttGenTime, ADDME) == 0); + } while(ATOMIC_CAS(&pAction->f_time, lastAct, + ((msg_t*)(pBatch->pElem[i].pUsrp))->ttGenTime, &pAction->mutCAS) == 0); } if(bProcessMarkMsgs) { pBatch->pElem[i].bFilterOK = 0; @@ -89,6 +89,7 @@ struct action_s { pthread_mutex_t mutActExec; /* mutex to guard actual execution of doAction for single-threaded modules */ uchar *pszName; /* action name (for documentation) */ int *pbShutdownImmediate;/* to facilitate shutdown, if var is 1, shut down immediately */ + DEF_ATOMIC_HELPER_MUT(mutCAS); }; diff --git a/runtime/atomic.h b/runtime/atomic.h index da0852fa..da544c4b 100644 --- a/runtime/atomic.h +++ b/runtime/atomic.h @@ -91,6 +91,21 @@ } static inline int + ATOMIC_CAS(int *data, int oldVal, int newVal, pthread_mutex_t *phlpmut) { + int bSuccess; + pthread_mutex_lock(phlpmut); + if(*data == oldVal) { + *data = newVal; + bSuccess = 1; + } else { + bSuccess = 0; + } + pthread_mutex_unlock(phlpmut); + return(bSuccess); + } + + + static inline int ATOMIC_CAS_VAL(int *data, int oldVal, int newVal, pthread_mutex_t *phlpmut) { int val; pthread_mutex_lock(phlpmut); |