mturk 2005/04/13 23:47:56
Modified: jni/java/org/apache/tomcat/jni Poll.java
jni/native/src network.c poll.c
Log:
Add time to live to socket Poller, so we can maintain resources.
The returned descriptor will be returned with
APR_POOLIN + APR_POLLHUP in case the recycle is needed.
Revision Changes Path
1.5 +3 -3
jakarta-tomcat-connectors/jni/java/org/apache/tomcat/jni/Poll.java
Index: Poll.java
===================================================================
RCS file:
/home/cvs/jakarta-tomcat-connectors/jni/java/org/apache/tomcat/jni/Poll.java,v
retrieving revision 1.4
retrieving revision 1.5
diff -u -r1.4 -r1.5
--- Poll.java 13 Apr 2005 13:16:55 -0000 1.4
+++ Poll.java 14 Apr 2005 06:47:56 -0000 1.5
@@ -60,9 +60,10 @@
* @param size The maximum number of descriptors that this pollset can
hold
* @param p The pool from which to allocate the pollset
* @param flags Optional flags to modify the operation of the pollset.
+ * @param ttl Maximum time to live for a particular socket.
* @return The pointer in which to return the newly created object
*/
- public static native long create(int size, long p, int flags)
+ public static native long create(int size, long p, int flags, long ttl)
throws Error;
/**
* Destroy a pollset object
@@ -98,8 +99,7 @@
* @return Number of signalled descriptors (output parameter)
*/
public static native int poll(long pollset, long timeout,
- long [] descriptors)
- throws Error;
+ long [] descriptors);
/**
* Return socket from poll descriptor
1.4 +6 -6 jakarta-tomcat-connectors/jni/native/src/network.c
Index: network.c
===================================================================
RCS file: /home/cvs/jakarta-tomcat-connectors/jni/native/src/network.c,v
retrieving revision 1.3
retrieving revision 1.4
diff -u -r1.3 -r1.4
--- network.c 13 Apr 2005 13:17:42 -0000 1.3
+++ network.c 14 Apr 2005 06:47:56 -0000 1.4
@@ -183,7 +183,7 @@
UNREFERENCED(o);
apr_socket_opt_get(s, APR_SO_NONBLOCK, &nb);
if (tosend > 0)
- nbytes = min(nbytes, (apr_size_t)tosend);
+ nbytes = min(nbytes - offset, (apr_size_t)tosend);
if (nb)
bytes = (*e)->GetPrimitiveArrayCritical(e, buf, NULL);
else
@@ -213,7 +213,7 @@
goto cleanup;
}
if (len > 0)
- nbytes = min(nbytes, (apr_size_t)len);
+ nbytes = min(nbytes - offset, (apr_size_t)len);
TCN_THROW_IF_ERR(apr_socket_send(s, bytes + offset, &nbytes), nbytes);
cleanup:
@@ -264,7 +264,7 @@
UNREFERENCED(o);
apr_socket_opt_get(s, APR_SO_NONBLOCK, &nb);
if (tosend > 0)
- nbytes = min(nbytes, (apr_size_t)tosend);
+ nbytes = min(nbytes - offset, (apr_size_t)tosend);
if (nb)
bytes = (*e)->GetPrimitiveArrayCritical(e, buf, NULL);
else
@@ -288,7 +288,7 @@
UNREFERENCED(o);
if (toread > 0)
- nbytes = min(nbytes, (apr_size_t)toread);
+ nbytes = min(nbytes - offset, (apr_size_t)toread);
TCN_THROW_IF_ERR(apr_socket_recv(s, bytes + offset, &nbytes), nbytes);
@@ -314,7 +314,7 @@
goto cleanup;
}
if (len > 0)
- nbytes = min(nbytes, (apr_size_t)len);
+ nbytes = min(nbytes - offset, (apr_size_t)len);
TCN_THROW_IF_ERR(apr_socket_recv(s, bytes + offset, &nbytes), nbytes);
@@ -333,7 +333,7 @@
UNREFERENCED(o);
if (toread > 0)
- nbytes = min(nbytes, (apr_size_t)toread);
+ nbytes = min(nbytes - offset, (apr_size_t)toread);
TCN_THROW_IF_ERR(apr_socket_recvfrom(f, s,
(apr_int32_t)flags, bytes + offset, &nbytes), nbytes);
1.4 +83 -17 jakarta-tomcat-connectors/jni/native/src/poll.c
Index: poll.c
===================================================================
RCS file: /home/cvs/jakarta-tomcat-connectors/jni/native/src/poll.c,v
retrieving revision 1.3
retrieving revision 1.4
diff -u -r1.3 -r1.4
--- poll.c 13 Apr 2005 13:17:42 -0000 1.3
+++ poll.c 14 Apr 2005 06:47:56 -0000 1.4
@@ -18,55 +18,86 @@
#include "apr_poll.h"
#include "tcn.h"
+/* Internal poll structure for queryset
+ */
+
+typedef struct tcn_pollset {
+ apr_pool_t *pool;
+ apr_int32_t nelts;
+ apr_int32_t nalloc;
+ apr_pollset_t *pollset;
+ apr_pollfd_t *query_set;
+ apr_time_t *query_add;
+ apr_interval_time_t max_ttl;
+} tcn_pollset_t;
TCN_IMPLEMENT_CALL(jlong, Poll, create)(TCN_STDARGS, jint size,
- jlong pool, jint flags)
+ jlong pool, jint flags,
+ jlong ttl)
{
apr_pool_t *p = J2P(pool, apr_pool_t *);
apr_pollset_t *pollset = NULL;
-
+ tcn_pollset_t *tps = NULL;
UNREFERENCED(o);
TCN_THROW_IF_ERR(apr_pollset_create(&pollset,
(apr_uint32_t)size, p, (apr_uint32_t)flags),
pollset);
+ tps = apr_palloc(p, sizeof(tcn_pollset_t));
+ tps->pollset = pollset;
+ tps->query_set = apr_palloc(p, size * sizeof(apr_pollfd_t));
+ tps->query_add = apr_palloc(p, size * sizeof(apr_time_t));
+ tps->nelts = 0;
+ tps->nalloc = size;
+ tps->pool = p;
+ tps->max_ttl = J2T(ttl);
+
cleanup:
- return P2J(pollset);
+ return P2J(tps);
}
TCN_IMPLEMENT_CALL(jint, Poll, destroy)(TCN_STDARGS, jlong pollset)
{
- apr_pollset_t *p = J2P(pollset, apr_pollset_t *);
+ tcn_pollset_t *p = J2P(pollset, tcn_pollset_t *);;
UNREFERENCED_STDARGS;;
- return (jint)apr_pollset_destroy(p);
+ return (jint)apr_pollset_destroy(p->pollset);
}
TCN_IMPLEMENT_CALL(jint, Poll, add)(TCN_STDARGS, jlong pollset,
jlong socket, jlong data,
jint reqevents)
{
- apr_pollset_t *p = J2P(pollset, apr_pollset_t *);
+ tcn_pollset_t *p = J2P(pollset, tcn_pollset_t *);
apr_pollfd_t fd;
+ apr_status_t rv;
UNREFERENCED_STDARGS;
+ if (p->nelts == p->nalloc) {
+ return APR_ENOMEM;
+ }
memset(&fd, 0, sizeof(apr_pollfd_t));
fd.desc_type = APR_POLL_SOCKET;
fd.reqevents = (apr_int16_t)reqevents;
fd.desc.s = J2P(socket, apr_socket_t *);
fd.client_data = J2P(data, void *);
-
- return (jint)apr_pollset_add(p, &fd);
+ if ((rv = apr_pollset_add(p->pollset, &fd)) == APR_SUCCESS) {
+ p->query_set[p->nelts] = fd;
+ p->query_add[p->nelts] = apr_time_now();
+ p->nelts++;
+ }
+ return (jint)rv;
}
TCN_IMPLEMENT_CALL(jint, Poll, remove)(TCN_STDARGS, jlong pollset,
jlong socket)
{
- apr_pollset_t *p = J2P(pollset, apr_pollset_t *);
+ tcn_pollset_t *p = J2P(pollset, tcn_pollset_t *);
apr_pollfd_t fd;
+ apr_int32_t i;
UNREFERENCED_STDARGS;;
@@ -74,29 +105,64 @@
fd.desc_type = APR_POLL_SOCKET;
fd.desc.s = J2P(socket, apr_socket_t *);
- return (jint)apr_pollset_remove(p, &fd);
+ for (i = 0; i < p->nelts; i++) {
+ if (fd.desc.s == p->query_set[i].desc.s) {
+ /* Found an instance of the fd: remove this and any other copies
*/
+ apr_int32_t dst = i;
+ apr_int32_t old_nelts = p->nelts;
+ p->nelts--;
+ for (i++; i < old_nelts; i++) {
+ if (fd.desc.s == p->query_set[i].desc.s) {
+ p->nelts--;
+ }
+ else {
+ p->query_set[dst] = p->query_set[i];
+ dst++;
+ }
+ }
+ break;
+ }
+ }
+
+ return (jint)apr_pollset_remove(p->pollset, &fd);
}
TCN_IMPLEMENT_CALL(jint, Poll, poll)(TCN_STDARGS, jlong pollset,
jlong timeout, jlongArray set)
{
const apr_pollfd_t *fd = NULL;
- apr_pollset_t *p = J2P(pollset, apr_pollset_t *);
+ tcn_pollset_t *p = J2P(pollset, tcn_pollset_t *);
jlong *pset = (*e)->GetLongArrayElements(e, set, NULL);
- apr_int32_t i, num = 0;
+ apr_int32_t n, i = 0, num = 0;
+ apr_status_t rv;
UNREFERENCED(o);
- TCN_THROW_IF_ERR(apr_pollset_poll(p, J2T(timeout),
- &num, &fd), num);
+ rv = apr_pollset_poll(p->pollset, J2T(timeout), &num, &fd);
-cleanup:
- if (num) {
+ if (rv == APR_SUCCESS && num > 0) {
for (i = 0; i < num; i++) {
pset[i] = P2J(fd);
fd ++;
}
- (*e)->ReleaseLongArrayElements(e, set, pset, 0);
}
+ /* In any case check for timeout sockets */
+ if (p->max_ttl > 0) {
+ apr_time_t now = apr_time_now();
+ /* TODO: Add thread mutex protection
+ * or make sure the Java part is synchronized.
+ */
+ for (n = 0; n < p->nelts; n++) {
+ if ((now - p->query_add[n]) > p->max_ttl) {
+ p->query_set[n].rtnevents = APR_POLLHUP | APR_POLLIN;
+ if (i < p->nelts) {
+ pset[i++] = P2J(&(p->query_set[n]));
+ num++;
+ }
+ }
+ }
+ }
+ if (num)
+ (*e)->ReleaseLongArrayElements(e, set, pset, 0);
else
(*e)->ReleaseLongArrayElements(e, set, pset, JNI_ABORT);
---------------------------------------------------------------------
To unsubscribe, e-mail: [EMAIL PROTECTED]
For additional commands, e-mail: [EMAIL PROTECTED]
