summaryrefslogtreecommitdiffstats
diff options
context:
space:
mode:
-rw-r--r--runtime/Makefile.am17
-rw-r--r--runtime/netstrm.c82
-rw-r--r--runtime/netstrm.h34
-rw-r--r--runtime/nsd.h20
-rw-r--r--runtime/nsd_ptcp.c187
-rw-r--r--runtime/nsd_ptcp.h2
-rw-r--r--runtime/nsdsel_ptcp.c216
-rw-r--r--runtime/nsdsel_ptcp.h46
-rw-r--r--runtime/nssel.c222
-rw-r--r--runtime/nssel.h55
-rw-r--r--runtime/rsyslog.h5
-rw-r--r--tcps_sess.c2
-rw-r--r--tcps_sess.h2
-rw-r--r--tcpsrv.c235
-rw-r--r--tcpsrv.h13
15 files changed, 785 insertions, 353 deletions
diff --git a/runtime/Makefile.am b/runtime/Makefile.am
index 077310c6..61ede1d7 100644
--- a/runtime/Makefile.am
+++ b/runtime/Makefile.am
@@ -81,7 +81,7 @@ lmregexp_la_LIBADD =
endif
if ENABLE_INET
-pkglib_LTLIBRARIES += lmnet.la lmnetstrm.la
+pkglib_LTLIBRARIES += lmnet.la lmnetstrm.la lmnssel.la
#
# network support
#
@@ -96,12 +96,25 @@ lmnetstrm_la_CPPFLAGS = $(pthreads_cflags) $(rsrt_cflags)
lmnetstrm_la_LDFLAGS = -module -avoid-version
lmnetstrm_la_LIBADD =
+# network stream select support (a helper class)
+lmnssel_la_SOURCES = nssel.c nssel.h
+lmnssel_la_CPPFLAGS = $(pthreads_cflags) $(rsrt_cflags)
+lmnssel_la_LDFLAGS = -module -avoid-version
+lmnssel_la_LIBADD =
+
# netstream drivers
-# plain tcp driver
+# plain tcp driver - main driver
pkglib_LTLIBRARIES += lmnsd_ptcp.la
lmnsd_ptcp_la_SOURCES = nsd_ptcp.c nsd_ptcp.h
lmnsd_ptcp_la_CPPFLAGS = $(pthreads_cflags) $(rsrt_cflags)
lmnsd_ptcp_la_LDFLAGS = -module -avoid-version
lmnsd_ptcp_la_LIBADD =
+
+# select interface for ptcp driver
+pkglib_LTLIBRARIES += lmnsdsel_ptcp.la
+lmnsdsel_ptcp_la_SOURCES = nsdsel_ptcp.c nsdsel_ptcp.h
+lmnsdsel_ptcp_la_CPPFLAGS = $(pthreads_cflags) $(rsrt_cflags)
+lmnsdsel_ptcp_la_LDFLAGS = -module -avoid-version
+lmnsdsel_ptcp_la_LIBADD =
endif # if ENABLE_INET
diff --git a/runtime/netstrm.c b/runtime/netstrm.c
index 274a92d7..bdd2636c 100644
--- a/runtime/netstrm.c
+++ b/runtime/netstrm.c
@@ -144,73 +144,65 @@ AbortDestruct(netstrm_t **ppThis)
}
-#if 0
-This is not yet working - wait until we arrive at the receiver side (distracts too much at the moment)
-
-/* accept an incoming connection request, pNsdLstn provides the "listen socket" on which we can
- * accept the new session.
- * rgerhards, 2008-03-17
+/* accept an incoming connection request
+ * The netstrm instance that had the incoming request must be provided. If
+ * the connection request succeeds, a new netstrm object is created and
+ * passed back to the caller. The caller is responsible for destructing it.
+ * pReq is the nsd_t obj that has the accept request.
+ * rgerhards, 2008-04-21
*/
static rsRetVal
-AcceptConnReq(netstrm_t **ppThis, nsd_t *pNsdLstn)
+AcceptConnReq(netstrm_t *pThis, nsd_t *pReqNsd, netstrm_t **ppNew)
{
- netstrm_t *pThis = NULL;
- nsd_t *pNsd;
+ netstrm_t *pNew = NULL;
+ nsd_t *pNewNsd = NULL;
DEFiRet;
- assert(ppThis != NULL);
+ ISOBJ_TYPE_assert(pThis, netstrm);
+ assert(pReqNsd != NULL);
+ assert(ppNew != NULL);
+
+ /* accept the new connection */
+ CHKiRet(pThis->Drvr.AcceptConnReq(pReqNsd, &pNewNsd));
/* construct our object so that we can use it... */
- CHKiRet(netstrmConstruct(&pThis));
-
- /* TODO: obtain hostname, normalize (callback?), save it */
- CHKiRet(FillRemHost(pThis, (struct sockaddr*) &addr));
-
- /* set the new socket to non-blocking IO */
- if((sockflags = fcntl(iNewSock, F_GETFL)) != -1) {
- sockflags |= O_NONBLOCK;
- /* SETFL could fail too, so get it caught by the subsequent
- * error check.
- */
- sockflags = fcntl(iNewSock, F_SETFL, sockflags);
- }
- if(sockflags == -1) {
- dbgprintf("error %d setting fcntl(O_NONBLOCK) on tcp socket %d", errno, iNewSock);
- ABORT_FINALIZE(RS_RET_IO_ERROR);
- }
+ CHKiRet(netstrmConstruct(&pNew));
- pThis->sock = iNewSock;
+ pNew->pDrvrData = pNewNsd;
+ if(pThis->pDrvrName == NULL) {
+ pNew->pDrvrName = NULL;
+ } else {
+ CHKmalloc(pNew->pDrvrName = (uchar*) strdup((char*)pThis->pDrvrName));
+ }
+ CHKiRet(loadDrvr(pNew));
- *ppThis = pThis;
+ *ppNew = pNew;
finalize_it:
if(iRet != RS_RET_OK) {
- if(pThis != NULL)
- netstrmDestruct(&pThis);
+ if(pNew != NULL)
+ netstrmDestruct(&pNew);
/* the close may be redundant, but that doesn't hurt... */
- if(iNewSock >= 0)
- close(iNewSock);
+ if(pNewNsd != NULL)
+ pThis->Drvr.Destruct(&pNewNsd);
}
-
RETiRet;
}
-#endif
-/* initialize the tcp socket for a listner
- * pLstnPort must point to a port name or number. NULL is NOT permitted
- * (hint: we need to be careful when we use this module together with librelp,
- * there NULL indicates the default port
- * default is used.
- * gerhards, 2008-03-17
+/* make the netstrm listen to specified port and IP.
+ * pLstnIP points to the port to listen to (NULL means "all"),
+ * iMaxSess has the maximum number of sessions permitted (this ist just a hint).
+ * pLstnPort must point to a port name or number. NULL is NOT permitted.
+ * rgerhards, 2008-04-22
*/
static rsRetVal
-LstnInit(netstrm_t *pThis, uchar *pLstnPort)
+LstnInit(netstrm_t *pThis, uchar *pLstnPort, uchar *pLstnIP, int iSessMax)
{
DEFiRet;
ISOBJ_TYPE_assert(pThis, netstrm);
assert(pLstnPort != NULL);
- CHKiRet(pThis->Drvr.LstnInit(pThis->pDrvrData, pLstnPort));
+ CHKiRet(pThis->Drvr.LstnInit(&pThis->parrLstn, &pThis->isizeLstnArr, pLstnPort, pLstnIP, iSessMax));
finalize_it:
RETiRet;
@@ -284,11 +276,11 @@ CODESTARTobjQueryInterface(netstrm)
pIf->ConstructFinalize = netstrmConstructFinalize;
pIf->Destruct = netstrmDestruct;
pIf->AbortDestruct = AbortDestruct;
- pIf->LstnInit = LstnInit;
- // TODO: add later: pIf->AcceptConnReq = AcceptConnReq;
pIf->Rcv = Rcv;
pIf->Send = Send;
pIf->Connect = Connect;
+ pIf->LstnInit = LstnInit;
+ pIf->AcceptConnReq = AcceptConnReq;
finalize_it:
ENDobjQueryInterface(netstrm)
diff --git a/runtime/netstrm.h b/runtime/netstrm.h
index 75b7c457..a3719f93 100644
--- a/runtime/netstrm.h
+++ b/runtime/netstrm.h
@@ -29,11 +29,34 @@
/* the netstrm object */
struct netstrm_s {
BEGINobjInstance; /* Data to implement generic object - MUST be the first data element! */
- nsd_if_t Drvr; /**< our stream driver */
nsd_t *pDrvrData; /**< the driver's data elements */
uchar *pDrvrName; /**< nsd driver name to use, or NULL if system default */
+ nsd_if_t Drvr; /**< our stream driver */
+ /* for listeners, we need to have the capablity to listen on multiple "sockets". This
+ * is needed to support IPv6. We do this by specifying an array of nsd_t objects to
+ * handle this case.
+ */
+ int isizeLstnArr;
+ nsd_t **parrLstn;
};
+/* a helper object enabling us to wait on a set of streams to become
+ * ready for IO - this is modelled after select(). We need this, because
+ * stream drivers may have different concepts. Consequently,
+ * the structure must contain nsd_t's from the same stream driver type
+ * only. This is implemented as a singly-linked list where every
+ * new element is added at the top of the list. -- rgerhards, 2008-04-22
+ */
+typedef struct netstrm_iowaiter_s netstrm_iowaiter_t;
+struct netstrm_iowaiter_s {
+ netstrm_iowaiter_t *pNext;
+ nsd_t *pNsd;
+ enum {
+ NETSTRM_IOWAIT_RD = 1,
+ NETSTRM_IOWAIT_WR = 2,
+ NETSTRM_IOWAIT_RDWR = 3
+ } waitOp; /**< the operation we wait for */
+};
/* interface */
BEGINinterface(netstrm) /* name must also be changed in ENDinterface macro! */
@@ -41,11 +64,16 @@ BEGINinterface(netstrm) /* name must also be changed in ENDinterface macro! */
rsRetVal (*ConstructFinalize)(netstrm_t *pThis);
rsRetVal (*Destruct)(netstrm_t **ppThis);
rsRetVal (*AbortDestruct)(netstrm_t **ppThis);
- rsRetVal (*LstnInit)(netstrm_t *pThis, unsigned char *pLstnPort);
- rsRetVal (*AcceptConnReq)(netstrm_t **ppThis, int sock);
+ rsRetVal (*LstnInit)(netstrm_t *pThis, uchar *pLstnPort, uchar *pLstnIP, int iSessMax);
+ rsRetVal (*AcceptConnReq)(netstrm_t *pThis, nsd_t *pReqNsd, netstrm_t **ppNew);
rsRetVal (*Rcv)(netstrm_t *pThis, uchar *pRcvBuf, ssize_t *pLenBuf);
rsRetVal (*Send)(netstrm_t *pThis, uchar *pBuf, ssize_t *pLenBuf);
rsRetVal (*Connect)(netstrm_t *pThis, int family, unsigned char *port, unsigned char *host);
+ rsRetVal (*SelectInit)(nsdsel_t **ppSel, netstrm_t *pThis);
+ rsRetVal (*SelectAdd)(nsdsel_t *pSel, netstrm_t *pThis);
+ rsRetVal (*SelectWait)(nsdsel_t *pSel, int *piNumReady);
+ rsRetVal (*SelectIsReady)(nsdsel_t *pSel, int *piNumReady);
+ rsRetVal (*SelectExit)(nsdsel_t **ppSel);
ENDinterface(netstrm)
#define netstrmCURR_IF_VERSION 1 /* increment whenever you change the interface structure! */
diff --git a/runtime/nsd.h b/runtime/nsd.h
index 52c36dcf..2fb883ac 100644
--- a/runtime/nsd.h
+++ b/runtime/nsd.h
@@ -28,6 +28,12 @@
#ifndef INCLUDED_NSD_H
#define INCLUDED_NSD_H
+enum nsdsel_waitOp_e {
+ NSDSEL_RD = 1,
+ NSDSEL_WR = 2,
+ NSDSEL_RDWR = 3
+}; /**< the operation we wait for */
+
/* nsd_t is actually obj_t (which is somewhat better than void* but in essence
* much the same).
*/
@@ -37,12 +43,22 @@ BEGINinterface(nsd) /* name must also be changed in ENDinterface macro! */
rsRetVal (*Construct)(nsd_t **ppThis);
rsRetVal (*Destruct)(nsd_t **ppThis);
rsRetVal (*Abort)(nsd_t *pThis);
- rsRetVal (*LstnInit)(nsd_t *pThis, unsigned char *pLstnPort);
- rsRetVal (*AcceptConnReq)(nsd_t **ppThis, int sock);
rsRetVal (*Rcv)(nsd_t *pThis, uchar *pRcvBuf, ssize_t *pLenBuf);
rsRetVal (*Send)(nsd_t *pThis, uchar *pBuf, ssize_t *pLenBuf);
rsRetVal (*Connect)(nsd_t *pThis, int family, unsigned char *port, unsigned char *host);
+ rsRetVal (*LstnInit)(nsd_t ***parrLstn, int *pLstnArrSize, uchar *pLstnPort, uchar *pLstnIP, int iSessMax);
+ rsRetVal (*AcceptConnReq)(nsd_t *pThis, nsd_t **ppThis);
ENDinterface(nsd)
#define nsdCURR_IF_VERSION 1 /* increment whenever you change the interface structure! */
+/* interface for the select call */
+BEGINinterface(nsdsel) /* name must also be changed in ENDinterface macro! */
+ rsRetVal (*Construct)(nsdsel_t **ppThis);
+ rsRetVal (*Destruct)(nsdsel_t **ppThis);
+ rsRetVal (*Add)(nsdsel_t *pNsdsel, nsd_t *pNsd, nsdsel_waitOp_t waitOp);
+ rsRetVal (*Select)(nsdsel_t *pNsdsel, int *piNumReady);
+ rsRetVal (*IsReady)(nsdsel_t *pNsdsel, nsd_t *pNsd, nsdsel_waitOp_t waitOp);
+ENDinterface(nsdsel)
+#define nsdselCURR_IF_VERSION 1 /* increment whenever you change the interface structure! */
+
#endif /* #ifndef INCLUDED_NSD_H */
diff --git a/runtime/nsd_ptcp.c b/runtime/nsd_ptcp.c
index 6f7dd04d..c737c168 100644
--- a/runtime/nsd_ptcp.c
+++ b/runtime/nsd_ptcp.c
@@ -56,30 +56,29 @@ DEFobjCurrIf(glbl)
DEFobjCurrIf(net)
+/* a few deinit helpers */
+
+/* close socket if open (may always be called) */
+static void
+sockClose(int *pSock)
+{
+ if(*pSock >= 0) {
+ close(*pSock);
+ *pSock = -1;
+ }
+}
+
/* Standard-Constructor
*/
BEGINobjConstruct(nsd_ptcp) /* be sure to specify the object type also in END macro! */
pThis->sock = -1;
- pThis->iSessMax = 500; /* default max nbr of sessions -TODO:make configurable--rgerhards, 2008-04-17*/
ENDobjConstruct(nsd_ptcp)
/* destructor for the nsd_ptcp object */
BEGINobjDestruct(nsd_ptcp) /* be sure to specify the object type also in END and CODESTART macros! */
- int i;
CODESTARTobjDestruct(nsd_ptcp)
- if(pThis->sock != -1) {
- close(pThis->sock);
- pThis->sock = -1;
- }
-
- if(pThis->socks != NULL) {
- /* if we have some sockets at this stage, we need to close them */
- for(i = 1 ; i <= pThis->socks[0] ; ++i)
- close(pThis->socks[i]);
- free(pThis->socks);
- }
-
+ sockClose(&pThis->sock);
if(pThis->pRemHostIP != NULL)
free(pThis->pRemHostIP);
if(pThis->pRemHostName != NULL)
@@ -189,33 +188,34 @@ finalize_it:
-/* accept an incoming connection request, sock provides the socket on which we can
- * accept the new session.
- * rgerhards, 2008-03-17
+/* accept an incoming connection request
+ * rgerhards, 2008-04-22
*/
static rsRetVal
-AcceptConnReq(nsd_t **ppThis, int sock)
+AcceptConnReq(nsd_t *pNsd, nsd_t **ppNew)
{
int sockflags;
+ nsd_ptcp_t *pThis = (nsd_ptcp_t*) pNsd;
struct sockaddr_storage addr;
socklen_t addrlen = sizeof(addr);
- nsd_ptcp_t *pThis = NULL;
+ nsd_ptcp_t *pNew = NULL;
int iNewSock = -1;
DEFiRet;
- assert(ppThis != NULL);
+ assert(ppNew != NULL);
+ ISOBJ_TYPE_assert(pThis, nsd_ptcp_t);
- iNewSock = accept(sock, (struct sockaddr*) &addr, &addrlen);
+ iNewSock = accept(pThis->sock, (struct sockaddr*) &addr, &addrlen);
if(iNewSock < 0) {
ABORT_FINALIZE(RS_RET_ACCEPT_ERR);
}
/* construct our object so that we can use it... */
- CHKiRet(nsd_ptcpConstruct(&pThis));
+ CHKiRet(nsd_ptcpConstruct(&pNew));
- CHKiRet(FillRemHost(pThis, (struct sockaddr*) &addr));
+ CHKiRet(FillRemHost(pNew, (struct sockaddr*) &addr));
- /* set the new socket to non-blocking IO */
+ /* set the new socket to non-blocking IO -TODO:do we really need to do this here? Do we always want it? */
if((sockflags = fcntl(iNewSock, F_GETFL)) != -1) {
sockflags |= O_NONBLOCK;
/* SETFL could fail too, so get it caught by the subsequent
@@ -228,41 +228,41 @@ AcceptConnReq(nsd_t **ppThis, int sock)
ABORT_FINALIZE(RS_RET_IO_ERROR);
}
- pThis->sock = iNewSock;
-
- *ppThis = (nsd_t*) pThis;
+ pNew->sock = iNewSock;
+ *ppNew = (nsd_t*) pNew;
finalize_it:
if(iRet != RS_RET_OK) {
- if(pThis != NULL)
- nsd_ptcpDestruct(&pThis);
+ if(pNew != NULL)
+ nsd_ptcpDestruct(&pNew);
/* the close may be redundant, but that doesn't hurt... */
- if(iNewSock >= 0)
- close(iNewSock);
+ sockClose(&iNewSock);
}
RETiRet;
}
-/* initialize the tcp socket for a listner
- * pLstnPort must point to a port name or number. NULL is NOT permitted
- * (hint: we need to be careful when we use this module together with librelp,
- * there NULL indicates the default port
- * default is used.
- * gerhards, 2008-03-17
+/* initialize tcp sockets for a listner. This function returns an array of nds_t
+ * objects. The size of this array is returend in pLstnArrSize.
+ * pLstnPort must point to a port name or number. NULL is NOT permitted. pLstnIP
+ * points to the port to listen to (NULL means "all"), iMaxSess has the maximum
+ * number of sessions permitted.
+ * rgerhards, 2008-04-22
*/
static rsRetVal
-LstnInit(nsd_t *pNsd, uchar *pLstnPort)
+LstnInit(nsd_t ***parrLstnNsd, int *pLstnArrSize, uchar *pLstnPort, uchar *pLstnIP, int iSessMax)
{
- nsd_ptcp_t *pThis = (nsd_ptcp_t*) pNsd;
- struct addrinfo hints, *res, *r;
- int error, maxs, *s, on = 1;
+ DEFiRet;
+ struct addrinfo hints, *res = NULL, *r;
+ nsd_ptcp_t **arrLstn = NULL;
+ int error, maxs, on = 1;
+ int sock;
int sockflags;
- DEFiRet;
- ISOBJ_TYPE_assert(pThis, nsd_ptcp);
+ assert(parrLstnNsd != NULL);
assert(pLstnPort != NULL);
+ assert(iSessMax >= 0);
dbgprintf("creating tcp listen socket on port %s\n", pLstnPort);
@@ -271,7 +271,7 @@ LstnInit(nsd_t *pNsd, uchar *pLstnPort)
hints.ai_family = glbl.GetDefPFFamily();
hints.ai_socktype = SOCK_STREAM;
- error = getaddrinfo(NULL, (char*) pLstnPort, &hints, &res);
+ error = getaddrinfo((char*)pLstnIP, (char*) pLstnPort, &hints, &res);
if(error) {
dbgprintf("error %d querying port '%s'\n", error, pLstnPort);
ABORT_FINALIZE(RS_RET_INVALID_PORT);
@@ -280,20 +280,14 @@ LstnInit(nsd_t *pNsd, uchar *pLstnPort)
/* Count max number of sockets we may open */
for(maxs = 0, r = res; r != NULL ; r = r->ai_next, maxs++)
/* EMPTY */;
- pThis->socks = malloc((maxs+1) * sizeof(int));
- if (pThis->socks == NULL) {
- dbgprintf("couldn't allocate memory for TCP listen sockets, suspending RELP message reception.");
- freeaddrinfo(res);
- ABORT_FINALIZE(RS_RET_OUT_OF_MEMORY);
- }
+ CHKmalloc(arrLstn = (nsd_ptcp_t**) malloc((maxs+1) * sizeof(nsd_ptcp_t*)));
- *pThis->socks = 0; /* num of sockets counter at start of array */
- s = pThis->socks + 1;
+ *pLstnArrSize = 0; /* num of sockets counter at start of array */
for(r = res; r != NULL ; r = r->ai_next) {
- *s = socket(r->ai_family, r->ai_socktype, r->ai_protocol);
- if (*s < 0) {
+ sock = socket(r->ai_family, r->ai_socktype, r->ai_protocol);
+ if(sock < 0) {
if(!(r->ai_family == PF_INET6 && errno == EAFNOSUPPORT))
- dbgprintf("creating tcp listen socket");
+ dbgprintf("error %d creating tcp listen socket", errno);
/* it is debatable if PF_INET with EAFNOSUPPORT should
* also be ignored...
*/
@@ -301,35 +295,32 @@ LstnInit(nsd_t *pNsd, uchar *pLstnPort)
}
#ifdef IPV6_V6ONLY
- if (r->ai_family == AF_INET6) {
+ if(r->ai_family == AF_INET6) {
int iOn = 1;
- if (setsockopt(*s, IPPROTO_IPV6, IPV6_V6ONLY,
+ if(setsockopt(sock, IPPROTO_IPV6, IPV6_V6ONLY,
(char *)&iOn, sizeof (iOn)) < 0) {
- close(*s);
- *s = -1;
- continue;
+ close(sock);
+ continue;
}
}
#endif
- if(setsockopt(*s, SOL_SOCKET, SO_REUSEADDR, (char *) &on, sizeof(on)) < 0 ) {
+ if(setsockopt(sock, SOL_SOCKET, SO_REUSEADDR, (char *) &on, sizeof(on)) < 0 ) {
dbgprintf("error %d setting tcp socket option\n", errno);
- close(*s);
- *s = -1;
+ close(sock);
continue;
}
/* We use non-blocking IO! */
- if((sockflags = fcntl(*s, F_GETFL)) != -1) {
+ if((sockflags = fcntl(sock, F_GETFL)) != -1) {
sockflags |= O_NONBLOCK;
/* SETFL could fail too, so get it caught by the subsequent
* error check.
*/
- sockflags = fcntl(*s, F_SETFL, sockflags);
+ sockflags = fcntl(sock, F_SETFL, sockflags);
}
if(sockflags == -1) {
dbgprintf("error %d setting fcntl(O_NONBLOCK) on tcp socket", errno);
- close(*s);
- *s = -1;
+ close(sock);
continue;
}
@@ -340,62 +331,75 @@ LstnInit(nsd_t *pNsd, uchar *pLstnPort)
*/
#ifndef BSD
if(net.should_use_so_bsdcompat()) {
- if (setsockopt(*s, SOL_SOCKET, SO_BSDCOMPAT,
+ if (setsockopt(sock, SOL_SOCKET, SO_BSDCOMPAT,
(char *) &on, sizeof(on)) < 0) {
errmsg.LogError(NO_ERRCODE, "TCP setsockopt(BSDCOMPAT)");
- close(*s);
- *s = -1;
+ close(sock);
continue;
}
}
#endif
- if( (bind(*s, r->ai_addr, r->ai_addrlen) < 0)
+ if( (bind(sock, r->ai_addr, r->ai_addrlen) < 0)
#ifndef IPV6_V6ONLY
&& (errno != EADDRINUSE)
#endif
) {
+ /* TODO: check if *we* bound the socket - else we *have* an error! */
dbgprintf("error %d while binding tcp socket", errno);
- close(*s);
- *s = -1;
+ close(sock);
continue;
}
- if(listen(*s,pThis->iSessMax / 10 + 5) < 0) {
+ if(listen(sock, iSessMax / 10 + 5) < 0) {
/* If the listen fails, it most probably fails because we ask
* for a too-large backlog. So in this case we first set back
* to a fixed, reasonable, limit that should work. Only if
* that fails, too, we give up.
*/
dbgprintf("listen with a backlog of %d failed - retrying with default of 32.",
- pThis->iSessMax / 10 + 5);
- if(listen(*s, 32) < 0) {
+ iSessMax / 10 + 5);
+ if(listen(sock, 32) < 0) {
dbgprintf("tcp listen error %d, suspending\n", errno);
- close(*s);
- *s = -1;
+ close(sock);
continue;
}
}
- (*pThis->socks)++;
- s++;
+ /* if we reach this point, we were able to obtain a valid socket, which we
+ * now can save to the array of listen sockets. -- rgerhards, 2008-04-22
+ */
+ CHKiRet(nsd_ptcpConstruct(arrLstn+*pLstnArrSize));
+ arrLstn[*pLstnArrSize]->sock = sock;
+ ++(*pLstnArrSize);
}
if(res != NULL)
freeaddrinfo(res);
- if(*pThis->socks != maxs)
- dbgprintf("We could initialize %d RELP TCP listen sockets out of %d we received "
- "- this may or may not be an error indication.\n", *pThis->socks, maxs);
+ if(*pLstnArrSize != maxs)
+ dbgprintf("We could initialize %d TCP listen sockets out of %d we received "
+ "- this may or may not be an error indication.\n", *pLstnArrSize, maxs);
- if(*pThis->socks == 0) {
- dbgprintf("No RELP TCP listen socket could successfully be initialized, "
- "message reception via RELP disabled.\n");
- free(pThis->socks);
+ if(*pLstnArrSize == 0) {
+ dbgprintf("No TCP listen sockets could successfully be initialized, "
+ "message reception disabled.\n");
ABORT_FINALIZE(RS_RET_COULD_NOT_BIND);
}
+ *parrLstnNsd = (nsd_t**) arrLstn;
+ arrLstn = NULL; /* prevent from being freed in error handler */
+
finalize_it:
+ if(iRet != RS_RET_OK) {
+ if(res != NULL)
+ freeaddrinfo(res);
+ if(arrLstn != NULL) {
+ for(maxs = 0 ; maxs < *pLstnArrSize ; ++maxs)
+ nsd_ptcpDestruct(arrLstn+*pLstnArrSize);
+ }
+ }
+
RETiRet;
}
@@ -494,10 +498,7 @@ finalize_it:
freeaddrinfo(res);
if(iRet != RS_RET_OK) {
- if(pThis->sock != -1) {
- close(pThis->sock);
- pThis->sock = -1;
- }
+ sockClose(&pThis->sock);
}
RETiRet;
@@ -519,10 +520,10 @@ CODESTARTobjQueryInterface(nsd_ptcp)
pIf->Construct = (rsRetVal(*)(nsd_t**)) nsd_ptcpConstruct;
pIf->Destruct = (rsRetVal(*)(nsd_t**)) nsd_ptcpDestruct;
pIf->Abort = Abort;
- pIf->LstnInit = LstnInit;
- pIf->AcceptConnReq = AcceptConnReq;
pIf->Rcv = Rcv;
pIf->Send = Send;
+ pIf->LstnInit = LstnInit;
+ pIf->AcceptConnReq = AcceptConnReq;
pIf->Connect = Connect;
finalize_it:
ENDobjQueryInterface(nsd_ptcp)
diff --git a/runtime/nsd_ptcp.h b/runtime/nsd_ptcp.h
index d4848314..ac11d528 100644
--- a/runtime/nsd_ptcp.h
+++ b/runtime/nsd_ptcp.h
@@ -33,8 +33,6 @@ struct nsd_ptcp_s {
uchar *pRemHostIP; /**< IP address of remote peer (currently used in server mode, only) */
uchar *pRemHostName; /**< host name of remote peer (currently used in server mode, only) */
int sock; /**< the socket we use for regular, single-socket, operations */
- int *socks; /**< the socket(s) we use for listeners, element 0 has nbr of socks */
- int iSessMax; /**< maximum number of sessions permitted */
};
/* interface is defined in nsd.h, we just implement it! */
diff --git a/runtime/nsdsel_ptcp.c b/runtime/nsdsel_ptcp.c
new file mode 100644
index 00000000..67f3c62a
--- /dev/null
+++ b/runtime/nsdsel_ptcp.c
@@ -0,0 +1,216 @@
+/* nsdsel_ptcp.c
+ *
+ * An implementation of the nsd select() interface for plain tcp sockets.
+ *
+ * Copyright 2008 Rainer Gerhards and Adiscon GmbH.
+ *
+ * This file is part of the rsyslog runtime library.
+ *
+ * The rsyslog runtime library is free software: you can redistribute it and/or modify
+ * it under the terms of the GNU Lesser General Public License as published by
+ * the Free Software Foundation, either version 3 of the License, or
+ * (at your option) any later version.
+ *
+ * The rsyslog runtime library is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+ * GNU Lesser General Public License for more details.
+ *
+ * You should have received a copy of the GNU Lesser General Public License
+ * along with the rsyslog runtime library. If not, see <http://www.gnu.org/licenses/>.
+ *
+ * A copy of the GPL can be found in the file "COPYING" in this distribution.
+ * A copy of the LGPL can be found in the file "COPYING.LESSER" in this distribution.
+ */
+#include "config.h"
+
+#include "rsyslog.h"
+#include <stdio.h>
+#include <stdarg.h>
+#include <stdlib.h>
+#include <assert.h>
+#include <errno.h>
+#include <string.h>
+#include <signal.h>
+#include <ctype.h>
+#include <netdb.h>
+#include <fnmatch.h>
+#include <fcntl.h>
+#include <unistd.h>
+#include <sys/select.h>
+
+#include "rsyslog.h"
+#include "module-template.h"
+#include "obj.h"
+#include "errmsg.h"
+#include "nsd_ptcp.h"
+#include "nsdsel_ptcp.h"
+
+MODULE_TYPE_LIB
+
+/* static data */
+DEFobjStaticHelpers
+DEFobjCurrIf(errmsg)
+DEFobjCurrIf(glbl)
+
+
+/* Standard-Constructor
+ */
+BEGINobjConstruct(nsdsel_ptcp) /* be sure to specify the object type also in END macro! */
+ pThis->maxfds = 0;
+ FD_ZERO(&pThis->readfds);
+ FD_ZERO(&pThis->writefds);
+ENDobjConstruct(nsdsel_ptcp)
+
+
+/* destructor for the nsdsel_ptcp object */
+BEGINobjDestruct(nsdsel_ptcp) /* be sure to specify the object type also in END and CODESTART macros! */
+CODESTARTobjDestruct(nsdsel_ptcp)
+ENDobjDestruct(nsdsel_ptcp)
+
+
+/* Add a socket to the select set */
+static rsRetVal
+Add(nsdsel_t *pNsdsel, nsd_t *pNsd, nsdsel_waitOp_t waitOp)
+{
+ DEFiRet;
+ nsdsel_ptcp_t *pThis = (nsdsel_ptcp_t*) pNsdsel;
+ nsd_ptcp_t *pSock = (nsd_ptcp_t*) pNsd;
+
+ ISOBJ_TYPE_assert(pSock, nsd_ptcp);
+ ISOBJ_TYPE_assert(pThis, nsdsel_ptcp);
+
+ switch(waitOp) {
+ case NSDSEL_RD:
+ FD_SET(pSock->sock, &pThis->readfds);
+ break;
+ case NSDSEL_WR:
+ FD_SET(pSock->sock, &pThis->writefds);
+ break;
+ case NSDSEL_RDWR:
+ FD_SET(pSock->sock, &pThis->readfds);
+ FD_SET(pSock->sock, &pThis->writefds);
+ break;
+ }
+
+ if(pSock->sock > pThis->maxfds)
+ pThis->maxfds = pSock->sock;
+
+ RETiRet;
+}
+
+
+/* perform the select() piNumReady returns how many descriptors are ready for IO
+ * TODO: add timeout!
+ */
+static rsRetVal
+Select(nsdsel_t *pNsdsel, int *piNumReady)
+{
+ DEFiRet;
+ int i;
+ nsdsel_ptcp_t *pThis = (nsdsel_ptcp_t*) pNsdsel;
+
+ ISOBJ_TYPE_assert(pThis, nsdsel_ptcp);
+ assert(piNumReady != NULL);
+
+ if(Debug) { // TODO: debug setting!
+ // TODO: name in dbgprintf!
+ dbgprintf("--------<NSDSEL_PTCP> calling select, active fds (max %d): ", pThis->maxfds);
+ for(i = 0; i <= pThis->maxfds; ++i)
+ if(FD_ISSET(i, &pThis->readfds) || FD_ISSET(i, &pThis->writefds))
+ dbgprintf("%d ", i);
+ dbgprintf("\n");
+ }
+
+ /* now do the select */
+ *piNumReady = select(pThis->maxfds+1, &pThis->readfds, &pThis->writefds, NULL, NULL);
+
+ RETiRet;
+}
+
+
+/* check if a socket is ready for IO */
+static rsRetVal
+IsReady(nsdsel_t *pNsdsel, nsd_t *pNsd, nsdsel_waitOp_t waitOp)
+{
+ DEFiRet;
+ nsdsel_ptcp_t *pThis = (nsdsel_ptcp_t*) pNsdsel;
+ nsd_ptcp_t *pSock = (nsd_ptcp_t*) pNsd;
+
+ ISOBJ_TYPE_assert(pThis, nsdsel_ptcp);
+ ISOBJ_TYPE_assert(pSock, nsd_ptcp);
+ RETiRet;
+}
+
+
+/* ------------------------------ end support for the select() interface ------------------------------ */
+
+
+/* queryInterface function */
+BEGINobjQueryInterface(nsdsel_ptcp)
+CODESTARTobjQueryInterface(nsdsel_ptcp)
+ if(pIf->ifVersion != nsdCURR_IF_VERSION) {/* check for current version, increment on each change */
+ ABORT_FINALIZE(RS_RET_INTERFACE_NOT_SUPPORTED);
+ }
+
+ /* ok, we have the right interface, so let's fill it
+ * Please note that we may also do some backwards-compatibility
+ * work here (if we can support an older interface version - that,
+ * of course, also affects the "if" above).
+ */
+ pIf->Construct = (rsRetVal(*)(nsdsel_t**)) nsdsel_ptcpConstruct;
+ pIf->Destruct = (rsRetVal(*)(nsdsel_t**)) nsdsel_ptcpDestruct;
+ pIf->Add = Add;
+ pIf->Select = Select;
+ pIf->IsReady = IsReady;
+finalize_it:
+ENDobjQueryInterface(nsdsel_ptcp)
+
+
+/* exit our class
+ */
+BEGINObjClassExit(nsdsel_ptcp, OBJ_IS_LOADABLE_MODULE) /* CHANGE class also in END MACRO! */
+CODESTARTObjClassExit(nsdsel_ptcp)
+ /* release objects we no longer need */
+ objRelease(glbl, CORE_COMPONENT);
+ objRelease(errmsg, CORE_COMPONENT);
+ENDObjClassExit(nsdsel_ptcp)
+
+
+/* Initialize the nsdsel_ptcp class. Must be called as the very first method
+ * before anything else is called inside this class.
+ * rgerhards, 2008-02-19
+ */
+BEGINObjClassInit(nsdsel_ptcp, 1, OBJ_IS_LOADABLE_MODULE) /* class, version */
+ /* request objects we use */
+ CHKiRet(objUse(errmsg, CORE_COMPONENT));
+ CHKiRet(objUse(glbl, CORE_COMPONENT));
+
+ /* set our own handlers */
+ENDObjClassInit(nsdsel_ptcp)
+
+
+/* --------------- here now comes the plumbing that makes as a library module --------------- */
+
+
+BEGINmodExit
+CODESTARTmodExit
+ nsdsel_ptcpClassExit();
+ENDmodExit
+
+
+BEGINqueryEtryPt
+CODESTARTqueryEtryPt
+CODEqueryEtryPt_STD_LIB_QUERIES
+ENDqueryEtryPt
+
+
+BEGINmodInit()
+CODESTARTmodInit
+ *ipIFVersProvided = CURR_MOD_IF_VERSION; /* we only support the current interface specification */
+
+ /* Initialize all classes that are in our module - this includes ourselfs */
+ CHKiRet(nsdsel_ptcpClassInit(pModInfo)); /* must be done after tcps_sess, as we use it */
+ENDmodInit
+/* vi:set ai:
+ */
diff --git a/runtime/nsdsel_ptcp.h b/runtime/nsdsel_ptcp.h
new file mode 100644
index 00000000..39294d7d
--- /dev/null
+++ b/runtime/nsdsel_ptcp.h
@@ -0,0 +1,46 @@
+/* An implementation of the nsd select interface for plain tcp sockets.
+ *
+ * Copyright 2008 Rainer Gerhards and Adiscon GmbH.
+ *
+ * This file is part of the rsyslog runtime library.
+ *
+ * The rsyslog runtime library is free software: you can redistribute it and/or modify
+ * it under the terms of the GNU Lesser General Public License as published by
+ * the Free Software Foundation, either version 3 of the License, or
+ * (at your option) any later version.
+ *
+ * The rsyslog runtime library is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+ * GNU Lesser General Public License for more details.
+ *
+ * You should have received a copy of the GNU Lesser General Public License
+ * along with the rsyslog runtime library. If not, see <http://www.gnu.org/licenses/>.
+ *
+ * A copy of the GPL can be found in the file "COPYING" in this distribution.
+ * A copy of the LGPL can be found in the file "COPYING.LESSER" in this distribution.
+ */
+
+#ifndef INCLUDED_NSDSEL_PTCP_H
+#define INCLUDED_NSDSEL_PTCP_H
+
+#include "nsd.h"
+typedef nsdsel_if_t nsdsel_ptcp_if_t; /* we just *implement* this interface */
+
+/* the nsdsel_ptcp object */
+struct nsdsel_ptcp_s {
+ BEGINobjInstance; /* Data to implement generic object - MUST be the first data element! */
+ int maxfds;
+ fd_set readfds;
+ fd_set writefds;
+};
+
+/* interface is defined in nsd.h, we just implement it! */
+
+/* prototypes */
+PROTOTYPEObj(nsdsel_ptcp);
+
+/* the name of our library binary */
+#define LM_NSDSEL_PTCP_FILENAME "lmnsdsel_ptcp"
+
+#endif /* #ifndef INCLUDED_NSDSEL_PTCP_H */
diff --git a/runtime/nssel.c b/runtime/nssel.c
new file mode 100644
index 00000000..f2844872
--- /dev/null
+++ b/runtime/nssel.c
@@ -0,0 +1,222 @@
+/* nssel.c
+ *
+ * The io waiter is a helper object enabling us to wait on a set of streams to become
+ * ready for IO - this is modelled after select(). We need this, because
+ * stream drivers may have different concepts. Consequently,
+ * the structure must contain nsd_t's from the same stream driver type
+ * only. This is implemented as a singly-linked list where every
+ * new element is added at the top of the list.
+ *
+ * Work on this module begun 2008-04-22 by Rainer Gerhards.
+ *
+ * Copyright 2008 Rainer Gerhards and Adiscon GmbH.
+ *
+ * This file is part of the rsyslog runtime library.
+ *
+ * The rsyslog runtime library is free software: you can redistribute it and/or modify
+ * it under the terms of the GNU Lesser General Public License as published by
+ * the Free Software Foundation, either version 3 of the License, or
+ * (at your option) any later version.
+ *
+ * The rsyslog runtime library is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+ * GNU Lesser General Public License for more details.
+ *
+ * You should have received a copy of the GNU Lesser General Public License
+ * along with the rsyslog runtime library. If not, see <http://www.gnu.org/licenses/>.
+ *
+ * A copy of the GPL can be found in the file "COPYING" in this distribution.
+ * A copy of the LGPL can be found in the file "COPYING.LESSER" in this distribution.
+ */
+#include "config.h"
+
+#include "rsyslog.h"
+#include <stdio.h>
+#include <stdlib.h>
+#include <assert.h>
+#include <errno.h>
+#include <string.h>
+
+#include "rsyslog.h"
+#include "obj.h"
+#include "module-template.h"
+#include "nssel.h"
+
+MODULE_TYPE_LIB
+
+/* static data */
+DEFobjStaticHelpers
+DEFobjCurrIf(glbl)
+
+
+/* load our low-level driver. This must be done before any
+ * driver-specific functions (allmost all...) can be carried
+ * out. Note that the driver's .ifIsLoaded is correctly
+ * initialized by calloc() and we depend on that.
+ * rgerhards, 2008-04-18
+ */
+static rsRetVal
+loadDrvr(nssel_t *pThis)
+{
+ uchar *pDrvrName;
+ DEFiRet;
+
+ pDrvrName = pThis->pDrvrName;
+ if(pDrvrName == NULL) /* if no drvr name is set, use system default */
+ pDrvrName = glbl.GetDfltNetstrmDrvr();
+
+ pThis->Drvr.ifVersion = nsdCURR_IF_VERSION;
+ /* The pDrvrName+2 below is a hack to obtain the object name. It
+ * safes us to have yet another variable with the name without "lm" in
+ * front of it. If we change the module load interface, we may re-think
+ * about this hack, but for the time being it is efficient and clean
+ * enough. -- rgerhards, 2008-04-18
+ */
+ CHKiRet(obj.UseObj(__FILE__, pDrvrName+2, pDrvrName, (void*) &pThis->Drvr));
+finalize_it:
+ RETiRet;
+}
+
+
+/* Standard-Constructor */
+BEGINobjConstruct(nssel) /* be sure to specify the object type also in END macro! */
+ENDobjConstruct(nssel)
+
+
+/* destructor for the nssel object */
+BEGINobjDestruct(nssel) /* be sure to specify the object type also in END and CODESTART macros! */
+CODESTARTobjDestruct(nssel)
+ENDobjDestruct(nssel)
+
+
+/* ConstructionFinalizer */
+static rsRetVal
+ConstructFinalize(nssel_t *pThis)
+{
+ DEFiRet;
+ ISOBJ_TYPE_assert(pThis, nssel);
+ CHKiRet(loadDrvr(pThis));
+ CHKiRet(pThis->Drvr.Construct(&pThis->pDrvrData));
+finalize_it:
+ RETiRet;
+}
+
+
+/* Add a stream object to the current IOW. Note that a single stream may
+ * have multiple "sockets" if it is a listener. If so, all of them are
+ * begin added.
+ */
+static rsRetVal
+Add(nssel_t *pThis, netstrm_t *pStrm)
+{
+ DEFiRet;
+
+ ISOBJ_TYPE_assert(pThis, nssel);
+ ISOBJ_TYPE_assert(pStrm, netstrm);
+
+
+finalize_it:
+ RETiRet;
+}
+
+
+/* wait for IO to happen on one of our netstreams. iNumReady has
+ * the number of ready "sockets" after the call. This function blocks
+ * until some are ready. EAGAIN is retried.
+ */
+static rsRetVal
+Wait(nssel_t *pThis, int *piNumReady)
+{
+ DEFiRet;
+ ISOBJ_TYPE_assert(pThis, nssel);
+ assert(piNumReady != NULL);
+ RETiRet;
+}
+
+
+/* Check if a stream is ready for IO. *piNumReady contains the remaining number
+ * of ready streams. Note that this function may say the stream is not ready
+ * but still decrement *piNumReady. This can happen when (e.g. with TLS) the low
+ * level driver requires some IO which is hidden from the upper layer point of view.
+ * rgerhards, 2008-04-23
+ */
+static rsRetVal
+IsReady(nssel_t *pThis, netstrm_t *pStrm, int *pbIsReady, int *piNumReady)
+{
+ DEFiRet;
+ ISOBJ_TYPE_assert(pThis, nssel);
+ ISOBJ_TYPE_assert(pStrm, netstrm);
+ assert(pbIsReady != NULL);
+ assert(piNumReady != NULL);
+ RETiRet;
+}
+
+
+/* queryInterface function */
+BEGINobjQueryInterface(nssel)
+CODESTARTobjQueryInterface(nssel)
+ if(pIf->ifVersion != nsselCURR_IF_VERSION) {/* check for current version, increment on each change */
+ ABORT_FINALIZE(RS_RET_INTERFACE_NOT_SUPPORTED);
+ }
+
+ /* ok, we have the right interface, so let's fill it
+ * Please note that we may also do some backwards-compatibility
+ * work here (if we can support an older interface version - that,
+ * of course, also affects the "if" above).
+ */
+ pIf->Construct = nsselConstruct;
+ pIf->ConstructFinalize = ConstructFinalize;
+ pIf->Destruct = nsselDestruct;
+ pIf->Add = Add;
+ pIf->Wait = Wait;
+ pIf->IsReady = IsReady;
+finalize_it:
+ENDobjQueryInterface(nssel)
+
+
+/* exit our class
+ */
+BEGINObjClassExit(nssel, OBJ_IS_LOADABLE_MODULE) /* CHANGE class also in END MACRO! */
+CODESTARTObjClassExit(nssel)
+ /* release objects we no longer need */
+ objRelease(glbl, CORE_COMPONENT);
+ENDObjClassExit(nssel)
+
+
+/* Initialize the nssel class. Must be called as the very first method
+ * before anything else is called inside this class.
+ * rgerhards, 2008-02-19
+ */
+BEGINObjClassInit(nssel, 1, OBJ_IS_CORE_MODULE) /* class, version */
+ /* request objects we use */
+ CHKiRet(objUse(glbl, CORE_COMPONENT));
+
+ /* set our own handlers */
+ENDObjClassInit(nssel)
+
+
+/* --------------- here now comes the plumbing that makes us a library module --------------- */
+
+
+BEGINmodExit
+CODESTARTmodExit
+ nsselClassExit();
+ENDmodExit
+
+
+BEGINqueryEtryPt
+CODESTARTqueryEtryPt
+CODEqueryEtryPt_STD_LIB_QUERIES
+ENDqueryEtryPt
+
+
+BEGINmodInit()
+CODESTARTmodInit
+ *ipIFVersProvided = CURR_MOD_IF_VERSION; /* we only support the current interface specification */
+
+ /* Initialize all classes that are in our module - this includes ourselfs */
+ CHKiRet(nsselClassInit(pModInfo)); /* must be done after tcps_sess, as we use it */
+ENDmodInit
+/* vi:set ai:
+ */
diff --git a/runtime/nssel.h b/runtime/nssel.h
new file mode 100644
index 00000000..16919b1f
--- /dev/null
+++ b/runtime/nssel.h
@@ -0,0 +1,55 @@
+/* Definitions for the nssel IO waiter.
+ *
+ * Copyright 2008 Rainer Gerhards and Adiscon GmbH.
+ *
+ * This file is part of the rsyslog runtime library.
+ *
+ * The rsyslog runtime library is free software: you can redistribute it and/or modify
+ * it under the terms of the GNU Lesser General Public License as published by
+ * the Free Software Foundation, either version 3 of the License, or
+ * (at your option) any later version.
+ *
+ * The rsyslog runtime library is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+ * GNU Lesser General Public License for more details.
+ *
+ * You should have received a copy of the GNU Lesser General Public License
+ * along with the rsyslog runtime library. If not, see <http://www.gnu.org/licenses/>.
+ *
+ * A copy of the GPL can be found in the file "COPYING" in this distribution.
+ * A copy of the LGPL can be found in the file "COPYING.LESSER" in this distribution.
+ */
+
+#ifndef INCLUDED_NSSEL_H
+#define INCLUDED_NSSEL_H
+
+#include "nsd.h"
+
+/* the nssel object */
+struct nssel_s {
+ BEGINobjInstance; /* Data to implement generic object - MUST be the first data element! */
+ nsd_t *pDrvrData; /**< the driver's data elements */
+ uchar *pDrvrName; /**< nsd driver name to use, or NULL if system default */
+ nsdsel_if_t Drvr; /**< our stream driver */
+};
+
+
+/* interface */
+BEGINinterface(nssel) /* name must also be changed in ENDinterface macro! */
+ rsRetVal (*Construct)(nssel_t **ppThis);
+ rsRetVal (*ConstructFinalize)(nssel_t *pThis);
+ rsRetVal (*Destruct)(nssel_t **ppThis);
+ rsRetVal (*Add)(nssel_t *pThis, netstrm_t *pStrm);
+ rsRetVal (*Wait)(nssel_t *pThis, int *pNumReady);
+ rsRetVal (*IsReady)(nssel_t *pThis, netstrm_t *pStrm, int *pbIsReady, int *pNumRead);
+ENDinterface(nssel)
+#define nsselCURR_IF_VERSION 1 /* increment whenever you change the interface structure! */
+
+/* prototypes */
+PROTOTYPEObj(nssel);
+
+/* the name of our library binary */
+#define LM_NSSEL_FILENAME "lmnssel"
+
+#endif /* #ifndef INCLUDED_NSSEL_H */
diff --git a/runtime/rsyslog.h b/runtime/rsyslog.h
index f7824006..4f62ca3c 100644
--- a/runtime/rsyslog.h
+++ b/runtime/rsyslog.h
@@ -65,8 +65,12 @@ typedef struct obj_s obj_t;
typedef struct filed selector_t;/* TODO: this so far resides in syslogd.c, think about modularization */
typedef struct NetAddr netAddr_t;
typedef struct netstrm_s netstrm_t;
+typedef struct nssel_s nssel_t;
+typedef enum nsdsel_waitOp_e nsdsel_waitOp_t;
typedef struct nsd_ptcp_s nsd_ptcp_t;
+typedef struct nsdsel_ptcp_s nsdsel_ptcp_t;
typedef obj_t nsd_t;
+typedef obj_t nsdsel_t;
typedef struct msg msg_t;
typedef struct interface_s interface_t;
typedef struct objInfo_s objInfo_t;
@@ -206,6 +210,7 @@ enum rsRetVal_ /** return value. All methods return this if not specified oth
RS_RET_INVALID_HNAME = -2075, /**< remote peer's hostname invalid or unobtainable */
RS_RET_INVALID_PORT = -2076, /**< invalid port value */
RS_RET_COULD_NOT_BIND = -2077, /**< could not bind socket, defunct */
+ RS_RET_MAX_SESS_REACHED = -2078, /**< max nbr of sessions reached, can not create more */
/* RainerScript error messages (range 1000.. 1999) */
RS_RET_SYSVAR_NOT_FOUND = 1001, /**< system variable could not be found (maybe misspelled) */
diff --git a/tcps_sess.c b/tcps_sess.c
index b5c9c31f..f66396e0 100644
--- a/tcps_sess.c
+++ b/tcps_sess.c
@@ -403,7 +403,7 @@ CODESTARTobjQueryInterface(tcps_sess)
pIf->SetUsrP = SetUsrP;
pIf->SetTcpsrv = SetTcpsrv;
pIf->SetHost = SetHost;
- pIf->SetSock = SetSock;
+ //pIf->SetSock = SetSock;
pIf->SetMsgIdx = SetMsgIdx;
finalize_it:
ENDobjQueryInterface(tcps_sess)
diff --git a/tcps_sess.h b/tcps_sess.h
index 1d45c482..102d91b5 100644
--- a/tcps_sess.h
+++ b/tcps_sess.h
@@ -61,7 +61,7 @@ BEGINinterface(tcps_sess) /* name must also be changed in ENDinterface macro! */
rsRetVal (*SetTcpsrv)(tcps_sess_t *pThis, struct tcpsrv_s *pSrv);
rsRetVal (*SetUsrP)(tcps_sess_t*, void*);
rsRetVal (*SetHost)(tcps_sess_t *pThis, uchar*);
- rsRetVal (*SetSock)(tcps_sess_t *pThis, int);
+ rsRetVal (*SetStrm)(tcps_sess_t *pThis, netstrm_t*);
rsRetVal (*SetMsgIdx)(tcps_sess_t *pThis, int);
ENDinterface(tcps_sess)
#define tcps_sessCURR_IF_VERSION 1 /* increment whenever you change the interface structure! */
diff --git a/tcpsrv.c b/tcpsrv.c
index 96048e31..a276bf19 100644
--- a/tcpsrv.c
+++ b/tcpsrv.c
@@ -63,6 +63,7 @@
#include "tcpsrv.h"
#include "obj.h"
#include "glbl.h"
+#include "netstrm.h"
#include "errmsg.h"
MODULE_TYPE_LIB
@@ -77,30 +78,7 @@ DEFobjCurrIf(glbl)
DEFobjCurrIf(tcps_sess)
DEFobjCurrIf(errmsg)
DEFobjCurrIf(net)
-
-
-
-/* code to free all sockets within a socket table.
- * A socket table is a descriptor table where the zero
- * element has the count of elements. This is used for
- * listening sockets. The socket table itself is also
- * freed.
- * A POINTER to this structure must be provided, thus
- * double indirection!
- * rgerhards, 2007-06-28
- */
-static void freeAllSockets(int **socks)
-{
- assert(socks != NULL);
- assert(*socks != NULL);
- while(**socks) {
- dbgprintf("Closing socket %d.\n", (*socks)[**socks]);
- close((*socks)[**socks]);
- (**socks)--;
- }
- free(*socks);
- *socks = NULL;
-}
+DEFobjCurrIf(netstrm)
/* configure TCP listener settings. This is called during command
@@ -236,33 +214,24 @@ static void deinit_tcp_listener(tcpsrv_t *pThis)
if(pThis->TCPLstnPort != NULL)
free(pThis->TCPLstnPort);
- /* finally close the listen sockets themselfs */
- freeAllSockets(&pThis->pSocksLstn);
+ /* finally close our listen stream */
+ netstrm.Destruct(&pThis->pLstn);
}
-/* Initialize TCP sockets (for listener)
- * This function returns either NULL (which means it failed) or
- * a pointer to an array of file descriptiors. If the pointer is
- * returned, the zeroest element [0] contains the count of valid
- * descriptors. The descriptors themself follow in range
- * [1] ... [num-descriptors]. It is guaranteed that each of these
- * descriptors is valid, at least when this function returns.
- * Please note that technically the array may be larger than the number
- * of valid pointers stored in it. The memory overhead is minimal, so
- * we do not bother to re-allocate an array of the exact size. Logically,
- * the array still contains the exactly correct number of descriptors.
- */
-static int *create_tcp_socket(tcpsrv_t *pThis)
+/* Initialize TCP sockets (for listener) and listens on them */
+static rsRetVal
+create_tcp_socket(tcpsrv_t *pThis)
{
- struct addrinfo hints, *res, *r;
- int error, maxs, *s, *socks, on = 1;
- char *TCPLstnPort;
+ DEFiRet;
+ uchar *TCPLstnPort;
ISOBJ_TYPE_assert(pThis, tcpsrv);
- if(!strcmp(pThis->TCPLstnPort, "0"))
- TCPLstnPort = "514";
+ if(!strcmp((char*)pThis->TCPLstnPort, "0"))
+ TCPLstnPort = (uchar*)"514";
+ // TODO: we need to enable the caller to set a port (based on who is
+ // using this, 514 may be totally unsuitable... --- rgerhards, 2008-04-22
/* use default - we can not do service db update, because there is
* no IANA-assignment for syslog/tcp. In the long term, we might
* re-use RFC 3195 port of 601, but that would probably break to
@@ -270,136 +239,13 @@ static int *create_tcp_socket(tcpsrv_t *pThis)
* rgerhards, 2007-06-28
*/
else
- TCPLstnPort = pThis->TCPLstnPort;
- dbgprintf("creating tcp socket on port %s\n", TCPLstnPort);
- memset(&hints, 0, sizeof(hints));
- hints.ai_flags = AI_PASSIVE | AI_NUMERICSERV;
- hints.ai_family = glbl.GetDefPFFamily();
- hints.ai_socktype = SOCK_STREAM;
-
- error = getaddrinfo(NULL, TCPLstnPort, &hints, &res);
- if(error) {
- errmsg.LogError(NO_ERRCODE, "%s", gai_strerror(error));
- return NULL;
- }
-
- /* Count max number of sockets we may open */
- for (maxs = 0, r = res; r != NULL ; r = r->ai_next, maxs++)
- /* EMPTY */;
- socks = malloc((maxs+1) * sizeof(int));
- if (socks == NULL) {
- errmsg.LogError(NO_ERRCODE, "couldn't allocate memory for TCP listen sockets, suspending TCP message reception.");
- freeaddrinfo(res);
- return NULL;
- }
-
- *socks = 0; /* num of sockets counter at start of array */
- s = socks + 1;
- for (r = res; r != NULL ; r = r->ai_next) {
- *s = socket(r->ai_family, r->ai_socktype, r->ai_protocol);
- if (*s < 0) {
- if(!(r->ai_family == PF_INET6 && errno == EAFNOSUPPORT))
- errmsg.LogError(NO_ERRCODE, "create_tcp_socket(), socket");
- /* it is debatable if PF_INET with EAFNOSUPPORT should
- * also be ignored...
- */
- continue;
- }
-
-#ifdef IPV6_V6ONLY
- if (r->ai_family == AF_INET6) {
- int iOn = 1;
- if (setsockopt(*s, IPPROTO_IPV6, IPV6_V6ONLY,
- (char *)&iOn, sizeof (iOn)) < 0) {
- errmsg.LogError(NO_ERRCODE, "TCP setsockopt");
- close(*s);
- *s = -1;
- continue;
- }
- }
-#endif
- if (setsockopt(*s, SOL_SOCKET, SO_REUSEADDR,
- (char *) &on, sizeof(on)) < 0 ) {
- errmsg.LogError(NO_ERRCODE, "TCP setsockopt(REUSEADDR)");
- close(*s);
- *s = -1;
- continue;
- }
-
- /* We need to enable BSD compatibility. Otherwise an attacker
- * could flood our log files by sending us tons of ICMP errors.
- */
-#ifndef OS_BSD
- if(net.should_use_so_bsdcompat()) {
- if (setsockopt(*s, SOL_SOCKET, SO_BSDCOMPAT,
- (char *) &on, sizeof(on)) < 0) {
- errmsg.LogError(NO_ERRCODE, "TCP setsockopt(BSDCOMPAT)");
- close(*s);
- *s = -1;
- continue;
- }
- }
-#endif
-
- if( (bind(*s, r->ai_addr, r->ai_addrlen) < 0)
-#ifndef IPV6_V6ONLY
- && (errno != EADDRINUSE)
-#endif
- ) {
- errmsg.LogError(NO_ERRCODE, "TCP bind");
- close(*s);
- *s = -1;
- continue;
- }
-
- if( listen(*s,pThis->iSessMax / 10 + 5) < 0) {
- /* If the listen fails, it most probably fails because we ask
- * for a too-large backlog. So in this case we first set back
- * to a fixed, reasonable, limit that should work. Only if
- * that fails, too, we give up.
- */
- errmsg.LogError(NO_ERRCODE, "listen with a backlog of %d failed - retrying with default of 32.",
- pThis->iSessMax / 10 + 5);
- if(listen(*s, 32) < 0) {
- errmsg.LogError(NO_ERRCODE, "TCP listen, suspending tcp inet");
- close(*s);
- *s = -1;
- continue;
- }
- }
+ TCPLstnPort = (uchar*)pThis->TCPLstnPort;
- (*socks)++;
- s++;
- }
-
- if(res != NULL)
- freeaddrinfo(res);
-
- if(Debug && *socks != maxs)
- dbgprintf("We could initialize %d TCP listen sockets out of %d we received "
- "- this may or may not be an error indication.\n", *socks, maxs);
-
- if(*socks == 0) {
- errmsg.LogError(NO_ERRCODE, "No TCP listen socket could successfully be initialized, "
- "message reception via TCP disabled.\n");
- free(socks);
- return(NULL);
- }
+ /* TODO: add capability to specify local listen address! */
+ CHKiRet(netstrm.LstnInit(pThis->pLstn, TCPLstnPort, NULL, pThis->iSessMax));
- /* OK, we had success. Now it is also time to
- * initialize our connections
- */
- if(TCPSessTblInit(pThis) != 0) {
- /* OK, we are in some trouble - we could not initialize the
- * session table, so we can not continue. We need to free all
- * we have assigned so far, because we can not really use it...
- */
- errmsg.LogError(NO_ERRCODE, "Could not initialize TCP session table, suspending TCP message reception.");
- freeAllSockets(&socks); /* prevent a socket leak */
- return(NULL);
- }
-
- return(socks);
+finalize_it:
+ RETiRet;
}
@@ -414,34 +260,26 @@ static int *create_tcp_socket(tcpsrv_t *pThis)
* rgerhards, 2008-03-02
*/
static rsRetVal
-SessAccept(tcpsrv_t *pThis, tcps_sess_t **ppSess, int fd)
+SessAccept(tcpsrv_t *pThis, tcps_sess_t **ppSess, nsd_t *pNsd)
{
DEFiRet;
tcps_sess_t *pSess;
- int newConn;
+ netstrm_t *pNewStrm = NULL;
int iSess = -1;
struct sockaddr_storage addr;
- socklen_t addrlen = sizeof(struct sockaddr_storage);
uchar fromHost[NI_MAXHOST];
uchar fromHostFQDN[NI_MAXHOST];
ISOBJ_TYPE_assert(pThis, tcpsrv);
- newConn = accept(fd, (struct sockaddr*) &addr, &addrlen);
- if (newConn < 0) {
- errmsg.LogError(NO_ERRCODE, "tcp accept, ignoring error and connection request");
- ABORT_FINALIZE(RS_RET_ERR); // TODO: better error code
- //was: return -1;
- }
+ CHKiRet(netstrm.AcceptConnReq(pThis->pLstn, pNsd, &pNewStrm));
/* Add to session list */
iSess = TCPSessTblFindFreeSpot(pThis);
if(iSess == -1) {
errno = 0;
errmsg.LogError(NO_ERRCODE, "too many tcp sessions - dropping incoming request");
- close(newConn);
- ABORT_FINALIZE(RS_RET_ERR); // TODO: better error code
- //was: return -1;
+ ABORT_FINALIZE(RS_RET_MAX_SESS_REACHED);
} else {
/* we found a free spot and can construct our session object */
CHKiRet(tcps_sess.Construct(&pSess));
@@ -455,7 +293,6 @@ SessAccept(tcpsrv_t *pThis, tcps_sess_t **ppSess, int fd)
* are now told to discard the connection request.
* Error message has been generated by cvthname.
*/
- close (newConn);
ABORT_FINALIZE(RS_RET_ERR); // TODO: better error code
}
@@ -469,10 +306,8 @@ SessAccept(tcpsrv_t *pThis, tcps_sess_t **ppSess, int fd)
dbgprintf("%s is not an allowed sender\n", (char *) fromHostFQDN);
if(glbl.GetOption_DisallowWarning()) {
errno = 0;
- errmsg.LogError(NO_ERRCODE, "TCP message from disallowed sender %s discarded",
- (char*)fromHost);
+ errmsg.LogError(NO_ERRCODE, "TCP message from disallowed sender %s discarded", fromHost);
}
- close(newConn);
ABORT_FINALIZE(RS_RET_HOST_NOT_PERMITTED);
}
@@ -480,7 +315,8 @@ SessAccept(tcpsrv_t *pThis, tcps_sess_t **ppSess, int fd)
* means we can finally fill in the session object.
*/
CHKiRet(tcps_sess.SetHost(pSess, fromHost));
- CHKiRet(tcps_sess.SetSock(pSess, newConn));
+ CHKiRet(tcps_sess.SetStrm(pSess, pNewStrm));
+ pNewStrm = NULL; /* prevent it from being freed in error handler, now done in tcps_sess! */
CHKiRet(tcps_sess.SetMsgIdx(pSess, 0));
CHKiRet(tcps_sess.ConstructFinalize(pSess));
@@ -499,6 +335,8 @@ finalize_it:
tcps_sess.Destruct(&pThis->pSessions[iSess]);
}
iSess = -1; // TODO: change this to be fully iRet compliant ;)
+ if(pNewStrm != NULL)
+ netstrm.Destruct(&pNewStrm);
}
RETiRet;
@@ -517,6 +355,7 @@ Run(tcpsrv_t *pThis)
int iTCPSess;
fd_set readfds;
tcps_sess_t *pNewSess;
+ nsdsel_t *pSel;
ISOBJ_TYPE_assert(pThis, tcpsrv);
@@ -525,11 +364,9 @@ Run(tcpsrv_t *pThis)
* right into the sleep below.
*/
while(1) {
- maxfds = 0;
- FD_ZERO (&readfds);
+ CHKiRet(nsdsel_ptcp.Construct(&pSel));
- /* Add the TCP listen sockets to the list of read descriptors.
- */
+ /* Add the TCP listen sockets to the list of read descriptors. */
if(pThis->pSocksLstn != NULL && *pThis->pSocksLstn) {
for (i = 0; i < *pThis->pSocksLstn; i++) {
/* The if() below is theoretically not needed, but I leave it in
@@ -537,8 +374,8 @@ Run(tcpsrv_t *pThis)
* feature is not yet supported by the current code base.
*/
if (pThis->pSocksLstn[i+1] != -1) {
- if(Debug)
- net.debugListenInfo(pThis->pSocksLstn[i+1], "TCP");
+ if(Debug) net.debugListenInfo(pThis->pSocksLstn[i+1], "TCP");
+ CHKiRet(nsdsel_ptcp.Add(pSel, ));
FD_SET(pThis->pSocksLstn[i+1], &readfds);
if(pThis->pSocksLstn[i+1]>maxfds) maxfds=pThis->pSocksLstn[i+1];
}
@@ -566,6 +403,7 @@ Run(tcpsrv_t *pThis)
}
/* wait for io to become ready */
+ /* this is the somewhat weak spot in our socket layer abstraction... */
nfds = select(maxfds+1, (fd_set *) &readfds, NULL, NULL, NULL);
for (i = 0; i < *pThis->pSocksLstn; i++) {
@@ -620,7 +458,6 @@ Run(tcpsrv_t *pThis)
/* Standard-Constructor
*/
BEGINobjConstruct(tcpsrv) /* be sure to specify the object type also in END macro! */
- pThis->pSocksLstn = NULL;
pThis->iSessMax = 200; /* TODO: useful default ;) */
ENDobjConstruct(tcpsrv)
@@ -632,7 +469,7 @@ tcpsrvConstructFinalize(tcpsrv_t __attribute__((unused)) *pThis)
{
DEFiRet;
ISOBJ_TYPE_assert(pThis, tcpsrv);
- pThis->pSocksLstn = pThis->OpenLstnSocks(pThis);
+ iRet = pThis->OpenLstnSocks(pThis);
RETiRet;
}
@@ -727,7 +564,7 @@ SetCBOnErrClose(tcpsrv_t *pThis, rsRetVal (*pCB)(tcps_sess_t*))
}
static rsRetVal
-SetCBOpenLstnSocks(tcpsrv_t *pThis, int* (*pCB)(tcpsrv_t*))
+SetCBOpenLstnSocks(tcpsrv_t *pThis, rsRetVal (*pCB)(tcpsrv_t*))
{
DEFiRet;
pThis->OpenLstnSocks = pCB;
@@ -793,6 +630,7 @@ CODESTARTObjClassExit(tcpsrv)
objRelease(conf, CORE_COMPONENT);
objRelease(glbl, CORE_COMPONENT);
objRelease(errmsg, CORE_COMPONENT);
+ objRelease(netstrm, LM_NET_FILENAME);
objRelease(net, LM_NET_FILENAME);
ENDObjClassExit(tcpsrv)
@@ -805,6 +643,7 @@ BEGINObjClassInit(tcpsrv, 1, OBJ_IS_LOADABLE_MODULE) /* class, version - CHANGE
/* request objects we use */
CHKiRet(objUse(errmsg, CORE_COMPONENT));
CHKiRet(objUse(net, LM_NET_FILENAME));
+ CHKiRet(objUse(netstrm, LM_NET_FILENAME));
CHKiRet(objUse(tcps_sess, DONT_LOAD_LIB));
CHKiRet(objUse(conf, CORE_COMPONENT));
CHKiRet(objUse(glbl, CORE_COMPONENT));
diff --git a/tcpsrv.h b/tcpsrv.h
index 20a7ea25..4cec906f 100644
--- a/tcpsrv.h
+++ b/tcpsrv.h
@@ -28,7 +28,8 @@
/* the tcpsrv object */
typedef struct tcpsrv_s {
BEGINobjInstance; /**< Data to implement generic object - MUST be the first data element! */
- int *pSocksLstn; /**< listen socket array for server [0] holds count */
+ //int *pSocksLstn; /**< listen socket array for server [0] holds count */
+ netstrm_t *pLstn; /**< our netstream listner (which may contain multiple "sockets" */
int iSessMax; /**< max number of sessions supported */
char *TCPLstnPort; /**< the port the listener shall listen on */
tcps_sess_t **pSessions;/**< array of all of our sessions */
@@ -36,7 +37,7 @@ typedef struct tcpsrv_s {
/* callbacks */
int (*pIsPermittedHost)(struct sockaddr *addr, char *fromHostFQDN, void*pUsrSrv, void*pUsrSess);
int (*pRcvData)(tcps_sess_t*, char*, size_t);
- int* (*OpenLstnSocks)(struct tcpsrv_s*);
+ rsRetVal (*OpenLstnSocks)(struct tcpsrv_s*);
rsRetVal (*pOnListenDeinit)(void*);
rsRetVal (*OnDestruct)(void*);
rsRetVal (*pOnRegularClose)(tcps_sess_t *pSess);
@@ -55,13 +56,13 @@ BEGINinterface(tcpsrv) /* name must also be changed in ENDinterface macro! */
rsRetVal (*ConstructFinalize)(tcpsrv_t __attribute__((unused)) *pThis);
rsRetVal (*Destruct)(tcpsrv_t **ppThis);
void (*configureTCPListen)(tcpsrv_t*, char *cOptarg);
- int (*SessAccept)(tcpsrv_t *pThis, tcps_sess_t**ppSess, int fd);
- int* (*create_tcp_socket)(tcpsrv_t *pThis);
+ rsRetVal (*SessAccept)(tcpsrv_t *pThis, tcps_sess_t **ppSess, nsd_t *pNsd);
+ rsRetVal (*create_tcp_socket)(tcpsrv_t *pThis);
rsRetVal (*Run)(tcpsrv_t *pThis);
/* set methods */
rsRetVal (*SetUsrP)(tcpsrv_t*, void*);
rsRetVal (*SetCBIsPermittedHost)(tcpsrv_t*, int (*) (struct sockaddr *addr, char*, void*, void*));
- rsRetVal (*SetCBOpenLstnSocks)(tcpsrv_t *, int* (*)(tcpsrv_t*));
+ rsRetVal (*SetCBOpenLstnSocks)(tcpsrv_t *, rsRetVal (*)(tcpsrv_t*));
rsRetVal (*SetCBRcvData)(tcpsrv_t *, int (*)(tcps_sess_t*, char*, size_t));
rsRetVal (*SetCBOnListenDeinit)(tcpsrv_t*, rsRetVal (*)(void*));
rsRetVal (*SetCBOnDestruct)(tcpsrv_t*, rsRetVal (*) (void*));
@@ -72,7 +73,7 @@ BEGINinterface(tcpsrv) /* name must also be changed in ENDinterface macro! */
rsRetVal (*SetCBOnSessDestruct)(tcpsrv_t*, rsRetVal (*) (void*));
rsRetVal (*SetCBOnSessConstructFinalize)(tcpsrv_t*, rsRetVal (*) (void*));
ENDinterface(tcpsrv)
-#define tcpsrvCURR_IF_VERSION 1 /* increment whenever you change the interface structure! */
+#define tcpsrvCURR_IF_VERSION 2 /* increment whenever you change the interface structure! */
/* prototypes */