You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@hc.apache.org by ol...@apache.org on 2016/06/05 10:50:55 UTC
svn commit: r1746898 - in
/httpcomponents/httpcore/trunk/httpcore5/src/main/java/org/apache/hc/core5/reactor:
AbstractIOReactor.java AbstractMultiworkerIOReactor.java BaseIOReactor.java
IOSessionImpl.java
Author: olegk
Date: Sun Jun 5 10:50:55 2016
New Revision: 1746898
URL: http://svn.apache.org/viewvc?rev=1746898&view=rev
Log:
Removed synchronization in IOSessionImpl; removed interest operation queueing
Modified:
httpcomponents/httpcore/trunk/httpcore5/src/main/java/org/apache/hc/core5/reactor/AbstractIOReactor.java
httpcomponents/httpcore/trunk/httpcore5/src/main/java/org/apache/hc/core5/reactor/AbstractMultiworkerIOReactor.java
httpcomponents/httpcore/trunk/httpcore5/src/main/java/org/apache/hc/core5/reactor/BaseIOReactor.java
httpcomponents/httpcore/trunk/httpcore5/src/main/java/org/apache/hc/core5/reactor/IOSessionImpl.java
Modified: httpcomponents/httpcore/trunk/httpcore5/src/main/java/org/apache/hc/core5/reactor/AbstractIOReactor.java
URL: http://svn.apache.org/viewvc/httpcomponents/httpcore/trunk/httpcore5/src/main/java/org/apache/hc/core5/reactor/AbstractIOReactor.java?rev=1746898&r1=1746897&r2=1746898&view=diff
==============================================================================
--- httpcomponents/httpcore/trunk/httpcore5/src/main/java/org/apache/hc/core5/reactor/AbstractIOReactor.java (original)
+++ httpcomponents/httpcore/trunk/httpcore5/src/main/java/org/apache/hc/core5/reactor/AbstractIOReactor.java Sun Jun 5 10:50:55 2016
@@ -43,7 +43,6 @@ import java.util.concurrent.ConcurrentLi
import org.apache.hc.core5.annotation.ThreadSafe;
import org.apache.hc.core5.util.Args;
-import org.apache.hc.core5.util.Asserts;
/**
* Generic implementation of {@link IOReactor} that can used as a subclass
@@ -59,10 +58,8 @@ public abstract class AbstractIOReactor
private final Object statusMutex;
private final long selectTimeout;
- private final boolean interestOpsQueueing;
private final Selector selector;
private final Set<IOSession> sessions;
- private final Queue<InterestOpEntry> interestOpsQueue;
private final Queue<IOSession> closedSessions;
private final Queue<ChannelEntry> newChannels;
@@ -73,26 +70,10 @@ public abstract class AbstractIOReactor
* @throws IOReactorException in case if a non-recoverable I/O error.
*/
public AbstractIOReactor(final long selectTimeout) throws IOReactorException {
- this(selectTimeout, false);
- }
-
- /**
- * Creates new AbstractIOReactor instance.
- *
- * @param selectTimeout the select timeout.
- * @param interestOpsQueueing Ops queueing flag.
- *
- * @throws IOReactorException in case if a non-recoverable I/O error.
- *
- * @since 4.1
- */
- public AbstractIOReactor(final long selectTimeout, final boolean interestOpsQueueing) throws IOReactorException {
super();
Args.positive(selectTimeout, "Select timeout");
this.selectTimeout = selectTimeout;
- this.interestOpsQueueing = interestOpsQueueing;
this.sessions = Collections.synchronizedSet(new HashSet<IOSession>());
- this.interestOpsQueue = new ConcurrentLinkedQueue<>();
this.closedSessions = new ConcurrentLinkedQueue<>();
this.newChannels = new ConcurrentLinkedQueue<>();
try {
@@ -200,15 +181,6 @@ public abstract class AbstractIOReactor
}
/**
- * Returns {@code true} if interest Ops queueing is enabled, {@code false} otherwise.
- *
- * @since 4.1
- */
- public boolean getInterestOpsQueueing() {
- return this.interestOpsQueueing;
- }
-
- /**
* Adds new channel entry. The channel will be asynchronously registered
* with the selector.
*
@@ -290,12 +262,6 @@ public abstract class AbstractIOReactor
&& this.sessions.isEmpty()) {
break;
}
-
- if (this.interestOpsQueueing) {
- // process all pending interestOps() operations
- processPendingInterestOps();
- }
-
}
} catch (final ClosedSelectorException ignore) {
@@ -386,21 +352,9 @@ public abstract class AbstractIOReactor
};
- InterestOpsCallback interestOpsCallback = null;
- if (this.interestOpsQueueing) {
- interestOpsCallback = new InterestOpsCallback() {
-
- @Override
- public void addInterestOps(final InterestOpEntry entry) {
- queueInterestOps(entry);
- }
-
- };
- }
-
final IOSession session;
try {
- session = new IOSessionImpl(key, interestOpsCallback, sessionClosedCallback);
+ session = new IOSessionImpl(key, sessionClosedCallback);
int timeout = 0;
try {
timeout = channel.socket().getSoTimeout();
@@ -443,35 +397,6 @@ public abstract class AbstractIOReactor
}
}
- private void processPendingInterestOps() {
- // validity check
- if (!this.interestOpsQueueing) {
- return;
- }
- InterestOpEntry entry;
- while ((entry = this.interestOpsQueue.poll()) != null) {
- // obtain the operation's details
- final SelectionKey key = entry.getSelectionKey();
- final int eventMask = entry.getEventMask();
- if (key.isValid()) {
- key.interestOps(eventMask);
- }
- }
- }
-
- private boolean queueInterestOps(final InterestOpEntry entry) {
- // validity checks
- Asserts.check(this.interestOpsQueueing, "Interest ops queueing not enabled");
- if (entry == null) {
- return false;
- }
-
- // add this operation to the interestOps() queue
- this.interestOpsQueue.add(entry);
-
- return true;
- }
-
/**
* Triggered to verify whether the I/O session associated with the
* given selection key has not timed out.
Modified: httpcomponents/httpcore/trunk/httpcore5/src/main/java/org/apache/hc/core5/reactor/AbstractMultiworkerIOReactor.java
URL: http://svn.apache.org/viewvc/httpcomponents/httpcore/trunk/httpcore5/src/main/java/org/apache/hc/core5/reactor/AbstractMultiworkerIOReactor.java?rev=1746898&r1=1746897&r2=1746898&view=diff
==============================================================================
--- httpcomponents/httpcore/trunk/httpcore5/src/main/java/org/apache/hc/core5/reactor/AbstractMultiworkerIOReactor.java (original)
+++ httpcomponents/httpcore/trunk/httpcore5/src/main/java/org/apache/hc/core5/reactor/AbstractMultiworkerIOReactor.java Sun Jun 5 10:50:55 2016
@@ -94,7 +94,6 @@ public abstract class AbstractMultiworke
protected final IOReactorConfig config;
protected final Selector selector;
protected final long selectTimeout;
- protected final boolean interestOpsQueueing;
private final int workerCount;
private final ThreadFactory threadFactory;
@@ -130,7 +129,6 @@ public abstract class AbstractMultiworke
throw new IOReactorException("Failure opening selector", ex);
}
this.selectTimeout = this.config.getSelectInterval();
- this.interestOpsQueueing = this.config.isInterestOpQueued();
this.statusLock = new Object();
if (threadFactory != null) {
this.threadFactory = threadFactory;
@@ -264,7 +262,7 @@ public abstract class AbstractMultiworke
this.status = IOReactorStatus.ACTIVE;
// Start I/O dispatchers
for (int i = 0; i < this.dispatchers.length; i++) {
- final BaseIOReactor dispatcher = new BaseIOReactor(this.selectTimeout, this.interestOpsQueueing);
+ final BaseIOReactor dispatcher = new BaseIOReactor(this.selectTimeout);
dispatcher.setExceptionHandler(exceptionHandler);
this.dispatchers[i] = dispatcher;
}
Modified: httpcomponents/httpcore/trunk/httpcore5/src/main/java/org/apache/hc/core5/reactor/BaseIOReactor.java
URL: http://svn.apache.org/viewvc/httpcomponents/httpcore/trunk/httpcore5/src/main/java/org/apache/hc/core5/reactor/BaseIOReactor.java?rev=1746898&r1=1746897&r2=1746898&view=diff
==============================================================================
--- httpcomponents/httpcore/trunk/httpcore5/src/main/java/org/apache/hc/core5/reactor/BaseIOReactor.java (original)
+++ httpcomponents/httpcore/trunk/httpcore5/src/main/java/org/apache/hc/core5/reactor/BaseIOReactor.java Sun Jun 5 10:50:55 2016
@@ -64,22 +64,7 @@ public class BaseIOReactor extends Abstr
* @throws IOReactorException in case if a non-recoverable I/O error.
*/
public BaseIOReactor(final long selectTimeout) throws IOReactorException {
- this(selectTimeout, false);
- }
-
- /**
- * Creates new BaseIOReactor instance.
- *
- * @param selectTimeout the select timeout.
- * @param interestOpsQueueing Ops queueing flag.
- *
- * @throws IOReactorException in case if a non-recoverable I/O error.
- *
- * @since 4.1
- */
- public BaseIOReactor(
- final long selectTimeout, final boolean interestOpsQueueing) throws IOReactorException {
- super(selectTimeout, interestOpsQueueing);
+ super(selectTimeout);
this.bufferingSessions = new HashSet<>();
this.timeoutCheckInterval = selectTimeout;
this.lastTimeoutCheck = System.currentTimeMillis();
Modified: httpcomponents/httpcore/trunk/httpcore5/src/main/java/org/apache/hc/core5/reactor/IOSessionImpl.java
URL: http://svn.apache.org/viewvc/httpcomponents/httpcore/trunk/httpcore5/src/main/java/org/apache/hc/core5/reactor/IOSessionImpl.java?rev=1746898&r1=1746897&r2=1746898&view=diff
==============================================================================
--- httpcomponents/httpcore/trunk/httpcore5/src/main/java/org/apache/hc/core5/reactor/IOSessionImpl.java (original)
+++ httpcomponents/httpcore/trunk/httpcore5/src/main/java/org/apache/hc/core5/reactor/IOSessionImpl.java Sun Jun 5 10:50:55 2016
@@ -34,9 +34,9 @@ import java.net.SocketAddress;
import java.nio.channels.ByteChannel;
import java.nio.channels.SelectionKey;
import java.nio.channels.SocketChannel;
-import java.util.Collections;
-import java.util.HashMap;
import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.atomic.AtomicInteger;
import org.apache.hc.core5.annotation.ThreadSafe;
import org.apache.hc.core5.util.Args;
@@ -52,43 +52,39 @@ public class IOSessionImpl implements IO
private final SelectionKey key;
private final ByteChannel channel;
private final Map<String, Object> attributes;
- private final InterestOpsCallback interestOpsCallback;
private final SessionClosedCallback sessionClosedCallback;
- private volatile int status;
- private volatile int currentEventMask;
+ private final long startedTime;
+ private final AtomicInteger status;
+ private final AtomicInteger eventMask;
+
private volatile SessionBufferStatus bufferStatus;
private volatile int socketTimeout;
- private final long startedTime;
-
- private long lastReadTime;
- private long lastWriteTime;
- private long lastAccessTime;
+ private volatile long lastReadTime;
+ private volatile long lastWriteTime;
+ private volatile long lastAccessTime;
/**
* Creates new instance of IOSessionImpl.
*
* @param key the selection key.
- * @param interestOpsCallback interestOps callback.
* @param sessionClosedCallback session closed callback.
*
* @since 4.1
*/
public IOSessionImpl(
final SelectionKey key,
- final InterestOpsCallback interestOpsCallback,
final SessionClosedCallback sessionClosedCallback) {
super();
Args.notNull(key, "Selection key");
this.key = key;
this.channel = (ByteChannel) this.key.channel();
- this.interestOpsCallback = interestOpsCallback;
this.sessionClosedCallback = sessionClosedCallback;
- this.attributes = Collections.synchronizedMap(new HashMap<String, Object>());
- this.currentEventMask = key.interestOps();
+ this.attributes = new ConcurrentHashMap<>();
this.socketTimeout = 0;
- this.status = ACTIVE;
+ this.eventMask = new AtomicInteger(key.interestOps());
+ this.status = new AtomicInteger(ACTIVE);
final long now = System.currentTimeMillis();
this.startedTime = now;
this.lastReadTime = now;
@@ -96,18 +92,6 @@ public class IOSessionImpl implements IO
this.lastAccessTime = now;
}
- /**
- * Creates new instance of IOSessionImpl.
- *
- * @param key the selection key.
- * @param sessionClosedCallback session closed callback.
- */
- public IOSessionImpl(
- final SelectionKey key,
- final SessionClosedCallback sessionClosedCallback) {
- this(key, null, sessionClosedCallback);
- }
-
@Override
public ByteChannel channel() {
return this.channel;
@@ -130,70 +114,55 @@ public class IOSessionImpl implements IO
}
@Override
- public synchronized int getEventMask() {
- return this.interestOpsCallback != null ? this.currentEventMask : this.key.interestOps();
+ public int getEventMask() {
+ return this.key.interestOps();
}
@Override
- public synchronized void setEventMask(final int ops) {
- if (this.status == CLOSED) {
+ public void setEventMask(final int newValue) {
+ if (this.status.get() == CLOSED) {
return;
}
- if (this.interestOpsCallback != null) {
- // update the current event mask
- this.currentEventMask = ops;
-
- // local variable
- final InterestOpEntry entry = new InterestOpEntry(this.key, this.currentEventMask);
-
- // add this operation to the interestOps() queue
- this.interestOpsCallback.addInterestOps(entry);
- } else {
- this.key.interestOps(ops);
+ final int currentValue = this.eventMask.get();
+ if (newValue == currentValue) {
+ return;
+ }
+ if (this.eventMask.compareAndSet(currentValue, newValue)) {
+ this.key.interestOps(newValue);
+ this.key.selector().wakeup();
}
- this.key.selector().wakeup();
}
@Override
- public synchronized void setEvent(final int op) {
- if (this.status == CLOSED) {
+ public void setEvent(final int op) {
+ if (this.status.get() == CLOSED) {
return;
}
- if (this.interestOpsCallback != null) {
- // update the current event mask
- this.currentEventMask |= op;
-
- // local variable
- final InterestOpEntry entry = new InterestOpEntry(this.key, this.currentEventMask);
-
- // add this operation to the interestOps() queue
- this.interestOpsCallback.addInterestOps(entry);
- } else {
- final int ops = this.key.interestOps();
- this.key.interestOps(ops | op);
+ for (;;) {
+ final int currentValue = this.eventMask.get();
+ final int newValue = currentValue | op;
+ if (this.eventMask.compareAndSet(currentValue, newValue)) {
+ this.key.interestOps(newValue);
+ this.key.selector().wakeup();
+ return;
+ }
}
- this.key.selector().wakeup();
}
@Override
- public synchronized void clearEvent(final int op) {
- if (this.status == CLOSED) {
+ public void clearEvent(final int op) {
+ if (this.status.get() == CLOSED) {
return;
}
- if (this.interestOpsCallback != null) {
- // update the current event mask
- this.currentEventMask &= ~op;
-
- // local variable
- final InterestOpEntry entry = new InterestOpEntry(this.key, this.currentEventMask);
-
- // add this operation to the interestOps() queue
- this.interestOpsCallback.addInterestOps(entry);
- } else {
- final int ops = this.key.interestOps();
- this.key.interestOps(ops & ~op);
+ for (;;) {
+ final int currentValue = this.eventMask.get();
+ final int newValue = currentValue & ~op;
+ if (this.eventMask.compareAndSet(currentValue, newValue)) {
+ this.key.interestOps(newValue);
+ this.key.selector().wakeup();
+ return;
+ }
}
- this.key.selector().wakeup();
}
@Override
@@ -202,40 +171,36 @@ public class IOSessionImpl implements IO
}
@Override
- public synchronized void setSocketTimeout(final int timeout) {
+ public void setSocketTimeout(final int timeout) {
this.socketTimeout = timeout;
this.lastAccessTime = System.currentTimeMillis();
}
@Override
- public synchronized void close() {
- if (this.status == CLOSED) {
- return;
- }
- this.status = CLOSED;
- this.key.cancel();
- try {
- this.key.channel().close();
- } catch (final IOException ex) {
- // Munching exceptions is not nice
- // but in this case it is justified
- }
- if (this.sessionClosedCallback != null) {
- this.sessionClosedCallback.sessionClosed(this);
- }
- if (this.key.selector().isOpen()) {
- this.key.selector().wakeup();
+ public void close() {
+ if (this.status.compareAndSet(ACTIVE, CLOSED)) {
+ this.key.cancel();
+ try {
+ this.key.channel().close();
+ } catch (final IOException ignore) {
+ }
+ if (this.sessionClosedCallback != null) {
+ this.sessionClosedCallback.sessionClosed(this);
+ }
+ if (this.key.selector().isOpen()) {
+ this.key.selector().wakeup();
+ }
}
}
@Override
public int getStatus() {
- return this.status;
+ return this.status.get();
}
@Override
public boolean isClosed() {
- return this.status == CLOSED;
+ return this.status.get() == CLOSED;
}
@Override
@@ -274,32 +239,36 @@ public class IOSessionImpl implements IO
@Override
public void setAttribute(final String name, final Object obj) {
- this.attributes.put(name, obj);
+ if (obj == null) {
+ this.attributes.remove(name);
+ } else {
+ this.attributes.put(name, obj);
+ }
}
- public synchronized long getStartedTime() {
+ public long getStartedTime() {
return this.startedTime;
}
- public synchronized long getLastReadTime() {
+ public long getLastReadTime() {
return this.lastReadTime;
}
- public synchronized long getLastWriteTime() {
+ public long getLastWriteTime() {
return this.lastWriteTime;
}
- public synchronized long getLastAccessTime() {
+ public long getLastAccessTime() {
return this.lastAccessTime;
}
- synchronized void resetLastRead() {
+ void resetLastRead() {
final long now = System.currentTimeMillis();
this.lastReadTime = now;
this.lastAccessTime = now;
}
- synchronized void resetLastWrite() {
+ void resetLastWrite() {
final long now = System.currentTimeMillis();
this.lastWriteTime = now;
this.lastAccessTime = now;
@@ -333,7 +302,15 @@ public class IOSessionImpl implements IO
}
@Override
- public synchronized String toString() {
+ public Socket getSocket() {
+ if (this.channel instanceof SocketChannel) {
+ return ((SocketChannel) this.channel).socket();
+ }
+ return null;
+ }
+
+ @Override
+ public String toString() {
final StringBuilder buffer = new StringBuilder();
final SocketAddress remoteAddress = getRemoteAddress();
final SocketAddress localAddress = getLocalAddress();
@@ -343,7 +320,7 @@ public class IOSessionImpl implements IO
formatAddress(buffer, remoteAddress);
}
buffer.append("[");
- switch (this.status) {
+ switch (this.status.get()) {
case ACTIVE:
buffer.append("ACTIVE");
break;
@@ -356,8 +333,7 @@ public class IOSessionImpl implements IO
}
buffer.append("][");
if (this.key.isValid()) {
- formatOps(buffer, this.interestOpsCallback != null ?
- this.currentEventMask : this.key.interestOps());
+ formatOps(buffer, this.key.interestOps());
buffer.append(":");
formatOps(buffer, this.key.readyOps());
}
@@ -365,12 +341,4 @@ public class IOSessionImpl implements IO
return buffer.toString();
}
- @Override
- public Socket getSocket() {
- if (this.channel instanceof SocketChannel) {
- return ((SocketChannel) this.channel).socket();
- }
- return null;
- }
-
}