You are viewing a plain text version of this content. The canonical link for it is here.
Posted to dev@tomcat.apache.org by ma...@apache.org on 2019/11/20 13:25:47 UTC
[tomcat] 02/05: Refactor APR Poller to remove use of multiple
pollsets
This is an automated email from the ASF dual-hosted git repository.
markt pushed a commit to branch 7.0.x
in repository https://gitbox.apache.org/repos/asf/tomcat.git
View the commit online:
https://github.com/apache/tomcat/commit/7e853199d4ce0db4cc173f80b39a616890b64dee
commit 7e853199d4ce0db4cc173f80b39a616890b64dee
Author: Mark Thomas <ma...@apache.org>
AuthorDate: Tue Nov 19 17:03:28 2019 +0000
Refactor APR Poller to remove use of multiple pollsets
---
java/org/apache/tomcat/util/net/AprEndpoint.java | 465 +++++++++--------------
webapps/docs/changelog.xml | 5 +
2 files changed, 187 insertions(+), 283 deletions(-)
diff --git a/java/org/apache/tomcat/util/net/AprEndpoint.java b/java/org/apache/tomcat/util/net/AprEndpoint.java
index a31430d..dfca17b 100644
--- a/java/org/apache/tomcat/util/net/AprEndpoint.java
+++ b/java/org/apache/tomcat/util/net/AprEndpoint.java
@@ -1409,39 +1409,17 @@ public class AprEndpoint extends AbstractEndpoint<Long> {
// ------------------------------------------------------ Poller Inner Class
- public class Poller implements Runnable {
+ public class Poller implements Runnable {
/**
- * Pointers to the pollers.
+ * Pointer to the poller.
*/
- protected long[] pollers = null;
+ private long aprPoller;
/**
* Actual poller size.
*/
- protected int actualPollerSize = 0;
-
- /**
- * Amount of spots left in the poller.
- */
- protected int[] pollerSpace = null;
-
- /**
- * Amount of low level pollers in use by this poller.
- */
- protected int pollerCount;
-
- /**
- * Timeout value for the poll call.
- */
- protected int pollerTime;
-
- /**
- * Variable poller timeout that adjusts depending on how many poll sets
- * are in use so that the total poll time across all poll sets remains
- * equal to pollTime.
- */
- private int nextPollerTime;
+ private int pollerSize = 0;
/**
* Root pool.
@@ -1495,60 +1473,23 @@ public class AprEndpoint extends AbstractEndpoint<Long> {
private volatile boolean pollerRunning = true;
/**
- * Create the poller. With some versions of APR, the maximum poller size
- * will be 62 (recompiling APR is necessary to remove this limitation).
+ * Create the poller.
*/
protected void init() {
pool = Pool.create(serverSockPool);
-
- // Single poller by default
- int defaultPollerSize = getMaxConnections();
-
- if ((OS.IS_WIN32 || OS.IS_WIN64) && (defaultPollerSize > 1024)) {
- // The maximum per poller to get reasonable performance is 1024
- // Adjust poller size so that it won't reach the limit. This is
- // a limitation of XP / Server 2003 that has been fixed in
- // Vista / Server 2008 onwards.
- actualPollerSize = 1024;
- } else {
- actualPollerSize = defaultPollerSize;
- }
-
- timeouts = new SocketTimeouts(defaultPollerSize);
+ pollerSize = getMaxConnections();
+ timeouts = new SocketTimeouts(pollerSize);
// At the moment, setting the timeout is useless, but it could get
// used again as the normal poller could be faster using maintain.
// It might not be worth bothering though.
- long pollset = allocatePoller(actualPollerSize, pool, -1);
- if (pollset == 0 && actualPollerSize > 1024) {
- actualPollerSize = 1024;
- pollset = allocatePoller(actualPollerSize, pool, -1);
- }
- if (pollset == 0) {
- actualPollerSize = 62;
- pollset = allocatePoller(actualPollerSize, pool, -1);
- }
-
- pollerCount = defaultPollerSize / actualPollerSize;
- pollerTime = pollTime / pollerCount;
- nextPollerTime = pollerTime;
-
- pollers = new long[pollerCount];
- pollers[0] = pollset;
- for (int i = 1; i < pollerCount; i++) {
- pollers[i] = allocatePoller(actualPollerSize, pool, -1);
- }
-
- pollerSpace = new int[pollerCount];
- for (int i = 0; i < pollerCount; i++) {
- pollerSpace[i] = actualPollerSize;
- }
+ aprPoller = allocatePoller(pollerSize, pool, -1);
- desc = new long[actualPollerSize * 2];
+ desc = new long[pollerSize * 2];
connectionCount.set(0);
- addList = new SocketList(defaultPollerSize);
- closeList = new SocketList(defaultPollerSize);
+ addList = new SocketList(pollerSize);
+ closeList = new SocketList(pollerSize);
}
@@ -1590,15 +1531,13 @@ public class AprEndpoint extends AbstractEndpoint<Long> {
}
addList.clear();
// Close all sockets still in the poller
- for (int i = 0; i < pollerCount; i++) {
- int rv = Poll.pollset(pollers[i], desc);
- if (rv > 0) {
- for (int n = 0; n < rv; n++) {
- boolean comet = connections.get(
- Long.valueOf(desc[n*2+1])).isComet();
- if (!comet || !processSocket(desc[n*2+1], SocketStatus.STOP)) {
- destroySocket(desc[n*2+1]);
- }
+ int rv = Poll.pollset(aprPoller, desc);
+ if (rv > 0) {
+ for (int n = 0; n < rv; n++) {
+ boolean comet = connections.get(
+ Long.valueOf(desc[n*2+1])).isComet();
+ if (!comet || !processSocket(desc[n*2+1], SocketStatus.STOP)) {
+ destroySocket(desc[n*2+1]);
}
}
}
@@ -1669,16 +1608,10 @@ public class AprEndpoint extends AbstractEndpoint<Long> {
* {@link Poller#run()}.
*/
protected boolean addToPoller(long socket, int events) {
- int rv = -1;
- for (int i = 0; i < pollers.length; i++) {
- if (pollerSpace[i] > 0) {
- rv = Poll.add(pollers[i], socket, events);
- if (rv == Status.APR_SUCCESS) {
- pollerSpace[i]--;
- connectionCount.incrementAndGet();
- return true;
- }
- }
+ int rv = Poll.add(aprPoller, socket, events);
+ if (rv == Status.APR_SUCCESS) {
+ connectionCount.incrementAndGet();
+ return true;
}
return false;
}
@@ -1708,19 +1641,12 @@ public class AprEndpoint extends AbstractEndpoint<Long> {
log.debug(sm.getString("endpoint.debug.pollerRemove",
Long.valueOf(socket)));
}
- int rv = -1;
- for (int i = 0; i < pollers.length; i++) {
- if (pollerSpace[i] < actualPollerSize) {
- rv = Poll.remove(pollers[i], socket);
- if (rv != Status.APR_NOTFOUND) {
- pollerSpace[i]++;
- connectionCount.decrementAndGet();
- if (log.isDebugEnabled()) {
- log.debug(sm.getString("endpoint.debug.pollerRemoved",
- Long.valueOf(socket)));
- }
- break;
- }
+ int rv = Poll.remove(aprPoller, socket);
+ if (rv != Status.APR_NOTFOUND) {
+ connectionCount.decrementAndGet();
+ if (log.isDebugEnabled()) {
+ log.debug(sm.getString("endpoint.debug.pollerRemoved",
+ Long.valueOf(socket)));
}
}
timeouts.remove(socket);
@@ -1765,15 +1691,13 @@ public class AprEndpoint extends AbstractEndpoint<Long> {
public String toString() {
StringBuffer buf = new StringBuffer();
buf.append("Poller");
- long[] res = new long[actualPollerSize * 2];
- for (int i = 0; i < pollers.length; i++) {
- int count = Poll.pollset(pollers[i], res);
- buf.append(" [ ");
- for (int j = 0; j < count; j++) {
- buf.append(desc[2*j+1]).append(" ");
- }
- buf.append("]");
+ long[] res = new long[pollerSize * 2];
+ int count = Poll.pollset(aprPoller, res);
+ buf.append(" [ ");
+ for (int j = 0; j < count; j++) {
+ buf.append(desc[2*j+1]).append(" ");
}
+ buf.append("]");
return buf.toString();
}
@@ -1909,126 +1833,100 @@ public class AprEndpoint extends AbstractEndpoint<Long> {
}
}
- // Poll for the specified interval
- for (int i = 0; i < pollers.length; i++) {
-
- // Flags to ask to reallocate the pool
- boolean reset = false;
- //ArrayList<Long> skip = null;
-
- int rv = 0;
- // Reset the nextPollerTime
- nextPollerTime = pollerTime;
- // Iterate on each pollers, but no need to poll empty pollers
- if (pollerSpace[i] < actualPollerSize) {
- rv = Poll.poll(pollers[i], nextPollerTime, desc, true);
- // Reset the nextPollerTime
- nextPollerTime = pollerTime;
- } else {
- // Skipping an empty poll set means skipping a wait
- // time of pollerTime microseconds. If most of the
- // poll sets are skipped then this loop will be
- // tighter than expected which could lead to higher
- // than expected CPU usage. Extending the
- // nextPollerTime ensures that this loop always
- // takes about the same time to execute.
- nextPollerTime += pollerTime;
- }
- if (rv > 0) {
- pollerSpace[i] += rv;
- connectionCount.addAndGet(-rv);
- for (int n = 0; n < rv; n++) {
- if (getLog().isDebugEnabled()) {
- log.debug(sm.getString(
- "endpoint.debug.pollerProcess",
- Long.valueOf(desc[n*2+1]),
+ // Flag to ask to reallocate the pool
+ boolean reset = false;
+
+ int rv = Poll.poll(aprPoller, pollTime, desc, true);
+ if (rv > 0) {
+ connectionCount.addAndGet(-rv);
+ for (int n = 0; n < rv; n++) {
+ if (getLog().isDebugEnabled()) {
+ log.debug(sm.getString(
+ "endpoint.debug.pollerProcess",
+ Long.valueOf(desc[n*2+1]),
+ Long.valueOf(desc[n*2])));
+ }
+ long timeout = timeouts.remove(desc[n*2+1]);
+ AprSocketWrapper wrapper = connections.get(
+ Long.valueOf(desc[n*2+1]));
+ if (wrapper == null) {
+ // Socket was closed in another thread while still in
+ // the Poller but wasn't removed from the Poller before
+ // new data arrived.
+ continue;
+ }
+ wrapper.pollerFlags = wrapper.pollerFlags & ~((int) desc[n*2]);
+ // Check for failed sockets and hand this socket off to a worker
+ if (wrapper.isComet()) {
+ // Event processes either a read or a write depending on what the poller returns
+ if (((desc[n*2] & Poll.APR_POLLHUP) == Poll.APR_POLLHUP)
+ || ((desc[n*2] & Poll.APR_POLLERR) == Poll.APR_POLLERR)
+ || ((desc[n*2] & Poll.APR_POLLNVAL) == Poll.APR_POLLNVAL)) {
+ if (!processSocket(desc[n*2+1], SocketStatus.ERROR)) {
+ // Close socket and clear pool
+ closeSocket(desc[n*2+1]);
+ }
+ } else if ((desc[n*2] & Poll.APR_POLLIN) == Poll.APR_POLLIN) {
+ if (wrapper.pollerFlags != 0) {
+ add(desc[n*2+1], 1, wrapper.pollerFlags);
+ }
+ if (!processSocket(desc[n*2+1], SocketStatus.OPEN_READ)) {
+ // Close socket and clear pool
+ closeSocket(desc[n*2+1]);
+ }
+ } else if ((desc[n*2] & Poll.APR_POLLOUT) == Poll.APR_POLLOUT) {
+ if (wrapper.pollerFlags != 0) {
+ add(desc[n*2+1], 1, wrapper.pollerFlags);
+ }
+ if (!processSocket(desc[n*2+1], SocketStatus.OPEN_WRITE)) {
+ // Close socket and clear pool
+ closeSocket(desc[n*2+1]);
+ }
+ } else {
+ // Unknown event
+ getLog().warn(sm.getString(
+ "endpoint.apr.pollUnknownEvent",
Long.valueOf(desc[n*2])));
+ if (!processSocket(desc[n*2+1], SocketStatus.ERROR)) {
+ // Close socket and clear pool
+ closeSocket(desc[n*2+1]);
+ }
}
- long timeout = timeouts.remove(desc[n*2+1]);
- AprSocketWrapper wrapper = connections.get(
- Long.valueOf(desc[n*2+1]));
- if (wrapper == null) {
- // Socket was closed in another thread while still in
- // the Poller but wasn't removed from the Poller before
- // new data arrived.
- continue;
- }
- wrapper.pollerFlags = wrapper.pollerFlags & ~((int) desc[n*2]);
- // Check for failed sockets and hand this socket off to a worker
- if (wrapper.isComet()) {
- // Event processes either a read or a write depending on what the poller returns
- if (((desc[n*2] & Poll.APR_POLLHUP) == Poll.APR_POLLHUP)
- || ((desc[n*2] & Poll.APR_POLLERR) == Poll.APR_POLLERR)
- || ((desc[n*2] & Poll.APR_POLLNVAL) == Poll.APR_POLLNVAL)) {
- if (!processSocket(desc[n*2+1], SocketStatus.ERROR)) {
- // Close socket and clear pool
- closeSocket(desc[n*2+1]);
- }
- } else if ((desc[n*2] & Poll.APR_POLLIN) == Poll.APR_POLLIN) {
- if (wrapper.pollerFlags != 0) {
- add(desc[n*2+1], 1, wrapper.pollerFlags);
- }
+ } else if (((desc[n*2] & Poll.APR_POLLHUP) == Poll.APR_POLLHUP)
+ || ((desc[n*2] & Poll.APR_POLLERR) == Poll.APR_POLLERR)
+ || ((desc[n*2] & Poll.APR_POLLNVAL) == Poll.APR_POLLNVAL)) {
+ if (wrapper.isUpgraded()) {
+ // Using non-blocking IO. Need to trigger error handling.
+ // Poller may return error codes plus the flags it was
+ // waiting for or it may just return an error code. By
+ // signalling read/write is possible, a read/write will be
+ // attempted, fail and that will trigger an exception the
+ // application will see.
+ // Check the return flags first, followed by what the socket
+ // was registered for
+ if ((desc[n*2] & Poll.APR_POLLIN) == Poll.APR_POLLIN) {
+ // Error probably occurred during a non-blocking read
if (!processSocket(desc[n*2+1], SocketStatus.OPEN_READ)) {
// Close socket and clear pool
closeSocket(desc[n*2+1]);
}
} else if ((desc[n*2] & Poll.APR_POLLOUT) == Poll.APR_POLLOUT) {
- if (wrapper.pollerFlags != 0) {
- add(desc[n*2+1], 1, wrapper.pollerFlags);
- }
+ // Error probably occurred during a non-blocking write
if (!processSocket(desc[n*2+1], SocketStatus.OPEN_WRITE)) {
// Close socket and clear pool
closeSocket(desc[n*2+1]);
}
- } else {
- // Unknown event
- getLog().warn(sm.getString(
- "endpoint.apr.pollUnknownEvent",
- Long.valueOf(desc[n*2])));
- if (!processSocket(desc[n*2+1], SocketStatus.ERROR)) {
+ } else if ((wrapper.pollerFlags & Poll.APR_POLLIN) == Poll.APR_POLLIN) {
+ // Can't tell what was happening when the error occurred but the
+ // socket is registered for non-blocking read so use that
+ if (!processSocket(desc[n*2+1], SocketStatus.OPEN_READ)) {
// Close socket and clear pool
closeSocket(desc[n*2+1]);
}
- }
- } else if (((desc[n*2] & Poll.APR_POLLHUP) == Poll.APR_POLLHUP)
- || ((desc[n*2] & Poll.APR_POLLERR) == Poll.APR_POLLERR)
- || ((desc[n*2] & Poll.APR_POLLNVAL) == Poll.APR_POLLNVAL)) {
- if (wrapper.isUpgraded()) {
- // Using non-blocking IO. Need to trigger error handling.
- // Poller may return error codes plus the flags it was
- // waiting for or it may just return an error code. By
- // signalling read/write is possible, a read/write will be
- // attempted, fail and that will trigger an exception the
- // application will see.
- // Check the return flags first, followed by what the socket
- // was registered for
- if ((desc[n*2] & Poll.APR_POLLIN) == Poll.APR_POLLIN) {
- // Error probably occurred during a non-blocking read
- if (!processSocket(desc[n*2+1], SocketStatus.OPEN_READ)) {
- // Close socket and clear pool
- closeSocket(desc[n*2+1]);
- }
- } else if ((desc[n*2] & Poll.APR_POLLOUT) == Poll.APR_POLLOUT) {
- // Error probably occurred during a non-blocking write
- if (!processSocket(desc[n*2+1], SocketStatus.OPEN_WRITE)) {
- // Close socket and clear pool
- closeSocket(desc[n*2+1]);
- }
- } else if ((wrapper.pollerFlags & Poll.APR_POLLIN) == Poll.APR_POLLIN) {
- // Can't tell what was happening when the error occurred but the
- // socket is registered for non-blocking read so use that
- if (!processSocket(desc[n*2+1], SocketStatus.OPEN_READ)) {
- // Close socket and clear pool
- closeSocket(desc[n*2+1]);
- }
- } else if ((wrapper.pollerFlags & Poll.APR_POLLOUT) == Poll.APR_POLLOUT) {
- // Can't tell what was happening when the error occurred but the
- // socket is registered for non-blocking write so use that
- if (!processSocket(desc[n*2+1], SocketStatus.OPEN_WRITE)) {
- // Close socket and clear pool
- closeSocket(desc[n*2+1]);
- }
- } else {
+ } else if ((wrapper.pollerFlags & Poll.APR_POLLOUT) == Poll.APR_POLLOUT) {
+ // Can't tell what was happening when the error occurred but the
+ // socket is registered for non-blocking write so use that
+ if (!processSocket(desc[n*2+1], SocketStatus.OPEN_WRITE)) {
// Close socket and clear pool
closeSocket(desc[n*2+1]);
}
@@ -2036,82 +1934,83 @@ public class AprEndpoint extends AbstractEndpoint<Long> {
// Close socket and clear pool
closeSocket(desc[n*2+1]);
}
- } else if (((desc[n*2] & Poll.APR_POLLIN) == Poll.APR_POLLIN)
- || ((desc[n*2] & Poll.APR_POLLOUT) == Poll.APR_POLLOUT)) {
- boolean error = false;
- if (((desc[n*2] & Poll.APR_POLLIN) == Poll.APR_POLLIN) &&
- !processSocket(desc[n*2+1], SocketStatus.OPEN_READ)) {
- error = true;
- // Close socket and clear pool
- closeSocket(desc[n*2+1]);
- }
- if (!error &&
- ((desc[n*2] & Poll.APR_POLLOUT) == Poll.APR_POLLOUT) &&
- !processSocket(desc[n*2+1], SocketStatus.OPEN_WRITE)) {
- // Close socket and clear pool
- error = true;
- closeSocket(desc[n*2+1]);
- }
- if (!error && wrapper.pollerFlags != 0) {
- // If socket was registered for multiple events but
- // only some of the occurred, re-register for the
- // remaining events.
- // timeout is the value of System.currentTimeMillis() that
- // was set as the point that the socket will timeout. When
- // adding to the poller, the timeout from now in
- // milliseconds is required.
- // So first, subtract the current timestamp
- if (timeout > 0) {
- timeout = timeout - System.currentTimeMillis();
- }
- // If the socket should have already expired by now,
- // re-add it with a very short timeout
- if (timeout <= 0) {
- timeout = 1;
- }
- // Should be impossible but just in case since timeout will
- // be cast to an int.
- if (timeout > Integer.MAX_VALUE) {
- timeout = Integer.MAX_VALUE;
- }
- add(desc[n*2+1], (int) timeout, wrapper.pollerFlags);
- }
} else {
- // Unknown event
- getLog().warn(sm.getString(
- "endpoint.apr.pollUnknownEvent",
- Long.valueOf(desc[n*2])));
// Close socket and clear pool
closeSocket(desc[n*2+1]);
}
- }
- } else if (rv < 0) {
- int errn = -rv;
- // Any non timeup or interrupted error is critical
- if ((errn != Status.TIMEUP) && (errn != Status.EINTR)) {
- if (errn > Status.APR_OS_START_USERERR) {
- errn -= Status.APR_OS_START_USERERR;
+ } else if (((desc[n*2] & Poll.APR_POLLIN) == Poll.APR_POLLIN)
+ || ((desc[n*2] & Poll.APR_POLLOUT) == Poll.APR_POLLOUT)) {
+ boolean error = false;
+ if (((desc[n*2] & Poll.APR_POLLIN) == Poll.APR_POLLIN) &&
+ !processSocket(desc[n*2+1], SocketStatus.OPEN_READ)) {
+ error = true;
+ // Close socket and clear pool
+ closeSocket(desc[n*2+1]);
+ }
+ if (!error &&
+ ((desc[n*2] & Poll.APR_POLLOUT) == Poll.APR_POLLOUT) &&
+ !processSocket(desc[n*2+1], SocketStatus.OPEN_WRITE)) {
+ // Close socket and clear pool
+ error = true;
+ closeSocket(desc[n*2+1]);
+ }
+ if (!error && wrapper.pollerFlags != 0) {
+ // If socket was registered for multiple events but
+ // only some of the occurred, re-register for the
+ // remaining events.
+ // timeout is the value of System.currentTimeMillis() that
+ // was set as the point that the socket will timeout. When
+ // adding to the poller, the timeout from now in
+ // milliseconds is required.
+ // So first, subtract the current timestamp
+ if (timeout > 0) {
+ timeout = timeout - System.currentTimeMillis();
+ }
+ // If the socket should have already expired by now,
+ // re-add it with a very short timeout
+ if (timeout <= 0) {
+ timeout = 1;
+ }
+ // Should be impossible but just in case since timeout will
+ // be cast to an int.
+ if (timeout > Integer.MAX_VALUE) {
+ timeout = Integer.MAX_VALUE;
+ }
+ add(desc[n*2+1], (int) timeout, wrapper.pollerFlags);
}
- getLog().error(sm.getString(
- "endpoint.apr.pollError",
- Integer.valueOf(errn),
- Error.strerror(errn)));
- // Destroy and reallocate the poller
- reset = true;
+ } else {
+ // Unknown event
+ getLog().warn(sm.getString(
+ "endpoint.apr.pollUnknownEvent",
+ Long.valueOf(desc[n*2])));
+ // Close socket and clear pool
+ closeSocket(desc[n*2+1]);
}
}
-
- if (reset && pollerRunning) {
- // Reallocate the current poller
- int count = Poll.pollset(pollers[i], desc);
- long newPoller = allocatePoller(actualPollerSize, pool, -1);
- // Don't restore connections for now, since I have not tested it
- pollerSpace[i] = actualPollerSize;
- connectionCount.addAndGet(-count);
- Poll.destroy(pollers[i]);
- pollers[i] = newPoller;
+ } else if (rv < 0) {
+ int errn = -rv;
+ // Any non timeup or interrupted error is critical
+ if ((errn != Status.TIMEUP) && (errn != Status.EINTR)) {
+ if (errn > Status.APR_OS_START_USERERR) {
+ errn -= Status.APR_OS_START_USERERR;
+ }
+ getLog().error(sm.getString(
+ "endpoint.apr.pollError",
+ Integer.valueOf(errn),
+ Error.strerror(errn)));
+ // Destroy and reallocate the poller
+ reset = true;
}
+ }
+ if (reset && pollerRunning) {
+ // Reallocate the current poller
+ int count = Poll.pollset(aprPoller, desc);
+ long newPoller = allocatePoller(pollerSize, pool, -1);
+ // Don't restore connections for now, since I have not tested it
+ connectionCount.addAndGet(-count);
+ Poll.destroy(aprPoller);
+ aprPoller = newPoller;
}
} catch (Throwable t) {
ExceptionUtils.handleThrowable(t);
diff --git a/webapps/docs/changelog.xml b/webapps/docs/changelog.xml
index 5b5ee4f..75ef806 100644
--- a/webapps/docs/changelog.xml
+++ b/webapps/docs/changelog.xml
@@ -117,6 +117,11 @@
request header to use the shared parsing code and reduce duplication.
(markt)
</fix>
+ <scode>
+ Refactor the APR poller to always use a single pollset now that the
+ Windows operating systems that required multiple smaller pollsets to be
+ used are no longer supported. (markt)
+ </scode>
</changelog>
</subsection>
<subsection name="Jasper">
---------------------------------------------------------------------
To unsubscribe, e-mail: dev-unsubscribe@tomcat.apache.org
For additional commands, e-mail: dev-help@tomcat.apache.org