{
+
+ private final int index;
+
+ private final String name;
+
+ private final AttributeInitFunction function;
+
+ AttributeKey(int index, String name, AttributeInitFunction function) {
+ this.name = name;
+ this.index = index;
+ this.function = function;
+ }
+
+ public int getIndex() {
+ return index;
+ }
+
+ public String getName() {
+ return name;
+ }
+
+ public AttributeInitFunction getFunction() {
+ return function;
+ }
+
+ @Override
+ public String toString() {
+ return "name: " + name + ", index: " + index;
+ }
+
+}
diff --git a/firenio-core/src/main/java/com/firenio/collection/AttributeMap.java b/firenio-core/src/main/java/com/firenio/collection/AttributeMap.java
new file mode 100644
index 000000000..d60515a57
--- /dev/null
+++ b/firenio-core/src/main/java/com/firenio/collection/AttributeMap.java
@@ -0,0 +1,150 @@
+/*
+ * Copyright 2015 The FireNio Project
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package com.firenio.collection;
+
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
+
+/**
+ * use for static variable, eg:
+ *
+ * static final AttributeKey KEY_NAME = AttributeMap.valueOfKey(Channel.class, "KEY_NAME");
+ *
+ */
+//TODO 考虑多个ChannelContext时，每个Context里Channel存储的数据key不一样，会造成内存浪费
+public abstract class AttributeMap {
+
+ private static final Map, AttributeKeys> INDEX_MAPPING = new ConcurrentHashMap<>();
+
+ private final Object[] attributes;
+
+ private volatile int memoryBarrier = 0;
+
+ public AttributeMap() {
+ AttributeKeys keys = getKeys();
+ if (keys == null) {
+ attributes = null;
+ return;
+ }
+ keys.initialized = true;
+ this.attributes = new Object[keys.getCounter()];
+ for (AttributeKey key : keys.keys.values()) {
+ AttributeInitFunction function = key.getFunction();
+ function.setValue(this, key.getIndex());
+ }
+ }
+
+ public T getAttribute(AttributeKey key) {
+ return (T) getAttribute(key.getIndex());
+ }
+
+ public T getAttributeUnsafe(AttributeKey key) {
+ return (T) getAttributeUnsafe(key.getIndex());
+ }
+
+ public Object getAttribute(int key) {
+ memoryBarrier = 1;
+ return attributes[key];
+ }
+
+ public Object getAttributeUnsafe(int key) {
+ return attributes[key];
+ }
+
+ public void setAttribute(AttributeKey key, Object value) {
+ attributes[key.getIndex()] = value;
+ }
+
+ public void setAttribute(int key, Object value) {
+ memoryBarrier = 1;
+ attributes[key] = value;
+ }
+
+ public void setAttributeUnsafe(int key, Object value) {
+ attributes[key] = value;
+ }
+
+ public static AttributeKeys getKeys(Class clazz) {
+ return INDEX_MAPPING.get(clazz);
+ }
+
+ public static AttributeKey valueOfKey(Class clazz, String name) {
+ return valueOfKey(clazz, name, null);
+ }
+
+ public static AttributeKey valueOfKey(Class clazz, String name, AttributeInitFunction function) {
+ AttributeKeys attributeKeys = INDEX_MAPPING.get(clazz);
+ if (attributeKeys == null) {
+ synchronized (INDEX_MAPPING) {
+ attributeKeys = INDEX_MAPPING.get(clazz);
+ if (attributeKeys == null) {
+ attributeKeys = new AttributeKeys();
+ INDEX_MAPPING.put(clazz, attributeKeys);
+ }
+ }
+ }
+ return attributeKeys.valueOf(name, function);
+ }
+
+ protected int getMemoryBarrier() {
+ return memoryBarrier;
+ }
+
+ protected abstract AttributeKeys getKeys();
+
+ public static class AttributeKeys {
+
+ final Map keys = new HashMap<>();
+ final Map ro_keys = Collections.unmodifiableMap(keys);
+
+ int counter = 0;
+ volatile boolean initialized = false;
+
+ synchronized AttributeKey valueOf(String name, AttributeInitFunction function) {
+ if (initialized) {
+ throw new RuntimeException("incorrect usage, AttributeKey can only be static constant");
+ }
+ if (function == null) {
+ function = NULL_VALUE_ATTRIBUTE_INIT_FUNCTION;
+ }
+ keys.put(name, new AttributeKey(counter++, name, function));
+ return keys.get(name);
+ }
+
+ public Map getKeys() {
+ return ro_keys;
+ }
+
+ public int getCounter() {
+ return counter;
+ }
+
+ }
+
+ public interface AttributeInitFunction {
+
+ void setValue(AttributeMap map, int key);
+
+ }
+
+ static final AttributeInitFunction NULL_VALUE_ATTRIBUTE_INIT_FUNCTION = new AttributeInitFunction() {
+ @Override
+ public void setValue(AttributeMap map, int key) { }
+ };
+
+}
diff --git a/firenio-core/src/main/java/com/firenio/collection/AttributesImpl.java b/firenio-core/src/main/java/com/firenio/collection/AttributesImpl.java
deleted file mode 100644
index d5d6f5f04..000000000
--- a/firenio-core/src/main/java/com/firenio/collection/AttributesImpl.java
+++ /dev/null
@@ -1,56 +0,0 @@
-/*
- * Copyright 2015 The FireNio Project
- *
- * Licensed under the Apache License, Version 2.0 (the "License");
- * you may not use this file except in compliance with the License.
- * You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-package com.firenio.collection;
-
-import java.util.HashMap;
-import java.util.Map;
-import java.util.Set;
-
-public class AttributesImpl implements Attributes {
-
- private final Map attributes = new HashMap<>();
-
- @Override
- public Map attributes() {
- return attributes;
- }
-
- @Override
- public void clearAttributes() {
- this.attributes.clear();
- }
-
- @Override
- public Object getAttribute(Object key) {
- return this.attributes.get(key);
- }
-
- @Override
- public Set
*/
private boolean isEnoughSslUnwrap(ByteBuf src) throws SSLException {
- if (src.remaining() < 5) {
+ if (src.readableBytes() < 5) {
return false;
}
- int pos = src.position();
+ int pos = src.readIndex();
// TLS - Check ContentType
int type = src.getUnsignedByte(pos);
if (type < 20 || type > 23) {
@@ -501,14 +535,14 @@ private boolean isEnoughSslUnwrap(ByteBuf src) throws SSLException {
throw NOT_TLS;
}
int len = packetLength + 5;
- if (src.remaining() < len) {
+ if (src.readableBytes() < len) {
return false;
}
if (len > SSL_PACKET_LIMIT) {
throw SSL_PACKET_OVER_LIMIT;
}
- src.markL();
- src.limit(pos + len);
+ src.markWriteIndex();
+ src.writeIndex(pos + len);
return true;
}
@@ -544,16 +578,17 @@ protected void read() throws Exception {
}
private void read_plain() throws Exception {
- ByteBuf src = eventLoop.getReadBuf();
+ final Channel ch = this;
+ final ProtocolCodec codec = this.codec;
+ ByteBuf dst = codec.getPlainReadBuf(eventLoop, ch);
for (; ; ) {
- src.clear();
- read_plain_remain(src);
- if (!read_data(src)) {
+ codec.readPlainRemain(ch, dst);
+ if (!read_data(dst)) {
return;
}
- boolean full = src.isFullLimit();
- accept(src);
- if (!full) {
+ accept(dst);
+ // for epoll et mode
+ if (dst.hasWritableBytes()) {
break;
}
}
@@ -567,78 +602,70 @@ private void read_ssl() throws Exception {
if (!read_data(src)) {
return;
}
- boolean full = src.isFullLimit();
for (; ; ) {
if (isEnoughSslUnwrap(src)) {
ByteBuf res = unwrap(src);
if (res != null) {
accept(res);
}
- src.resetL();
- if (!src.hasRemaining()) {
+ src.resetWriteIndex();
+ if (!src.hasReadableBytes()) {
break;
}
} else {
- if (src.hasRemaining()) {
+ if (src.hasReadableBytes()) {
slice_remain_ssl(src);
}
break;
}
}
- if (!full) {
+ // for epoll et mode
+ if (src.hasWritableBytes()) {
break;
}
}
}
- private boolean read_data(ByteBuf src) {
- int len = native_read();
+ private boolean read_data(ByteBuf dst) {
+ int len = native_read(dst);
if (len < 1) {
if (len == -1) {
Util.close(this);
return false;
}
- store_remain(src.flip());
+ store_remain(dst);
return false;
}
- if (Native.EPOLL_AVAILABLE) {
- src.absLimit(src.absPos() + len);
- src.absPos(0);
- } else {
- src.reverse();
- src.flip();
- }
+ dst.skipWrite(len);
return true;
}
private void store_remain(ByteBuf src) {
- if (src.hasRemaining()) {
+ if (src.hasReadableBytes()) {
if (enable_ssl) {
slice_remain_ssl(src);
} else {
- slice_remain_plain(src);
+ codec.storePlainReadRemain(this, src);
}
}
}
- private void read_plain_remain(ByteBuf dst) {
- ByteBuf remaining_buf = this.plain_remain_buf;
- if (remaining_buf == null) {
- return;
+ protected void read_plain_remain(ByteBuf dst) {
+ ByteBuf remain = this.plain_remain_buf;
+ if (remain != null) {
+ dst.writeBytes(remain);
+ remain.release();
+ this.plain_remain_buf = null;
}
- dst.putBytes(remaining_buf);
- remaining_buf.release();
- this.plain_remain_buf = null;
}
private void read_ssl_remain(ByteBuf dst) {
- ByteBuf remaining_buf = this.ssl_remain_buf;
- if (remaining_buf == null) {
- return;
+ ByteBuf remain = this.ssl_remain_buf;
+ if (remain != null) {
+ dst.writeBytes(remain);
+ remain.release();
+ this.ssl_remain_buf = null;
}
- dst.putBytes(remaining_buf);
- remaining_buf.release();
- this.ssl_remain_buf = null;
}
public void release(Frame frame) {
@@ -716,21 +743,26 @@ private void safe_close() {
release(ssl_remain_buf);
release(plain_remain_buf);
close_channel();
+ channel_establish();
fire_closed();
stop_context();
}
}
+ private void channel_establish() {
+ context.channelEstablish(this, CLOSED_CHANNEL);
+ }
+
abstract void close_channel();
public abstract void setOption(int name, int value) throws IOException;
private ByteBuf slice_remain(ByteBuf src) {
- int remain = src.remaining();
+ int remain = src.readableBytes();
if (remain > 0) {
ByteBuf remaining = alloc().allocate(remain);
- remaining.putBytes(src);
- return remaining.flip();
+ remaining.writeBytes(src);
+ return remaining;
} else {
return null;
}
@@ -744,15 +776,16 @@ private void stop_context() {
//FIXME 部分buf不需要swap
private ByteBuf swap(ByteBufAllocator allocator, ByteBuf buf) {
- ByteBuf out = allocator.allocate(buf.limit());
- out.putBytes(buf);
- return out.flip();
+ // use writeIndex instead of readableBytes because of the buf readIndex always be zero.
+ ByteBuf out = allocator.allocate(buf.writeIndex());
+ out.writeBytes(buf);
+ return out;
}
private void sync_buf(SSLEngineResult result, ByteBuf src, ByteBuf dst) {
//FIXME 同步。。。。。
- src.reverse();
- dst.reverse();
+ src.reverseRead();
+ dst.reverseWrite();
// int bytesConsumed = result.bytesConsumed();
// int bytesProduced = result.bytesProduced();
// if (bytesConsumed > 0) {
@@ -772,9 +805,8 @@ private ByteBuf unwrap(ByteBuf src) throws IOException {
SSLEngine ssl_engine = getSSLEngine();
ByteBuf dst = FastThreadLocal.get().getSslUnwrapBuf();
if (ssl_handshake_finished) {
- dst.clear();
- read_plain_remain(dst);
- SSLEngineResult result = ssl_engine.unwrap(src.nioBuffer(), dst.nioBuffer());
+ codec.readPlainRemain(this, dst);
+ SSLEngineResult result = ssl_engine.unwrap(src.nioReadBuffer(), dst.nioWriteBuffer());
if (result.getStatus() == Status.BUFFER_OVERFLOW) {
//why throw an exception here instead of handle it?
//the getSslUnwrapBuf will return an thread local buffer for unwrap,
@@ -783,12 +815,12 @@ private ByteBuf unwrap(ByteBuf src) throws IOException {
//one EventLoop only have one buffer,but before do unwrap, every channel maybe cached a large
//buffer under SSL_UNWRAP_BUFFER_SIZE,I do not think it is a good way to cached much memory in
//channel, it is not friendly for load much channels in one system, if you get exception here,
- //you may need find a way to limit you frame size,or cache your incomplete frame's data to
+ //you may need find a way to writeIndex you frame size,or cache your incomplete frame's data to
//file system or others way.
throw SSL_UNWRAP_OVER_LIMIT;
}
sync_buf(result, src, dst);
- return dst.flip();
+ return dst;
} else {
for (; ; ) {
dst.clear();
@@ -805,10 +837,20 @@ private ByteBuf unwrap(ByteBuf src) throws IOException {
finish_handshake();
return null;
} else if (handshakeStatus == HandshakeStatus.NEED_UNWRAP) {
- if (src.hasRemaining()) {
+ if (src.hasReadableBytes()) {
continue;
}
return null;
+ } else if (handshakeStatus == HandshakeStatus.NOT_HANDSHAKING) {
+ // NOTICE: this handle may not the NOT_HANDSHAKING expected
+ // NOT_HANDSHAKING is mean closed in wildfly-openssl.unwrap, so do not do finish_handshake
+ // see: https://stackoverflow.com/questions/31149383/difference-between-not-handshaking-and-finished
+ if (SSL_DEBUG) {
+ logger.error("unwrap handle status: NOT_HANDSHAKING({})", this);
+ }
+ throw SSL_UNWRAP_CLOSED;
+ } else if (result.getStatus() == Status.BUFFER_OVERFLOW) {
+ throw SSL_UNWRAP_OVER_LIMIT;
}
}
}
@@ -816,7 +858,7 @@ private ByteBuf unwrap(ByteBuf src) throws IOException {
private static SSLEngineResult unwrap(SSLEngine ssl_engine, ByteBuf src, ByteBuf dst) throws SSLException {
try {
- return ssl_engine.unwrap(src.nioBuffer(), dst.nioBuffer());
+ return ssl_engine.unwrap(src.nioReadBuffer(), dst.nioWriteBuffer());
} catch (SSLException e) {
logger.error(e.getMessage());
debugException(logger, e);
@@ -832,64 +874,68 @@ private ByteBuf wrap(ByteBuf src) throws IOException {
if (ssl_handshake_finished) {
byte sslWrapExt = this.ssl_wrap_ext;
if (sslWrapExt == 0) {
- out = alloc.allocate(guess_wrap_out(src.remaining(), 0xff + 1));
+ out = alloc.allocate(guess_wrap_out(src.readableBytes(), 0xff + 1));
} else {
- out = alloc.allocate(guess_wrap_out(src.remaining(), sslWrapExt & 0xff));
+ out = alloc.allocate(guess_wrap_out(src.readableBytes(), sslWrapExt & 0xff));
}
final int SSL_PACKET_BUFFER_SIZE = SslContext.SSL_PACKET_BUFFER_SIZE;
for (; ; ) {
- SSLEngineResult result = engine.wrap(src.nioBuffer(), out.nioBuffer());
+ SSLEngineResult result = engine.wrap(src.nioReadBuffer(), out.nioWriteBuffer());
Status status = result.getStatus();
sync_buf(result, src, out);
if (status == Status.CLOSED) {
- return out.flip();
+ return out;
} else if (status == Status.BUFFER_OVERFLOW) {
out.expansion(out.capacity() + SSL_PACKET_BUFFER_SIZE);
continue;
} else {
- if (src.hasRemaining()) {
+ if (src.hasReadableBytes()) {
continue;
}
if (sslWrapExt == 0) {
- int srcLen = src.limit();
- int outLen = out.position();
+ int srcLen = src.writeIndex();
+ int outLen = out.readIndex();
int y = ((srcLen + 1) / SSL_PACKET_BUFFER_SIZE) + 1;
int u = ((outLen - srcLen) / y) * 2;
this.ssl_wrap_ext = (byte) u;
}
- return out.flip();
+ return out;
}
}
} else {
ByteBuf dst = FastThreadLocal.get().getSslWrapBuf();
for (; ; ) {
dst.clear();
- SSLEngineResult result = engine.wrap(src.nioBuffer(), dst.nioBuffer());
+ SSLEngineResult result = engine.wrap(src.nioReadBuffer(), dst.nioWriteBuffer());
Status status = result.getStatus();
HandshakeStatus handshakeStatus = result.getHandshakeStatus();
sync_buf(result, src, dst);
if (status == Status.CLOSED) {
- return swap(alloc, dst.flip());
+ return swap(alloc, dst);
}
if (handshakeStatus == HandshakeStatus.NEED_UNWRAP) {
if (out != null) {
- out.putBytes(dst.flip());
- return out.flip();
+ out.writeBytes(dst);
+ return out;
}
- return swap(alloc, dst.flip());
+ return swap(alloc, dst);
} else if (handshakeStatus == HandshakeStatus.NEED_WRAP) {
if (out == null) {
out = alloc.allocate(256);
}
- out.putBytes(dst.flip());
+ out.writeBytes(dst);
continue;
} else if (handshakeStatus == HandshakeStatus.FINISHED) {
finish_handshake();
if (out != null) {
- out.putBytes(dst.flip());
- return out.flip();
+ out.writeBytes(dst);
+ return out;
}
- return swap(alloc, dst.flip());
+ return swap(alloc, dst);
+ } else if (handshakeStatus == HandshakeStatus.NOT_HANDSHAKING) {
+ // NOTICE: this handle may not the NOT_HANDSHAKING expected
+ // It is shouldn't here to have "NOT_HANDSHAKING", because of the ssl is closed?
+ throw SSL_WRAP_CLOSED;
} else if (handshakeStatus == HandshakeStatus.NEED_TASK) {
run_delegated_tasks(engine);
continue;
@@ -911,7 +957,7 @@ public void write(ByteBuf buf) {
ByteBuf old = buf;
try {
buf = wrap(old);
- } catch (Exception e) {
+ } catch (Throwable e) {
debugException(logger, e);
} finally {
old.release();
@@ -942,9 +988,14 @@ public void writeAndFlush(Frame frame) throws Exception {
flush();
}
+ @Override
+ protected AttributeKeys getKeys() {
+ return getKeys(Channel.class);
+ }
+
abstract boolean isInterestWrite();
- abstract int native_read();
+ abstract int native_read(ByteBuf dst);
static final class EpollChannel extends Channel {
@@ -977,9 +1028,9 @@ boolean isInterestWrite() {
}
@Override
- int native_read() {
- ByteBuf buf = eventLoop.getReadBuf();
- return Native.read(fd, eventLoop.getBufAddress() + buf.absPos(), buf.remaining());
+ int native_read(ByteBuf dst) {
+ long address = dst.address() + dst.absWriteIndex();
+ return Native.read(fd, address, dst.writableBytes());
}
@Override
@@ -1010,13 +1061,13 @@ int write() {
}
if (cw_len == 1) {
ByteBuf buf = cwb_array[0];
- long address = buf.address() + buf.absPos();
- int len = Native.write(fd, address, buf.remaining());
+ long address = buf.address() + buf.absReadIndex();
+ int len = Native.write(fd, address, buf.readableBytes());
if (len == -1) {
return -1;
}
- buf.skip(len);
- if (buf.hasRemaining()) {
+ buf.skipRead(len);
+ if (buf.hasReadableBytes()) {
this.current_wbs_len = 1;
this.interestWrite = true;
return 0;
@@ -1034,9 +1085,9 @@ int write() {
long iov_pos = iovec;
for (int i = 0; i < cw_len; i++) {
ByteBuf buf = cwb_array[i];
- Unsafe.putLong(iov_pos, buf.address() + buf.absPos());
+ Unsafe.putLong(iov_pos, buf.address() + buf.absReadIndex());
iov_pos += 8;
- Unsafe.putLong(iov_pos, buf.remaining());
+ Unsafe.putLong(iov_pos, buf.readableBytes());
iov_pos += 8;
}
long len = Native.writev(fd, iovec, cw_len);
@@ -1045,14 +1096,14 @@ int write() {
}
for (int i = 0; i < cw_len; i++) {
ByteBuf buf = cwb_array[i];
- int r = buf.remaining();
+ int r = buf.readableBytes();
if (len < r) {
- buf.skip((int) len);
+ buf.skipRead((int) len);
int remain = cw_len - i;
System.arraycopy(cwb_array, i, cwb_array, 0, remain);
fill_null(cwb_array, remain, cw_len);
this.interestWrite = true;
- this.current_wbs_len = remain;
+ this.current_wbs_len = (byte) remain;
return 0;
} else {
len -= r;
@@ -1133,7 +1184,7 @@ boolean isInterestWrite() {
private int native_write(ByteBuffer src) {
try {
return channel.write(src);
- } catch (IOException e) {
+ } catch (Throwable e) {
return -1;
}
}
@@ -1141,16 +1192,16 @@ private int native_write(ByteBuffer src) {
private long native_write(ByteBuffer[] srcs, int len) {
try {
return channel.write(srcs, 0, len);
- } catch (IOException e) {
+ } catch (Throwable e) {
return -1;
}
}
@Override
- int native_read() {
+ int native_read(ByteBuf dst) {
try {
- return channel.read(eventLoop.getReadBuf().nioBuffer());
- } catch (IOException e) {
+ return channel.read(dst.nioWriteBuffer());
+ } catch (Throwable e) {
return -1;
}
}
@@ -1189,14 +1240,14 @@ int write() {
}
if (cwb_len == 1) {
ByteBuf buf = cwb_array[0];
- ByteBuffer nioBuf = buf.nioBuffer();
+ ByteBuffer nioBuf = buf.nioReadBuffer();
int len = native_write(nioBuf);
if (len == -1) {
return -1;
}
if (nioBuf.hasRemaining()) {
this.current_wbs_len = 1;
- buf.reverse();
+ buf.reverseRead();
interestWrite();
return 0;
} else {
@@ -1210,7 +1261,7 @@ int write() {
}
} else {
for (int i = 0; i < cwb_len; i++) {
- wb_array[i] = cwb_array[i].nioBuffer();
+ wb_array[i] = cwb_array[i].nioReadBuffer();
}
long len = native_write(wb_array, cwb_len);
if (len == -1) {
@@ -1219,13 +1270,13 @@ int write() {
for (int i = 0; i < cwb_len; i++) {
ByteBuf buf = cwb_array[i];
if (wb_array[i].hasRemaining()) {
- buf.reverse();
+ buf.reverseRead();
int remain = cwb_len - i;
System.arraycopy(cwb_array, i, cwb_array, 0, remain);
fill_null(cwb_array, remain, cwb_len);
fill_null(wb_array, i, cwb_len);
interestWrite();
- this.current_wbs_len = remain;
+ this.current_wbs_len = (byte) remain;
return 0;
} else {
wb_array[i] = null;
@@ -1279,7 +1330,7 @@ private static Field OPENSSL_DESTROYED() {
static boolean isOpensslEngineDestroyed(SSLEngine sslEngine) {
try {
- return ((Integer) OPENSSL_DESTROYED.get(sslEngine)) == 1;
+ return ((Integer) OPENSSL_DESTROYED.get(sslEngine)) != 0;
} catch (IllegalAccessException e) {
throw new RuntimeException(e);
}
diff --git a/firenio-core/src/main/java/com/firenio/component/ChannelAcceptor.java b/firenio-core/src/main/java/com/firenio/component/ChannelAcceptor.java
index 6cc7f69aa..11a2d09f2 100644
--- a/firenio-core/src/main/java/com/firenio/component/ChannelAcceptor.java
+++ b/firenio-core/src/main/java/com/firenio/component/ChannelAcceptor.java
@@ -146,6 +146,11 @@ public synchronized void unbind() {
Util.stop(this);
}
+ @Override
+ public String toString() {
+ return (unsafe instanceof EpollAcceptorUnsafe ? "Epoll" : "Jdk") + "Acceptor(" + getServerAddress() + ")";
+ }
+
static abstract class AcceptorUnsafe implements Closeable {
abstract void bind(NioEventLoop el, ChannelAcceptor a, int backlog) throws IOException;
diff --git a/firenio-core/src/main/java/com/firenio/component/ChannelActiveListener.java b/firenio-core/src/main/java/com/firenio/component/ChannelActiveListener.java
index 152cbb3b5..53fee7631 100644
--- a/firenio-core/src/main/java/com/firenio/component/ChannelActiveListener.java
+++ b/firenio-core/src/main/java/com/firenio/component/ChannelActiveListener.java
@@ -24,7 +24,7 @@ public class ChannelActiveListener implements ChannelIdleListener {
private Logger logger = LoggerFactory.getLogger(getClass());
@Override
- public void channelIdled(Channel ch, long lastIdleTime, long currentTime) {
+ public void channelIdled(Channel ch, long lastIdleTime) {
if (ch.getLastAccessTime() < lastIdleTime) {
logger.info("disconnect(hb) {}", ch);
Util.close(ch);
diff --git a/firenio-core/src/main/java/com/firenio/component/ChannelAliveListener.java b/firenio-core/src/main/java/com/firenio/component/ChannelAliveListener.java
index 2c15dcd27..8185cf173 100644
--- a/firenio-core/src/main/java/com/firenio/component/ChannelAliveListener.java
+++ b/firenio-core/src/main/java/com/firenio/component/ChannelAliveListener.java
@@ -24,7 +24,7 @@ public class ChannelAliveListener implements ChannelIdleListener {
private Logger logger = LoggerFactory.getLogger(getClass());
@Override
- public void channelIdled(Channel ch, long lastIdleTime, long currentTime) {
+ public void channelIdled(Channel ch, long lastIdleTime) {
if (ch.getLastAccessTime() < lastIdleTime && matched(ch)) {
logger.info("disconnect(hb) {}", ch);
Util.close(ch);
diff --git a/firenio-core/src/main/java/com/firenio/component/ChannelConnector.java b/firenio-core/src/main/java/com/firenio/component/ChannelConnector.java
index a238a5a5d..76c2b2847 100644
--- a/firenio-core/src/main/java/com/firenio/component/ChannelConnector.java
+++ b/firenio-core/src/main/java/com/firenio/component/ChannelConnector.java
@@ -19,6 +19,7 @@
import java.io.IOException;
import java.net.InetAddress;
import java.nio.channels.SelectionKey;
+import java.nio.channels.Selector;
import java.nio.channels.SocketChannel;
import com.firenio.Develop;
@@ -39,12 +40,12 @@
public final class ChannelConnector extends ChannelContext implements Closeable {
private static final Logger logger = LoggerFactory.getLogger(ChannelConnector.class);
- private volatile Callback callback;
- private volatile boolean callback_ed = true;
private Channel ch;
private NioEventLoop eventLoop;
- private volatile DelayedQueue.DelayTask timeoutTask;
private ConnectorUnsafe unsafe;
+ private volatile Callback callback;
+ private volatile boolean callback_ed = true;
+ private volatile DelayedQueue.DelayTask timeoutTask;
public ChannelConnector(int port) {
this("127.0.0.1", port);
@@ -64,11 +65,6 @@ public ChannelConnector(NioEventLoopGroup group, String host, int port) {
if (!group.isSharable() && !group.isRunning()) {
group.setEventLoopSize(1);
}
- if (Native.EPOLL_AVAILABLE) {
- unsafe = new EpollConnectorUnsafe();
- } else {
- unsafe = new JavaConnectorUnsafe();
- }
}
public ChannelConnector(String host, int port) {
@@ -78,9 +74,9 @@ public ChannelConnector(String host, int port) {
@Override
protected void channelEstablish(Channel ch, Throwable ex) {
if (!callback_ed) {
- this.unsafe.channelEstablish(ch, eventLoop, ex);
this.ch = ch;
this.callback_ed = true;
+ this.unsafe.channelEstablish(ch, ex);
this.timeoutTask.cancel();
try {
this.callback.call(ch, ex);
@@ -94,10 +90,6 @@ protected void channelEstablish(Channel ch, Throwable ex) {
public synchronized void close() {
Util.close(ch);
Util.stop(this);
- if (!getProcessorGroup().isSharable()) {
- this.eventLoop = null;
- }
- this.ch = null;
}
public synchronized Channel connect() throws Exception {
@@ -118,12 +110,17 @@ public synchronized void connect(Callback callback, long timeout) throw
throw new IOException("connect is pending");
}
this.callback_ed = false;
- this.timeoutTask = new TimeoutTask(this, timeout);
this.callback = callback;
+ this.timeoutTask = new TimeoutTask(this, timeout);
this.getProcessorGroup().setContext(this);
+ if (Native.EPOLL_AVAILABLE) {
+ unsafe = new EpollConnectorUnsafe();
+ } else {
+ unsafe = new JavaConnectorUnsafe();
+ }
Util.start(getProcessorGroup());
Util.start(this);
- if (eventLoop == null) {
+ if (eventLoop == null || !eventLoop.isRunning()) {
eventLoop = getProcessorGroup().getNext();
}
boolean submitted = this.eventLoop.submit(new Runnable() {
@@ -152,17 +149,9 @@ public synchronized Channel connect(long timeout) throws Exception {
// If your application blocking here, check if you are blocking the io thread.
// Notice that do not blocking io thread at any time.
callback.await();
- if (callback.isTimeout()) {
- Util.close(this);
- throw new TimeoutException("connect to " + getServerAddress() + " time out");
- }
if (callback.isFailed()) {
Util.close(this);
- Throwable ex = callback.getThrowable();
- if (ex instanceof Exception) {
- throw (Exception) callback.getThrowable();
- }
- throw new IOException("connect failed", ex);
+ throw new IOException("connect failed", callback.getThrowable());
}
return getChannel();
}
@@ -202,14 +191,15 @@ static abstract class ConnectorUnsafe {
abstract void connect(ChannelConnector ctx, NioEventLoop el) throws IOException;
- abstract void channelEstablish(Channel ch, NioEventLoop el, Throwable ex);
+ abstract void channelEstablish(Channel ch, Throwable ex);
}
static final class EpollConnectorUnsafe extends ConnectorUnsafe {
- private int fd = -1;
- private String remoteAddr;
+ private int fd = -1;
+ private String remoteAddr;
+ private EpollEventLoop el;
@Override
void connect(ChannelConnector ctx, NioEventLoop el) throws IOException {
@@ -219,28 +209,28 @@ void connect(ChannelConnector ctx, NioEventLoop el) throws IOException {
int fd = Native.connect(host.getHostAddress(), ctx.getPort());
Native.throwException(fd);
this.fd = fd;
- el.schedule(ctx.timeoutTask);
+ this.el = un;
un.ctxs.put(fd, ctx);
+ el.schedule(ctx.timeoutTask);
int res = Native.epoll_add(un.epfd, fd, Native.EPOLL_OUT);
Native.throwException(res);
}
@Override
- void channelEstablish(Channel ch, NioEventLoop el, Throwable ex) {
+ void channelEstablish(Channel ch, Throwable ex) {
if (ex != null && fd != -1) {
- EpollEventLoop un = (EpollEventLoop) el;
- un.ctxs.remove(fd);
+ el.ctxs.remove(fd);
if (Develop.NATIVE_DEBUG) {
- int res = Native.epoll_del(un.epfd, fd);
+ int res = Native.epoll_del(el.epfd, fd);
if (res == -1) {
- logger.error("cancel...epfd:{},fd:{}", un.epfd, fd);
+ logger.error("cancel...epfd:{},fd:{}", el.epfd, fd);
}
res = Native.close(fd);
if (res == -1) {
logger.error("cancel...fd:{}", fd);
}
} else {
- Native.epoll_del(un.epfd, fd);
+ Native.epoll_del(el.epfd, fd);
Native.close(fd);
}
this.fd = -1;
@@ -251,25 +241,22 @@ String getRemoteAddr() {
return remoteAddr;
}
- int getFd() {
- return fd;
- }
-
}
static final class JavaConnectorUnsafe extends ConnectorUnsafe {
private SocketChannel javaChannel;
+ private SelectionKey selectionKey;
@Override
void connect(ChannelConnector ctx, NioEventLoop el) throws IOException {
- Util.close(javaChannel);
this.javaChannel = SocketChannel.open();
this.javaChannel.configureBlocking(false);
if (!javaChannel.connect(ctx.getServerAddress())) {
el.schedule(ctx.timeoutTask);
JavaEventLoop elUnsafe = (JavaEventLoop) el;
- javaChannel.register(elUnsafe.getSelector(), SelectionKey.OP_CONNECT, ctx);
+ Selector selector = elUnsafe.getSelector();
+ this.selectionKey = javaChannel.register(selector, SelectionKey.OP_CONNECT, ctx);
}
}
@@ -278,16 +265,10 @@ SocketChannel getSelectableChannel() {
}
@Override
- void channelEstablish(Channel ch, NioEventLoop el, Throwable ex) {
- final SocketChannel channel = this.javaChannel;
- if (ex != null && channel != null) {
- final JavaEventLoop un = (JavaEventLoop) el;
- SelectionKey key = channel.keyFor(un.getSelector());
- if (key != null) {
- key.cancel();
- }
- Util.close(channel);
- this.javaChannel = null;
+ void channelEstablish(Channel ch, Throwable ex) {
+ if (ex != null) {
+ selectionKey.cancel();
+ Util.close(javaChannel);
}
}
diff --git a/firenio-core/src/main/java/com/firenio/component/ChannelContext.java b/firenio-core/src/main/java/com/firenio/component/ChannelContext.java
index ebff8940e..50e4e9d34 100644
--- a/firenio-core/src/main/java/com/firenio/component/ChannelContext.java
+++ b/firenio-core/src/main/java/com/firenio/component/ChannelContext.java
@@ -33,6 +33,7 @@
import com.firenio.common.Assert;
import com.firenio.common.FileUtil;
import com.firenio.common.Properties;
+import com.firenio.common.Unsafe;
import com.firenio.common.Util;
import com.firenio.concurrent.EventLoop;
import com.firenio.concurrent.EventLoopGroup;
@@ -148,19 +149,29 @@ protected void doStart() throws Exception {
}
if (isEnableSsl()) {
sb.setLength(0);
- for (String p : SslContext.ENABLED_PROTOCOLS) {
+ for (String p : sslContext.getProtocols()) {
sb.append(p);
sb.append(',');
sb.append(' ');
}
sb.setLength(sb.length() - 2);
- logger.info("ssl default protocols : [ {} ]", sb.toString());
+ logger.info("ssl protocols : [ {} ]", sb.toString());
+ }
+ if (isEnableSsl()) {
+ sb.setLength(0);
+ for (String p : sslContext.getCipherSuites()) {
+ sb.append(p);
+ sb.append(',');
+ sb.append(' ');
+ }
+ sb.setLength(sb.length() - 2);
+ logger.info("ssl cipher suites : [ {} ]", sb.toString());
}
}
}
private String getByteBufPoolType(NioEventLoopGroup g) {
- if (Options.isEnableUnsafeBuf()) {
+ if (Unsafe.UNSAFE_BUF_AVAILABLE) {
return "unsafe";
}
return g.isEnableMemoryPoolDirect() ? "direct" : "heap";
diff --git a/firenio-core/src/main/java/com/firenio/component/ChannelIdleListener.java b/firenio-core/src/main/java/com/firenio/component/ChannelIdleListener.java
index ae5c1da1d..5f337df55 100644
--- a/firenio-core/src/main/java/com/firenio/component/ChannelIdleListener.java
+++ b/firenio-core/src/main/java/com/firenio/component/ChannelIdleListener.java
@@ -19,6 +19,6 @@
public interface ChannelIdleListener extends EventListener {
- void channelIdled(Channel ch, long lastIdleTime, long currentTime);
+ void channelIdled(Channel ch, long lastIdleTime);
}
diff --git a/firenio-core/src/main/java/com/firenio/component/ChannelManager.java b/firenio-core/src/main/java/com/firenio/component/ChannelManager.java
index 96ca37278..51d2c531d 100644
--- a/firenio-core/src/main/java/com/firenio/component/ChannelManager.java
+++ b/firenio-core/src/main/java/com/firenio/component/ChannelManager.java
@@ -44,18 +44,26 @@ public static void broadcast(ByteBuf buf, Collection chs) {
}
}
+ public void broadcast(ByteBuf buf) {
+ broadcast(buf, channels.values());
+ }
+
public static void broadcast(Frame frame, Collection chs) throws Exception {
if (chs.size() == 0) {
+ frame.release();
return;
}
Channel ch = chs.iterator().next();
- if (ch != null) {
- broadcast(ch.encode(frame), chs);
+ if (ch == null) {
+ frame.release();
+ return;
+ }
+ try {
+ ByteBuf buf = ch.getCodec().encode(ch, frame);
+ broadcast(buf, chs);
+ } finally {
+ frame.release();
}
- }
-
- public void broadcast(ByteBuf buf) {
- broadcast(buf, channels.values());
}
public void broadcast(Frame frame) throws Exception {
diff --git a/firenio-core/src/main/java/com/firenio/component/ExceptionCaughtHandle.java b/firenio-core/src/main/java/com/firenio/component/ExceptionCaughtHandle.java
index afeb8ffa4..85dfb56d0 100644
--- a/firenio-core/src/main/java/com/firenio/component/ExceptionCaughtHandle.java
+++ b/firenio-core/src/main/java/com/firenio/component/ExceptionCaughtHandle.java
@@ -20,6 +20,6 @@
*/
public interface ExceptionCaughtHandle {
- void exceptionCaught(Channel ch, Frame frame, Exception ex);
+ void exceptionCaught(Channel ch, Frame frame, Throwable ex);
}
diff --git a/firenio-core/src/main/java/com/firenio/component/FastThreadLocal.java b/firenio-core/src/main/java/com/firenio/component/FastThreadLocal.java
index 5a24128ec..252f1ca4b 100644
--- a/firenio-core/src/main/java/com/firenio/component/FastThreadLocal.java
+++ b/firenio-core/src/main/java/com/firenio/component/FastThreadLocal.java
@@ -23,28 +23,27 @@
import java.util.IdentityHashMap;
import java.util.List;
import java.util.Map;
-import java.util.concurrent.atomic.AtomicInteger;
import com.firenio.buffer.ByteBuf;
-import com.firenio.collection.AttributesImpl;
+import com.firenio.collection.AttributeKey;
+import com.firenio.collection.AttributeMap;
import com.firenio.common.Util;
/**
* @author wangkai
*/
-public final class FastThreadLocal extends AttributesImpl {
+public final class FastThreadLocal extends AttributeMap {
- private static final AtomicInteger indexedVarsIndex = new AtomicInteger(0);
- private static final int maxIndexedVarsSize = 32;
- private static final ThreadLocal slowThreadLocal = new ThreadLocal<>();
+ private static final ThreadLocal slowThreadLocal = new ThreadLocal<>();
- private byte[] bytes32 = new byte[32];
- private Map charsetDecoders = new IdentityHashMap<>();
- private Map charsetEncoders = new IdentityHashMap<>();
- private Object[] indexedVariables = new Object[maxIndexedVarsSize];
+ private byte[] bytes32 = new byte[32];
+ private Map charsetDecoders = new IdentityHashMap<>();
+ private Map charsetEncoders = new IdentityHashMap<>();
private ByteBuf sslUnwrapBuf;
private ByteBuf sslWrapBuf;
- private StringBuilder stringBuilder = new StringBuilder(512);
+ private StringBuilder stringBuilder = new StringBuilder(512);
+ private List list = new ArrayList<>();
+ private Map map = new HashMap<>();
FastThreadLocal() {}
@@ -75,15 +74,12 @@ public static FastThreadLocal get() {
}
}
- public static int nextIndexedVariablesIndex() {
- if (indexedVarsIndex.get() >= maxIndexedVarsSize) {
- return -1;
- }
- int index = indexedVarsIndex.getAndIncrement();
- if (index >= maxIndexedVarsSize) {
- return -1;
- }
- return index;
+ public static AttributeKey valueOfKey(String name) {
+ return AttributeMap.valueOfKey(FastThreadLocal.class, name);
+ }
+
+ public static AttributeKey valueOfKey(String name, AttributeInitFunction function) {
+ return AttributeMap.valueOfKey(FastThreadLocal.class, name, function);
}
private void destroy0() {
@@ -113,40 +109,16 @@ public CharsetEncoder getCharsetEncoder(Charset charset) {
return encoder;
}
- public Object getIndexedVariable(int index) {
- return indexedVariables[index];
- }
-
- public List getList(int index) {
- List list = (List) getIndexedVariable(index);
- if (list == null) {
- list = new ArrayList<>();
- setIndexedVariable(index, list);
- }
- list.clear();
- return list;
- }
-
- public Map getMap(int index) {
- Map map = (Map) getIndexedVariable(index);
- if (map == null) {
- map = new HashMap<>();
- setIndexedVariable(index, map);
- }
- map.clear();
- return map;
- }
-
public ByteBuf getSslUnwrapBuf() {
if (sslUnwrapBuf == null) {
- sslUnwrapBuf = ByteBuf.direct(SslContext.SSL_UNWRAP_BUFFER_SIZE);
+ sslUnwrapBuf = ByteBuf.buffer(SslContext.SSL_UNWRAP_BUFFER_SIZE);
}
return sslUnwrapBuf;
}
public ByteBuf getSslWrapBuf() {
if (sslWrapBuf == null) {
- sslWrapBuf = ByteBuf.direct(SslContext.SSL_PACKET_BUFFER_SIZE);
+ sslWrapBuf = ByteBuf.buffer(SslContext.SSL_PACKET_BUFFER_SIZE);
}
return sslWrapBuf;
}
@@ -156,8 +128,18 @@ public StringBuilder getStringBuilder() {
return stringBuilder;
}
- public void setIndexedVariable(int index, Object value) {
- indexedVariables[index] = value;
+ public List getList() {
+ list.clear();
+ return list;
}
+ public Map getMap() {
+ map.clear();
+ return map;
+ }
+
+ @Override
+ protected AttributeKeys getKeys() {
+ return getKeys(FastThreadLocal.class);
+ }
}
diff --git a/firenio-core/src/main/java/com/firenio/component/Frame.java b/firenio-core/src/main/java/com/firenio/component/Frame.java
index 36191b8ef..4efa9263b 100644
--- a/firenio-core/src/main/java/com/firenio/component/Frame.java
+++ b/firenio-core/src/main/java/com/firenio/component/Frame.java
@@ -81,7 +81,7 @@ public void setBytes(byte[] bytes, int off, int len, Channel ch) {
}
public void setBytes(int header, byte[] bytes, int off, int len) {
- this.content = ByteBuf.heap(header + len).skip(header);
+ this.content = ByteBuf.buffer(header + len).skipWrite(header);
write(bytes, off, len);
}
@@ -98,7 +98,7 @@ public void write(byte[] bytes, int off, int len) {
if (c == null) {
throw new NullPointerException("do setContent(buf) before write");
}
- c.putBytes(bytes, off, len);
+ c.writeBytes(bytes, off, len);
}
public void write(String text, Channel ch) {
diff --git a/firenio-core/src/main/java/com/firenio/component/IoEventHandle.java b/firenio-core/src/main/java/com/firenio/component/IoEventHandle.java
index afec0a0db..d9512b0fe 100644
--- a/firenio-core/src/main/java/com/firenio/component/IoEventHandle.java
+++ b/firenio-core/src/main/java/com/firenio/component/IoEventHandle.java
@@ -25,7 +25,7 @@ public abstract class IoEventHandle implements ExceptionCaughtHandle {
public abstract void accept(Channel ch, Frame frame) throws Exception;
@Override
- public void exceptionCaught(Channel ch, Frame frame, Exception ex) {
+ public void exceptionCaught(Channel ch, Frame frame, Throwable ex) {
logger.error(ex.getMessage(), ex);
}
diff --git a/firenio-core/src/main/java/com/firenio/component/LoggerExceptionCaughtHandle.java b/firenio-core/src/main/java/com/firenio/component/LoggerExceptionCaughtHandle.java
index 2a79c4145..0e71daacd 100644
--- a/firenio-core/src/main/java/com/firenio/component/LoggerExceptionCaughtHandle.java
+++ b/firenio-core/src/main/java/com/firenio/component/LoggerExceptionCaughtHandle.java
@@ -26,7 +26,7 @@ public class LoggerExceptionCaughtHandle implements ExceptionCaughtHandle {
private Logger logger = LoggerFactory.getLogger(getClass());
@Override
- public void exceptionCaught(Channel ch, Frame frame, Exception ex) {
+ public void exceptionCaught(Channel ch, Frame frame, Throwable ex) {
logger.error(ex.getMessage(), ex);
}
diff --git a/firenio-core/src/main/java/com/firenio/component/Native.java b/firenio-core/src/main/java/com/firenio/component/Native.java
index 1504a2f05..0ff6545cc 100644
--- a/firenio-core/src/main/java/com/firenio/component/Native.java
+++ b/firenio-core/src/main/java/com/firenio/component/Native.java
@@ -2,13 +2,11 @@
import java.io.File;
import java.io.IOException;
-import java.io.InputStream;
import com.firenio.Develop;
import com.firenio.Options;
import com.firenio.common.FileUtil;
import com.firenio.common.Unsafe;
-import com.firenio.common.Util;
import com.firenio.log.Logger;
import com.firenio.log.LoggerFactory;
@@ -18,54 +16,71 @@ public class Native {
//gcc -shared -g -fPIC -m64 -o obj/native.o src/Native.cpp
//gcc -shared -fPIC -m64 -o obj/native.o src/Native.cpp
- public static final boolean EPOLL_AVAILABLE;
- public static final int EPOLL_ERR;
- public static final int EPOLL_ET;
- public static final int EPOLL_IN;
- public static final int EPOLL_IN_ET;
- public static final int EPOLL_IN_OUT;
- public static final int EPOLL_IN_OUT_ET;
- public static final int EPOLL_OUT;
- public static final int EPOLL_OUT_ET;
- public static final int EPOLL_HUP;
- public static final int EPOLL_RD_HUP;
- public static final String[] ERRORS;
- public static final boolean IS_LINUX;
+ public static final int SEEK_SET = 0; /* Seek from beginning of file. */
+ public static final int SEEK_CUR = 1; /* Seek from current position. */
+ public static final int SEEK_END = 2; /* Seek from end of file. */
+ public static final int SEEK_DATA = 3; /* Seek to next data. */
+ public static final int SEEK_HOLE = 4; /* Seek to next hole. */
+
+ public static final int O_ACCMODE = 0003;
+ public static final int O_RDONLY = 00;
+ public static final int O_WRONLY = 01;
+ public static final int O_RDWR = 02;
+ public static final int O_CREAT = 0100;
+ public static final int O_EXCL = 0200;
+ public static final int O_NOCTTY = 0400;
+ public static final int O_TRUNC = 01000;
+ public static final int O_APPEND = 02000;
+ public static final int O_NONBLOCK = 04000;
+ public static final int O_DIRECT = 040000;
+ public static final int O_NDELAY = O_NONBLOCK;
+ public static final int O_SYNC = 04010000;
+ public static final int O_FSYNC = O_SYNC;
+ public static final int O_ASYNC = 020000;
+
+ public static final int EPOLL_ERR = 1 << 3;
+ public static final int EPOLL_ET = 1 << 31;
+ public static final int EPOLL_IN = 1 << 0;
+ public static final int EPOLL_OUT = 1 << 2;
+ public static final int EPOLL_HUP = 1 << 4;
+ public static final int EPOLL_RD_HUP = 1 << 13;
+ public static final int EPOLL_IN_ET = EPOLL_IN | EPOLL_ET;
+ public static final int EPOLL_IN_OUT = EPOLL_IN | EPOLL_OUT;
+ public static final int EPOLL_OUT_ET = EPOLL_OUT | EPOLL_ET;
+ public static final int EPOLL_IN_OUT_ET = EPOLL_IN_ET | EPOLL_OUT_ET;
+ public static final boolean EPOLL_AVAILABLE = try_load_epoll();
public static final int SIZEOF_EPOLL_EVENT;
public static final int SIZEOF_SOCK_ADDR_IN;
- private static final Logger logger = LoggerFactory.getLogger(Native.class);
+ public static final String[] ERRORS;
+ private static final Logger logger = LoggerFactory.getLogger(Native.class);
static {
- EPOLL_ET = 1 << 31;
- EPOLL_IN = 1 << 0;
- EPOLL_OUT = 1 << 2;
- EPOLL_ERR = 1 << 3;
- EPOLL_HUP = 1 << 4;
- EPOLL_RD_HUP = 1 << 13;
- EPOLL_IN_OUT = EPOLL_IN | EPOLL_OUT;
- EPOLL_IN_ET = EPOLL_IN | EPOLL_ET;
- EPOLL_OUT_ET = EPOLL_OUT | EPOLL_ET;
- EPOLL_IN_OUT_ET = EPOLL_IN_ET | EPOLL_OUT_ET;
- IS_LINUX = isLinux();
- EPOLL_AVAILABLE = IS_LINUX && tryLoadEpoll();
if (EPOLL_AVAILABLE) {
- ERRORS = new String[256];
- byte[] temp = new byte[1024];
- for (int i = 0; i < ERRORS.length; i++) {
- int len = strerrno(i, temp);
- ERRORS[i] = new String(temp, 0, len);
- }
+ ERRORS = load_errors();
SIZEOF_EPOLL_EVENT = size_of_epoll_event();
SIZEOF_SOCK_ADDR_IN = size_of_sockaddr_in();
} else {
+ ERRORS = null;
SIZEOF_EPOLL_EVENT = -1;
SIZEOF_SOCK_ADDR_IN = -1;
- ERRORS = null;
}
}
- private static boolean tryLoadEpoll() {
- if (Options.isEnableEpoll()) {
+ private static String[] load_errors() {
+ String[] errors = new String[256];
+ byte[] temp = new byte[1024];
+ for (int i = 0; i < errors.length; i++) {
+ int len = strerrno(i, temp);
+ errors[i] = new String(temp, 0, len);
+ }
+ return errors;
+ }
+
+ private static boolean try_load_epoll() {
+ if (!Unsafe.DIRECT_BUFFER_AVAILABLE) {
+ return false;
+ }
+ if (Unsafe.IS_LINUX && Options.isEnableEpoll()) {
boolean epollLoaded = false;
if (Develop.EPOLL_DEBUG) {
logger.info("load epoll from:{}", Develop.EPOLL_PATH);
@@ -76,7 +91,7 @@ private static boolean tryLoadEpoll() {
}
} else {
try {
- loadNative("native.o");
+ load_native("native.o");
epollLoaded = true;
} catch (Throwable e) {
if (Develop.NATIVE_DEBUG) {
@@ -92,7 +107,7 @@ private static boolean tryLoadEpoll() {
return true;
}
if (Develop.NATIVE_DEBUG) {
- logger.error("epoll creat failed:" + Native.err_str());
+ logger.error("epoll create failed:" + Native.err_str());
}
} catch (Throwable e) {
}
@@ -101,12 +116,20 @@ private static boolean tryLoadEpoll() {
return false;
}
+ private static void load_native(String name) throws IOException {
+ byte[] native_data = FileUtil.readBytesByCls(name, Native.class.getClassLoader());
+ File native_tmp = File.createTempFile(name, ".o");
+ FileUtil.writeByFile(native_tmp, native_data);
+ System.load(native_tmp.getAbsolutePath());
+ native_tmp.deleteOnExit();
+ }
+
public static int close_event() {
return EPOLL_ERR | EPOLL_HUP | EPOLL_RD_HUP;
}
public static int accept(int epfd, int listen_fd, long address) {
- return printException(accept0(epfd, listen_fd, address));
+ return print_exception(accept0(epfd, listen_fd, address));
}
public static int bind(String host, int port, int backlog) {
@@ -117,11 +140,11 @@ public static int close(int fd) {
if (fd == -1) {
return -1;
}
- return printException(close0(fd));
+ return print_exception(close0(fd));
}
public static int connect(String host, int port) {
- return printException(connect0(host, port));
+ return print_exception(connect0(host, port));
}
public static int epoll_add(int epfd, int fd, int state) {
@@ -136,19 +159,17 @@ public static int epoll_del(int epfd, int fd) {
if (fd == -1) {
return -1;
}
- return printException(epoll_del0(epfd, fd));
+ return print_exception(epoll_del0(epfd, fd));
}
public static int epoll_mod(int epfd, int fd, int state) {
- return printException(epoll_mod0(epfd, fd, state));
+ return print_exception(epoll_mod0(epfd, fd, state));
}
- public static int epoll_wait(int fd, long address, int maxEvents, long timeout) {
- return printException(epoll_wait0(fd, address, maxEvents, timeout));
+ public static int epoll_wait(int fd, long address, int max_events, long timeout) {
+ return print_exception(epoll_wait0(fd, address, max_events, timeout));
}
- public static native int errno();
-
public static String err_str() {
return err_str(errno());
}
@@ -162,24 +183,11 @@ private static String err_str(int errno) {
}
public static int event_fd_read(int fd) {
- return printException(event_fd_read0(fd));
+ return print_exception(event_fd_read0(fd));
}
public static int event_fd_write(int fd, long value) {
- return printException(event_fd_write0(fd, value));
- }
-
- private static boolean isLinux() {
- return Util.getStringProperty("os.name", "").toLowerCase().startsWith("lin");
- }
-
- private static void loadNative(String name) throws IOException {
- InputStream in = Native.class.getClassLoader().getResourceAsStream(name);
- File tmpFile = File.createTempFile(name, ".o");
- byte[] data = FileUtil.inputStream2ByteArray(in);
- FileUtil.writeByFile(tmpFile, data);
- System.load(tmpFile.getAbsolutePath());
- tmpFile.deleteOnExit();
+ return print_exception(event_fd_write0(fd, value));
}
public static long new_epoll_event_array(int size) {
@@ -191,10 +199,14 @@ public static int new_event_fd() {
}
public static int read(int fd, long address, int len) {
- return printException(read0(fd, address, len));
+ return print_exception(read0(fd, address, len));
}
- private static int printException(int res) {
+ public static int pread(int fd, long address, int len, long pos) {
+ return print_exception(pread0(fd, address, len, pos));
+ }
+
+ private static int print_exception(int res) {
if (Develop.NATIVE_DEBUG && res == -1) {
int errno = errno();
if (errno != 0) {
@@ -205,11 +217,16 @@ private static int printException(int res) {
return res;
}
- public static native int size_of_epoll_event();
-
- public static native int size_of_sockaddr_in();
-
- public static native int strerrno(int no, byte[] buf);
+ private static long print_exception(long res) {
+ if (Develop.NATIVE_DEBUG && res == -1) {
+ int errno = errno();
+ if (errno != 0) {
+ IOException e = new IOException(err_str(errno));
+ logger.error(e.getMessage(), e);
+ }
+ }
+ return res;
+ }
public static int throwException(int res) throws IOException {
if (res == -1) {
@@ -226,30 +243,39 @@ public static int throwRuntimeException(int res) {
}
public static int write(int fd, long address, int len) {
- return printException(write0(fd, address, len));
+ return print_exception(write0(fd, address, len));
+ }
+
+ public static int pwrite(int fd, long address, int len, long pos) {
+ return print_exception(pwrite0(fd, address, len, pos));
}
public static int get_port(int fd) {
- return Short.reverseBytes((short) printException(get_port0(fd))) & 0xffff;
+ return Short.reverseBytes((short) print_exception(get_port0(fd))) & 0xffff;
}
public static long writev(int fd, long iovec, int count) {
- return printException(writev0(fd, iovec, count));
+ return print_exception(writev0(fd, iovec, count));
}
public static int set_socket_opt(int fd, int type, int name, int value) {
- return printException(set_socket_opt0(fd, type, name, value));
+ return print_exception(set_socket_opt0(fd, type, name, value));
}
public static boolean finish_connect(int fd) {
- int type = SocketOptions.SOL_SOCKET >> 16;
+ int type = SocketOptions.RAW_SOL_SOCKET;
int res = get_socket_opt0(fd, type, SocketOptions.SO_ERROR & 0xff);
if (res != 0) {
if (res == -1) {
- printException(res);
+ print_exception(res);
} else {
- IOException e = new IOException(err_str(res & 0x7fffffff));
- logger.error(e.getMessage(), e);
+ if (Develop.NATIVE_DEBUG) {
+ if (res < 0) {
+ res = -res;
+ }
+ IOException e = new IOException(err_str(res));
+ logger.error(e.getMessage(), e);
+ }
}
return false;
}
@@ -257,10 +283,35 @@ public static boolean finish_connect(int fd) {
}
public static int get_socket_opt(int fd, int type, int name) {
- return printException(get_socket_opt0(fd, type, name));
+ return print_exception(get_socket_opt0(fd, type, name));
+ }
+
+ public static int open(String path, int op, int pem) {
+ return print_exception(open0(path, op, pem));
+ }
+
+ public static long posix_memalign_allocate(int len, int align) {
+ return print_exception(posix_memalign_allocate0(len, align));
+ }
+
+ public static long file_length(int fd) {
+ return print_exception(file_length0(fd));
+ }
+
+ public static long lseek(int fd, long off, int seek_mode) {
+ return print_exception(lseek0(fd, off, seek_mode));
}
- //---------------------------------------------------------------------------------------------------------
+ // epoll----------------------------------------------------------------------------------------
+
+ public static native int errno();
+
+ private static native int size_of_epoll_event();
+
+ private static native int size_of_sockaddr_in();
+
+ private static native int strerrno(int no, byte[] buf);
+
private static native int get_socket_opt0(int fd, int type, int name);
private static native int set_socket_opt0(int fd, int type, int name, int value);
@@ -281,7 +332,7 @@ public static int get_socket_opt(int fd, int type, int name) {
private static native int epoll_mod0(int epfd, int fd, int state);
- private static native int epoll_wait0(int fd, long address, int maxEvents, long timeout);
+ private static native int epoll_wait0(int fd, long address, int max_events, long timeout);
private static native int event_fd_read0(int fd);
@@ -297,4 +348,18 @@ public static int get_socket_opt(int fd, int type, int name) {
private static native int writev0(int fd, long iovec, int count);
+ // direct io----------------------------------------------------------------------------------------
+
+ private static native int open0(String path, int op, int pem);
+
+ private static native long posix_memalign_allocate0(int len, int align);
+
+ private static native long file_length0(int fd);
+
+ private static native long lseek0(int fd, long off, int seek_mode);
+
+ private static native int pread0(int fd, long address, int len, long off);
+
+ private static native int pwrite0(int fd, long address, int len, long off);
+
}
diff --git a/firenio-core/src/main/java/com/firenio/component/NioEventLoop.java b/firenio-core/src/main/java/com/firenio/component/NioEventLoop.java
index 6dc7a0200..4c2dd2083 100644
--- a/firenio-core/src/main/java/com/firenio/component/NioEventLoop.java
+++ b/firenio-core/src/main/java/com/firenio/component/NioEventLoop.java
@@ -29,10 +29,8 @@
import java.security.PrivilegedAction;
import java.util.AbstractSet;
import java.util.Arrays;
-import java.util.HashMap;
import java.util.Iterator;
import java.util.List;
-import java.util.Map;
import java.util.Set;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;
@@ -43,8 +41,11 @@
import com.firenio.buffer.ByteBuf;
import com.firenio.buffer.ByteBufAllocator;
import com.firenio.collection.ArrayListStack;
-import com.firenio.collection.Attributes;
+import com.firenio.collection.AttributeKey;
+import com.firenio.collection.AttributeMap;
+import com.firenio.collection.AttributeMap.AttributeInitFunction;
import com.firenio.collection.DelayedQueue;
+import com.firenio.collection.IntArray;
import com.firenio.collection.IntMap;
import com.firenio.collection.LinkedBQStack;
import com.firenio.collection.Stack;
@@ -64,28 +65,33 @@
/**
* @author wangkai
*/
-public abstract class NioEventLoop extends EventLoop implements Attributes {
+public abstract class NioEventLoop extends EventLoop {
static final boolean CHANNEL_READ_FIRST = Options.isChannelReadFirst();
static final Logger logger = NEW_LOGGER();
static final IOException NOT_FINISH_CONNECT = NOT_FINISH_CONNECT();
static final IOException OVER_CH_SIZE_LIMIT = OVER_CH_SIZE_LIMIT();
+ // NOTICE USE_HAS_TASK can make a memory barrier for io thread,
+ // please do not change this value to false unless you are really need.
+ // I am not sure if the select()/select(n) has memory barrier,
+ // it can be set USE_HAS_TASK to false if there is a memory barrier.
static final boolean USE_HAS_TASK = true;
final ByteBufAllocator alloc;
- final Map attributes = new HashMap<>();
final ByteBuf buf;
- final IntMap channels = new IntMap<>(4096);
+ final AttributeMap attributeMap = new NioEventLoopAttributeMap();
+ final IntMap channels = new IntMap<>(4096);
+ final IntArray close_channels = new IntArray();
final int ch_size_limit;
- final DelayedQueue delayed_queue = new DelayedQueue();
- final BlockingQueue events = new LinkedBlockingQueue<>();
+ final DelayedQueue delayed_queue = new DelayedQueue();
+ final BlockingQueue events = new LinkedBlockingQueue<>();
final NioEventLoopGroup group;
final int index;
- final AtomicInteger selecting = new AtomicInteger();
+ final AtomicInteger selecting = new AtomicInteger();
final boolean sharable;
final long buf_address;
final boolean acceptor;
- volatile boolean has_task = false;
+ volatile boolean has_task = false;
NioEventLoop(NioEventLoopGroup group, int index, String threadName) {
super(threadName);
@@ -97,43 +103,43 @@ public abstract class NioEventLoop extends EventLoop implements Attributes {
this.ch_size_limit = group.getChannelSizeLimit();
int channelReadBuffer = group.getChannelReadBuffer();
if (channelReadBuffer > 0) {
- this.buf = ByteBuf.direct(channelReadBuffer);
- this.buf_address = Unsafe.address(buf.getNioBuffer());
+ this.buf = ByteBuf.buffer(channelReadBuffer);
+ this.buf_address = buf.address();
} else {
this.buf = null;
this.buf_address = -1;
}
}
- private static void channel_idle(ChannelIdleListener l, Channel ch, long lastIdleTime, long currentTime) {
+ private static void channel_idle(ChannelIdleListener l, Channel ch, long lastIdleTime) {
try {
- l.channelIdled(ch, lastIdleTime, currentTime);
- } catch (Exception e) {
+ l.channelIdled(ch, lastIdleTime);
+ } catch (Throwable e) {
logger.error(e.getMessage(), e);
}
}
- private static void channel_idle(ChannelContext context, IntMap channels, long last_idle_time, long current_time) {
+ private static void channel_idle(ChannelContext context, IntMap channels, long last_idle_time) {
List ls = context.getChannelIdleEventListeners();
for (int i = 0; i < ls.size(); i++) {
ChannelIdleListener l = ls.get(i);
for (channels.scan(); channels.hasNext(); ) {
- Channel ch = channels.nextValue();
- channel_idle(l, ch, last_idle_time, current_time);
+ Channel ch = channels.value();
+ channel_idle(l, ch, last_idle_time);
}
}
}
- private static void channel_idle_share(IntMap channels, long last_idle_time, long current_time) {
+ private static void channel_idle_share(IntMap channels, long last_idle_time) {
for (channels.scan(); channels.hasNext(); ) {
- Channel ch = channels.nextValue();
+ Channel ch = channels.value();
ChannelContext context = ch.getContext();
List ls = context.getChannelIdleEventListeners();
if (ls.size() == 1) {
- channel_idle(ls.get(0), ch, last_idle_time, current_time);
+ channel_idle(ls.get(0), ch, last_idle_time);
} else {
for (int i = 0; i < ls.size(); i++) {
- channel_idle(ls.get(i), ch, last_idle_time, current_time);
+ channel_idle(ls.get(i), ch, last_idle_time);
}
}
}
@@ -163,7 +169,7 @@ static IOException NOT_FINISH_CONNECT() {
}
static IOException OVER_CH_SIZE_LIMIT() {
- return Util.unknownStackTrace(new IOException("over channel size limit"), NioEventLoop.class, "register_channel(...)");
+ return Util.unknownStackTrace(new IOException("over channel size writeIndex"), NioEventLoop.class, "register_channel(...)");
}
static void read_exception_caught(Channel ch, Throwable ex) {
@@ -178,32 +184,23 @@ public ByteBufAllocator alloc() {
return alloc;
}
- @Override
- public Map attributes() {
- return attributes;
- }
-
//FIXME ..optimize sharable group
- private void channel_idle(long last_idle_time, long current_time) {
+ private void channel_idle(long last_idle_time) {
IntMap channels = this.channels;
if (channels.isEmpty()) {
return;
}
if (sharable) {
- channel_idle_share(channels, last_idle_time, current_time);
+ channel_idle_share(channels, last_idle_time);
} else {
- channel_idle(group.getContext(), channels, last_idle_time, current_time);
+ channel_idle(group.getContext(), channels, last_idle_time);
}
}
- @Override
- public void clearAttributes() {
- this.attributes.clear();
- }
-
private void close_channels() {
+ IntMap channels = this.channels;
for (channels.scan(); channels.hasNext(); ) {
- Util.close(channels.nextValue());
+ Util.close(channels.value());
}
}
@@ -211,23 +208,16 @@ protected long getBufAddress() {
return buf_address;
}
- @Override
- public Object getAttribute(Object key) {
- return this.attributes.get(key);
- }
-
- @Override
- public Set