jetty-server/src/test/java/org/eclipse/jetty/server/ThreadStarvationTest.java - jetty - Git at Google

 //
 //  ========================================================================
 //  Copyright (c) 1995-2017 Mort Bay Consulting Pty. Ltd.
 //  ------------------------------------------------------------------------
 //  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.server;

 import static org.hamcrest.Matchers.containsString;
 import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.assertFalse;
 import static org.junit.Assert.assertThat;
 import static org.junit.Assert.assertTrue;

 import java.io.IOException;
 import java.io.InputStream;
 import java.io.OutputStream;
 import java.net.Socket;
 import java.nio.charset.StandardCharsets;
 import java.nio.file.Path;
 import java.util.ArrayList;
 import java.util.List;
 import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.Executor;

 import javax.net.ssl.HttpsURLConnection;
 import javax.net.ssl.SSLContext;
 import javax.servlet.ServletException;
 import javax.servlet.http.HttpServletRequest;
 import javax.servlet.http.HttpServletResponse;

 import org.eclipse.jetty.io.ByteBufferPool;
 import org.eclipse.jetty.io.LeakTrackingByteBufferPool;
 import org.eclipse.jetty.io.ManagedSelector;
 import org.eclipse.jetty.io.MappedByteBufferPool;
 import org.eclipse.jetty.server.handler.AbstractHandler;
 import org.eclipse.jetty.toolchain.test.MavenTestingUtils;
 import org.eclipse.jetty.toolchain.test.TestTracker;
 import org.eclipse.jetty.util.IO;
 import org.eclipse.jetty.util.ssl.SslContextFactory;
 import org.eclipse.jetty.util.thread.ExecutionStrategy;
 import org.eclipse.jetty.util.thread.QueuedThreadPool;
 import org.eclipse.jetty.util.thread.Scheduler;
 import org.eclipse.jetty.util.thread.strategy.ExecuteProduceConsume;
 import org.eclipse.jetty.util.thread.strategy.ProduceExecuteConsume;
 import org.junit.After;
 import org.junit.Rule;
 import org.junit.Test;
 import org.junit.runner.RunWith;
 import org.junit.runners.Parameterized;

 @RunWith(Parameterized.class)
 public class ThreadStarvationTest
 {
     @Rule
     public TestTracker tracker = new TestTracker();

     interface ConnectorProvider {
         ServerConnector newConnector(Server server, int acceptors, int selectors);
     }

     interface ClientSocketProvider {
         Socket newSocket(String host, int port) throws IOException;
     }

     @Parameterized.Parameters(name = "{0}")
     public static List<Object[]> params()
     {
         List<Object[]> params = new ArrayList<>();

         // HTTP
         ConnectorProvider http = (server, acceptors, selectors) -> new ServerConnector(server, acceptors, selectors);
         ClientSocketProvider httpClient = (host, port) -> new Socket(host, port);
         params.add(new Object[]{ "http", http, httpClient });

         // HTTPS/SSL/TLS
         ConnectorProvider https = (server, acceptors, selectors) -> {
             Path keystorePath = MavenTestingUtils.getTestResourcePath("keystore");
             SslContextFactory sslContextFactory = new SslContextFactory();
             sslContextFactory.setKeyStorePath(keystorePath.toString());
             sslContextFactory.setKeyStorePassword("storepwd");
             sslContextFactory.setKeyManagerPassword("keypwd");
             sslContextFactory.setTrustStorePath(keystorePath.toString());
             sslContextFactory.setTrustStorePassword("storepwd");
             ByteBufferPool pool = new LeakTrackingByteBufferPool(new MappedByteBufferPool.Tagged());

             HttpConnectionFactory httpConnectionFactory = new HttpConnectionFactory();
             ServerConnector connector = new ServerConnector(server,(Executor)null,(Scheduler)null,
                     pool, acceptors, selectors,
                     AbstractConnectionFactory.getFactories(sslContextFactory,httpConnectionFactory));
             SecureRequestCustomizer secureRequestCustomer = new SecureRequestCustomizer();
             secureRequestCustomer.setSslSessionAttribute("SSL_SESSION");
             httpConnectionFactory.getHttpConfiguration().addCustomizer(secureRequestCustomer);
             return connector;
         };
         ClientSocketProvider httpsClient = new ClientSocketProvider()
         {
             private SSLContext sslContext;
             {
                 try
                 {
                     HttpsURLConnection.setDefaultHostnameVerifier((hostname, session)-> true);
                     sslContext = SSLContext.getInstance("TLS");
                     sslContext.init(null, SslContextFactory.TRUST_ALL_CERTS, new java.security.SecureRandom());
                     HttpsURLConnection.setDefaultSSLSocketFactory(sslContext.getSocketFactory());
                 }
                 catch(Exception e)
                 {
                     e.printStackTrace();
                     throw new RuntimeException(e);
                 }
             }

             @Override
             public Socket newSocket(String host, int port) throws IOException
             {
                 return sslContext.getSocketFactory().createSocket(host,port);
             }
         };
         params.add(new Object[]{ "https/ssl/tls", https, httpsClient });

         return params;
     }

     private final ConnectorProvider connectorProvider;
     private final ClientSocketProvider clientSocketProvider;
     private QueuedThreadPool _threadPool;
     private Server _server;
     private ServerConnector _connector;
     private int _availableThreads;

     public ThreadStarvationTest(String testType, ConnectorProvider connectorProvider, ClientSocketProvider clientSocketProvider)
     {
         this.connectorProvider = connectorProvider;
         this.clientSocketProvider = clientSocketProvider;
     }

     private Server prepareServer(Handler handler)
     {
         int threads = 4;
         _threadPool = new QueuedThreadPool();
         _threadPool.setMinThreads(threads);
         _threadPool.setMaxThreads(threads);
         _threadPool.setDetailedDump(true);
         _server = new Server(_threadPool);
         int acceptors = 1;
         int selectors = 1;
         _connector = connectorProvider.newConnector(_server, acceptors, selectors);
         _server.addConnector(_connector);
         _server.setHandler(handler);
         _availableThreads = threads - acceptors - selectors;
         return _server;
     }

     @After
     public void dispose() throws Exception
     {
         _server.stop();
     }

     @Test
     public void testReadInput() throws Exception
     {
         prepareServer(new ReadHandler()).start();

         Socket client = clientSocketProvider.newSocket("localhost", _connector.getLocalPort());

         OutputStream os = client.getOutputStream();
         InputStream is = client.getInputStream();

         String request = "" +
                 "GET / HTTP/1.0\r\n" +
                 "Host: localhost\r\n" +
                 "Content-Length: 10\r\n" +
                 "\r\n" +
                 "0123456789\r\n";
         os.write(request.getBytes(StandardCharsets.UTF_8));
         os.flush();

         String response = IO.toString(is);
         assertEquals(-1, is.read());
         assertThat(response, containsString("200 OK"));
         assertThat(response, containsString("Read Input 10"));
     }

     @Test
     public void testEPCStarvation() throws Exception
     {
         testStarvation(new ExecuteProduceConsume.Factory());
     }

     @Test
     public void testPECStarvation() throws Exception
     {
         testStarvation(new ProduceExecuteConsume.Factory());
     }

     private void testStarvation(ExecutionStrategy.Factory executionFactory) throws Exception
     {
         prepareServer(new ReadHandler());
         _connector.setExecutionStrategyFactory(executionFactory);
         _server.start();

         Socket[] client = new Socket[_availableThreads + 1];
         OutputStream[] os = new OutputStream[client.length];
         InputStream[] is = new InputStream[client.length];

         for (int i = 0; i < client.length; i++)
         {
             client[i] = clientSocketProvider.newSocket("localhost", _connector.getLocalPort());
             client[i].setSoTimeout(10000);

             os[i] = client[i].getOutputStream();
             is[i] = client[i].getInputStream();

             String request = "" +
                     "PUT / HTTP/1.0\r\n" +
                     "host: localhost\r\n" +
                     "content-length: 10\r\n" +
                     "\r\n" +
                     "1";
             os[i].write(request.getBytes(StandardCharsets.UTF_8));
             os[i].flush();
         }

         Thread.sleep(500);

         for (int i = 0; i < client.length; i++)
         {
             os[i].write(("234567890\r\n").getBytes(StandardCharsets.UTF_8));
             os[i].flush();
         }

         Thread.sleep(500);

         for (int i = 0; i < client.length; i++)
         {
             String response = IO.toString(is[i]);
             assertEquals(-1, is[i].read());
             assertThat(response, containsString("200 OK"));
             assertThat(response, containsString("Read Input 10"));
         }
     }

     @Test
     public void testEPCExitsLowThreadsMode() throws Exception
     {
         prepareServer(new ReadHandler());
         _threadPool.setMaxThreads(5);
         _connector.setExecutionStrategyFactory(new ExecuteProduceConsume.Factory());
         _server.start();

         // Three idle threads in the pool here.
         // The server will accept the socket in normal mode.
         Socket client = clientSocketProvider.newSocket("localhost", _connector.getLocalPort());
         client.setSoTimeout(10000);

         Thread.sleep(500);

         // Now steal two threads.
         CountDownLatch[] latches = new CountDownLatch[2];
         for (int i = 0; i < latches.length; ++i)
         {
             CountDownLatch latch = latches[i] = new CountDownLatch(1);
             _threadPool.execute(() ->
             {
                 try
                 {
                     latch.await();
                 }
                 catch (InterruptedException ignored)
                 {
                 }
             });
         }

         InputStream is = client.getInputStream();
         OutputStream os = client.getOutputStream();

         String request = "" +
                 "PUT / HTTP/1.0\r\n" +
                 "Host: localhost\r\n" +
                 "Content-Length: 10\r\n" +
                 "\r\n" +
                 "1";
         os.write(request.getBytes(StandardCharsets.UTF_8));
         os.flush();

         Thread.sleep(500);

         // Request did not send the whole body, Handler
         // is blocked reading, zero idle threads here,
         // EPC is in low threads mode.
         for (ManagedSelector selector : _connector.getSelectorManager().getBeans(ManagedSelector.class))
         {
             ExecuteProduceConsume executionStrategy = (ExecuteProduceConsume)selector.getExecutionStrategy();
             assertTrue(executionStrategy.isLowOnThreads());
         }

         // Release the stolen threads.
         for (CountDownLatch latch : latches)
             latch.countDown();
         Thread.sleep(500);

         // Send the rest of the body to unblock the reader thread.
         // This will be run directly by the selector thread,
         // which then will exit the low threads mode.
         os.write("234567890".getBytes(StandardCharsets.UTF_8));
         os.flush();

         Thread.sleep(500);

         for (ManagedSelector selector : _connector.getSelectorManager().getBeans(ManagedSelector.class))
         {
             ExecuteProduceConsume executionStrategy = (ExecuteProduceConsume)selector.getExecutionStrategy();
             assertFalse(executionStrategy.isLowOnThreads());
         }
     }

     protected static class ReadHandler extends AbstractHandler
     {
         @Override
         public void handle(String target, Request baseRequest, HttpServletRequest request, HttpServletResponse response) throws IOException, ServletException
         {
             baseRequest.setHandled(true);
             response.setStatus(200);

             int l = request.getContentLength();
             int r = 0;
             while (r < l)
             {
                 if (request.getInputStream().read() >= 0)
                     r++;
             }

             response.getOutputStream().write(("Read Input " + r + "\r\n").getBytes());
         }
     }
 }
	//
	// ========================================================================
	// Copyright (c) 1995-2017 Mort Bay Consulting Pty. Ltd.
	// ------------------------------------------------------------------------
	// 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.server;

	import static org.hamcrest.Matchers.containsString;
	import static org.junit.Assert.assertEquals;
	import static org.junit.Assert.assertFalse;
	import static org.junit.Assert.assertThat;
	import static org.junit.Assert.assertTrue;

	import java.io.IOException;
	import java.io.InputStream;
	import java.io.OutputStream;
	import java.net.Socket;
	import java.nio.charset.StandardCharsets;
	import java.nio.file.Path;
	import java.util.ArrayList;
	import java.util.List;
	import java.util.concurrent.CountDownLatch;
	import java.util.concurrent.Executor;

	import javax.net.ssl.HttpsURLConnection;
	import javax.net.ssl.SSLContext;
	import javax.servlet.ServletException;
	import javax.servlet.http.HttpServletRequest;
	import javax.servlet.http.HttpServletResponse;

	import org.eclipse.jetty.io.ByteBufferPool;
	import org.eclipse.jetty.io.LeakTrackingByteBufferPool;
	import org.eclipse.jetty.io.ManagedSelector;
	import org.eclipse.jetty.io.MappedByteBufferPool;
	import org.eclipse.jetty.server.handler.AbstractHandler;
	import org.eclipse.jetty.toolchain.test.MavenTestingUtils;
	import org.eclipse.jetty.toolchain.test.TestTracker;
	import org.eclipse.jetty.util.IO;
	import org.eclipse.jetty.util.ssl.SslContextFactory;
	import org.eclipse.jetty.util.thread.ExecutionStrategy;
	import org.eclipse.jetty.util.thread.QueuedThreadPool;
	import org.eclipse.jetty.util.thread.Scheduler;
	import org.eclipse.jetty.util.thread.strategy.ExecuteProduceConsume;
	import org.eclipse.jetty.util.thread.strategy.ProduceExecuteConsume;
	import org.junit.After;
	import org.junit.Rule;
	import org.junit.Test;
	import org.junit.runner.RunWith;
	import org.junit.runners.Parameterized;

	@RunWith(Parameterized.class)
	public class ThreadStarvationTest
	{
	@Rule
	public TestTracker tracker = new TestTracker();

	interface ConnectorProvider {
	ServerConnector newConnector(Server server, int acceptors, int selectors);
	}

	interface ClientSocketProvider {
	Socket newSocket(String host, int port) throws IOException;
	}

	@Parameterized.Parameters(name = "{0}")
	public static List<Object[]> params()
	{
	List<Object[]> params = new ArrayList<>();

	// HTTP
	ConnectorProvider http = (server, acceptors, selectors) -> new ServerConnector(server, acceptors, selectors);
	ClientSocketProvider httpClient = (host, port) -> new Socket(host, port);
	params.add(new Object[]{ "http", http, httpClient });

	// HTTPS/SSL/TLS
	ConnectorProvider https = (server, acceptors, selectors) -> {
	Path keystorePath = MavenTestingUtils.getTestResourcePath("keystore");
	SslContextFactory sslContextFactory = new SslContextFactory();
	sslContextFactory.setKeyStorePath(keystorePath.toString());
	sslContextFactory.setKeyStorePassword("storepwd");
	sslContextFactory.setKeyManagerPassword("keypwd");
	sslContextFactory.setTrustStorePath(keystorePath.toString());
	sslContextFactory.setTrustStorePassword("storepwd");
	ByteBufferPool pool = new LeakTrackingByteBufferPool(new MappedByteBufferPool.Tagged());

	HttpConnectionFactory httpConnectionFactory = new HttpConnectionFactory();
	ServerConnector connector = new ServerConnector(server,(Executor)null,(Scheduler)null,
	pool, acceptors, selectors,
	AbstractConnectionFactory.getFactories(sslContextFactory,httpConnectionFactory));
	SecureRequestCustomizer secureRequestCustomer = new SecureRequestCustomizer();
	secureRequestCustomer.setSslSessionAttribute("SSL_SESSION");
	httpConnectionFactory.getHttpConfiguration().addCustomizer(secureRequestCustomer);
	return connector;
	};
	ClientSocketProvider httpsClient = new ClientSocketProvider()
	{
	private SSLContext sslContext;
	{
	try
	{
	HttpsURLConnection.setDefaultHostnameVerifier((hostname, session)-> true);
	sslContext = SSLContext.getInstance("TLS");
	sslContext.init(null, SslContextFactory.TRUST_ALL_CERTS, new java.security.SecureRandom());
	HttpsURLConnection.setDefaultSSLSocketFactory(sslContext.getSocketFactory());
	}
	catch(Exception e)
	{
	e.printStackTrace();
	throw new RuntimeException(e);
	}
	}

	@Override
	public Socket newSocket(String host, int port) throws IOException
	{
	return sslContext.getSocketFactory().createSocket(host,port);
	}
	};
	params.add(new Object[]{ "https/ssl/tls", https, httpsClient });

	return params;
	}

	private final ConnectorProvider connectorProvider;
	private final ClientSocketProvider clientSocketProvider;
	private QueuedThreadPool _threadPool;
	private Server _server;
	private ServerConnector _connector;
	private int _availableThreads;

	public ThreadStarvationTest(String testType, ConnectorProvider connectorProvider, ClientSocketProvider clientSocketProvider)
	{
	this.connectorProvider = connectorProvider;
	this.clientSocketProvider = clientSocketProvider;
	}

	private Server prepareServer(Handler handler)
	{
	int threads = 4;
	_threadPool = new QueuedThreadPool();
	_threadPool.setMinThreads(threads);
	_threadPool.setMaxThreads(threads);
	_threadPool.setDetailedDump(true);
	_server = new Server(_threadPool);
	int acceptors = 1;
	int selectors = 1;
	_connector = connectorProvider.newConnector(_server, acceptors, selectors);
	_server.addConnector(_connector);
	_server.setHandler(handler);
	_availableThreads = threads - acceptors - selectors;
	return _server;
	}

	@After
	public void dispose() throws Exception
	{
	_server.stop();
	}

	@Test
	public void testReadInput() throws Exception
	{
	prepareServer(new ReadHandler()).start();

	Socket client = clientSocketProvider.newSocket("localhost", _connector.getLocalPort());

	OutputStream os = client.getOutputStream();
	InputStream is = client.getInputStream();

	String request = "" +
	"GET / HTTP/1.0\r\n" +
	"Host: localhost\r\n" +
	"Content-Length: 10\r\n" +
	"\r\n" +
	"0123456789\r\n";
	os.write(request.getBytes(StandardCharsets.UTF_8));
	os.flush();

	String response = IO.toString(is);
	assertEquals(-1, is.read());
	assertThat(response, containsString("200 OK"));
	assertThat(response, containsString("Read Input 10"));
	}

	@Test
	public void testEPCStarvation() throws Exception
	{
	testStarvation(new ExecuteProduceConsume.Factory());
	}

	@Test
	public void testPECStarvation() throws Exception
	{
	testStarvation(new ProduceExecuteConsume.Factory());
	}

	private void testStarvation(ExecutionStrategy.Factory executionFactory) throws Exception
	{
	prepareServer(new ReadHandler());
	_connector.setExecutionStrategyFactory(executionFactory);
	_server.start();

	Socket[] client = new Socket[_availableThreads + 1];
	OutputStream[] os = new OutputStream[client.length];
	InputStream[] is = new InputStream[client.length];

	for (int i = 0; i < client.length; i++)
	{
	client[i] = clientSocketProvider.newSocket("localhost", _connector.getLocalPort());
	client[i].setSoTimeout(10000);

	os[i] = client[i].getOutputStream();
	is[i] = client[i].getInputStream();

	String request = "" +
	"PUT / HTTP/1.0\r\n" +
	"host: localhost\r\n" +
	"content-length: 10\r\n" +
	"\r\n" +
	"1";
	os[i].write(request.getBytes(StandardCharsets.UTF_8));
	os[i].flush();
	}

	Thread.sleep(500);

	for (int i = 0; i < client.length; i++)
	{
	os[i].write(("234567890\r\n").getBytes(StandardCharsets.UTF_8));
	os[i].flush();
	}

	Thread.sleep(500);

	for (int i = 0; i < client.length; i++)
	{
	String response = IO.toString(is[i]);
	assertEquals(-1, is[i].read());
	assertThat(response, containsString("200 OK"));
	assertThat(response, containsString("Read Input 10"));
	}
	}

	@Test
	public void testEPCExitsLowThreadsMode() throws Exception
	{
	prepareServer(new ReadHandler());
	_threadPool.setMaxThreads(5);
	_connector.setExecutionStrategyFactory(new ExecuteProduceConsume.Factory());
	_server.start();

	// Three idle threads in the pool here.
	// The server will accept the socket in normal mode.
	Socket client = clientSocketProvider.newSocket("localhost", _connector.getLocalPort());
	client.setSoTimeout(10000);

	Thread.sleep(500);

	// Now steal two threads.
	CountDownLatch[] latches = new CountDownLatch[2];
	for (int i = 0; i < latches.length; ++i)
	{
	CountDownLatch latch = latches[i] = new CountDownLatch(1);
	_threadPool.execute(() ->
	{
	try
	{
	latch.await();
	}
	catch (InterruptedException ignored)
	{
	}
	});
	}

	InputStream is = client.getInputStream();
	OutputStream os = client.getOutputStream();

	String request = "" +
	"PUT / HTTP/1.0\r\n" +
	"Host: localhost\r\n" +
	"Content-Length: 10\r\n" +
	"\r\n" +
	"1";
	os.write(request.getBytes(StandardCharsets.UTF_8));
	os.flush();

	Thread.sleep(500);

	// Request did not send the whole body, Handler
	// is blocked reading, zero idle threads here,
	// EPC is in low threads mode.
	for (ManagedSelector selector : _connector.getSelectorManager().getBeans(ManagedSelector.class))
	{
	ExecuteProduceConsume executionStrategy = (ExecuteProduceConsume)selector.getExecutionStrategy();
	assertTrue(executionStrategy.isLowOnThreads());
	}

	// Release the stolen threads.
	for (CountDownLatch latch : latches)
	latch.countDown();
	Thread.sleep(500);

	// Send the rest of the body to unblock the reader thread.
	// This will be run directly by the selector thread,
	// which then will exit the low threads mode.
	os.write("234567890".getBytes(StandardCharsets.UTF_8));
	os.flush();

	Thread.sleep(500);

	for (ManagedSelector selector : _connector.getSelectorManager().getBeans(ManagedSelector.class))
	{
	ExecuteProduceConsume executionStrategy = (ExecuteProduceConsume)selector.getExecutionStrategy();
	assertFalse(executionStrategy.isLowOnThreads());
	}
	}

	protected static class ReadHandler extends AbstractHandler
	{
	@Override
	public void handle(String target, Request baseRequest, HttpServletRequest request, HttpServletResponse response) throws IOException, ServletException
	{
	baseRequest.setHandled(true);
	response.setStatus(200);

	int l = request.getContentLength();
	int r = 0;
	while (r < l)
	{
	if (request.getInputStream().read() >= 0)
	r++;
	}

	response.getOutputStream().write(("Read Input " + r + "\r\n").getBytes());
	}
	}
	}