Skip to content

Commit

Permalink
469936 - Remove usages of SpinLock.
Browse files Browse the repository at this point in the history 
Causes high CPU usage when contended, and the JVM can do better with
its own spin lock and biased locking.
  • Loading branch information
@sbordet
sbordet committed Jun 11, 2015
1 parent 9306477 commit 2c26e82
Show file tree
Hide file tree
Showing 5 changed files with 94 additions and 251 deletions.
90 changes: 78 additions & 12 deletions jetty-client/src/main/java/org/eclipse/jetty/client/ConnectionPool.java
View file
Expand Up @@ -26,6 +26,7 @@
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingDeque;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.locks.ReentrantLock;

import org.eclipse.jetty.client.api.Connection;
import org.eclipse.jetty.client.api.Destination;
Expand All @@ -35,15 +36,14 @@
import org.eclipse.jetty.util.component.Dumpable;
import org.eclipse.jetty.util.log.Log;
import org.eclipse.jetty.util.log.Logger;
import org.eclipse.jetty.util.thread.SpinLock;
import org.eclipse.jetty.util.thread.Sweeper;

public class ConnectionPool implements Closeable, Dumpable, Sweeper.Sweepable
{
protected static final Logger LOG = Log.getLogger(ConnectionPool.class);

private final AtomicInteger connectionCount = new AtomicInteger();
private final SpinLock lock = new SpinLock();
private final ReentrantLock lock = new ReentrantLock();
private final Destination destination;
private final int maxConnections;
private final Promise<Connection> requester;
Expand Down Expand Up @@ -136,25 +136,38 @@ public void failed(Throwable x)
protected void idleCreated(Connection connection)
{
boolean idle;
try (SpinLock.Lock lock = this.lock.lock())
final ReentrantLock lock = this.lock;
lock.lock();
try
{
// Use "cold" new connections as last.
idle = idleConnections.offerLast(connection);
}
finally
{
lock.unlock();
}

idle(connection, idle);
}

private Connection activateIdle()
{
boolean acquired;
Connection connection;
try (SpinLock.Lock lock = this.lock.lock())
final ReentrantLock lock = this.lock;
lock.lock();
try
{
connection = idleConnections.pollFirst();
if (connection == null)
return null;
acquired = activeConnections.offer(connection);
}
finally
{
lock.unlock();
}

if (acquired)
{
Expand All @@ -179,13 +192,20 @@ protected void acquired(Connection connection)
public boolean release(Connection connection)
{
boolean idle;
try (SpinLock.Lock lock = this.lock.lock())
final ReentrantLock lock = this.lock;
lock.lock();
try
{
if (!activeConnections.remove(connection))
return false;
// Make sure we use "hot" connections first.
idle = idleConnections.offerFirst(connection);
}
finally
{
lock.unlock();
}

released(connection);
return idle(connection, idle);
}
Expand Down Expand Up @@ -215,11 +235,18 @@ public boolean remove(Connection connection)
{
boolean activeRemoved;
boolean idleRemoved;
try (SpinLock.Lock lock = this.lock.lock())
final ReentrantLock lock = this.lock;
lock.lock();
try
{
activeRemoved = activeConnections.remove(connection);
idleRemoved = idleConnections.remove(connection);
}
finally
{
lock.unlock();
}

if (activeRemoved)
released(connection);
boolean removed = activeRemoved || idleRemoved;
Expand All @@ -234,18 +261,30 @@ public boolean remove(Connection connection)

public boolean isActive(Connection connection)
{
try (SpinLock.Lock lock = this.lock.lock())
final ReentrantLock lock = this.lock;
lock.lock();
try
{
return activeConnections.contains(connection);
}
finally
{
lock.unlock();
}
}

public boolean isIdle(Connection connection)
{
try (SpinLock.Lock lock = this.lock.lock())
final ReentrantLock lock = this.lock;
lock.lock();
try
{
return idleConnections.contains(connection);
}
finally
{
lock.unlock();
}
}

public boolean isEmpty()
Expand All @@ -257,13 +296,20 @@ public void close()
{
List<Connection> idles = new ArrayList<>();
List<Connection> actives = new ArrayList<>();
try (SpinLock.Lock lock = this.lock.lock())
final ReentrantLock lock = this.lock;
lock.lock();
try
{
idles.addAll(idleConnections);
idleConnections.clear();
actives.addAll(activeConnections);
activeConnections.clear();
}
finally
{
lock.unlock();
}

connectionCount.set(0);

for (Connection connection : idles)
Expand All @@ -285,11 +331,18 @@ public void dump(Appendable out, String indent) throws IOException
{
List<Connection> actives = new ArrayList<>();
List<Connection> idles = new ArrayList<>();
try (SpinLock.Lock lock = this.lock.lock())
final ReentrantLock lock = this.lock;
lock.lock();
try
{
actives.addAll(activeConnections);
idles.addAll(idleConnections);
}
finally
{
lock.unlock();
}

ContainerLifeCycle.dumpObject(out, this);
ContainerLifeCycle.dump(out, indent, actives, idles);
}
Expand All @@ -298,14 +351,20 @@ public void dump(Appendable out, String indent) throws IOException
public boolean sweep()
{
List<Sweeper.Sweepable> toSweep = new ArrayList<>();
try (SpinLock.Lock lock = this.lock.lock())
final ReentrantLock lock = this.lock;
lock.lock();
try
{
for (Connection connection : getActiveConnections())
{
if (connection instanceof Sweeper.Sweepable)
toSweep.add(((Sweeper.Sweepable)connection));
}
}
finally
{
lock.unlock();
}

for (Sweeper.Sweepable candidate : toSweep)
{
Expand All @@ -329,11 +388,18 @@ public String toString()
{
int activeSize;
int idleSize;
try (SpinLock.Lock lock = this.lock.lock())
final ReentrantLock lock = this.lock;
lock.lock();
try
{
activeSize = activeConnections.size();
idleSize = idleConnections.size();
}
finally
{
lock.unlock();
}

return String.format("%s[c=%d/%d,a=%d,i=%d]",
getClass().getSimpleName(),
connectionCount.get(),
Expand Down
9 changes: 4 additions & 5 deletions jetty-client/src/main/java/org/eclipse/jetty/client/HttpChannel.java
View file
Expand Up @@ -21,13 +21,11 @@
import org.eclipse.jetty.client.api.Result;
import org.eclipse.jetty.util.log.Log;
import org.eclipse.jetty.util.log.Logger;
import org.eclipse.jetty.util.thread.SpinLock;

public abstract class HttpChannel
{
protected static final Logger LOG = Log.getLogger(HttpChannel.class);

private final SpinLock _lock = new SpinLock();
private final HttpDestination _destination;
private HttpExchange _exchange;

Expand All @@ -53,7 +51,7 @@ public boolean associate(HttpExchange exchange)
{
boolean result = false;
boolean abort = true;
try (SpinLock.Lock lock = _lock.lock())
synchronized (this)
{
if (_exchange == null)
{
Expand All @@ -76,7 +74,7 @@ public boolean associate(HttpExchange exchange)
public boolean disassociate(HttpExchange exchange)
{
boolean result = false;
try (SpinLock.Lock lock = _lock.lock())
synchronized (this)
{
HttpExchange existing = _exchange;
_exchange = null;
Expand All @@ -86,14 +84,15 @@ public boolean disassociate(HttpExchange exchange)
result = true;
}
}

if (LOG.isDebugEnabled())
LOG.debug("{} disassociated {} from {}", exchange, result, this);
return result;
}

public HttpExchange getHttpExchange()
{
try (SpinLock.Lock lock = _lock.lock())
synchronized (this)
{
return _exchange;
}
Expand Down
26 changes: 12 additions & 14 deletions jetty-client/src/main/java/org/eclipse/jetty/client/HttpExchange.java
View file
Expand Up @@ -24,7 +24,6 @@
import org.eclipse.jetty.client.api.Result;
import org.eclipse.jetty.util.log.Log;
import org.eclipse.jetty.util.log.Logger;
import org.eclipse.jetty.util.thread.SpinLock;

public class HttpExchange
{
Expand All @@ -34,7 +33,6 @@ public class HttpExchange
private final HttpRequest request;
private final List<Response.ResponseListener> listeners;
private final HttpResponse response;
private final SpinLock _lock = new SpinLock();
private State requestState = State.PENDING;
private State responseState = State.PENDING;
private HttpChannel _channel;
Expand Down Expand Up @@ -64,7 +62,7 @@ public HttpRequest getRequest()

public Throwable getRequestFailure()
{
try (SpinLock.Lock lock = _lock.lock())
synchronized (this)
{
return requestFailure;
}
Expand All @@ -82,7 +80,7 @@ public HttpResponse getResponse()

public Throwable getResponseFailure()
{
try (SpinLock.Lock lock = _lock.lock())
synchronized (this)
{
return responseFailure;
}
Expand All @@ -99,7 +97,7 @@ boolean associate(HttpChannel channel)
{
boolean result = false;
boolean abort = false;
try (SpinLock.Lock lock = _lock.lock())
synchronized (this)
{
// Only associate if the exchange state is initial,
// as the exchange could be already failed.
Expand All @@ -123,7 +121,7 @@ boolean associate(HttpChannel channel)
void disassociate(HttpChannel channel)
{
boolean abort = false;
try (SpinLock.Lock lock = _lock.lock())
synchronized (this)
{
if (_channel != channel || requestState != State.TERMINATED || responseState != State.TERMINATED)
abort = true;
Expand All @@ -136,15 +134,15 @@ void disassociate(HttpChannel channel)

private HttpChannel getHttpChannel()
{
try (SpinLock.Lock lock = _lock.lock())
synchronized (this)
{
return _channel;
}
}

public boolean requestComplete(Throwable failure)
{
try (SpinLock.Lock lock = _lock.lock())
synchronized (this)
{
return completeRequest(failure);
}
Expand All @@ -163,7 +161,7 @@ private boolean completeRequest(Throwable failure)

public boolean responseComplete(Throwable failure)
{
try (SpinLock.Lock lock = _lock.lock())
synchronized (this)
{
return completeResponse(failure);
}
Expand All @@ -183,7 +181,7 @@ private boolean completeResponse(Throwable failure)
public Result terminateRequest()
{
Result result = null;
try (SpinLock.Lock lock = _lock.lock())
synchronized (this)
{
if (requestState == State.COMPLETED)
requestState = State.TERMINATED;
Expand All @@ -200,7 +198,7 @@ public Result terminateRequest()
public Result terminateResponse()
{
Result result = null;
try (SpinLock.Lock lock = _lock.lock())
synchronized (this)
{
if (responseState == State.COMPLETED)
responseState = State.TERMINATED;
Expand All @@ -220,7 +218,7 @@ public boolean abort(Throwable failure)
// This will avoid that this exchange can be associated to a channel.
boolean abortRequest;
boolean abortResponse;
try (SpinLock.Lock lock = _lock.lock())
synchronized (this)
{
abortRequest = completeRequest(failure);
abortResponse = completeResponse(failure);
Expand Down Expand Up @@ -273,7 +271,7 @@ private void notifyFailureComplete(Throwable failure)

public void resetResponse()
{
try (SpinLock.Lock lock = _lock.lock())
synchronized (this)
{
responseState = State.PENDING;
responseFailure = null;
Expand All @@ -290,7 +288,7 @@ public void proceed(Throwable failure)
@Override
public String toString()
{
try (SpinLock.Lock lock = _lock.lock())
synchronized (this)
{
return String.format("%s@%x req=%s/%s@%h res=%s/%s@%h",
HttpExchange.class.getSimpleName(),
Expand Down

0 comments on commit 2c26e82

Please sign in to comment.