Merge branch 'master' of git://anonscm.debian.org/collab-maint/rsyslog

Conflicts:
	debian/changelog
	debian/patches/series
	debian/rules

1 files changed, 53 insertions, 11 deletions

diff --git a/tools/ompipe.c b/tools/ompipe.c
index 420e2b1..c94568b 100644
--- a/tools/ompipe.c
+++ b/tools/ompipe.c
@@ -12,7 +12,7 @@
  * NOTE: read comments in module-template.h to understand how this pipe
  *       works!
  *
- * Copyright 2007-2012 Rainer Gerhards and Adiscon GmbH.
+ * Copyright 2007-2014 Rainer Gerhards and Adiscon GmbH.
  *
  * This file is part of rsyslog.
  *
@@ -69,9 +69,14 @@ typedef struct _instanceData {
 	uchar	*pipe;	/* pipe or template name (display only) */
 	uchar	*tplName;       /* format template to use */
 	short	fd;		/* pipe descriptor for (current) pipe */
+	pthread_mutex_t mutWrite; /* guard against multiple instances writing to same pipe */
 	sbool	bHadError;	/* did we already have/report an error on this pipe? */
 } instanceData;
 
+typedef struct wrkrInstanceData {
+	instanceData *pData;
+} wrkrInstanceData_t;
+
 typedef struct configSettings_s {
 	EMPTY_STRUCT
 } configSettings_t;
@@ -154,7 +159,7 @@ preparePipe(instanceData *pData)
 		if(!pData->bHadError) {
 			char errStr[1024];
 			rs_strerror_r(errno, errStr, sizeof(errStr));
-			errmsg.LogError(0, RS_RET_NO_FILE_ACCESS, "Could no open output pipe '%s': %s",
+			errmsg.LogError(0, RS_RET_NO_FILE_ACCESS, "Could not open output pipe '%s': %s",
 				        pData->pipe, errStr);
 			pData->bHadError = 1;
 		}
@@ -276,25 +281,66 @@ CODESTARTcreateInstance
 	pData->pipe = NULL;
 	pData->fd = -1;
 	pData->bHadError = 0;
+	pthread_mutex_init(&pData->mutWrite, NULL);
 ENDcreateInstance
 
 
+BEGINcreateWrkrInstance
+CODESTARTcreateWrkrInstance
+ENDcreateWrkrInstance
+
+
 BEGINfreeInstance
 CODESTARTfreeInstance
+	pthread_mutex_destroy(&pData->mutWrite);
 	free(pData->pipe);
 	if(pData->fd != -1)
 		close(pData->fd);
 ENDfreeInstance
 
 
+BEGINfreeWrkrInstance
+CODESTARTfreeWrkrInstance
+ENDfreeWrkrInstance
+
+
 BEGINtryResume
+	instanceData *__restrict__ const pData = pWrkrData->pData;
+	fd_set wrds;
+	struct timeval tv;
+	int ready;
 CODESTARTtryResume
+	if(pData->fd == -1) {
+		rsRetVal iRetLocal;
+		iRetLocal = preparePipe(pData);
+		if((iRetLocal != RS_RET_OK) || (pData->fd == -1))
+			ABORT_FINALIZE(RS_RET_SUSPENDED);
+	} else {
+		/* we can reach this if the pipe is full, so we need
+		 * to check if we can write again. /dev/xconsole is the
+		 * ugly example of why this is necessary.
+		 */
+		FD_ZERO(&wrds);
+		FD_SET(pData->fd, &wrds);
+		tv.tv_sec = 0;
+		tv.tv_usec = 0;
+		ready = select(pData->fd+1, NULL, &wrds, NULL, &tv);
+		DBGPRINTF("ompipe: tryResume: ready to write fd %d: %d\n", pData->fd, ready);
+		if(ready != 1)
+			ABORT_FINALIZE(RS_RET_SUSPENDED);
+	}
+finalize_it:
 ENDtryResume
 
 BEGINdoAction
+	instanceData *pData;
 CODESTARTdoAction
-	DBGPRINTF(" (%s)\n", pData->pipe);
+	pData = pWrkrData->pData;
+	DBGPRINTF("ompipe: writing to %s\n", pData->pipe);
+	/* this module is single-threaded by nature */
+	pthread_mutex_lock(&pData->mutWrite);
 	iRet = writePipe(ppString, pData);
+	pthread_mutex_unlock(&pData->mutWrite);
 ENDdoAction
 
 
@@ -329,14 +375,9 @@ CODESTARTnewActInst
 		}
 	}
 
-	if(pData->tplName == NULL) {
-		CHKiRet(OMSRsetEntry(*ppOMSR, 0, (uchar*) "RSYSLOG_FileFormat",
-			OMSR_NO_RQD_TPL_OPTS));
-	} else {
-		CHKiRet(OMSRsetEntry(*ppOMSR, 0,
-			(uchar*) strdup((char*) pData->tplName),
-			OMSR_NO_RQD_TPL_OPTS));
-	}
+	CHKiRet(OMSRsetEntry(*ppOMSR, 0, (uchar*)strdup((pData->tplName == NULL) ? 
+						"RSYSLOG_FileFormat" : (char*)pData->tplName),
+						OMSR_NO_RQD_TPL_OPTS));
 CODE_STD_FINALIZERnewActInst
 	cnfparamvalsDestruct(pvals, &actpblk);
 ENDnewActInst
@@ -386,6 +427,7 @@ ENDmodExit
 BEGINqueryEtryPt
 CODESTARTqueryEtryPt
 CODEqueryEtryPt_STD_OMOD_QUERIES
+CODEqueryEtryPt_STD_OMOD8_QUERIES
 CODEqueryEtryPt_doHUP
 CODEqueryEtryPt_STD_CONF2_QUERIES
 CODEqueryEtryPt_STD_CONF2_CNFNAME_QUERIES