SerinaNya/KLALB
1
1
Fork 0
forked from KNEMC/KLALB
Code Pull Requests Activity

优化代码，性能暴涨

Browse Source
This commit is contained in:
Administrator
2025-01-29 14:01:11 +08:00
parent e243203535
commit 6de65562be
130 changed files with 9138 additions and 1020 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/org/kne/cloud/network/klalb/KLALBVirtualSocketImpl.java
+424 -118
View File
@@ -34,10 +34,12 @@ import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.LinkedBlockingDeque;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.Semaphore;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicLong;
import java.util.concurrent.atomic.AtomicReference;
import java.util.concurrent.atomic.AtomicReferenceArray;
import java.util.concurrent.locks.Lock;
@@ -54,10 +56,12 @@ import java.util.zip.Inflater;
import java.util.zip.InflaterInputStream;
import org.kne.acclerate.FastLib;
import org.kne.cloud.network.PortPair;
import org.kne.cloud.network.SpeedLimiter;
import org.kne.cloud.network.ThreadTool;
import org.kne.cloud.network.VirtualSocketImpl;
import org.kne.cloud.network.monitor.SpeedAndTrafficAndDelayMonitorDataImpl;
import org.kne.concurrent.SpinLock;
import org.kne.debug.TimeDebugger;
import org.kne.io.Data;
@@ -82,6 +86,8 @@ public class KLALBVirtualSocketImpl extends VirtualSocketImpl implements Bindabl
private SpeedAndTrafficAndDelayMonitorDataImpl socketMonitor=new SpeedAndTrafficAndDelayMonitorDataImpl();
private SpeedAndTrafficAndDelayMonitorDataImpl socketRawMonitor=new SpeedAndTrafficAndDelayMonitorDataImpl();
private KLALBController controller;
protected int getInputchachesize() {
@@ -100,16 +106,29 @@ public class KLALBVirtualSocketImpl extends VirtualSocketImpl implements Bindabl
this.outputchachesize = outputchachesize;
}
//private Speed
//60000 30 30
private static final int MTU=8192;
private volatile int inputchachesize = MTU * 2000;
private volatile int outputchachesize = MTU * 50;
private volatile int reallimit = MTU * 40;
//private SpeedLimiter spdlmt=new SpeedLimiter(1024*1024);
//4000 100 100
//10000 500 500
private volatile int inputchachesize = MTU * 4000;
private volatile int outputchachesize = MTU * 240;//500
private volatile int reallimit = MTU * 240;
private final long MIN_RTTVAR=50000000L;
private final long MIN_LIMIT_SPEED=64*1024L;
private volatile long rcvSpeed=MIN_LIMIT_SPEED;
private volatile long requestSpeed=MIN_LIMIT_SPEED;
private SpeedLimiter spdlmt=new SpeedLimiter(MIN_LIMIT_SPEED,1000000L);
private volatile long congressSpeed=0;
private volatile double congressFactor=2;
//private double[] congressFactors=new double[] {0.95,0.95,0.95,1.2,0.8};
//private int congressFactorsState=0;
//private volatile double maxutilization=0.8;
private volatile boolean nodelay=false;
private volatile long delaytime=1;
private volatile boolean nodelay=true;
private volatile long delaytime=2;
public long getDelaytime() {
return delaytime;
@@ -131,9 +150,13 @@ public class KLALBVirtualSocketImpl extends VirtualSocketImpl implements Bindabl
private ReentrantLock backlogQueuelock=new ReentrantLock();
private volatile Thread sendDequeLock;
private Queue<DATATPacket> sendDeque = new ArrayBlockingQueue<DATATPacket>(20000);
private Queue<DATATPacket> recvQueue = new ConcurrentLinkedQueue<DATATPacket>();
private AtomicInteger recvQueueUsed=new AtomicInteger(0);
//private AtomicInteger recvCounter=new AtomicInteger(0);
private Map<Long,DATATPacket> sendmap=new ConcurrentHashMap();
private AtomicInteger sendmapWindowUsed=new AtomicInteger(0);
//private ReentrantReadWriteLock sendmaplock=new ReentrantReadWriteLock();
private volatile Thread sendthread;
@@ -142,27 +165,29 @@ public class KLALBVirtualSocketImpl extends VirtualSocketImpl implements Bindabl
private class SendCheckTask extends TimerTask{
public void run() {
/*sendmaplock.writeLock().lock();
try {*/
try {
Collection<DATATPacket> cdp=sendmap.values();
for (Iterator iterator = cdp.iterator(); iterator.hasNext();) {
DATATPacket dtp = (DATATPacket) iterator.next();
try {
long x=System.nanoTime();
long dt=x-dtp.resendtimer;
long limit= (long) (Math.pow(2, dtp.getSendRecord().size()-1)*(RTTMin*20+100000L));
long limit= (long) (Math.pow(2, dtp.getSendCounter()-1)*(RTO));
if(dt>limit) {
if(dtp.getSendRecord().size()>=10) {
if(dtp.getSendCounter()>=10) {
throw new IOException("send error!");
}
/*if(reallimit>LIMIT*2) {
reallimit=reallimit-4096;
System.out.println(reallimit+" -4096");
}*/
/*if(spdlmt.getLimitspeed()>LIMIT*2)
spdlmt.setLimitspeed(spdlmt.getLimitspeed()-4096);*/
if(dtp.isDisposed())
continue;
dtp.setPriority(4);
controller.sendPacketToAddress((Inet6Address) remoteaddr,dtp,1);
long length= dtp.getLength();
socketRawMonitor.getOutTrafficAL().addAndGet(length);
spdlmt.forceTransmit(length);
controller.sendPacketToAddress((Inet6Address) remoteaddr,0,dtp,1);
//System.out.println("第"+(dtp.getSendRecord().size()-1)+"次重传："+dtp+" "+dt+">"+limit);
dtp.resendtimer=x;
@@ -178,10 +203,11 @@ public class KLALBVirtualSocketImpl extends VirtualSocketImpl implements Bindabl
}
}
/*}finally {
sendmaplock.writeLock().unlock();
}*/
}catch(Exception e) {
e.printStackTrace();
}
if(portp!=null)
updateBandwidthReq(requestSpeed);
}
}
@@ -191,11 +217,12 @@ public class KLALBVirtualSocketImpl extends VirtualSocketImpl implements Bindabl
@Override
public void run() {
if(sendDeque.isEmpty()) {
try {if(getLocalPort()!=0&&getPort()!=0)
if(recvQueue.isEmpty()) {
try {
if(getLocalPort()!=0&&getPort()!=0)
if(remoteaddr instanceof Inet6Address&&(!remoteaddr.isAnyLocalAddress()))
controller.sendPacketToAddress((Inet6Address) remoteaddr, new ACKTPacket(getLocalPort(),getPort(), -1,
true,0));
controller.sendPacketToAddress((Inet6Address) remoteaddr,0, new ACKTPacket(getLocalPort(),getPort(), -1,
true,false,socketMonitor.getInSpeedMax(),0));
} catch (IOException e) {
try {
close0(true);
@@ -215,15 +242,23 @@ public class KLALBVirtualSocketImpl extends VirtualSocketImpl implements Bindabl
return backlogQueue != null;
}
//private int inputcross = 0;
private List<Object> inputchache = new ArrayList<>();
private ReentrantLock inputchachelock=new ReentrantLock();
//private List<Object> inputchache = new RangeArrayList<>();
private Map<Long,DATATPacket> inputchache=new ConcurrentHashMap();
private Lock inputchachelock=new SpinLock();
private long inputcount = 0;
private volatile boolean avaliable = true;
private AtomicBoolean firstUpdate=new AtomicBoolean(true);
private volatile long RTTMin=1000000000L;
private volatile long RTTVar=1000000000L;
private volatile long RTTAvg=1000000000L;
private volatile long RTO=1000000000L;
private volatile long QueueingAvg=1000000000L;
private volatile long RunningSpeed=spdlmt.getLimitspeed();
private long RTTMin=1000000000L;
private volatile boolean ignoreBindCheck;
private boolean connected;
@@ -307,7 +342,9 @@ public class KLALBVirtualSocketImpl extends VirtualSocketImpl implements Bindabl
protected void connect(InetAddress address, int port) throws IOException {
connect(new InetSocketAddress(address, port), 10000);
}
private PortPair portp;
@Override
protected void connect(SocketAddress address, int timeout) throws IOException {
if(!ignoreBindCheck)
@@ -317,13 +354,14 @@ public class KLALBVirtualSocketImpl extends VirtualSocketImpl implements Bindabl
port = ((InetSocketAddress) address).getPort();
this.address=this.remoteaddr = (Inet6Address) ((InetSocketAddress) address).getAddress();
if(connected)
throw new SocketException("already connected");
controller.getStreamPortBinder().connect(this);
connectionPending=true;
controller.getResendTimer().schedule(sendCheckTask, 50, 50);
controller.getResendTimer().schedule(sendCheckTask, 10, 10);
//controller.sendPacketToAddress((Inet6Address) this.remoteaddr, new SYNTPacket(localport, port), 0,1);
try {
getKVSIOutputStream().write(compress);
@@ -345,6 +383,16 @@ public class KLALBVirtualSocketImpl extends VirtualSocketImpl implements Bindabl
}
connected=true;
PortPair portpx=new PortPair(localport, port);
controller.registerDistUpdateConsumer(remoteaddr, portpx, (c)->{
//System.out.println(c);
spdlmt.setLimitspeed(Math.max(MIN_LIMIT_SPEED,c));
});
updateBandwidthReq(requestSpeed);
this.portp=portpx;
}catch(SocketTimeoutException e) {
throw new SocketTimeoutException("connect time out");
}catch(NoRouteToHostException e) {
@@ -357,6 +405,11 @@ public class KLALBVirtualSocketImpl extends VirtualSocketImpl implements Bindabl
controller.getResendTimer().schedule(flowControlTask, 5000, 5000);
}
private void updateBandwidthReq(long requestSpeed) {
if(remoteaddr!=null&&portp!=null)
controller.updateBandwidthRequest(remoteaddr, portp,requestSpeed );
}
public boolean isConnectionPending() {
return connectionPending;
}
@@ -424,7 +477,7 @@ public class KLALBVirtualSocketImpl extends VirtualSocketImpl implements Bindabl
kvsi.remoteaddr=(Inet6Address) isa.getAddress();
kvsi.bind(localaddr, localport,true);
kvsi.accept((KLALBRemoteLine)p[1],(KLALBPacket) p[2]);
kvsi.accept((Inet6Address)p[1],(KLALBPacket) p[2]);
kvsi.connect(isa, 5000);
/*kvsi.port = isa.getPort();
kvsi.localport = localport;
@@ -463,26 +516,30 @@ public class KLALBVirtualSocketImpl extends VirtualSocketImpl implements Bindabl
while (true) {
if (isClosed())
throw new SocketException("Socket is closed");
DATATPacket dtp2 = sendDeque.poll();
DATATPacket dtp2 = recvQueue.poll();
if (dtp2 != null) {
recvQueueUsed.addAndGet(-dtp2.getSize());
//System.out.println("PULL:"+dtp2);
dataPack = dtp2;
socketMonitor.getInTrafficAL().addAndGet(dtp2.getSize());
socketMonitor.getInPacketCounterAL().incrementAndGet();
controller.getDatatMonitor().getInTrafficAL().addAndGet(dtp2.getSize());
checkFlowControl(dtp2);
controller.getDatatMonitor().getInPacketCounterAL().incrementAndGet();
//checkFlowControl(dtp2);
break;
}
sendDequeLock=Thread.currentThread();
LockSupport.parkNanos(1000000);
}
//System.out.println("sorted:"+dataPack);
return dataPack;
}
private void checkFlowControl(DATATPacket dtp2) throws IOException {
if(sendDeque.size() >= inputchachesize/MTU-4) {
controller.sendPacketToAddress(remoteaddr, new ACKTPacket(dtp2.getDport(), dtp2.getSport(), dtp2.getNumber(),
/*private void checkFlowControl(DATATPacket dtp2) throws IOException {
if(recvQueue.size() >= inputchachesize/MTU-4) {
controller.sendPacketToAddress(remoteaddr,0,KLALB_PROTOCOL_NUMBER, new ACKTPacket(dtp2.getDport(), dtp2.getSport(), dtp2.getNumber(),
true,dtp2.getSendcount()));
}
}
}*/
@Override
@@ -501,6 +558,8 @@ public class KLALBVirtualSocketImpl extends VirtualSocketImpl implements Bindabl
dst.limit(dst.position()+len);
dst.put( dataPack.getDataBuffer().get()) ;
if(!dataPack.getDataBuffer().hasRemaining()) {
dataPack.putTimePassport("unpacked");
dataPack.printPassport();
dataPack.dispose();
dataPack=null;
}
@@ -523,6 +582,8 @@ public class KLALBVirtualSocketImpl extends VirtualSocketImpl implements Bindabl
i+=min;
//b[off + i]= dtp.getData()[count++] ;
if(!dataPack.getDataBuffer().hasRemaining()) {
dataPack.putTimePassport("unpacked");
dataPack.printPassport();
dataPack.dispose();
dataPack=null;
}
@@ -558,6 +619,8 @@ public class KLALBVirtualSocketImpl extends VirtualSocketImpl implements Bindabl
len = Math.min(len, available());
b[off]= dataPack.getDataBuffer().get() ;
if(!dataPack.getDataBuffer().hasRemaining()) {
dataPack.putTimePassport("unpacked");
dataPack.printPassport();
dataPack.dispose();
dataPack=null;
}
@@ -578,6 +641,8 @@ public class KLALBVirtualSocketImpl extends VirtualSocketImpl implements Bindabl
i+=min;
//b[off + i]= dtp.getData()[count++] ;
if(!dataPack.getDataBuffer().hasRemaining()) {
dataPack.putTimePassport("unpacked");
dataPack.printPassport();
dataPack.dispose();
dataPack=null;
}
@@ -601,13 +666,13 @@ public class KLALBVirtualSocketImpl extends VirtualSocketImpl implements Bindabl
}
@Override
public int available() throws IOException {
AtomicInteger i = new AtomicInteger(0);
sendDeque.forEach((V) -> {
i.addAndGet(V.getSize());
});
int i = recvQueueUsed.get();
/*for(DATATPacket V:recvQueue) {
i+=(V.getSize());
}*/
if (dataPack != null)
i.addAndGet(dataPack.getDataBuffer().remaining());
return i.get();
i+=dataPack.getDataBuffer().remaining();
return i;
}
@Override
@@ -629,7 +694,6 @@ public class KLALBVirtualSocketImpl extends VirtualSocketImpl implements Bindabl
this.compress = compress;
}
private static final int MTU=60000;
protected class KVSIOutputStream extends OutputStream implements WritableByteChannel{
DATATPacket dataPack=new DATATPacket(localport, port, outputcount++,MTU);
@@ -730,7 +794,7 @@ public class KLALBVirtualSocketImpl extends VirtualSocketImpl implements Bindabl
throw new SocketException("Socket is closed");
if(sendmap.isEmpty())
break;
if(timeout!=0&&(System.nanoTime()-start>timeout*1000000))
if(timeout!=0&&(System.nanoTime()-start>timeout*1000000L))
throw new SocketTimeoutException("wait for acknowledged timout");
sendthread=Thread.currentThread();
LockSupport.parkNanos(1000000L);
@@ -794,6 +858,7 @@ public class KLALBVirtualSocketImpl extends VirtualSocketImpl implements Bindabl
}
//TimeDebugger td=new TimeDebugger();
//td.putTime("start");
dataPack.putTimePassport("packed");
while (!avaliable) {
try {
Thread.sleep(1);
@@ -801,35 +866,46 @@ public class KLALBVirtualSocketImpl extends VirtualSocketImpl implements Bindabl
e.printStackTrace();
}
}
dataPack.putTimePassport("waitForAvaliable");
//td.putTime("waitForAvaliable");
while(true){
if (isClosed())
throw new SocketException("Socket is closed");
//System.out.println(sendmap.size());
boolean b=sendmap.size()<=reallimit/MTU;
boolean b=sendmapWindowUsed.get()<=reallimit;
//boolean b=sendmap.size()<=reallimit/MTU;
if(b)
break;
sendthread=Thread.currentThread();
LockSupport.parkNanos(1000000L);
}
dataPack.putTimePassport("waitForWindow");
//td.putTime("waitForCache");
DATATPacket pack=dataPack;
dataPack=new DATATPacket(localport, port, outputcount++,MTU);
dataPack.putTimePassport("buildHeader");
pack.getDataBuffer().flip();
//System.out.println(pack.getDataBuffer());
//cacheCreateTime=System.nanoTime();
//td.putTime("flushBuffer");
//spdlmt.transmit(count);
spdlmt.transmit(pack.getDataBuffer().limit());
socketMonitor.getOutTrafficAL().addAndGet(pack.getDataBuffer().limit());
socketMonitor.getOutPacketCounterAL().incrementAndGet();
controller.getDatatMonitor().getOutTrafficAL().addAndGet(pack.getDataBuffer().limit());
controller.getDatatMonitor().getOutPacketCounterAL().incrementAndGet();
//td.putTime("doStatistic");
pack.setPriority(5);
pack.resendtimer=System.nanoTime();
controller.sendPacketToAddress(remoteaddr,pack);
socketRawMonitor.getOutTrafficAL().addAndGet(pack.getLength());
controller.sendPacketToAddress(remoteaddr,0,pack);
//td.putTime("doSend");
/* sendmaplock.readLock().lock();
try{*/
pack.resendtimer=System.nanoTime();
sendmap.put(pack.getNumber(),pack);
DATATPacket prv= sendmap.put(pack.getNumber(),pack);
sendmapWindowUsed.addAndGet(pack.getSize());
if(prv!=null)
sendmapWindowUsed.addAndGet(-prv.getSize());
//System.out.println("PUSH:"+pack);
/*}finally {
sendmaplock.readLock().unlock();
@@ -849,20 +925,24 @@ public class KLALBVirtualSocketImpl extends VirtualSocketImpl implements Bindabl
}
private void close0() throws IOException {
//outputclosed=true;
if(isClosed())
return;
olock.lock();
try {
flush0();
}finally {
olock.unlock();
}
DATATPacket pack=new DATATPacket(localport, port, outputcount++,MTU);
pack.setPriority(5);
pack.getDataBuffer(). flip();
controller.sendPacketToAddress(remoteaddr,pack);
controller.sendPacketToAddress(remoteaddr,0,pack);
/*sendmaplock.readLock().lock();
try{*/
pack.resendtimer=System.nanoTime();
sendmap.put(pack.getNumber(),pack);
sendmapWindowUsed.addAndGet(pack.getSize());
/*}finally {
sendmaplock.readLock().unlock();
}*/
@@ -958,6 +1038,7 @@ public class KLALBVirtualSocketImpl extends VirtualSocketImpl implements Bindabl
private void close0(boolean b) throws IOException {
if ( !isClosed()) {
closed = true;
if(!isListening() ) {
sendCheckTask.cancel();
flowControlTask.cancel();
@@ -965,11 +1046,17 @@ public class KLALBVirtualSocketImpl extends VirtualSocketImpl implements Bindabl
if(remoteaddr!=null)
try {
controller.sendPacketToAddress(remoteaddr, new RSTPacket(super.localport, super.port),
controller.sendPacketToAddress(remoteaddr,0, new RSTPacket(super.localport, super.port),
2);
} catch (NoRouteToHostException e) {
}
controller.getStreamPortBinder().disconnect(this);
if(portp!=null) {
PortPair portpx=portp;
portp=null;
controller.updateBandwidthRequest(remoteaddr, portpx, 0L);
controller.registerDistUpdateConsumer(remoteaddr, portpx, null);
}
}else {
//System.out.println("unlisten"+getLocalPort());
controller.getStreamPortBinder().unlisten(this);
@@ -1001,18 +1088,22 @@ public void associateSocketChannel(SocketChannel b) throws IOException{
InputStream is=getInputStream();
OutputStream os=getOutputStream();
if((!(is instanceof KVSIInputStream))||(!(os instanceof KVSIOutputStream))) {
throw new SocketException("cant use association because compress is enabled");
throw new CannotAssociateException("cant use association because compress is enabled");
}
KVSIInputStream kis=(KVSIInputStream) is;
KVSIOutputStream kos=(KVSIOutputStream) os;
Thread t1=ThreadTool.makeVThreadIfSupport("本地发送线程", ()->{
boolean onError=false;
try {
DATATPacket dp;
while((dp=kis.nextPacket()).getSize()!=0) {
b.write(dp.getDataBuffer());
dp.putTimePassport("unpacked");
dp.printPassport();
dp.dispose();
}
}catch(IOException e) {
onError=true;
e.printStackTrace();
}finally {
try {
@@ -1025,23 +1116,68 @@ public void associateSocketChannel(SocketChannel b) throws IOException{
} catch (IOException e) {
e.printStackTrace();
}
if(true) {
try {
b.close();
} catch (IOException e) {
// TODO 自动生成的 catch 块
e.printStackTrace();
}
try {
close();
} catch (IOException e) {
// TODO 自动生成的 catch 块
e.printStackTrace();
}
}
}
});
Thread t2=ThreadTool.makeVThreadIfSupport("本地接收线程", ()->{
boolean onError=false;
try {
ByteBuffer tst=ByteBuffer.allocateDirect(MTU);
while(true) {
if(b.read(kos.dataPack.getDataBuffer())==-1) {
if(nodelay) {
if(b.read(kos.dataPack.getDataBuffer())==-1) {
break;
}
kos.flush0();
}else {
if(b.read(tst)==-1) {
break;
}
tst.flip();
kos.olock.lock();
try {
kos.flush0();
kos.dataPack.getDataBuffer().put(tst);
/*if(b.read(kos.dataPack.getDataBuffer())==-1) {
break;
}*/
}finally {
kos.olock.unlock();
}
if(kos.dataPack.getDataBuffer().hasRemaining()) {
kos.olock.lock();
try {
kos.flush();
}finally {
kos.olock.unlock();
}
}else {
kos.olock.lock();
try {
kos.flush0();
}finally {
kos.olock.unlock();
}
}
tst.clear();
}
}
}catch(IOException e) {
onError=true;
e.printStackTrace();
}finally {
try {
@@ -1054,6 +1190,20 @@ public void associateSocketChannel(SocketChannel b) throws IOException{
} catch (IOException e) {
e.printStackTrace();
}
if(true) {
try {
b.close();
} catch (IOException e) {
// TODO 自动生成的 catch 块
e.printStackTrace();
}
try {
close();
} catch (IOException e) {
// TODO 自动生成的 catch 块
e.printStackTrace();
}
}
}
});
t1.start();
@@ -1064,6 +1214,7 @@ public void associateSocketChannel(SocketChannel b) throws IOException{
} catch (InterruptedException e) {
e.printStackTrace();
}
b.close();
close();
}
@@ -1077,18 +1228,22 @@ public void associateSocketChannel(SocketChannel b) throws IOException{
InputStream is=getInputStream();
OutputStream os=getOutputStream();
if((!(is instanceof KVSIInputStream))||(!(os instanceof KVSIOutputStream))) {
throw new SocketException("cant use association because compress is enabled");
throw new CannotAssociateException("cant use association because compress is enabled");
}
KVSIInputStream kis=(KVSIInputStream) is;
KVSIOutputStream kos=(KVSIOutputStream) os;
Thread t1=ThreadTool.makeVThreadIfSupport("本地发送线程", ()->{
boolean onError=false;
try {
DATATPacket dp;
while((dp=kis.nextPacket()).getSize()!=0) {
orc.write(dp.getDataBuffer());
dp.putTimePassport("unpacked");
dp.printPassport();
dp.dispose();
}
}catch(IOException e) {
onError=true;
e.printStackTrace();
}finally {
try {
@@ -1101,23 +1256,68 @@ public void associateSocketChannel(SocketChannel b) throws IOException{
} catch (IOException e) {
e.printStackTrace();
}
if(true) {
try {
associateSocket.close();
} catch (IOException e) {
// TODO 自动生成的 catch 块
e.printStackTrace();
}
try {
close();
} catch (IOException e) {
// TODO 自动生成的 catch 块
e.printStackTrace();
}
}
}
});
Thread t2=ThreadTool.makeVThreadIfSupport("本地接收线程", ()->{
boolean onError=false;
try {
ByteBuffer tst=ByteBuffer.allocateDirect(MTU);
while(true) {
if(irc.read(kos.dataPack.getDataBuffer())==-1) {
break;
}
if(nodelay) {
if(irc.read(kos.dataPack.getDataBuffer())==-1) {
break;
}
kos.flush0();
}else {
if(irc.read(tst)==-1) {
break;
}
tst.flip();
kos.olock.lock();
try {
kos.dataPack.getDataBuffer().put(tst);
/*if(irc.read(kos.dataPack.getDataBuffer())==-1) {
break;
}*/
}finally {
kos.olock.unlock();
}
if(kos.dataPack.getDataBuffer().hasRemaining()) {
kos.olock.lock();
try {
kos.flush();
}finally {
kos.olock.unlock();
}
}else {
kos.olock.lock();
try {
kos.flush0();
}finally {
kos.olock.unlock();
}
}
tst.clear();
}
}
}catch(IOException e) {
onError=true;
e.printStackTrace();
}finally {
try {
@@ -1130,6 +1330,20 @@ public void associateSocketChannel(SocketChannel b) throws IOException{
} catch (IOException e) {
e.printStackTrace();
}
if(true) {
try {
associateSocket.close();
} catch (IOException e) {
// TODO 自动生成的 catch 块
e.printStackTrace();
}
try {
close();
} catch (IOException e) {
// TODO 自动生成的 catch 块
e.printStackTrace();
}
}
}
});
t1.start();
@@ -1140,11 +1354,12 @@ public void associateSocketChannel(SocketChannel b) throws IOException{
} catch (InterruptedException e) {
e.printStackTrace();
}
associateSocket.close();
close();
}
@Override
public void accept(KLALBRemoteLine from, KLALBPacket u) {
public void accept(Inet6Address from, KLALBPacket u) {
try {
//System.out.println(this+" "+u);
switch (u.getType()) {
@@ -1155,13 +1370,15 @@ public void associateSocketChannel(SocketChannel b) throws IOException{
break;
case KLALBPacket.DATAT:
DATATPacket dtp = (DATATPacket) u;
socketRawMonitor.getInTrafficAL().addAndGet(dtp.getLength());
if(isListening()) {
if(dtp.getNumber()==0) {
backlogQueuelock.lock();
try{
//controller.getStreamPortBinder().checkIsConnected(new Pair);
InetSocketAddress is=new InetSocketAddress(from.getRemoteVaddr(), dtp.getSport());
InetSocketAddress is=new InetSocketAddress(from, dtp.getSport());
AtomicBoolean ab=new AtomicBoolean(true);
for (Iterator iterator = backlogQueue.iterator(); iterator.hasNext();) {
Object[] objects = (Object[]) iterator.next();
@@ -1171,7 +1388,7 @@ public void associateSocketChannel(SocketChannel b) throws IOException{
}
}
if(ab.get()) {
if(controller.getStreamPortBinder().checkIsConnect(this,new InetSocketAddress(from.getRemoteVaddr(), dtp.getSport()))) {
if(controller.getStreamPortBinder().checkIsConnect(this,new InetSocketAddress(from, dtp.getSport()))) {
ab.set(false);
}
}
@@ -1182,7 +1399,7 @@ public void associateSocketChannel(SocketChannel b) throws IOException{
/* controller.sendPacketToAddress(from.getRemoteVaddr(), new ACKTPacket(dtp.getDport(), dtp.getSport(),dtp.getNumber(),true,0),
0,2);*/
}else {
controller.sendPacketToAddress(from.getRemoteVaddr(), new RSTPacket(dtp.getDport(), dtp.getSport()),
controller.sendPacketToAddress(from,0, new RSTPacket(dtp.getDport(), dtp.getSport()),
2);
}
@@ -1191,96 +1408,185 @@ public void associateSocketChannel(SocketChannel b) throws IOException{
backlogQueuelock.unlock();
}
}else {
controller.sendPacketToAddress(from.getRemoteVaddr(), new RSTPacket(dtp.getDport(), dtp.getSport()),
controller.sendPacketToAddress(from,0, new RSTPacket(dtp.getDport(), dtp.getSport()),
2);
}
}else {
controller.sendPacketToAddress(from.getRemoteVaddr(), new ACKTPacket(dtp.getDport(), dtp.getSport(), dtp.getNumber(),
sendDeque.size() < inputchachesize/MTU,dtp.getSendcount()), 1);
inputchachelock.lock();
try{
if (dtp.getNumber() >= inputcount) {
int currindex=(int) (dtp.getNumber()-inputcount);
//System.out.println(dtp.isCE());
controller.sendPacketToAddress(from,0, new ACKTPacket(dtp.getDport(), dtp.getSport(), dtp.getNumber(),
recvQueueUsed.get() < inputchachesize,dtp.isCE(),socketMonitor.getInSpeedMax(),dtp.getSendcount()), 1);
boolean added=false;
long number=dtp.getNumber();
if (number >= inputcount) {
inputchache.putIfAbsent(number, dtp);
/*inputchachelock.lock();
try{
int currindex=(int) (number-inputcount);
int reqsize=1+currindex;
while(reqsize>inputchache.size()) {
inputchache.add(new AtomicInteger());
}
for (int i = 0; i < currindex; i++) {
Object o=inputchache.get(i);
if(o instanceof AtomicInteger) {
((AtomicInteger) o).incrementAndGet();
/*if(((AtomicInteger) o).get()==30) {
controller.sendPacketToAddress(from.getRemoteVaddr(), new NACKTPacket(dtp.getDport(), dtp.getSport(), inputcount+i),1);
//System.out.println("请求快速重传："+(inputcount+i));
}*/
}
}
inputchache.set(currindex, dtp);
Iterator<Object>itr=inputchache.iterator();
while (itr.hasNext()) {
Object datatPacket = itr.next();
if(datatPacket instanceof DATATPacket) {
itr.remove();
sendDeque.add((DATATPacket) datatPacket);
LockSupport.unpark(sendDequeLock);
inputcount++;
}else {
break;
}
inputchache.add(null);
}
if(currindex>=0) {
DATATPacket old=(DATATPacket) inputchache.get(currindex);
if(old==null) {
inputchache.set(currindex, dtp);
}else {
dtp.dispose();
}
}
int spos=inputchache.size();
for (int i = 0; i < inputchache.size(); i++) {
DATATPacket datatPacket=(DATATPacket) inputchache.get(i);
if(datatPacket==null) {
spos=i;
break;
}
}
for (int i = 0; i < spos; i++) {
DATATPacket datatPacket = (DATATPacket)inputchache.get(i);
recvQueueUsed.addAndGet(datatPacket.getSize());
recvQueue.add(datatPacket);
added=true;
inputcount++;
}
((RangeArrayList)inputchache).removeRange(0,spos);
}finally{
inputchachelock.unlock();
}*/
}else {
dtp.dispose();
}
}finally{
inputchachelock.lock();
try {
DATATPacket datatPacket = inputchache.remove(inputcount);
if(datatPacket!=null) {
inputcount++;
recvQueueUsed.addAndGet(datatPacket.getSize());
recvQueue.add(datatPacket);
added=true;
}
}finally {
inputchachelock.unlock();
}
if(added) {
LockSupport.unpark(sendDequeLock);
}
//dbg.println(from.getMonitor()+","+dtp.getNumber());
}
break;
case KLALBPacket.ACKT:
ACKTPacket ackt = (ACKTPacket) u;
if(isListening()) {
controller.sendPacketToAddress(from.getRemoteVaddr(), new RSTPacket(ackt.getDport(), ackt.getSport()),
controller.sendPacketToAddress(from,0, new RSTPacket(ackt.getDport(), ackt.getSport()),
2);
}else {
avaliable = ackt.isAvaliable();
rcvSpeed=ackt.getRcvSpeed();
/*if(rcvSpeed>=congressSpeed) {
congressSpeed=rcvSpeed;
}else {
congressSpeed=(congressSpeed*99+rcvSpeed)/100;
}*/
if(ackt.isCongress()) {
congressFactor=1.2;
if(reallimit>MTU*2) {
reallimit-=8192;
//System.out.println(reallimit+" -2048");
}
}else {
if(sendmapWindowUsed.get()*2L>=reallimit) {
reallimit+=1024;
//System.out.println(reallimit+" +1024");
}
}
requestSpeed=(long) (Math.max(MIN_LIMIT_SPEED, rcvSpeed)*congressFactor);
//System.out.println(spdlmt.getLimitspeed()/1024+"K "+ackt.getRcvSpeed()/1024+"K");
DATATPacket kl=null;
/*sendmaplock.readLock().lock();
try{*/
kl=sendmap.remove(ackt.getNumber());
/* }finally {
sendmaplock.readLock().unlock();
}*/
if(kl!=null) {
sendmapWindowUsed.addAndGet(-kl.getSize());
}
if(kl!=null) {
if(sendthread!=null)
LockSupport.unpark(sendthread);
//controller.removeFromSend(from.getRemoteVaddr(),kl);
if(kl.getSendRecord().size()==1) {
if(kl.getSendCounter()==1) {
long RTTC=ackt.getRcvtime()- kl.getSndtime();
if(RTTC<=RTTMin) {
RTTMin=RTTC;
}else {
RTTMin= (RTTMin*9999+RTTC)/10000;
RTTMin= (RTTMin*99999+RTTC)/100000;
}
/*if(RTTC>RTTMin*2) {
if(reallimit>MTU*3) {
reallimit=reallimit-512;
}
long queueing=RTTC-RTTMin;
QueueingAvg=(QueueingAvg*99999+queueing)/100000;
if(firstUpdate.compareAndSet(true, false)) {
RTTAvg=RTTC;
RTTVar=RTTC/2;
}else {
RTTVar=(RTTVar*3+Math.abs(RTTAvg-RTTC))/4;
RTTAvg= (RTTAvg*7+RTTC)/8;
}
RTO=RTTAvg+Math.max(MIN_RTTVAR, RTTVar*4);//RTTVar*4
/*if(RTTC>RTO) {
congressFactor=Math.min( 0.9,congressFactor);
}else {
if(reallimit<outputchachesize)
reallimit+=512;
}*/
if(congressFactor<1.1)
congressFactor+=0.001;
} */
//System.out.println(congressFactor);
//spdlmt.setLimitspeed(1024*1024);
//reallimit=Math.max(MTU*2,(int) (congressSpeed*RTTMin*4/1000000000L));
//System.out.println(reallimit/MTU);
/*long nspd=(long) (congressSpeed*congressFactor);
spdlmt.setLimitspeed(Math.max(nspd,MIN_LIMIT_SPEED));*/
// System.out.println("RwqSpeed:"+(requestSpeed/1024)+"K MaxSpeed:"+(congressSpeed/1024)+"K LimitSpeed:"+(spdlmt.getLimitspeed()/1024)+"K");
//System.out.println("RTTMin:"+RTTMin/1000000L+"ms RTTAvg:"+RTTAvg/1000000L+"ms");
}
kl.dispose();
}
}
ackt.dispose();
break;
@@ -1302,7 +1608,7 @@ public void associateSocketChannel(SocketChannel b) throws IOException{
//System.out.println("快速重传："+st);
st.setPriority(4);
controller.sendPacketToAddress(remoteaddr,st);
controller.sendPacketToAddress(remoteaddr,0,st);
}
break;
default: