diff --git a/src/org/kne/cloud/network/klalb/Consts.java b/src/org/kne/cloud/network/klalb/Consts.java index 9a25c5a..88cd51e 100644 --- a/src/org/kne/cloud/network/klalb/Consts.java +++ b/src/org/kne/cloud/network/klalb/Consts.java @@ -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; } diff --git a/src/org/kne/cloud/network/klalb/IOThreadManager.java b/src/org/kne/cloud/network/klalb/IOThreadManager.java index b665e54..2eb1d0f 100644 --- a/src/org/kne/cloud/network/klalb/IOThreadManager.java +++ b/src/org/kne/cloud/network/klalb/IOThreadManager.java @@ -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; + } + } diff --git a/src/org/kne/cloud/network/klalb/KLALBBlock.java b/src/org/kne/cloud/network/klalb/KLALBBlock.java index 9aba7ac..ef2ec9c 100644 --- a/src/org/kne/cloud/network/klalb/KLALBBlock.java +++ b/src/org/kne/cloud/network/klalb/KLALBBlock.java @@ -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{ 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: @@ -55,5 +62,18 @@ public class KLALBBlock { } return sb.toString(); } + + + + @Override + public int compareTo(KLALBBlock o) { + if(o.number>number) { + return -1; + }else if(o.number localtcps = new Vector<>(); + private Map localtcps = new ConcurrentHashMap<>(); private List remotetcps = new Vector<>(); private PredicateacceptSYN; @@ -80,7 +82,7 @@ public class KLALBCore { this.acceptSYN = acceptSYN; } - public List getLocaltcps() { + public Map 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,12 +169,64 @@ 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) { - + public void submitDataBlock(KLALBBlock kb) throws InterruptedException { + while(true) { + synchronized (remotetcps) { + Collections.sort(remotetcps, new Comparator() { + @Override + public int compare(RemoteTCPConnection o1, RemoteTCPConnection o2) { + if(o1.getDelay()>o2.getDelay()) { + return 1; + }else if(o1.getDelay() iterator = remotetcps.iterator(); iterator.hasNext();) { + RemoteTCPConnection remoteTCPConnection = (RemoteTCPConnection) iterator.next(); + BlockingDeque bdq=remoteTCPConnection.getSendDeque(); + if(bdq.size()<1) { + bdq.add(kb); + return; + } + } + } + Thread.sleep(1); + } + } public void submitDataBlockNoDelay(KLALBBlock kb) { synchronized (remotetcps) { @@ -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 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()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); + } + } } diff --git a/src/org/kne/cloud/network/klalb/KLALBServer.java b/src/org/kne/cloud/network/klalb/KLALBServer.java index 8f2d9a1..686d29c 100644 --- a/src/org/kne/cloud/network/klalb/KLALBServer.java +++ b/src/org/kne/cloud/network/klalb/KLALBServer.java @@ -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(); diff --git a/src/org/kne/cloud/network/klalb/LocalTCPConnection.java b/src/org/kne/cloud/network/klalb/LocalTCPConnection.java index 7dd49c2..25e90d7 100644 --- a/src/org/kne/cloud/network/klalb/LocalTCPConnection.java +++ b/src/org/kne/cloud/network/klalb/LocalTCPConnection.java @@ -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 inputcache = new TreeSet<>(); +private List 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 getInputcache() { + return inputcache; + } + public List getOutputcache() { + return outputcache; + } + public long getInputcount() { + return inputcount; + } + public long getOutputcount() { + return outputcount; + } + public void unpackBlock(KLALBBlock data) throws IOException { + if(data.number 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 BlockingDequesendDeque=new LinkedBlockingDeque<>(); + public BlockingDeque getSendDeque() { + return sendDeque; + } + private volatile int pcu=0; + public void setPeerCacheUsed(int cacheused) { + pcu= cacheused; + } + public int getPeerCacheUsed() { + return pcu; + } } diff --git a/src/org/kne/cloud/network/klalb/RemoteTCPConnection.java b/src/org/kne/cloud/network/klalb/RemoteTCPConnection.java index 6a041f1..f19a000 100644 --- a/src/org/kne/cloud/network/klalb/RemoteTCPConnection.java +++ b/src/org/kne/cloud/network/klalb/RemoteTCPConnection.java @@ -10,46 +10,55 @@ import java.util.concurrent.LinkedBlockingDeque; 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; } - + 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) { - dout.writeInt(data.data.length); - dout.write(data.data); - }else 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) { 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); + 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) { + 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(); + klb.pingtime = din.readLong(); break; case 2: - klb.lservice=din.readUTF(); - klb.lipport=new IPPort(din.readUTF()); + klb.lservice = din.readUTF(); + klb.lipport = new IPPort(din.readUTF()); } + } else { + klb.cacheused = din.readInt(); } - System.out.println("RECEIVE:"+klb); + 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 BlockingDequesendDeque=new LinkedBlockingDeque<>(); + + private BlockingDeque sendDeque = new LinkedBlockingDeque<>(); + public BlockingDeque getSendDeque() { return sendDeque; } diff --git a/src/org/kne/cloud/network/klalb/Tunnel.java b/src/org/kne/cloud/network/klalb/Tunnel.java index 0003d64..452ff41 100644 --- a/src/org/kne/cloud/network/klalb/Tunnel.java +++ b/src/org/kne/cloud/network/klalb/Tunnel.java @@ -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 entry = (Entry) 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); diff --git a/src/org/kne/cloud/network/klalb/UUID1.java b/src/org/kne/cloud/network/klalb/UUID1.java new file mode 100644 index 0000000..411397a --- /dev/null +++ b/src/org/kne/cloud/network/klalb/UUID1.java @@ -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; + } +}