bugfix: regression caused more locking action in msg.c than necessary

also: bugfix: mutexes used to similate atomic instructions were not destructed

6 files changed, 39 insertions, 43 deletions

diff --git a/ChangeLog b/ChangeLog
index 34a64bfe..f62ac6ad 100644
--- a/ChangeLog
+++ b/ChangeLog
@@ -12,6 +12,8 @@ Version 5.5.6  [DEVEL] (rgerhards), 2010-06-??
   * RSYSLOG_TraditionalFileFormat
   * RSYSLOG_ForwardFormat
   * RSYSLOG_TraditionalForwardFormat
+- bugfix: mutexes used to similate atomic instructions were not destructed
+- bugfix: regression caused more locking action in msg.c than necessary
 ---------------------------------------------------------------------------
 Version 5.5.5  [DEVEL] (rgerhards), 2010-05-20
 - added new cancel-reduced action thread termination method
@@ -23,7 +25,7 @@ Version 5.5.5  [DEVEL] (rgerhards), 2010-05-20
 ---------------------------------------------------------------------------
 Version 5.5.4  [DEVEL] (rgerhards), 2010-05-03
 - This version offers full support for Solaris on Intel and Sparc
-- bugfix: problems with atomic operations emulaton
+- bugfix: problems with atomic operations emulation
   replaced atomic operation emulation with new code. The previous code
   seemed to have some issue and also limited concurrency severely. The
   whole atomic operation emulation has been rewritten.
diff --git a/action.c b/action.c
index c8622764..0fac6cf8 100644
--- a/action.c
+++ b/action.c
@@ -955,7 +955,6 @@ submitBatch(action_t *pAction, batch_t *pBatch, int nElem, int *pbShutdownImmedi
 		} else {
 			if(nElem == 1) {
 				pBatch->pElem[pBatch->iDoneUpTo++].state = BATCH_STATE_BAD;
-// TODO: This is a mark, remove when no longer needed - Here was the bug, postincrement needs to be used, not preinc
 				bDone = 1;
 			} else {
 				/* retry with half as much. Depth is log_2 batchsize, so recursion is not too deep */
diff --git a/doc/highperf.txt b/doc/highperf.txt
new file mode 100644
index 00000000..5f9481e1
--- /dev/null
+++ b/doc/highperf.txt
@@ -0,0 +1,4 @@
+links to high performance papers:
+
+- http://www.kegel.com/c10k.html
+- http://pl.atyp.us/content/tech/servers.html (**)
diff --git a/runtime/atomic.h b/runtime/atomic.h
index d8cdff7b..e5fafe04 100644
--- a/runtime/atomic.h
+++ b/runtime/atomic.h
@@ -42,7 +42,7 @@
 #	define ATOMIC_SUB(data, val, phlpmut) __sync_fetch_and_sub(data, val)
 #	define ATOMIC_ADD(data, val) __sync_fetch_and_add(&(data), val)
 #	define ATOMIC_INC(data, phlpmut) ((void) __sync_fetch_and_add(data, 1))
-#	define ATOMIC_INC_AND_FETCH(data) __sync_fetch_and_add(&(data), 1)
+#	define ATOMIC_INC_AND_FETCH(data, phlpmut) __sync_fetch_and_add(data, 1)
 #	define ATOMIC_DEC(data, phlpmut) ((void) __sync_sub_and_fetch(data, 1))
 #	define ATOMIC_DEC_AND_FETCH(data, phlpmut) __sync_sub_and_fetch(data, 1)
 #	define ATOMIC_FETCH_32BIT(data, phlpmut) ((unsigned) __sync_fetch_and_and(data, 0xffffffff))
@@ -109,6 +109,15 @@
 	}
 
 	static inline int
+	ATOMIC_INC_AND_FETCH(int *data, pthread_mutex_t *phlpmut) {
+		int val;
+		pthread_mutex_lock(phlpmut);
+		val = ++(*data);
+		pthread_mutex_unlock(phlpmut);
+		return(val);
+	}
+
+	static inline int
 	ATOMIC_DEC_AND_FETCH(int *data, pthread_mutex_t *phlpmut) {
 		int val;
 		pthread_mutex_lock(phlpmut);
@@ -139,7 +148,7 @@
 #endif
 #	define DEF_ATOMIC_HELPER_MUT(x)  pthread_mutex_t x
 #	define INIT_ATOMIC_HELPER_MUT(x) pthread_mutex_init(&(x), NULL)
-#	define DESTROY_ATOMIC_HELPER_MUT(x) pthread_mutex_init(&(x), NULL)
+#	define DESTROY_ATOMIC_HELPER_MUT(x) pthread_mutex_destroy(&(x))
 
 #	define PREFER_ATOMIC_INC(data) ((void) ++data)
 
diff --git a/runtime/msg.c b/runtime/msg.c
index dc354947..1b188263 100644
--- a/runtime/msg.c
+++ b/runtime/msg.c
@@ -278,6 +278,11 @@ static char *syslog_severity_names[8] = { "emerg", "alert", "crit", "err", "warn
 static char *syslog_number_names[24] = { "0", "1", "2", "3", "4", "5", "6", "7", "8", "9", "10", "11", "12", "13", "14",
 					 "15", "16", "17", "18", "19", "20", "21", "22", "23" };
 
+/* global variables */
+#if defined(HAVE_MALLOC_TRIM) && !defined(HAVE_ATOMIC_BUILTINS)
+static pthread_mutex_t mutTrimCtr;	 /* mutex to handle malloc trim */
+#endif
+
 /* some forward declarations */
 static int getAPPNAMELen(msg_t *pM, sbool bLockMutex);
 
@@ -784,6 +789,9 @@ static inline void freeHOSTNAME(msg_t *pThis)
 
 BEGINobjDestruct(msg) /* be sure to specify the object type also in END and CODESTART macros! */
 	int currRefCount;
+#	if HAVE_MALLOC_TRIM
+	int currCnt;
+#	endif
 CODESTARTobjDestruct(msg)
 	/* DEV Debugging only ! dbgprintf("msgDestruct\t0x%lx, Ref now: %d\n", (unsigned long)pThis, pThis->iRefCount - 1); */
 #	ifdef HAVE_ATOMIC_BUILTINS
@@ -843,18 +851,10 @@ CODESTARTobjDestruct(msg)
 			 * that we trim too often when the counter wraps.
 			 */
 			static unsigned iTrimCtr = 1;
-#	ifdef HAVE_ATOMICS
-			if(ATOMIC_INC_AND_FETCH(iTrimCtr) % 100000 == 0) {
-				malloc_trim(128*1024);
-			}
-#	else
-static pthread_mutex_t mutTrimCtr = PTHREAD_MUTEX_INITIALIZER;
-			d_pthread_mutex_lock(&mutTrimCtr);
-			if(iTrimCtr++ % 100000 == 0) {
+			currCnt = ATOMIC_INC_AND_FETCH(&iTrimCtr, &mutTrimCtr);
+			if(currCnt % 100000 == 0) {
 				malloc_trim(128*1024);
 			}
-			d_pthread_mutex_unlock(&mutTrimCtr);
-#	endif
 		}
 #		endif
 	} else {
@@ -3185,6 +3185,10 @@ BEGINObjClassInit(msg, 1, OBJ_IS_CORE_MODULE)
 	funcUnlock = MsgLockingDummy;
 	funcDeleteMutex = MsgLockingDummy;
 	funcMsgPrepareEnqueue = MsgLockingDummy;
+	/* some more inits */
+#	if HAVE_MALLOC_TRIM
+	INIT_ATOMIC_HELPER_MUT(mutTrimCtr);
+#	endif
 ENDObjClassInit(msg)
 /* vim:set ai:
  */
diff --git a/tools/omfile.c b/tools/omfile.c
index 5e49afa9..92089a59 100644
--- a/tools/omfile.c
+++ b/tools/omfile.c
@@ -92,33 +92,15 @@ static uint64 clockFileAccess = 0;
 static unsigned clockFileAccess = 0;
 #endif
 /* and the "tick" function */
-#ifdef HAVE_ATOMIC_BUILTINS
-static inline uint64
-getClockFileAccess(void)
-{
-	return ATOMIC_INC_AND_FETCH(clockFileAccess);
-}
-#else
-/* if we do not have atomics, we need to guard this via a mutex.
- * the reason is that otherwise cache lookups may fail. That function
- * requires the highest current value, and we can not provide that
- * without using atomic instructions and mutexes.
- * rgerhards, 2010-03-05
- */
+#ifndef HAVE_ATOMIC_BUILTINS
 static pthread_mutex_t mutClock;
-static uint64
+#endif
+static inline uint64
 getClockFileAccess(void)
 {
-	uint64 retVal;
-
-	BEGINfunc
-	d_pthread_mutex_lock(&mutClock);
-	retVal = ++clockFileAccess;
-	d_pthread_mutex_unlock(&mutClock);
-	ENDfunc
-	return retVal;
+	return ATOMIC_INC_AND_FETCH(&clockFileAccess, &mutClock);
 }
-#endif  /* #ifdef HAVE_ATOMIC_BUILTINS */
+
 
 /* The following structure is a dynafile name cache entry.
  */
@@ -879,9 +861,7 @@ CODESTARTmodExit
 	objRelease(errmsg, CORE_COMPONENT);
 	objRelease(strm, CORE_COMPONENT);
 	free(pszFileDfltTplName);
-#	ifndef HAVE_ATOMIC_BUILTINS
-	pthread_mutex_destroy(&mutClock);
-#	endif
+	DESTROY_ATOMIC_HELPER_MUT(mutClock);
 ENDmodExit
 
 
@@ -900,9 +880,7 @@ CODEmodInit_QueryRegCFSLineHdlr
 	CHKiRet(objUse(errmsg, CORE_COMPONENT));
 	CHKiRet(objUse(strm, CORE_COMPONENT));
 
-#	ifndef HAVE_ATOMIC_BUILTINS
-	pthread_mutex_init(&mutClock, NULL);
-#	endif
+	INIT_ATOMIC_HELPER_MUT(mutClock);
 
 	INITChkCoreFeature(bCoreSupportsBatching, CORE_FEATURE_BATCHING);
 	DBGPRINTF("omfile: %susing transactional output interface.\n", bCoreSupportsBatching ? "" : "not ");