mirror of
https://github.com/ParkerTenBroeck/coroutines.git
synced 2026-06-07 05:08:51 -04:00
moved selector code into own class
This commit is contained in:
ParkerTenBroeck
parent
93a764c986
commit
f99beb721b
5 changed files with 270 additions and 185 deletions
57
src/async_runtime/SelectorThread.java
Normal file
57
src/async_runtime/SelectorThread.java
Normal file
|
|
@ -0,0 +1,57 @@
|
|||
package async_runtime;
|
||||
|
||||
import java.io.IOException;
|
||||
import java.nio.channels.*;
|
||||
import java.util.ArrayDeque;
|
||||
|
||||
public abstract class SelectorThread<T extends SelectableChannel, A> extends Thread{
|
||||
|
||||
private final Selector selector;
|
||||
|
||||
private record ToRegister<T, A>(T sc, int ops, A waker){}
|
||||
private final ArrayDeque<ToRegister<T, A>> to_register = new ArrayDeque<>();
|
||||
|
||||
|
||||
public SelectorThread(String name) throws IOException {
|
||||
selector = Selector.open();
|
||||
this.setName(name + " Polling Thread");
|
||||
this.setDaemon(true);
|
||||
this.start();
|
||||
}
|
||||
|
||||
public abstract void handle(SelectionKey key, T t, A a) throws IOException;
|
||||
|
||||
public void register(T sc, int ops, A waker){
|
||||
synchronized (to_register){
|
||||
to_register.add(new ToRegister<>(sc, ops, waker));
|
||||
}
|
||||
selector.wakeup();
|
||||
}
|
||||
|
||||
@SuppressWarnings("unchecked")
|
||||
@Override
|
||||
public void run() {
|
||||
while(!Thread.currentThread().isInterrupted()){
|
||||
try{
|
||||
synchronized (to_register){
|
||||
while(!to_register.isEmpty()){
|
||||
var to = to_register.poll();
|
||||
to.sc.register(selector, to.ops, to.waker);
|
||||
}
|
||||
}
|
||||
|
||||
selector.select();
|
||||
var keys = selector.selectedKeys().iterator();
|
||||
|
||||
while (keys.hasNext()) {
|
||||
SelectionKey key = keys.next();
|
||||
keys.remove();
|
||||
|
||||
handle(key, (T)key.channel(), (A)key.attachment());
|
||||
}
|
||||
}catch (Exception e){
|
||||
e.printStackTrace();
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
122
src/async_runtime/net/DatagramSocket.java
Normal file
122
src/async_runtime/net/DatagramSocket.java
Normal file
|
|
@ -0,0 +1,122 @@
|
|||
package async_runtime.net;
|
||||
|
||||
import async_runtime.SelectorThread;
|
||||
import future.Future;
|
||||
import future.Waker;
|
||||
|
||||
import java.io.IOException;
|
||||
import java.net.InetSocketAddress;
|
||||
import java.net.SocketAddress;
|
||||
import java.net.SocketOption;
|
||||
import java.nio.ByteBuffer;
|
||||
import java.nio.channels.DatagramChannel;
|
||||
import java.nio.channels.SelectionKey;
|
||||
|
||||
public class DatagramSocket implements AutoCloseable{
|
||||
private final static SelectorThread<DatagramChannel, Waker> SELECTOR;
|
||||
|
||||
static {
|
||||
try {
|
||||
SELECTOR = new SelectorThread<>("DatagramSocket") {
|
||||
@Override
|
||||
public void handle(SelectionKey key, DatagramChannel c, Waker w) {
|
||||
if (!key.isValid()) {
|
||||
}else if(key.isAcceptable()){
|
||||
}else if(key.isConnectable()){
|
||||
w.wake();
|
||||
}else if(key.isReadable()){
|
||||
w.wake();
|
||||
}else if(key.isWritable()){
|
||||
w.wake();
|
||||
}
|
||||
}
|
||||
};
|
||||
} catch (IOException e) {
|
||||
throw new RuntimeException(e);
|
||||
}
|
||||
}
|
||||
|
||||
private final DatagramChannel socket;
|
||||
|
||||
protected DatagramSocket(DatagramChannel sc){
|
||||
this.socket = sc;
|
||||
}
|
||||
|
||||
public static DatagramSocket open(InetSocketAddress inet) throws IOException {
|
||||
var socket = DatagramChannel.open();
|
||||
socket.configureBlocking(false);
|
||||
return new DatagramSocket(socket);
|
||||
}
|
||||
|
||||
public DatagramSocket bind(InetSocketAddress inet) throws IOException {
|
||||
socket.bind(inet);
|
||||
return this;
|
||||
}
|
||||
|
||||
public DatagramSocket connect(InetSocketAddress inet) throws IOException {
|
||||
socket.connect(inet);
|
||||
return this;
|
||||
}
|
||||
|
||||
public DatagramSocket disconnect() throws IOException{
|
||||
socket.disconnect();
|
||||
return this;
|
||||
}
|
||||
|
||||
public <T> DatagramSocket set_options(SocketOption<T> option, T value) throws IOException{
|
||||
socket.setOption(option, value);
|
||||
return this;
|
||||
}
|
||||
|
||||
public SocketAddress local_address() throws IOException {
|
||||
return socket.getLocalAddress();
|
||||
}
|
||||
|
||||
public SocketAddress remote_address() throws IOException {
|
||||
return socket.getRemoteAddress();
|
||||
}
|
||||
|
||||
public Future<Integer, IOException> write(ByteBuffer buffer){
|
||||
return waker -> {
|
||||
var wrote = socket.write(buffer);
|
||||
if(wrote!=0) return wrote;
|
||||
SELECTOR.register(socket, SelectionKey.OP_WRITE, waker);
|
||||
return Future.Pending.INSTANCE;
|
||||
};
|
||||
}
|
||||
|
||||
public Future<Integer, IOException> write_all(ByteBuffer buffer){
|
||||
var wrote = buffer.remaining();
|
||||
return waker -> {
|
||||
socket.write(buffer);
|
||||
if(!buffer.hasRemaining()) return wrote;
|
||||
SELECTOR.register(socket, SelectionKey.OP_WRITE, waker);
|
||||
return Future.Pending.INSTANCE;
|
||||
};
|
||||
}
|
||||
|
||||
public Future<Integer, IOException> read(ByteBuffer buffer){
|
||||
return waker -> {
|
||||
var read = socket.read(buffer);
|
||||
if(read!=0) return read;
|
||||
SELECTOR.register(socket, SelectionKey.OP_READ, waker);
|
||||
return Future.Pending.INSTANCE;
|
||||
};
|
||||
}
|
||||
|
||||
public Future<Integer, IOException> read_all(ByteBuffer buffer){
|
||||
int read = buffer.remaining();
|
||||
return waker -> {
|
||||
var read_now = socket.read(buffer);
|
||||
if(read_now ==-1)throw new IOException("Reached EOS while filling buffer");
|
||||
if(!buffer.hasRemaining()) return read;
|
||||
SELECTOR.register(socket, SelectionKey.OP_READ, waker);
|
||||
return Future.Pending.INSTANCE;
|
||||
};
|
||||
}
|
||||
|
||||
@Override
|
||||
public void close() throws Exception {
|
||||
socket.close();
|
||||
}
|
||||
}
|
||||
|
|
@ -1,67 +1,35 @@
|
|||
package async_runtime.net;
|
||||
|
||||
import async_runtime.SelectorThread;
|
||||
import future.Future;
|
||||
import future.Waker;
|
||||
|
||||
import java.io.IOException;
|
||||
import java.net.InetSocketAddress;
|
||||
import java.net.SocketAddress;
|
||||
import java.net.SocketOption;
|
||||
import java.nio.channels.SelectionKey;
|
||||
import java.nio.channels.Selector;
|
||||
import java.nio.channels.ServerSocketChannel;
|
||||
import java.util.ArrayDeque;
|
||||
|
||||
public class ServerSocket implements AutoCloseable{
|
||||
|
||||
private final static Selector SELECTOR;
|
||||
private record ToRegister(ServerSocketChannel sc, int ops, Waker waker){}
|
||||
private final static ArrayDeque<ToRegister> to_register = new ArrayDeque<>();
|
||||
private final static SelectorThread<ServerSocketChannel, Waker> SELECTOR;
|
||||
|
||||
static {
|
||||
try {
|
||||
SELECTOR = Selector.open();
|
||||
} catch (IOException e) {
|
||||
throw new RuntimeException(e);
|
||||
}
|
||||
|
||||
var thread = new Thread(() -> {
|
||||
while(!Thread.currentThread().isInterrupted()){
|
||||
try{
|
||||
synchronized (to_register){
|
||||
while(!to_register.isEmpty()){
|
||||
var to = to_register.poll();
|
||||
to.sc.register(SELECTOR, to.ops, to.waker);
|
||||
}
|
||||
}
|
||||
|
||||
SELECTOR.select();
|
||||
var keys = SELECTOR.selectedKeys().iterator();
|
||||
|
||||
while (keys.hasNext()) {
|
||||
SelectionKey key = keys.next();
|
||||
keys.remove();
|
||||
var c = (ServerSocketChannel)key.channel();
|
||||
var w = (Waker)key.attachment();
|
||||
|
||||
SELECTOR = new SelectorThread<>("ServerSocket") {
|
||||
@Override
|
||||
public void handle(SelectionKey key, ServerSocketChannel c, Waker w) {
|
||||
if (!key.isValid()) {
|
||||
}else if(key.isAcceptable()){
|
||||
w.wake();
|
||||
}
|
||||
}
|
||||
}catch (Exception e){
|
||||
e.printStackTrace();
|
||||
};
|
||||
} catch (IOException e) {
|
||||
throw new RuntimeException(e);
|
||||
}
|
||||
}
|
||||
});
|
||||
thread.setName("ServerSocket Polling Thread");
|
||||
thread.setDaemon(true);
|
||||
thread.start();
|
||||
}
|
||||
|
||||
private static void register(ServerSocketChannel sc, int ops, Waker waker){
|
||||
synchronized (to_register){
|
||||
to_register.add(new ToRegister(sc, ops, waker));
|
||||
}
|
||||
SELECTOR.wakeup();
|
||||
}
|
||||
|
||||
private final ServerSocketChannel socket;
|
||||
|
||||
|
|
@ -77,27 +45,23 @@ public class ServerSocket implements AutoCloseable{
|
|||
}
|
||||
|
||||
public Future<Socket, IOException> accept(){
|
||||
return new Future<>() {
|
||||
@Override
|
||||
public Object poll(Waker waker) throws IOException {
|
||||
var socc = socket.accept();
|
||||
if(socc==null) {
|
||||
register(socket, SelectionKey.OP_ACCEPT, waker);
|
||||
return Pending.INSTANCE;
|
||||
return waker -> {
|
||||
var accepted = socket.accept();
|
||||
if(accepted==null) {
|
||||
SELECTOR.register(socket, SelectionKey.OP_ACCEPT, waker);
|
||||
return Future.Pending.INSTANCE;
|
||||
}
|
||||
socc.configureBlocking(false);
|
||||
return new Socket(socc);
|
||||
accepted.configureBlocking(false);
|
||||
return new Socket(accepted);
|
||||
};
|
||||
}
|
||||
|
||||
@Override
|
||||
public void cancel() {
|
||||
try {
|
||||
if(socket!=null) socket.close();
|
||||
} catch (IOException e) {
|
||||
throw new RuntimeException(e);
|
||||
public <T> void set_options(SocketOption<T> option, T value) throws IOException{
|
||||
socket.setOption(option, value);
|
||||
}
|
||||
}
|
||||
};
|
||||
|
||||
public SocketAddress local_address() throws IOException {
|
||||
return socket.getLocalAddress();
|
||||
}
|
||||
|
||||
@Override
|
||||
|
|
|
|||
|
|
@ -1,47 +1,25 @@
|
|||
package async_runtime.net;
|
||||
|
||||
import async_runtime.SelectorThread;
|
||||
import future.Future;
|
||||
import future.Waker;
|
||||
|
||||
import java.io.IOException;
|
||||
import java.net.InetSocketAddress;
|
||||
import java.net.SocketAddress;
|
||||
import java.net.SocketOption;
|
||||
import java.nio.ByteBuffer;
|
||||
import java.nio.channels.SelectionKey;
|
||||
import java.nio.channels.Selector;
|
||||
import java.nio.channels.SocketChannel;
|
||||
import java.util.ArrayDeque;
|
||||
|
||||
public class Socket implements AutoCloseable{
|
||||
private final static SelectorThread<SocketChannel, Waker> SELECTOR;
|
||||
|
||||
private final static Selector SELECTOR;
|
||||
private record ToRegister(SocketChannel sc, int ops, Waker waker){}
|
||||
private final static ArrayDeque<ToRegister> to_register = new ArrayDeque<>();
|
||||
static {
|
||||
try {
|
||||
SELECTOR = Selector.open();
|
||||
} catch (IOException e) {
|
||||
throw new RuntimeException(e);
|
||||
}
|
||||
|
||||
var thread = new Thread(() -> {
|
||||
while(!Thread.currentThread().isInterrupted()){
|
||||
try{
|
||||
synchronized (to_register){
|
||||
while(!to_register.isEmpty()){
|
||||
var to = to_register.poll();
|
||||
to.sc.register(SELECTOR, to.ops, to.waker);
|
||||
}
|
||||
}
|
||||
|
||||
SELECTOR.select();
|
||||
var keys = SELECTOR.selectedKeys().iterator();
|
||||
|
||||
while (keys.hasNext()) {
|
||||
SelectionKey key = keys.next();
|
||||
keys.remove();
|
||||
var c = (SocketChannel)key.channel();
|
||||
var w = (Waker)key.attachment();
|
||||
|
||||
SELECTOR = new SelectorThread<>("Socket") {
|
||||
@Override
|
||||
public void handle(SelectionKey key, SocketChannel c, Waker w) throws IOException {
|
||||
if (!key.isValid()) {
|
||||
}else if(key.isAcceptable()){
|
||||
}else if(key.isConnectable()){
|
||||
|
|
@ -53,22 +31,11 @@ public class Socket implements AutoCloseable{
|
|||
w.wake();
|
||||
}
|
||||
}
|
||||
}catch (Exception e){
|
||||
e.printStackTrace();
|
||||
};
|
||||
} catch (IOException e) {
|
||||
throw new RuntimeException(e);
|
||||
}
|
||||
}
|
||||
});
|
||||
thread.setName("Socket Polling Thread");
|
||||
thread.setDaemon(true);
|
||||
thread.start();
|
||||
}
|
||||
|
||||
private static void register(SocketChannel sc, int ops, Waker waker){
|
||||
synchronized (to_register){
|
||||
to_register.add(new ToRegister(sc, ops, waker));
|
||||
}
|
||||
SELECTOR.wakeup();
|
||||
}
|
||||
|
||||
private final SocketChannel socket;
|
||||
|
||||
|
|
@ -86,88 +53,65 @@ public class Socket implements AutoCloseable{
|
|||
socket.configureBlocking(false);
|
||||
var connected = socket.connect(inet);
|
||||
if(!connected) {
|
||||
register(socket, SelectionKey.OP_CONNECT, waker);
|
||||
SELECTOR.register(socket, SelectionKey.OP_CONNECT, waker);
|
||||
return Pending.INSTANCE;
|
||||
}
|
||||
}
|
||||
if(socket.isConnected()) return new Socket(socket);
|
||||
return Pending.INSTANCE;
|
||||
}
|
||||
};
|
||||
}
|
||||
|
||||
@Override
|
||||
public void cancel() {
|
||||
try {
|
||||
if(socket!=null) socket.close();
|
||||
} catch (IOException e) {
|
||||
throw new RuntimeException(e);
|
||||
public <T> Socket set_options(SocketOption<T> option, T value) throws IOException{
|
||||
socket.setOption(option, value);
|
||||
return this;
|
||||
}
|
||||
|
||||
public SocketAddress local_address() throws IOException {
|
||||
return socket.getLocalAddress();
|
||||
}
|
||||
|
||||
public SocketAddress remote_address() throws IOException {
|
||||
return socket.getRemoteAddress();
|
||||
}
|
||||
|
||||
public Future<Integer, IOException> write(ByteBuffer buffer){
|
||||
return waker -> {
|
||||
var wrote = socket.write(buffer);
|
||||
if(wrote!=0) return wrote;
|
||||
SELECTOR.register(socket, SelectionKey.OP_WRITE, waker);
|
||||
return Future.Pending.INSTANCE;
|
||||
};
|
||||
}
|
||||
|
||||
public Future<Integer, IOException> write_all(ByteBuffer buffer){
|
||||
return new Future<>() {
|
||||
int wrote = 0;
|
||||
@Override
|
||||
public Object poll(Waker waker) throws IOException {
|
||||
wrote += socket.write(buffer);
|
||||
var wrote = buffer.remaining();
|
||||
return waker -> {
|
||||
socket.write(buffer);
|
||||
if(!buffer.hasRemaining()) return wrote;
|
||||
register(socket, SelectionKey.OP_WRITE, waker);
|
||||
return Pending.INSTANCE;
|
||||
}
|
||||
|
||||
@Override
|
||||
public void cancel() {
|
||||
try {
|
||||
if(socket!=null) socket.close();
|
||||
} catch (IOException e) {
|
||||
throw new RuntimeException(e);
|
||||
}
|
||||
}
|
||||
SELECTOR.register(socket, SelectionKey.OP_WRITE, waker);
|
||||
return Future.Pending.INSTANCE;
|
||||
};
|
||||
}
|
||||
|
||||
public Future<Integer, IOException> read(ByteBuffer buffer){
|
||||
return new Future<>() {
|
||||
int read = 0;
|
||||
@Override
|
||||
public Object poll(Waker waker) throws IOException {
|
||||
read += socket.read(buffer);
|
||||
if(read>0) return read;
|
||||
register(socket, SelectionKey.OP_READ, waker);
|
||||
return Pending.INSTANCE;
|
||||
}
|
||||
|
||||
@Override
|
||||
public void cancel() {
|
||||
try {
|
||||
if(socket!=null) socket.close();
|
||||
} catch (IOException e) {
|
||||
throw new RuntimeException(e);
|
||||
}
|
||||
}
|
||||
return waker -> {
|
||||
var read = socket.read(buffer);
|
||||
if(read!=0) return read;
|
||||
SELECTOR.register(socket, SelectionKey.OP_READ, waker);
|
||||
return Future.Pending.INSTANCE;
|
||||
};
|
||||
}
|
||||
|
||||
public Future<Integer, IOException> read_all(ByteBuffer buffer){
|
||||
return new Future<>() {
|
||||
int read = 0;
|
||||
@Override
|
||||
public Object poll(Waker waker) throws IOException {
|
||||
read += socket.read(buffer);
|
||||
int read = buffer.remaining();
|
||||
return waker -> {
|
||||
var read_now = socket.read(buffer);
|
||||
if(read_now ==-1)throw new IOException("Reached EOS while filling buffer");
|
||||
if(!buffer.hasRemaining()) return read;
|
||||
register(socket, SelectionKey.OP_READ, waker);
|
||||
return Pending.INSTANCE;
|
||||
}
|
||||
|
||||
@Override
|
||||
public void cancel() {
|
||||
try {
|
||||
if(socket!=null) socket.close();
|
||||
} catch (IOException e) {
|
||||
throw new RuntimeException(e);
|
||||
}
|
||||
}
|
||||
SELECTOR.register(socket, SelectionKey.OP_READ, waker);
|
||||
return Future.Pending.INSTANCE;
|
||||
};
|
||||
}
|
||||
|
||||
|
|
|
|||
|
|
@ -2,9 +2,7 @@ package future;
|
|||
|
||||
public interface Future<R, E extends Throwable> {
|
||||
|
||||
default Object poll(Waker waker) throws E{
|
||||
return Pending.INSTANCE;
|
||||
}
|
||||
Object poll(Waker waker) throws E;
|
||||
|
||||
default R await() throws E{
|
||||
throw new RuntimeException("NO!");
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue