增加流量控制

增加自动按延迟选择线路
This commit is contained in:
Administrator
2022-12-14 17:27:04 +08:00
parent 5d7324db0e
commit e93a5815b4
10 changed files with 431 additions and 71 deletions
+3 -1
View File
@@ -1,9 +1,11 @@
package org.kne.cloud.network.klalb;
public class Consts {
public static final int BLOCKSIZE=32768;
public static final int BLOCKSIZE=65536;
public static final long PINGTIMENS=10000000000L;
public static final double A = 0.125;
public static final int SO_TIMEOUT = 20000;
public static final long SN_KEEP = 60000000000L;
public static final int MAX_RESEND = 20;
public static final long UACK_TIME = 10000000000L;
}
@@ -15,35 +15,109 @@ import org.kne.cloud.network.mport.ServiceElement;
import org.kne.cloud.network.mport.ThreadTool;
public class IOThreadManager {
private KLALBCore klc = new KLALBCore(5000);
private KLALBCore klc = new KLALBCore(2000);
private boolean isopen = true;
public void handleLocal(LocalTCPConnection tc, ServiceElement se,boolean syn) {
klc.getLocaltcps().add(tc);
public IOThreadManager() {
}
public void handleLocal(LocalTCPConnection tc, boolean syn) {
klc.getLocaltcps().put(tc.getCuid(), tc);
try {
if(syn) {
KLALBBlock sbk=new KLALBBlock();
sbk.cuid=tc.getCuid();
sbk.number=0;
sbk.command=2;
sbk.lservice=se.proc;
sbk.lipport=se.ipport;
if (syn) {
KLALBBlock sbk = new KLALBBlock();
sbk.cuid = tc.getCuid();
sbk.number = 0;
sbk.command = 2;
sbk.lservice = tc.getServiceElement().proc;
sbk.lipport = tc.getServiceElement().ipport;
klc.submitDataBlockNoDelay(sbk);
}
Thread lt = ThreadTool.makeVThreadIfSupport("数据包计时重发线程", () -> {
try {
while (true) {
klc.outputTimer(tc);
}
} catch (InterruptedException e) {
}
});
Thread ls = ThreadTool.makeVThreadIfSupport("本地发送线程", () -> {
try {
while (true) {
KLALBBlock rd=klc.getDataBlock(tc);
if(rd.data==null) {
break;
}
tc.unpackBlock(rd);
}
} catch (InterruptedException e) {
} catch (IOException e) {
lt.interrupt();
tc.getOutputcache().clear();
KLALBBlock sbk = new KLALBBlock();
sbk.cuid = tc.getCuid();
sbk.number = 0;
sbk.command = 3;
klc.submitDataBlockNoDelay(sbk);
e.printStackTrace();
}finally {
try {
tc.getDout().close();
} catch (IOException e) {
e.printStackTrace();
}
}
});
Thread lr = ThreadTool.makeVThreadIfSupport("本地接收线程", () -> {
try {
while (true) {
KLALBBlock ks = tc.packBlock();
klc.putDataBlock(tc, ks);
if (ks.data == null) {
break;
}
}
} catch (IOException e) {
ls.interrupt();
lt.interrupt();
tc.getOutputcache().clear();
KLALBBlock sbk = new KLALBBlock();
sbk.cuid = tc.getCuid();
sbk.number = 0;
sbk.command = 3;
klc.submitDataBlockNoDelay(sbk);
e.printStackTrace();
} catch (InterruptedException e) {
} finally {
try {
tc.getDin().close();
} catch (IOException e) {
e.printStackTrace();
}
}
});
ls.start();
lr.start();
ls.join();
lt.start();
lr.join();
ls.interrupt();
klc.waitOutput(tc);
lt.interrupt();
} catch (InterruptedException e) {
e.printStackTrace();
} finally {
klc.getLocaltcps().remove(tc);
klc.getLocaltcps().remove(tc.getCuid());
}
}
@@ -57,7 +131,6 @@ public class IOThreadManager {
klc.remoteSend(tc);
}
} catch (InterruptedException e) {
e.printStackTrace();
} catch (IOException e) {
e.printStackTrace();
}
@@ -67,14 +140,15 @@ public class IOThreadManager {
while (true) {
klc.remoteReceive(tc);
}
} catch (IOException e) {
} catch (Exception e) {
rs.interrupt();
e.printStackTrace();
}
});
rs.start();
rr.start();
rs.join();
rr.join();
rs.interrupt();
} catch (InterruptedException e) {
e.printStackTrace();
} finally {
@@ -90,4 +164,8 @@ public class IOThreadManager {
return isopen;
}
public KLALBCore getCore() {
return klc;
}
}
@@ -7,7 +7,7 @@ import java.util.concurrent.atomic.AtomicLong;
import org.kne.cloud.network.mport.IPPort;
public class KLALBBlock {
public class KLALBBlock implements Comparable<KLALBBlock>{
private static AtomicLong sng=new AtomicLong(0);
public long sn;//每个数据包的唯一编号
@@ -16,10 +16,13 @@ public class KLALBBlock {
public byte[]data;//数据内容
public int command;//功能标记 0=PING 1=PONG 2=SYN 3=RST
public long pingtime;//PING计时器
public int cacheused;
public String lservice;
public IPPort lipport;
public transient volatile long sendtime;
public transient volatile int resend;
public KLALBBlock() {
@@ -34,9 +37,13 @@ public class KLALBBlock {
sb.append(cuid);
sb.append(' ');
if(number>0) {
if(data==null) {
sb.append("DATA").append(number).append(':').append("EOF");
}else {
sb.append("DATA").append(number).append(':').append(data.length);
}
}else if(number<0) {
sb.append("ACK").append(-number);
sb.append("ACK").append(-number).append(':').append(cacheused);
}else {
switch(command) {
case 0:
@@ -56,4 +63,17 @@ public class KLALBBlock {
return sb.toString();
}
@Override
public int compareTo(KLALBBlock o) {
if(o.number>number) {
return -1;
}else if(o.number<number){
return 1;
}else {
return 0;
}
}
}
@@ -2,11 +2,13 @@ package org.kne.cloud.network.klalb;
import java.io.BufferedInputStream;
import java.io.BufferedOutputStream;
import java.io.EOFException;
import java.io.File;
import java.io.IOException;
import java.net.ConnectException;
import java.net.MalformedURLException;
import java.net.Socket;
import java.net.SocketTimeoutException;
import java.net.URI;
import java.net.URL;
import java.net.UnknownHostException;
@@ -37,8 +39,9 @@ public class KLALBClient {
LocalTCPConnection tcc = null;
try {
tcc = new LocalTCPConnection(s);
tcc.setServiceElement(sel);
System.out.println("TCP"+tcc.getCuid()+"已连接");
iom.handleLocal(tcc,sel,true);
iom.handleLocal(tcc,true);
System.out.println("TCP"+tcc.getCuid()+"已关闭");
} catch (IOException e) {
e.printStackTrace();
@@ -64,12 +67,17 @@ public class KLALBClient {
tc.getDout().writeLong(suid.getMostSignificantBits());
tc.getDout().writeLong(suid.getLeastSignificantBits());
tc.getDout().flush();
int v=tc.getDin().read();
if(v==-1) {
throw new EOFException();
}
System.out.println(tll+":连接成功");
iom.handleRemote(tc);
System.out.println(tll+":连接断开");
} catch (UnknownHostException e) {
e.printStackTrace();
}catch(ConnectException e) {
}catch(ConnectException|SocketTimeoutException e) {
}catch(EOFException e) {
} catch (IOException e) {
e.printStackTrace();
}finally {
+115 -4
View File
@@ -8,6 +8,7 @@ import java.io.OutputStream;
import java.io.StreamCorruptedException;
import java.util.ArrayList;
import java.util.Collections;
import java.util.Comparator;
import java.util.HashSet;
import java.util.Iterator;
import java.util.LinkedList;
@@ -15,12 +16,14 @@ import java.util.List;
import java.util.Map;
import java.util.Objects;
import java.util.Queue;
import java.util.Random;
import java.util.Set;
import java.util.TreeMap;
import java.util.UUID;
import java.util.Vector;
import java.util.WeakHashMap;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingDeque;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ExecutorService;
@@ -35,8 +38,7 @@ public class KLALBCore {
private static final UUID ZERO_UUID=new UUID(0,0);
private volatile int cacheblocks;
private List<LocalTCPConnection> localtcps = new Vector<>();
private Map<UUID,LocalTCPConnection> localtcps = new ConcurrentHashMap<>();
private List<RemoteTCPConnection> remotetcps = new Vector<>();
private Predicate<KLALBBlock>acceptSYN;
@@ -80,7 +82,7 @@ public class KLALBCore {
this.acceptSYN = acceptSYN;
}
public List<LocalTCPConnection> getLocaltcps() {
public Map<UUID, LocalTCPConnection> getLocaltcps() {
return localtcps;
}
@@ -108,6 +110,7 @@ public class KLALBCore {
}
KLALBBlock k=tc.getSendDeque().poll();
if(k!=null) {
//if(localtcps.containsKey(k.cuid)||k.cuid.equals(ZERO_UUID))
tc.sendBlock(k);
}
}
@@ -115,6 +118,9 @@ public class KLALBCore {
KLALBBlock brc=null;
for(;;) {
brc=tc.receiveBlock();
if(brc.number!=0) {
break;
}
SN s=new SN(brc.sn);
synchronized (rsns) {
if(!rsns.contains(s)) {
@@ -163,11 +169,63 @@ public class KLALBCore {
submitDataBlockNoDelay(rst);
}
}else if(brc.command==3) {
LocalTCPConnection ltc=localtcps.get(brc.cuid);
if(ltc!=null) {
ltc.close();
}
}
}else if(brc.number>0){
LocalTCPConnection ltcs=localtcps.get(brc.cuid);
if(ltcs!=null) {
ltcs.getSendDeque().add(brc);
KLALBBlock ack=new KLALBBlock();
ack.cuid=brc.cuid;
ack.number=-brc.number;
ack.cacheused=ltcs.getSendDeque().size();
submitDataBlockNoDelay(ack);
}
}else {
long nx=-brc.number;
LocalTCPConnection ltc=localtcps.get(brc.cuid);
if(ltc!=null) {
ltc.getOutputcache().removeIf((b) -> {
return b.number == nx;
});
ltc.setPeerCacheUsed(brc.cacheused);
}
}
}
}
public void submitDataBlock(KLALBBlock kb) throws InterruptedException {
while(true) {
synchronized (remotetcps) {
Collections.sort(remotetcps, new Comparator<RemoteTCPConnection>() {
@Override
public int compare(RemoteTCPConnection o1, RemoteTCPConnection o2) {
if(o1.getDelay()>o2.getDelay()) {
return 1;
}else if(o1.getDelay()<o2.getDelay()){
return -1;
}else {
return 0;
}
}
});
for (Iterator<RemoteTCPConnection> iterator = remotetcps.iterator(); iterator.hasNext();) {
RemoteTCPConnection remoteTCPConnection = (RemoteTCPConnection) iterator.next();
BlockingDeque<KLALBBlock> bdq=remoteTCPConnection.getSendDeque();
if(bdq.size()<1) {
bdq.add(kb);
return;
}
}
}
Thread.sleep(1);
}
public void submitDataBlock(KLALBBlock kb) {
}
public void submitDataBlockNoDelay(KLALBBlock kb) {
@@ -178,4 +236,57 @@ public class KLALBCore {
}
}
}
public void putDataBlock(LocalTCPConnection tc, KLALBBlock ks) throws InterruptedException {
while (!tc.getOutputcache().isEmpty()&&(tc.getOutputcount()-tc.getOutputcache().get(0).number>cacheblocks)) {
Thread.sleep(1);
}
submitDataBlock(ks);
ks.sendtime=System.nanoTime();
tc.getOutputcache().add(ks);
while(tc.getPeerCacheUsed()>cacheblocks) {
Thread.sleep(1);
}
}
private volatile long uackt=System.nanoTime();
public void outputTimer(LocalTCPConnection tc) throws InterruptedException {
List<KLALBBlock> l=tc.getOutputcache();
synchronized( l) {
for (int i = 0; i < l.size(); i++) {
KLALBBlock block=l.get(i);
long timex = (System.nanoTime() - block.sendtime) / 1000000;
if (timex > 200+1000*i) {
submitDataBlock(block);
block.sendtime = System.nanoTime();
block.resend++;
if(block.resend>=Consts.MAX_RESEND) {
l.remove(i);
i--;
}
System.out.println("³¬Ê±ÖØ´«£º"+block);
}
}
}
long curr=System.nanoTime();
if(tc.getSendDeque().size()<cacheblocks&&curr-uackt>Consts.UACK_TIME) {
uackt=curr;
KLALBBlock ack=new KLALBBlock();
ack.cuid=tc.getCuid();
ack.number=Long.MIN_VALUE;
ack.cacheused=tc.getSendDeque().size();
submitDataBlockNoDelay(ack);
}
Thread.sleep(10);
}
public KLALBBlock getDataBlock(LocalTCPConnection tc) throws InterruptedException {
return tc.getSendDeque().take();
}
public void waitOutput(LocalTCPConnection tc) throws InterruptedException {
while(tc.getOutputcache().size()>0) {
Thread.sleep(10);
}
}
}
@@ -16,6 +16,7 @@ import java.util.WeakHashMap;
import org.kne.cloud.network.mport.IPPort;
import org.kne.cloud.network.mport.ServiceElement;
import org.kne.cloud.network.mport.ThreadTool;
public class KLALBServer {
@@ -66,9 +67,30 @@ public class KLALBServer {
nx=whm.get(uid);
}else {
nx=new IOThreadManager();
IOThreadManager n1=nx;
nx.getCore().setAcceptSYN((b)->{
try {
Socket soc=new Socket(b.lipport.getIp(),b.lipport.getPort());
LocalTCPConnection ltc=new LocalTCPConnection(soc, b.cuid);
ThreadTool.makeVThreadIfSupport("局域网连接监视线程", ()->{
try {
n1.handleLocal(ltc, false);
}finally {
n1.close();
}
}).start();
return true;
} catch (IOException e) {
System.out.println("连接本地服务"+b.lipport+"失败,请检查你的本地服务");
e.printStackTrace();
}
return false;
});
whm.put(uid, nx);
}
}
tcc.getDout().write(0);
tcc.getDout().flush();
nx.handleRemote(tcc);
}catch(IOException e) {
e.printStackTrace();
@@ -2,11 +2,22 @@ package org.kne.cloud.network.klalb;
import java.io.IOException;
import java.net.Socket;
import java.net.UnknownHostException;
import java.util.UUID;
import java.util.*;
import java.util.concurrent.BlockingDeque;
import java.util.concurrent.LinkedBlockingDeque;
import org.kne.cloud.network.mport.ServiceElement;
public class LocalTCPConnection extends TCPConnection {
private UUID cuid;
private ServiceElement serviceElement;
private Set<KLALBBlock> inputcache = new TreeSet<>();
private List<KLALBBlock> outputcache = new Vector<>();
private volatile long inputcount = 1;
private volatile long outputcount = 1;
public LocalTCPConnection(Socket s) throws IOException {
super(s);
cuid=UUID.randomUUID();
@@ -21,5 +32,68 @@ private UUID cuid;
public void setCuid(UUID cuid) {
this.cuid = cuid;
}
public ServiceElement getServiceElement() {
return serviceElement;
}
public void setServiceElement(ServiceElement serviceElement) {
this.serviceElement = serviceElement;
}
public Set<KLALBBlock> getInputcache() {
return inputcache;
}
public List<KLALBBlock> getOutputcache() {
return outputcache;
}
public long getInputcount() {
return inputcount;
}
public long getOutputcount() {
return outputcount;
}
public void unpackBlock(KLALBBlock data) throws IOException {
if(data.number<inputcount) {
return;
}
inputcache.add(data);
for (Iterator<KLALBBlock> iterator = inputcache.iterator(); iterator.hasNext();) {
KLALBBlock klalbBlock = (KLALBBlock) iterator.next();
if(klalbBlock.number==inputcount) {
iterator.remove();
getDout().write(klalbBlock.data);
getDout().flush();
inputcount++;
}else {
break;
}
}
}
public KLALBBlock packBlock() throws IOException {
byte[]dat=new byte[Consts.BLOCKSIZE];
int len=getDin().read(dat);
KLALBBlock pb=new KLALBBlock();
pb.cuid=cuid;
pb.number=outputcount++;
if(len==-1) {
pb.data=null;
}else if(len==dat.length){
pb.data=dat;
}else {
pb.data=Arrays.copyOf(dat, len);
}
return pb;
}
private BlockingDeque<KLALBBlock>sendDeque=new LinkedBlockingDeque<>();
public BlockingDeque<KLALBBlock> getSendDeque() {
return sendDeque;
}
private volatile int pcu=0;
public void setPeerCacheUsed(int cacheused) {
pcu= cacheused;
}
public int getPeerCacheUsed() {
return pcu;
}
}
@@ -12,15 +12,18 @@ import org.kne.cloud.network.mport.IPPort;
public class RemoteTCPConnection extends TCPConnection {
public RemoteTCPConnection(Tunnel t) throws UnknownHostException, IOException {
this(t,t.connectClientSocket());
this(t, t.connectClientSocket());
}
public RemoteTCPConnection(Tunnel t,Socket s) throws UnknownHostException, IOException {
public RemoteTCPConnection(Tunnel t, Socket s) throws UnknownHostException, IOException {
super(s);
this.tunnel=t;
this.tunnel = t;
s.setSoTimeout(Consts.SO_TIMEOUT);
}
private long delay=-1;
private long delay = Long.MAX_VALUE;
private Tunnel tunnel;
public Tunnel getTunnel() {
return tunnel;
}
@@ -28,28 +31,34 @@ public class RemoteTCPConnection extends TCPConnection {
public void setTunnel(Tunnel tunnel) {
this.tunnel = tunnel;
}
public long getDelay() {
return delay;
}
public void setDelay(long delay) {
this.delay = delay;
if(tunnel!=null) {
if (tunnel != null) {
tunnel.setDelay(delay);
}
}
public void sendBlock(KLALBBlock data) throws IOException {
dout.writeLong(data.sn);
dout.writeLong(data.cuid.getMostSignificantBits());
dout.writeLong(data.cuid.getLeastSignificantBits());
dout.writeLong(data.number);
if(data.number>0) {
if (data.number > 0) {
if (data.data == null) {
dout.writeInt(-1);
} else {
dout.writeInt(data.data.length);
dout.write(data.data);
}else if(data.number==0) {
}
} else if (data.number == 0) {
dout.write(data.command);
switch(data.command) {
switch (data.command) {
case 0:
case 1:
dout.writeLong(data.pingtime);
@@ -59,51 +68,62 @@ public class RemoteTCPConnection extends TCPConnection {
dout.writeUTF(data.lipport.toString());
break;
}
} else {
dout.writeInt(data.cacheused);
}
dout.flush();
System.out.println("SEND:"+data);
}
public KLALBBlock receiveBlock() throws IOException {
KLALBBlock klb=new KLALBBlock();
klb.sn=din.readLong();
klb.cuid=new UUID(din.readLong(), din.readLong());
klb.number=din.readLong();
if(klb.number>0) {
int size=din.readInt();
byte[]d=new byte[size];
din.readFully(d);
klb.data=d;
}else if(klb.number==0) {
klb.command=din.read();
switch(klb.command) {
case 0:
case 1:
klb.pingtime=din.readLong();
break;
case 2:
klb.lservice=din.readUTF();
klb.lipport=new IPPort(din.readUTF());
}
System.out.println("SEND:" + data);
}
System.out.println("RECEIVE:"+klb);
public KLALBBlock receiveBlock() throws IOException {
KLALBBlock klb = new KLALBBlock();
klb.sn = din.readLong();
klb.cuid = new UUID(din.readLong(), din.readLong());
klb.number = din.readLong();
if (klb.number > 0) {
int size = din.readInt();
if (size == -1) {
klb.data = null;
} else {
byte[] d = new byte[size];
din.readFully(d);
klb.data = d;
}
} else if (klb.number == 0) {
klb.command = din.read();
switch (klb.command) {
case 0:
case 1:
klb.pingtime = din.readLong();
break;
case 2:
klb.lservice = din.readUTF();
klb.lipport = new IPPort(din.readUTF());
}
} else {
klb.cacheused = din.readInt();
}
System.out.println("RECEIVE:" + klb);
return klb;
}
private volatile long time=System.nanoTime();
private volatile long time = System.nanoTime();
public boolean checkPingTime() {
long cu=System.nanoTime();
if(cu-time>Consts.PINGTIMENS) {
time=cu;
long cu = System.nanoTime();
if (cu - time > Consts.PINGTIMENS) {
time = cu;
return true;
}else {
} else {
return false;
}
}
private BlockingDeque<KLALBBlock>sendDeque=new LinkedBlockingDeque<>();
private BlockingDeque<KLALBBlock> sendDeque = new LinkedBlockingDeque<>();
public BlockingDeque<KLALBBlock> getSendDeque() {
return sendDeque;
}
+2 -2
View File
@@ -24,7 +24,7 @@ public class Tunnel {
public IPPort getIpport() {
return ipport;
}
private long delay=-1;
private long delay=Long.MAX_VALUE;
public long getDelay() {
return delay;
}
@@ -100,7 +100,7 @@ public class Tunnel {
for (Iterator iterator = s.iterator(); iterator.hasNext();) {
Entry<IPPort, Tunnel> entry = (Entry<IPPort, Tunnel>) iterator.next();
Tunnel tll=entry.getValue();
System.out.println(tll.getName()+" "+((tll.getDelay()==-1)?"δ֪":tll.getDelay()/1000000+"ms")+" Á¬½ÓÊý"+tll.ati);
System.out.println(tll.getName()+" "+((tll.getDelay()==Long.MAX_VALUE)?"δ֪":tll.getDelay()/1000000+"ms")+" Á¬½ÓÊý"+tll.ati);
}
}
private AtomicInteger ati=new AtomicInteger(0);
@@ -0,0 +1,25 @@
package org.kne.cloud.network.klalb;
import java.time.Duration;
import java.time.LocalDateTime;
import java.util.Random;
public class UUID1 {
private static long get64LeastSignificantBitsForVersion1() {
Random random = new Random();
long random63BitLong = random.nextLong() & 0x3FFFFFFFFFFFFFFFL;
long variant3BitFlag = 0x8000000000000000L;
return random63BitLong + variant3BitFlag;
}
private static long get64MostSignificantBitsForVersion1() {
LocalDateTime start = LocalDateTime.of(1582, 10, 15, 0, 0, 0);
Duration duration = Duration.between(start, LocalDateTime.now());
long seconds = duration.getSeconds();
long nanos = duration.getNano();
long timeForUuidIn100Nanos = seconds * 10000000 + nanos * 100;
long least12SignificatBitOfTime = (timeForUuidIn100Nanos & 0x000000000000FFFFL) >> 4;
long version = 1 << 12;
return (timeForUuidIn100Nanos & 0xFFFFFFFFFFFF0000L) + version + least12SignificatBitOfTime;
}
}