diff --git a/jetty-client/src/main/java/org/eclipse/jetty/client/AbstractConnectionPool.java b/jetty-client/src/main/java/org/eclipse/jetty/client/AbstractConnectionPool.java index 7d4258d41240..6ec1ee474254 100644 --- a/jetty-client/src/main/java/org/eclipse/jetty/client/AbstractConnectionPool.java +++ b/jetty-client/src/main/java/org/eclipse/jetty/client/AbstractConnectionPool.java @@ -78,7 +78,7 @@ public CompletableFuture preCreateConnections(int connectionCount) CompletableFuture[] futures = new CompletableFuture[connectionCount]; for (int i = 0; i < connectionCount; i++) { - futures[i] = tryCreateReturningFuture(pool.getMaxEntries()); + futures[i] = tryCreateAsync(getMaxConnectionCount()); } return CompletableFuture.allOf(futures); } @@ -175,6 +175,8 @@ public Connection acquire() */ protected Connection acquire(boolean create) { + if (LOG.isDebugEnabled()) + LOG.debug("Acquiring create={} on {}", create, this); Connection connection = activate(); if (connection == null && create) { @@ -196,20 +198,21 @@ protected Connection acquire(boolean create) */ protected void tryCreate(int maxPending) { - tryCreateReturningFuture(maxPending); + tryCreateAsync(maxPending); } - private CompletableFuture tryCreateReturningFuture(int maxPending) + private CompletableFuture tryCreateAsync(int maxPending) { + int connectionCount = getConnectionCount(); if (LOG.isDebugEnabled()) - LOG.debug("tryCreate {}/{} connections {}/{} pending", pool.size(), pool.getMaxEntries(), getPendingConnectionCount(), maxPending); + LOG.debug("Try creating connection {}/{} with {}/{} pending", connectionCount, getMaxConnectionCount(), getPendingConnectionCount(), maxPending); Pool.Entry entry = pool.reserve(maxPending); if (entry == null) return CompletableFuture.completedFuture(null); if (LOG.isDebugEnabled()) - LOG.debug("newConnection {}/{} connections {}/{} pending", pool.size(), pool.getMaxEntries(), getPendingConnectionCount(), maxPending); + LOG.debug("Creating connection {}/{}", connectionCount, getMaxConnectionCount()); CompletableFuture future = new CompletableFuture<>(); destination.newConnection(new Promise() @@ -218,7 +221,7 @@ private CompletableFuture tryCreateReturningFuture(int maxPending) public void succeeded(Connection connection) { if (LOG.isDebugEnabled()) - LOG.debug("Connection {}/{} creation succeeded {}", pool.size(), pool.getMaxEntries(), connection); + LOG.debug("Connection {}/{} creation succeeded {}", connectionCount, getMaxConnectionCount(), connection); if (!(connection instanceof Attachable)) { failed(new IllegalArgumentException("Invalid connection object: " + connection)); @@ -236,7 +239,7 @@ public void succeeded(Connection connection) public void failed(Throwable x) { if (LOG.isDebugEnabled()) - LOG.debug("Connection " + pool.size() + "/" + pool.getMaxEntries() + " creation failed", x); + LOG.debug("Connection {}/{} creation failed", connectionCount, getMaxConnectionCount(), x); entry.remove(); future.completeExceptionally(x); requester.failed(x); @@ -257,7 +260,7 @@ protected Connection activate() if (entry != null) { if (LOG.isDebugEnabled()) - LOG.debug("activated {}", entry); + LOG.debug("Activated {} {}", entry, pool); Connection connection = entry.getPooled(); acquired(connection); return connection; @@ -275,8 +278,6 @@ public boolean isActive(Connection connection) Pool.Entry entry = (Pool.Entry)attachable.getAttachment(); if (entry == null) return false; - if (LOG.isDebugEnabled()) - LOG.debug("isActive {}", entry); return !entry.isIdle(); } @@ -300,7 +301,7 @@ protected boolean deactivate(Connection connection) return true; boolean reusable = pool.release(entry); if (LOG.isDebugEnabled()) - LOG.debug("Released ({}) {}", reusable, entry); + LOG.debug("Released ({}) {} {}", reusable, entry, pool); if (reusable) return true; remove(connection); @@ -325,7 +326,7 @@ protected boolean remove(Connection connection, boolean force) attachable.setAttachment(null); boolean removed = pool.remove(entry); if (LOG.isDebugEnabled()) - LOG.debug("Removed ({}) {}", removed, entry); + LOG.debug("Removed ({}) {} {}", removed, entry, pool); if (removed || force) { released(connection); diff --git a/jetty-client/src/main/java/org/eclipse/jetty/client/HttpDestination.java b/jetty-client/src/main/java/org/eclipse/jetty/client/HttpDestination.java index 59ff82753f8e..6ca7165b350c 100644 --- a/jetty-client/src/main/java/org/eclipse/jetty/client/HttpDestination.java +++ b/jetty-client/src/main/java/org/eclipse/jetty/client/HttpDestination.java @@ -384,7 +384,7 @@ private ProcessResult process(Connection connection) { // Aggressively send other queued requests // in case connections are multiplexed. - return getHttpExchanges().size() > 0 ? ProcessResult.CONTINUE : ProcessResult.FINISH; + return getQueuedRequestCount() > 0 ? ProcessResult.CONTINUE : ProcessResult.FINISH; } if (LOG.isDebugEnabled()) @@ -438,7 +438,7 @@ public void release(Connection connection) { if (connectionPool.isActive(connection)) { - // trigger the next request after releasing the connection + // Trigger the next request after releasing the connection. if (connectionPool.release(connection)) send(false); else diff --git a/jetty-client/src/main/java/org/eclipse/jetty/client/http/HttpChannelOverHTTP.java b/jetty-client/src/main/java/org/eclipse/jetty/client/http/HttpChannelOverHTTP.java index fa358b0d0374..3d6904b1488b 100644 --- a/jetty-client/src/main/java/org/eclipse/jetty/client/http/HttpChannelOverHTTP.java +++ b/jetty-client/src/main/java/org/eclipse/jetty/client/http/HttpChannelOverHTTP.java @@ -40,7 +40,6 @@ public class HttpChannelOverHTTP extends HttpChannel private final HttpConnectionOverHTTP connection; private final HttpSenderOverHTTP sender; private final HttpReceiverOverHTTP receiver; - private final LongAdder inMessages = new LongAdder(); private final LongAdder outMessages = new LongAdder(); public HttpChannelOverHTTP(HttpConnectionOverHTTP connection) @@ -129,7 +128,6 @@ public Result exchangeTerminating(HttpExchange exchange, Result result) public void receive() { - inMessages.increment(); receiver.receive(); } @@ -185,7 +183,7 @@ else if (sender.isShutdown()) protected long getMessagesIn() { - return inMessages.longValue(); + return receiver.getMessagesIn(); } protected long getMessagesOut() diff --git a/jetty-client/src/main/java/org/eclipse/jetty/client/http/HttpReceiverOverHTTP.java b/jetty-client/src/main/java/org/eclipse/jetty/client/http/HttpReceiverOverHTTP.java index e5bf9eec51b9..b6350461cd27 100644 --- a/jetty-client/src/main/java/org/eclipse/jetty/client/http/HttpReceiverOverHTTP.java +++ b/jetty-client/src/main/java/org/eclipse/jetty/client/http/HttpReceiverOverHTTP.java @@ -20,6 +20,7 @@ import java.io.EOFException; import java.nio.ByteBuffer; +import java.util.concurrent.atomic.LongAdder; import org.eclipse.jetty.client.HttpClient; import org.eclipse.jetty.client.HttpExchange; @@ -40,6 +41,7 @@ public class HttpReceiverOverHTTP extends HttpReceiver implements HttpParser.ResponseHandler { + private final LongAdder inMessages = new LongAdder(); private final HttpParser parser; private RetainableByteBuffer networkBuffer; private boolean shutdown; @@ -333,9 +335,11 @@ public boolean messageComplete() return false; int status = exchange.getResponse().getStatus(); - if (status != HttpStatus.CONTINUE_100) + { + inMessages.increment(); complete = true; + } return !responseSuccess(exchange); } @@ -376,6 +380,11 @@ private void failAndClose(Throwable failure) getHttpConnection().close(failure); } + long getMessagesIn() + { + return inMessages.longValue(); + } + @Override public String toString() { diff --git a/jetty-client/src/test/java/org/eclipse/jetty/client/ConnectionPoolTest.java b/jetty-client/src/test/java/org/eclipse/jetty/client/ConnectionPoolTest.java index 56d13899941a..334b9b6a6d6e 100644 --- a/jetty-client/src/test/java/org/eclipse/jetty/client/ConnectionPoolTest.java +++ b/jetty-client/src/test/java/org/eclipse/jetty/client/ConnectionPoolTest.java @@ -48,7 +48,6 @@ import org.eclipse.jetty.util.thread.QueuedThreadPool; import org.hamcrest.Matchers; import org.junit.jupiter.api.AfterEach; -import org.junit.jupiter.api.Assumptions; import org.junit.jupiter.params.ParameterizedTest; import org.junit.jupiter.params.provider.MethodSource; @@ -63,10 +62,15 @@ public class ConnectionPoolTest private HttpClient client; public static Stream pools() + { + return Stream.concat(poolsNoRoundRobin(), + Stream.of(new ConnectionPoolFactory("round-robin", destination -> new RoundRobinConnectionPool(destination, destination.getHttpClient().getMaxConnectionsPerDestination(), destination)))); + } + + public static Stream poolsNoRoundRobin() { return Stream.of( new ConnectionPoolFactory("duplex", destination -> new DuplexConnectionPool(destination, destination.getHttpClient().getMaxConnectionsPerDestination(), destination)), - new ConnectionPoolFactory("round-robin", destination -> new RoundRobinConnectionPool(destination, destination.getHttpClient().getMaxConnectionsPerDestination(), destination)), new ConnectionPoolFactory("multiplex", destination -> new MultiplexConnectionPool(destination, destination.getHttpClient().getMaxConnectionsPerDestination(), destination, 1)) ); } @@ -295,11 +299,11 @@ public void resolve(String host, int port, Promise> prom } @ParameterizedTest - @MethodSource("pools") + @MethodSource("poolsNoRoundRobin") public void testConcurrentRequestsDontOpenTooManyConnections(ConnectionPoolFactory factory) throws Exception { - // Round robin connection pool does open a few more connections than expected. - Assumptions.assumeFalse(factory.name.equals("round-robin")); + // Round robin connection pool does open a few more + // connections than expected, exclude it from this test. startServer(new EmptyServerHandler()); diff --git a/jetty-io/src/main/java/org/eclipse/jetty/io/IncludeExcludeConnectionStatistics.java b/jetty-io/src/main/java/org/eclipse/jetty/io/IncludeExcludeConnectionStatistics.java new file mode 100644 index 000000000000..364a48726cfd --- /dev/null +++ b/jetty-io/src/main/java/org/eclipse/jetty/io/IncludeExcludeConnectionStatistics.java @@ -0,0 +1,114 @@ +// +// ======================================================================== +// Copyright (c) 1995-2020 Mort Bay Consulting Pty Ltd and others. +// ------------------------------------------------------------------------ +// All rights reserved. This program and the accompanying materials +// are made available under the terms of the Eclipse Public License v1.0 +// and Apache License v2.0 which accompanies this distribution. +// +// The Eclipse Public License is available at +// http://www.eclipse.org/legal/epl-v10.html +// +// The Apache License v2.0 is available at +// http://www.opensource.org/licenses/apache2.0.php +// +// You may elect to redistribute this code under either of these licenses. +// ======================================================================== +// + +package org.eclipse.jetty.io; + +import java.util.AbstractSet; +import java.util.HashSet; +import java.util.Iterator; +import java.util.Set; +import java.util.function.Predicate; + +import org.eclipse.jetty.util.IncludeExcludeSet; + +public class IncludeExcludeConnectionStatistics extends ConnectionStatistics +{ + private final IncludeExcludeSet, Connection> _set = new IncludeExcludeSet<>(ConnectionSet.class); + + public void include(String className) throws ClassNotFoundException + { + _set.include(connectionForName(className)); + } + + public void include(Class clazz) + { + _set.include(clazz); + } + + public void exclude(String className) throws ClassNotFoundException + { + _set.exclude(connectionForName(className)); + } + + public void exclude(Class clazz) + { + _set.exclude(clazz); + } + + private Class connectionForName(String className) throws ClassNotFoundException + { + Class aClass = Class.forName(className); + if (!Connection.class.isAssignableFrom(aClass)) + throw new IllegalArgumentException("Class is not a Connection"); + + @SuppressWarnings("unchecked") + Class connectionClass = (Class)aClass; + return connectionClass; + } + + @Override + public void onOpened(Connection connection) + { + if (_set.test(connection)) + super.onOpened(connection); + } + + @Override + public void onClosed(Connection connection) + { + if (_set.test(connection)) + super.onClosed(connection); + } + + public static class ConnectionSet extends AbstractSet> implements Predicate + { + private final Set> set = new HashSet<>(); + + @Override + public boolean add(Class aClass) + { + return set.add(aClass); + } + + @Override + public boolean remove(Object o) + { + return set.remove(o); + } + + @Override + public Iterator> iterator() + { + return set.iterator(); + } + + @Override + public int size() + { + return set.size(); + } + + @Override + public boolean test(Connection connection) + { + if (connection == null) + return false; + return set.stream().anyMatch(c -> c.isAssignableFrom(connection.getClass())); + } + } +} diff --git a/jetty-proxy/src/test/java/org/eclipse/jetty/proxy/AsyncMiddleManServletTest.java b/jetty-proxy/src/test/java/org/eclipse/jetty/proxy/AsyncMiddleManServletTest.java index 847e4735be35..aa189e4928b4 100644 --- a/jetty-proxy/src/test/java/org/eclipse/jetty/proxy/AsyncMiddleManServletTest.java +++ b/jetty-proxy/src/test/java/org/eclipse/jetty/proxy/AsyncMiddleManServletTest.java @@ -78,6 +78,7 @@ import org.eclipse.jetty.util.log.StacklessLogging; import org.eclipse.jetty.util.thread.QueuedThreadPool; import org.junit.jupiter.api.AfterEach; +import org.junit.jupiter.api.Disabled; import org.junit.jupiter.api.Test; import org.junit.jupiter.api.condition.OS; @@ -978,6 +979,7 @@ public boolean transform(Source source, Sink sink) throws IOException assertArrayEquals(data, response.getContent()); } + @Disabled("See issue #3974") @Test public void testAfterContentTransformerOverflowingToDisk() throws Exception { diff --git a/jetty-server/src/main/java/org/eclipse/jetty/server/CustomRequestLog.java b/jetty-server/src/main/java/org/eclipse/jetty/server/CustomRequestLog.java index 5abc31e34700..59a9ca9453b1 100644 --- a/jetty-server/src/main/java/org/eclipse/jetty/server/CustomRequestLog.java +++ b/jetty-server/src/main/java/org/eclipse/jetty/server/CustomRequestLog.java @@ -30,6 +30,7 @@ import java.util.function.Supplier; import java.util.regex.Matcher; import java.util.regex.Pattern; +import java.util.stream.Collectors; import javax.servlet.http.Cookie; import javax.servlet.http.HttpServletRequest; @@ -274,18 +275,15 @@ public class CustomRequestLog extends ContainerLifeCycle implements RequestLog protected static final Logger LOG = Log.getLogger(CustomRequestLog.class); public static final String DEFAULT_DATE_FORMAT = "dd/MMM/yyyy:HH:mm:ss ZZZ"; - public static final String NCSA_FORMAT = "%{client}a - %u %t \"%r\" %s %O"; - public static final String EXTENDED_NCSA_FORMAT = "%{client}a - %u %t \"%r\" %s %O \"%{Referer}i\" \"%{User-Agent}i\""; - - private static ThreadLocal _buffers = ThreadLocal.withInitial(() -> new StringBuilder(256)); + public static final String EXTENDED_NCSA_FORMAT = NCSA_FORMAT + " \"%{Referer}i\" \"%{User-Agent}i\""; + private static final ThreadLocal _buffers = ThreadLocal.withInitial(() -> new StringBuilder(256)); - private String[] _ignorePaths; - private transient PathMappings _ignorePathMap; - - private RequestLog.Writer _requestLogWriter; + private final RequestLog.Writer _requestLogWriter; private final MethodHandle _logHandle; private final String _formatString; + private transient PathMappings _ignorePathMap; + private String[] _ignorePaths; public CustomRequestLog(RequestLog.Writer writer, String formatString) { @@ -445,8 +443,10 @@ private MethodHandle getLogHandle(String formatString) throws NoSuchMethodExcept { if (t.isLiteralString()) logHandle = updateLogHandle(logHandle, append, t.literal); - else + else if (t.isPercentCode()) logHandle = updateLogHandle(logHandle, append, lookup, t.code, t.arg, t.modifiers, t.negated); + else + throw new IllegalStateException("bad token " + t); } return logHandle; @@ -479,6 +479,7 @@ private static List getTokens(String formatString) String arg = m.group("ARG"); String modifierString = m.group("MOD"); + List modifiers = null; boolean negated = false; if (modifierString != null) { @@ -487,9 +488,14 @@ private static List getTokens(String formatString) modifierString = modifierString.substring(1); negated = true; } + + modifiers = new QuotedCSV(modifierString) + .getValues() + .stream() + .map(Integer::parseInt) + .collect(Collectors.toList()); } - List modifiers = new QuotedCSV(modifierString).getValues(); tokens.add(new Token(code, arg, modifiers, negated)); } else if (m.group("LITERAL") != null) @@ -517,18 +523,17 @@ private static class Token { public final String code; public final String arg; - public final List modifiers; + public final List modifiers; public final boolean negated; public final String literal; - public Token(String code, String arg, List modifiers, boolean negated) + public Token(String code, String arg, List modifiers, boolean negated) { this.code = code; this.arg = arg; this.modifiers = modifiers; this.negated = negated; - this.literal = null; } @@ -538,7 +543,6 @@ public Token(String literal) this.arg = null; this.modifiers = null; this.negated = false; - this.literal = literal; } @@ -553,26 +557,21 @@ public boolean isPercentCode() } } - private MethodHandle updateLogHandle(MethodHandle logHandle, MethodHandle append, String literal) + @SuppressWarnings("unused") + private static boolean modify(List modifiers, Boolean negated, StringBuilder b, Request request, Response response) { - return foldArguments(logHandle, dropArguments(dropArguments(append.bindTo(literal), 1, Request.class), 2, Response.class)); + if (negated) + return !modifiers.contains(response.getStatus()); + else + return modifiers.contains(response.getStatus()); } - //TODO use integer comparisons instead of strings - private static boolean modify(List modifiers, Boolean negated, StringBuilder b, Request request, Response response) + private MethodHandle updateLogHandle(MethodHandle logHandle, MethodHandle append, String literal) { - String responseCode = Integer.toString(response.getStatus()); - if (negated) - { - return (!modifiers.contains(responseCode)); - } - else - { - return (modifiers.contains(responseCode)); - } + return foldArguments(logHandle, dropArguments(dropArguments(append.bindTo(literal), 1, Request.class), 2, Response.class)); } - private MethodHandle updateLogHandle(MethodHandle logHandle, MethodHandle append, MethodHandles.Lookup lookup, String code, String arg, List modifiers, boolean negated) throws NoSuchMethodException, IllegalAccessException + private MethodHandle updateLogHandle(MethodHandle logHandle, MethodHandle append, MethodHandles.Lookup lookup, String code, String arg, List modifiers, boolean negated) throws NoSuchMethodException, IllegalAccessException { MethodType logType = methodType(void.class, StringBuilder.class, Request.class, Response.class); MethodType logTypeArg = methodType(void.class, String.class, StringBuilder.class, Request.class, Response.class); @@ -908,12 +907,12 @@ else if ("d".equals(arg)) if (modifiers != null && !modifiers.isEmpty()) { - MethodHandle modifierTest = lookup.findStatic(CustomRequestLog.class, "modify", methodType(Boolean.TYPE, List.class, Boolean.class, StringBuilder.class, Request.class, Response.class)); - MethodHandle dash = updateLogHandle(logHandle, append, "-"); MethodHandle log = foldArguments(logHandle, specificHandle); - modifierTest = modifierTest.bindTo(modifiers).bindTo(negated); + MethodHandle modifierTest = lookup.findStatic(CustomRequestLog.class, "modify", + methodType(Boolean.TYPE, List.class, Boolean.class, StringBuilder.class, Request.class, Response.class)); + modifierTest = modifierTest.bindTo(modifiers).bindTo(negated); return MethodHandles.guardWithTest(modifierTest, log, dash); } @@ -921,57 +920,67 @@ else if ("d".equals(arg)) } //-----------------------------------------------------------------------------------// - + @SuppressWarnings("unused") private static void logNothing(StringBuilder b, Request request, Response response) { } + @SuppressWarnings("unused") private static void logServerHost(StringBuilder b, Request request, Response response) { append(b, request.getServerName()); } + @SuppressWarnings("unused") private static void logClientHost(StringBuilder b, Request request, Response response) { append(b, request.getRemoteHost()); } + @SuppressWarnings("unused") private static void logLocalHost(StringBuilder b, Request request, Response response) { append(b, request.getHttpChannel().getEndPoint().getLocalAddress().getAddress().getHostAddress()); } + @SuppressWarnings("unused") private static void logRemoteHost(StringBuilder b, Request request, Response response) { append(b, request.getHttpChannel().getEndPoint().getRemoteAddress().getAddress().getHostAddress()); } + @SuppressWarnings("unused") private static void logServerPort(StringBuilder b, Request request, Response response) { b.append(request.getServerPort()); } + @SuppressWarnings("unused") private static void logClientPort(StringBuilder b, Request request, Response response) { b.append(request.getRemotePort()); } + @SuppressWarnings("unused") private static void logLocalPort(StringBuilder b, Request request, Response response) { b.append(request.getHttpChannel().getEndPoint().getLocalAddress().getPort()); } + @SuppressWarnings("unused") private static void logRemotePort(StringBuilder b, Request request, Response response) { b.append(request.getHttpChannel().getEndPoint().getRemoteAddress().getPort()); } + @SuppressWarnings("unused") private static void logResponseSize(StringBuilder b, Request request, Response response) { long written = response.getHttpChannel().getBytesWritten(); b.append(written); } + @SuppressWarnings("unused") private static void logResponseSizeCLF(StringBuilder b, Request request, Response response) { long written = response.getHttpChannel().getBytesWritten(); @@ -981,11 +990,13 @@ private static void logResponseSizeCLF(StringBuilder b, Request request, Respons b.append(written); } + @SuppressWarnings("unused") private static void logBytesSent(StringBuilder b, Request request, Response response) { b.append(response.getHttpChannel().getBytesWritten()); } + @SuppressWarnings("unused") private static void logBytesSentCLF(StringBuilder b, Request request, Response response) { long sent = response.getHttpChannel().getBytesWritten(); @@ -995,38 +1006,39 @@ private static void logBytesSentCLF(StringBuilder b, Request request, Response r b.append(sent); } + @SuppressWarnings("unused") private static void logBytesReceived(StringBuilder b, Request request, Response response) { - //todo this be content received rather than consumed - b.append(request.getHttpInput().getContentConsumed()); + b.append(request.getHttpInput().getContentReceived()); } + @SuppressWarnings("unused") private static void logBytesReceivedCLF(StringBuilder b, Request request, Response response) { - //todo this be content received rather than consumed - long received = request.getHttpInput().getContentConsumed(); + long received = request.getHttpInput().getContentReceived(); if (received == 0) b.append('-'); else b.append(received); } + @SuppressWarnings("unused") private static void logBytesTransferred(StringBuilder b, Request request, Response response) { - //todo this be content received rather than consumed - b.append(request.getHttpInput().getContentConsumed() + response.getHttpOutput().getWritten()); + b.append(request.getHttpInput().getContentReceived() + response.getHttpOutput().getWritten()); } + @SuppressWarnings("unused") private static void logBytesTransferredCLF(StringBuilder b, Request request, Response response) { - //todo this be content received rather than consumed - long transferred = request.getHttpInput().getContentConsumed() + response.getHttpOutput().getWritten(); + long transferred = request.getHttpInput().getContentReceived() + response.getHttpOutput().getWritten(); if (transferred == 0) b.append('-'); else b.append(transferred); } + @SuppressWarnings("unused") private static void logRequestCookie(String arg, StringBuilder b, Request request, Response response) { Cookie[] cookies = request.getCookies(); @@ -1045,6 +1057,7 @@ private static void logRequestCookie(String arg, StringBuilder b, Request reques b.append('-'); } + @SuppressWarnings("unused") private static void logRequestCookies(StringBuilder b, Request request, Response response) { Cookie[] cookies = request.getCookies(); @@ -1063,11 +1076,13 @@ private static void logRequestCookies(StringBuilder b, Request request, Response } } + @SuppressWarnings("unused") private static void logEnvironmentVar(String arg, StringBuilder b, Request request, Response response) { append(b, System.getenv(arg)); } + @SuppressWarnings("unused") private static void logFilename(StringBuilder b, Request request, Response response) { UserIdentity.Scope scope = request.getUserIdentityScope(); @@ -1082,16 +1097,19 @@ private static void logFilename(StringBuilder b, Request request, Response respo } } + @SuppressWarnings("unused") private static void logRequestProtocol(StringBuilder b, Request request, Response response) { append(b, request.getProtocol()); } + @SuppressWarnings("unused") private static void logRequestHeader(String arg, StringBuilder b, Request request, Response response) { append(b, request.getHeader(arg)); } + @SuppressWarnings("unused") private static void logKeepAliveRequests(StringBuilder b, Request request, Response response) { long requests = request.getHttpChannel().getConnection().getMessagesIn(); @@ -1101,21 +1119,25 @@ private static void logKeepAliveRequests(StringBuilder b, Request request, Respo b.append('-'); } + @SuppressWarnings("unused") private static void logRequestMethod(StringBuilder b, Request request, Response response) { append(b, request.getMethod()); } + @SuppressWarnings("unused") private static void logResponseHeader(String arg, StringBuilder b, Request request, Response response) { append(b, response.getHeader(arg)); } + @SuppressWarnings("unused") private static void logQueryString(StringBuilder b, Request request, Response response) { append(b, "?" + request.getQueryString()); } + @SuppressWarnings("unused") private static void logRequestFirstLine(StringBuilder b, Request request, Response response) { append(b, request.getMethod()); @@ -1125,17 +1147,19 @@ private static void logRequestFirstLine(StringBuilder b, Request request, Respon append(b, request.getProtocol()); } + @SuppressWarnings("unused") private static void logRequestHandler(StringBuilder b, Request request, Response response) { append(b, request.getServletName()); } + @SuppressWarnings("unused") private static void logResponseStatus(StringBuilder b, Request request, Response response) { - //todo can getCommittedMetaData be null? check what happens when its aborted b.append(response.getCommittedMetaData().getStatus()); } + @SuppressWarnings("unused") private static void logRequestTime(DateCache dateCache, StringBuilder b, Request request, Response response) { b.append('['); @@ -1143,6 +1167,7 @@ private static void logRequestTime(DateCache dateCache, StringBuilder b, Request b.append(']'); } + @SuppressWarnings("unused") private static void logLatencyMicroseconds(StringBuilder b, Request request, Response response) { long currentTime = System.currentTimeMillis(); @@ -1154,38 +1179,45 @@ private static void logLatencyMicroseconds(StringBuilder b, Request request, Res b.append(latencyUs); } + @SuppressWarnings("unused") private static void logLatencyMilliseconds(StringBuilder b, Request request, Response response) { long latency = System.currentTimeMillis() - request.getTimeStamp(); b.append(latency); } + @SuppressWarnings("unused") private static void logLatencySeconds(StringBuilder b, Request request, Response response) { long latency = System.currentTimeMillis() - request.getTimeStamp(); b.append(TimeUnit.MILLISECONDS.toSeconds(latency)); } + @SuppressWarnings("unused") private static void logRequestAuthentication(StringBuilder b, Request request, Response response) { append(b, getAuthentication(request, false)); } + @SuppressWarnings("unused") private static void logRequestAuthenticationWithDeferred(StringBuilder b, Request request, Response response) { append(b, getAuthentication(request, true)); } + @SuppressWarnings("unused") private static void logUrlRequestPath(StringBuilder b, Request request, Response response) { append(b, request.getRequestURI()); } + @SuppressWarnings("unused") private static void logConnectionStatus(StringBuilder b, Request request, Response response) { b.append(request.getHttpChannel().isResponseCompleted() ? (request.getHttpChannel().isPersistent() ? '+' : '-') : 'X'); } + @SuppressWarnings("unused") private static void logRequestTrailer(String arg, StringBuilder b, Request request, Response response) { HttpFields trailers = request.getTrailers(); @@ -1195,6 +1227,7 @@ private static void logRequestTrailer(String arg, StringBuilder b, Request reque b.append('-'); } + @SuppressWarnings("unused") private static void logResponseTrailer(String arg, StringBuilder b, Request request, Response response) { Supplier supplier = response.getTrailers(); diff --git a/jetty-server/src/main/java/org/eclipse/jetty/server/HttpInput.java b/jetty-server/src/main/java/org/eclipse/jetty/server/HttpInput.java index bdd90f0a0cd4..3eabed665d39 100644 --- a/jetty-server/src/main/java/org/eclipse/jetty/server/HttpInput.java +++ b/jetty-server/src/main/java/org/eclipse/jetty/server/HttpInput.java @@ -640,6 +640,14 @@ public long getContentConsumed() } } + public long getContentReceived() + { + synchronized (_inputQ) + { + return _contentArrived; + } + } + /** * This method should be called to signal that an EOF has been detected before all the expected content arrived. *

diff --git a/jetty-util/src/main/java/org/eclipse/jetty/util/Pool.java b/jetty-util/src/main/java/org/eclipse/jetty/util/Pool.java index ca16fa382081..b00981f08313 100644 --- a/jetty-util/src/main/java/org/eclipse/jetty/util/Pool.java +++ b/jetty-util/src/main/java/org/eclipse/jetty/util/Pool.java @@ -362,7 +362,12 @@ public void dump(Appendable out, String indent) throws IOException @Override public String toString() { - return getClass().getSimpleName() + " size=" + sharedList.size() + " closed=" + closed + " entries=" + sharedList; + return String.format("%s@%x[size=%d closed=%s entries=%s]", + getClass().getSimpleName(), + hashCode(), + sharedList.size(), + closed, + sharedList); } public class Entry @@ -544,11 +549,13 @@ public int getUsageCount() public String toString() { long encoded = state.get(); - return String.format("%s@%x{usage=%d,multiplex=%d,pooled=%s}", + return String.format("%s@%x{usage=%d/%d,multiplex=%d/%d,pooled=%s}", getClass().getSimpleName(), hashCode(), AtomicBiInteger.getHi(encoded), + getMaxUsageCount(), AtomicBiInteger.getLo(encoded), + getMaxMultiplex(), pooled); } } diff --git a/jetty-websocket/javax-websocket-client-impl/src/main/java/org/eclipse/jetty/websocket/jsr356/JsrSessionTracker.java b/jetty-websocket/javax-websocket-client-impl/src/main/java/org/eclipse/jetty/websocket/jsr356/JsrSessionTracker.java index 6af4830c59b4..303e77c8c902 100644 --- a/jetty-websocket/javax-websocket-client-impl/src/main/java/org/eclipse/jetty/websocket/jsr356/JsrSessionTracker.java +++ b/jetty-websocket/javax-websocket-client-impl/src/main/java/org/eclipse/jetty/websocket/jsr356/JsrSessionTracker.java @@ -18,16 +18,19 @@ package org.eclipse.jetty.websocket.jsr356; +import java.io.IOException; import java.util.Collections; import java.util.HashSet; import java.util.Set; import java.util.concurrent.ConcurrentHashMap; import javax.websocket.Session; +import org.eclipse.jetty.util.annotation.ManagedAttribute; import org.eclipse.jetty.util.component.AbstractLifeCycle; +import org.eclipse.jetty.util.component.Dumpable; import org.eclipse.jetty.util.component.LifeCycle; -public class JsrSessionTracker extends AbstractLifeCycle implements JsrSessionListener +public class JsrSessionTracker extends AbstractLifeCycle implements JsrSessionListener, Dumpable { private final Set sessions = Collections.newSetFromMap(new ConcurrentHashMap<>()); @@ -57,4 +60,16 @@ protected void doStop() throws Exception } super.doStop(); } + + @ManagedAttribute("Total number of active WebSocket Sessions") + public int getNumSessions() + { + return sessions.size(); + } + + @Override + public void dump(Appendable out, String indent) throws IOException + { + Dumpable.dumpObjects(out, indent, this, sessions); + } } diff --git a/jetty-websocket/jetty-websocket-tests/pom.xml b/jetty-websocket/jetty-websocket-tests/pom.xml index c27e463eb4bd..d9fbabf13a61 100644 --- a/jetty-websocket/jetty-websocket-tests/pom.xml +++ b/jetty-websocket/jetty-websocket-tests/pom.xml @@ -66,5 +66,11 @@ jetty-test-helper test + + org.eclipse.jetty + jetty-jmx + ${project.version} + test + diff --git a/jetty-websocket/jetty-websocket-tests/src/test/java/org/eclipse/jetty/websocket/tests/WebSocketStatsTest.java b/jetty-websocket/jetty-websocket-tests/src/test/java/org/eclipse/jetty/websocket/tests/WebSocketStatsTest.java index f096535dcbe4..da0741c55b98 100644 --- a/jetty-websocket/jetty-websocket-tests/src/test/java/org/eclipse/jetty/websocket/tests/WebSocketStatsTest.java +++ b/jetty-websocket/jetty-websocket-tests/src/test/java/org/eclipse/jetty/websocket/tests/WebSocketStatsTest.java @@ -18,6 +18,7 @@ package org.eclipse.jetty.websocket.tests; +import java.lang.management.ManagementFactory; import java.net.URI; import java.nio.ByteBuffer; import java.util.concurrent.CountDownLatch; @@ -26,9 +27,9 @@ import org.eclipse.jetty.io.ByteBufferPool; import org.eclipse.jetty.io.Connection; -import org.eclipse.jetty.io.ConnectionStatistics; +import org.eclipse.jetty.io.IncludeExcludeConnectionStatistics; import org.eclipse.jetty.io.MappedByteBufferPool; -import org.eclipse.jetty.server.HttpConnection; +import org.eclipse.jetty.jmx.MBeanContainer; import org.eclipse.jetty.server.Server; import org.eclipse.jetty.server.ServerConnector; import org.eclipse.jetty.servlet.ServletContextHandler; @@ -41,8 +42,8 @@ import org.eclipse.jetty.websocket.common.WebSocketFrame; import org.eclipse.jetty.websocket.common.frames.TextFrame; import org.eclipse.jetty.websocket.common.io.AbstractWebSocketConnection; -import org.eclipse.jetty.websocket.servlet.WebSocketServlet; -import org.eclipse.jetty.websocket.servlet.WebSocketServletFactory; +import org.eclipse.jetty.websocket.server.NativeWebSocketServletContainerInitializer; +import org.eclipse.jetty.websocket.server.WebSocketUpgradeFilter; import org.junit.jupiter.api.AfterEach; import org.junit.jupiter.api.BeforeEach; import org.junit.jupiter.api.Test; @@ -53,51 +54,47 @@ public class WebSocketStatsTest { - public static class MyWebSocketServlet extends WebSocketServlet - { - @Override - public void configure(WebSocketServletFactory factory) - { - factory.setCreator((req, resp) -> new EchoSocket()); - } - } - + private final CountDownLatch wsConnectionClosed = new CountDownLatch(1); private Server server; private ServerConnector connector; private WebSocketClient client; - private ConnectionStatistics statistics; - private CountDownLatch wsUpgradeComplete = new CountDownLatch(1); - private CountDownLatch wsConnectionClosed = new CountDownLatch(1); + private IncludeExcludeConnectionStatistics statistics; @BeforeEach public void start() throws Exception { - statistics = new ConnectionStatistics() + statistics = new IncludeExcludeConnectionStatistics(); + statistics.include(AbstractWebSocketConnection.class); + + Connection.Listener.Adapter wsCloseListener = new Connection.Listener.Adapter() { @Override public void onClosed(Connection connection) { - super.onClosed(connection); - if (connection instanceof AbstractWebSocketConnection) wsConnectionClosed.countDown(); - else if (connection instanceof HttpConnection) - wsUpgradeComplete.countDown(); } }; server = new Server(); connector = new ServerConnector(server); connector.addBean(statistics); + connector.addBean(wsCloseListener); server.addConnector(connector); ServletContextHandler contextHandler = new ServletContextHandler(ServletContextHandler.SESSIONS); contextHandler.setContextPath("/"); - contextHandler.addServlet(MyWebSocketServlet.class, "/testPath"); + NativeWebSocketServletContainerInitializer.configure(contextHandler, (context, container) -> + container.addMapping("/", EchoSocket.class)); + WebSocketUpgradeFilter.configure(contextHandler); server.setHandler(contextHandler); client = new WebSocketClient(); + // Setup JMX. + MBeanContainer mbeanContainer = new MBeanContainer(ManagementFactory.getPlatformMBeanServer()); + server.addBean(mbeanContainer); + server.start(); client.start(); } @@ -105,8 +102,8 @@ else if (connection instanceof HttpConnection) @AfterEach public void stop() throws Exception { - client.stop(); server.stop(); + client.stop(); } long getFrameByteSize(WebSocketFrame frame) @@ -122,22 +119,14 @@ long getFrameByteSize(WebSocketFrame frame) @Test public void echoStatsTest() throws Exception { - URI uri = URI.create("ws://localhost:" + connector.getLocalPort() + "/testPath"); + URI uri = URI.create("ws://localhost:" + connector.getLocalPort() + "/"); EventSocket socket = new EventSocket(); Future connect = client.connect(socket, uri); - final long numMessages = 10000; + final long numMessages = 1000; final String msgText = "hello world"; - - long upgradeSentBytes; - long upgradeReceivedBytes; - try (Session session = connect.get(5, TimeUnit.SECONDS)) { - wsUpgradeComplete.await(5, TimeUnit.SECONDS); - upgradeSentBytes = statistics.getSentBytes(); - upgradeReceivedBytes = statistics.getReceivedBytes(); - for (int i = 0; i < numMessages; i++) { session.getRemote().sendString(msgText); @@ -150,18 +139,18 @@ public void echoStatsTest() throws Exception assertThat(statistics.getConnectionsMax(), is(1L)); assertThat(statistics.getConnections(), is(0L)); - assertThat(statistics.getSentMessages(), is(numMessages + 2L)); - assertThat(statistics.getReceivedMessages(), is(numMessages + 2L)); + // Sent and received all of the echo messages + 1 for the close frame. + assertThat(statistics.getSentMessages(), is(numMessages + 1L)); + assertThat(statistics.getReceivedMessages(), is(numMessages + 1L)); WebSocketFrame textFrame = new TextFrame().setPayload(msgText); WebSocketFrame closeFrame = new CloseInfo(socket.closeCode, socket.closeReason).asFrame(); - final long textFrameSize = getFrameByteSize(textFrame); final long closeFrameSize = getFrameByteSize(closeFrame); final int maskSize = 4; // We use 4 byte mask for client frames - final long expectedSent = upgradeSentBytes + numMessages * textFrameSize + closeFrameSize; - final long expectedReceived = upgradeReceivedBytes + numMessages * (textFrameSize + maskSize) + closeFrameSize + maskSize; + final long expectedSent = numMessages * textFrameSize + closeFrameSize; + final long expectedReceived = numMessages * (textFrameSize + maskSize) + (closeFrameSize + maskSize); assertThat("stats.sendBytes", statistics.getSentBytes(), is(expectedSent)); assertThat("stats.receivedBytes", statistics.getReceivedBytes(), is(expectedReceived)); diff --git a/jetty-websocket/websocket-common/src/main/java/org/eclipse/jetty/websocket/common/SessionTracker.java b/jetty-websocket/websocket-common/src/main/java/org/eclipse/jetty/websocket/common/SessionTracker.java index 635f1a9d2ce1..c75f1b103d65 100644 --- a/jetty-websocket/websocket-common/src/main/java/org/eclipse/jetty/websocket/common/SessionTracker.java +++ b/jetty-websocket/websocket-common/src/main/java/org/eclipse/jetty/websocket/common/SessionTracker.java @@ -24,6 +24,7 @@ import java.util.Set; import java.util.concurrent.ConcurrentHashMap; +import org.eclipse.jetty.util.annotation.ManagedAttribute; import org.eclipse.jetty.util.component.AbstractLifeCycle; import org.eclipse.jetty.util.component.Dumpable; import org.eclipse.jetty.util.component.LifeCycle; @@ -61,6 +62,12 @@ protected void doStop() throws Exception super.doStop(); } + @ManagedAttribute("Total number of active WebSocket Sessions") + public int getNumSessions() + { + return sessions.size(); + } + @Override public void dump(Appendable out, String indent) throws IOException { diff --git a/tests/test-http-client-transport/src/test/java/org/eclipse/jetty/http/client/RoundRobinConnectionPoolTest.java b/tests/test-http-client-transport/src/test/java/org/eclipse/jetty/http/client/RoundRobinConnectionPoolTest.java index e5d20c427f8a..45b0a512df78 100644 --- a/tests/test-http-client-transport/src/test/java/org/eclipse/jetty/http/client/RoundRobinConnectionPoolTest.java +++ b/tests/test-http-client-transport/src/test/java/org/eclipse/jetty/http/client/RoundRobinConnectionPoolTest.java @@ -19,8 +19,9 @@ package org.eclipse.jetty.http.client; import java.io.IOException; -import java.util.ArrayList; import java.util.List; +import java.util.Map; +import java.util.concurrent.CopyOnWriteArrayList; import java.util.concurrent.CountDownLatch; import java.util.concurrent.CyclicBarrier; import java.util.concurrent.TimeUnit; @@ -56,7 +57,7 @@ public void testRoundRobin(Transport transport) throws Exception { init(transport); AtomicBoolean record = new AtomicBoolean(); - List remotePorts = new ArrayList<>(); + List remotePorts = new CopyOnWriteArrayList<>(); scenario.start(new EmptyServerHandler() { @Override @@ -115,8 +116,7 @@ public void testMultiplex(Transport transport) throws Exception int maxConnections = 3; int count = maxConnections * maxMultiplex; - AtomicBoolean record = new AtomicBoolean(); - List remotePorts = new ArrayList<>(); + List remotePorts = new CopyOnWriteArrayList<>(); AtomicReference requestLatch = new AtomicReference<>(); CountDownLatch serverLatch = new CountDownLatch(count); CyclicBarrier barrier = new CyclicBarrier(count + 1); @@ -127,13 +127,10 @@ protected void service(String target, Request jettyRequest, HttpServletRequest r { try { - if (record.get()) - { - remotePorts.add(request.getRemotePort()); - requestLatch.get().countDown(); - serverLatch.countDown(); - barrier.await(); - } + remotePorts.add(request.getRemotePort()); + requestLatch.get().countDown(); + serverLatch.countDown(); + barrier.await(); } catch (Exception x) { @@ -144,17 +141,9 @@ protected void service(String target, Request jettyRequest, HttpServletRequest r scenario.client.getTransport().setConnectionPoolFactory(destination -> new RoundRobinConnectionPool(destination, maxConnections, destination, maxMultiplex)); - // Prime the connections, so that they are all opened - // before we actually test the round robin behavior. - for (int i = 0; i < maxConnections; ++i) - { - ContentResponse response = scenario.client.newRequest(scenario.newURI()) - .timeout(5, TimeUnit.SECONDS) - .send(); - assertEquals(HttpStatus.OK_200, response.getStatus()); - } + // Do not prime the connections, to see if the behavior is + // correct even if the connections are not pre-created. - record.set(true); CountDownLatch clientLatch = new CountDownLatch(count); AtomicInteger requests = new AtomicInteger(); for (int i = 0; i < count; ++i)