diff --git a/src/org/kne/cloud/network/klalb/Consts.java b/src/org/kne/cloud/network/klalb/Consts.java index 066a209..c941e0f 100644 --- a/src/org/kne/cloud/network/klalb/Consts.java +++ b/src/org/kne/cloud/network/klalb/Consts.java @@ -1,13 +1,14 @@ package org.kne.cloud.network.klalb; public class Consts { - public static final int BLOCKSIZE=16384; + //public static final int BLOCKSIZE=8192; public static final long PINGTIMENS=1000000000L; - public static final int SO_TIMEOUT = 5000; + public static final int SO_TIMEOUT = 2000; public static final long SN_KEEP = 60000000000L; - public static final int MAX_RESEND = 10; - public static final long UACK_TIME = 5000000000L; + public static final int MAX_RESEND = 20; + public static final long UACK_TIME = 2000000000L; public static final int CONNECT_TIMEOUT=10000; public static final double MAX_QUEUE_TIME=500; public static final long KEEP_TIME = 60000000000L; + public static final int PRE_SO_TIMEOUT = 10000; } diff --git a/src/org/kne/cloud/network/klalb/KLALBBlock.java b/src/org/kne/cloud/network/klalb/KLALBBlock.java index 5cc04ad..7ddcdc0 100644 --- a/src/org/kne/cloud/network/klalb/KLALBBlock.java +++ b/src/org/kne/cloud/network/klalb/KLALBBlock.java @@ -8,15 +8,15 @@ import java.util.concurrent.atomic.AtomicLong; import org.kne.cloud.network.mport.IPPort; public class KLALBBlock implements Comparable{ - private static AtomicLong sng=new AtomicLong(1); + //private static AtomicLong sng=new AtomicLong(1); - public volatile long sn;//每个数据包的唯一编号 + //public volatile long sn;//每个数据包的唯一编号 public volatile int cuid;//用于识别数据包的stream ID号 public volatile long number;//数据包的编号,用于排序 public volatile byte[]data;//数据内容 - public volatile int command;//功能标记 0=PING 1=PONG 2=SYN 3=RST 4=KEEP + public volatile int command;//功能标记 0=PING 1=PONG 2=SYN 3=RST 4=KEEP 5=SYK public volatile long pingtime;//PING计时器 - public volatile int cacheused; + public volatile boolean cacheused; public volatile String lservice; public volatile IPPort lipport; @@ -24,12 +24,12 @@ public class KLALBBlock implements Comparable{ public transient volatile long sendtime; public transient volatile int resend; - public volatile long sendtimeForRTT; + //public volatile long sendtimeForRTT; - public KLALBBlock() { + /*public KLALBBlock() { sn=sng.getAndIncrement(); - } + }*/ @@ -60,6 +60,9 @@ public class KLALBBlock implements Comparable{ case 3: sb.append("RST"); break; + case 5: + sb.append("SYK"); + break; } } return sb.toString(); diff --git a/src/org/kne/cloud/network/klalb/KLALBClient.java b/src/org/kne/cloud/network/klalb/KLALBClient.java index 9fff01f..a767cb5 100644 --- a/src/org/kne/cloud/network/klalb/KLALBClient.java +++ b/src/org/kne/cloud/network/klalb/KLALBClient.java @@ -36,7 +36,7 @@ import com.google.gson.stream.JsonReader; public class KLALBClient { private UUID suid=UUID.randomUUID(); - private IOThreadManager iom=new IOThreadManager(); + private KLALBController iom=new KLALBController(); private List tls=new ArrayList<>(); private List tcpl=new ArrayList<>(); private List pet=new ArrayList<>(); @@ -114,7 +114,7 @@ public class KLALBClient { } try { //System.out.println(re); - Thread.sleep(1000*(1< { try { while (true) { @@ -114,15 +108,11 @@ public class IOThreadManager { } catch (IOException|InterruptedException e) { e.printStackTrace(); AB.set(false); - ls.interrupt(); + /*ls.interrupt(); lt.interrupt(); - tc.getOutputcache().clear(); + tc.getOutputcache().clear();*/ tc.close(); - KLALBBlock sbk = new KLALBBlock(); - sbk.cuid = tc.getCuid(); - sbk.number = 0; - sbk.command = 3; - klc.submitDataBlockNoDelay(sbk); + klc.closeLocal(tc); } finally { try { tc.getDin().close(); @@ -136,19 +126,14 @@ public class IOThreadManager { //System.out.println("xxxxxxxxxxxxxxxxxxxxxxxxxx"); //System.exit(123); AB.set(false); - ls.interrupt(); + /*ls.interrupt(); lt.interrupt(); - tc.getOutputcache().clear(); + tc.getOutputcache().clear();*/ tc.close(); + klc.closeLocal(tc); }); if (syn) { - KLALBBlock sbk = new KLALBBlock(); - sbk.cuid = tc.getCuid(); - sbk.number = 0; - sbk.command = 2; - sbk.lservice = tc.getServiceElement().name; - sbk.lipport = tc.getServiceElement().ipport; - klc.submitAckBlockNoDelay(sbk,1); + klc.connectLocal(tc); } ls.start(); lr.start(); @@ -161,8 +146,12 @@ public class IOThreadManager { } catch (InterruptedException e) { e.printStackTrace(); + } catch (IOException e) { + // TODO 自动生成的 catch 块 + e.printStackTrace(); } finally { - klc.getLocaltcps().remove(tc.getCuid()); + klc.closeLocal(tc); + //klc.getLocaltcps().remove(tc.getCuid()); } } @@ -170,7 +159,7 @@ public class IOThreadManager { if(!isopen) { return; } - klc.getRemotetcps().add(tc); + klc.openRemote(tc); try { Thread rs = ThreadTool.makeVThreadIfSupport("远程发送线程", () -> { @@ -200,19 +189,8 @@ public class IOThreadManager { } catch (InterruptedException e) { e.printStackTrace(); } finally { - klc.getRemotetcps().remove(tc); - Queue sq=tc.getSendDeque(); - KLALBBlock tmp; - while((tmp=sq.poll())!=null) { - try { - klc.submitDataBlock(tmp); - } catch (InterruptedException e) { - e.printStackTrace(); - } catch (IOException e) { - // TODO 自动生成的 catch 块 - e.printStackTrace(); - } - } + klc.closeRemote(tc); + // checkRemotes(); } @@ -260,4 +238,5 @@ public class IOThreadManager { return klc; } + } diff --git a/src/org/kne/cloud/network/klalb/KLALBCore.java b/src/org/kne/cloud/network/klalb/KLALBCore.java index 83da240..f135e41 100644 --- a/src/org/kne/cloud/network/klalb/KLALBCore.java +++ b/src/org/kne/cloud/network/klalb/KLALBCore.java @@ -6,6 +6,8 @@ import java.io.IOException; import java.io.InputStream; import java.io.OutputStream; import java.io.StreamCorruptedException; +import java.lang.invoke.ConstantCallSite; +import java.net.ConnectException; import java.util.ArrayList; import java.util.Collections; import java.util.Comparator; @@ -33,21 +35,25 @@ import java.util.concurrent.Executors; import java.util.concurrent.LinkedBlockingQueue; import java.util.concurrent.atomic.AtomicBoolean; import java.util.concurrent.atomic.AtomicInteger; +import java.util.function.BiConsumer; +import java.util.function.Consumer; import java.util.function.Predicate; import java.util.zip.DataFormatException; import org.kne.cloud.network.mport.ThreadTool; public class KLALBCore { - private volatile long RTT=1000000000; + /*private volatile long RTT=1000000000; private volatile long RTO=1000000000; private volatile long DevRTT=0; - - private volatile int cacheblocks; + */ + //private volatile int cacheblocks; private Map localtcps = new ConcurrentHashMap<>(); + private Setlocaltcpsconnecting=Collections.synchronizedSet(new HashSet<>()); + private List remotetcps = new Vector<>(); - private PredicateacceptSYN; + private BiConsumer>acceptSYN; /*private volatile long rsntime=System.nanoTime(); @@ -81,11 +87,11 @@ public class KLALBCore { }*/ - public Predicate getAcceptSYN() { + public BiConsumer> getAcceptSYN() { return acceptSYN; } - public void setAcceptSYN(Predicate acceptSYN) { + public void setAcceptSYN(BiConsumer> acceptSYN) { this.acceptSYN = acceptSYN; } @@ -97,9 +103,6 @@ public class KLALBCore { return remotetcps; } - public KLALBCore(int cachesize) { - cacheblocks=cachesize; - } public void remoteSend(RemoteTCPConnection tc) throws InterruptedException, IOException { for(;;) { @@ -112,13 +115,12 @@ public class KLALBCore { tc.sendBlock(pdb); tc.updateTraffics(); } - if(!tc.getNDSendDeque().isEmpty()||!tc.getSendDeque().isEmpty()) + if(!tc.isEmpty()) break; - synchronized (tc.getSendDeque()) { - tc.getSendDeque().wait(10); - } + tc.lock(10); } - KLALBBlock k0=tc.getNDSendDeque().poll(); + KLALBBlock kbv=tc.getNextBlock(); + /*KLALBBlock k0=tc.getNDSendDeque().poll(); if(k0!=null) { if(k0.cuid==0&&k0.number==0&&k0.command==1) { k0.pingtime+=System.nanoTime()- k0.sendtime; @@ -132,6 +134,12 @@ public class KLALBCore { tc.sendBlock(k); } + }*/ + if(kbv!=null) { + if(kbv.cuid==0&&kbv.number==0&&kbv.command==1) { + kbv.pingtime+=System.nanoTime()- kbv.sendtime; + } + tc.sendBlock(kbv); } } public void remoteReceive(RemoteTCPConnection tc) throws IOException { @@ -181,11 +189,13 @@ public class KLALBCore { pdb.command=1; pdb.pingtime=brc.pingtime; pdb.sendtime=brc.sendtime; - tc.getNDSendDeque().offer(pdb); + tc.addBlock(pdb, 0); + //tc.getNDSendDeque().offer(pdb); }else if(brc.command==1) { long cur=System.nanoTime(); long delay=cur-brc.pingtime; setDelayAvg(tc,delay); + tc.setSoTimeout(Consts.SO_TIMEOUT); } } }else { @@ -193,23 +203,53 @@ public class KLALBCore { if(brc.command==0) { }else if(brc.command==1) { }else if(brc.command==2) { - if(!localtcps.containsKey(brc.cuid)) { - - if(acceptSYN==null||!acceptSYN.test(brc)) { + if(!localtcps.containsKey(brc.cuid)&&!localtcpsconnecting.contains(brc.cuid)) { + localtcpsconnecting.add(brc.cuid); + KLALBBlock brcx=brc; + ThreadTool.makeVThreadIfSupport("局域网连接监视线程", ()->{ + try { + if(acceptSYN==null) { KLALBBlock rst=new KLALBBlock(); - rst.cuid=brc.cuid; + rst.cuid=brcx.cuid; rst.number=0; rst.command=3; submitDataBlockNoDelay(rst); + }else {acceptSYN.accept(brcx,(Consumer)(c)->{ + if(c) { + KLALBBlock syk=new KLALBBlock(); + syk.cuid=brcx.cuid; + syk.number=0; + syk.command=5; + submitDataBlockNoDelay(syk); + }else { + KLALBBlock rst=new KLALBBlock(); + rst.cuid=brcx.cuid; + rst.number=0; + rst.command=3; + submitDataBlockNoDelay(rst); + } + }); + } + }finally { + localtcpsconnecting.remove(brcx.cuid); + } + }).start(); } }else if(brc.command==3) { LocalTCPConnection ltc=localtcps.get(brc.cuid); if(ltc!=null) { //ltc.close(); + ltc.setIsrst(true); ltc.getRSTHook().accept(ltc); } + }else if(brc.command==5) { + LocalTCPConnection ltc=localtcps.get(brc.cuid); + + if(ltc!=null) { + ltc.setIssyk(true); + } } }else if(brc.number>0){ LocalTCPConnection ltcs=localtcps.get(brc.cuid); @@ -217,9 +257,41 @@ public class KLALBCore { KLALBBlock ack=new KLALBBlock(); ack.cuid=brc.cuid; ack.number=-brc.number; - ack.cacheused=ltcs.getSendDeque().size(); + ack.cacheused=ltcs.getSendDeque().size()>ltcs.getInputcachesize()/ltcs.getBlocksize(); submitAckBlockNoDelay(ack,3); - ltcs.getSendDeque().add(brc); + + List ic=ltcs.getInputcache(); + synchronized(ic) { + if(brc.number>=ltcs.getInputcount()) { + ic.add(brc); + while(true) { + KLALBBlock kkb=null; + for (int i = 0; i < ic.size(); i++) { + KLALBBlock klalbBlock = ic.get(i); + if(klalbBlock.number==ltcs.getInputcount()) { + ic.remove(i); + i--; + kkb=klalbBlock; + break; + } + } + if(kkb==null) + break; + ltcs.getSendDeque().add(kkb); + ltcs.incInputcount(); + } + /*for (Iterator iterator = ltcs.getInputcache().iterator(); iterator.hasNext();) { + KLALBBlock klalbBlock = (KLALBBlock) iterator.next(); + if(klalbBlock.number==ltcs.getInputcount()) { + iterator.remove(); + ltcs.getSendDeque().add(klalbBlock); + ltcs.incInputcount(); + }else { + break; + } + }*/ + } + } } @@ -227,7 +299,7 @@ public class KLALBCore { long nx=-brc.number; LocalTCPConnection ltc=localtcps.get(brc.cuid); if(ltc!=null) { - List l=ltc.getOutputcache(); + /*List l=ltc.getOutputcache(); synchronized (l) { for (Iterator iterator = l.iterator(); iterator.hasNext();) { KLALBBlock klalbBlock = (KLALBBlock) iterator.next(); @@ -240,13 +312,13 @@ public class KLALBCore { //System.out.println(RTO/1000000); } } - } + }*/ - /*ltc.getOutputcache().removeIf((b) -> { + ltc.getOutputcache().removeIf((b) -> { return b.number == nx; - });*/ - ltc.setPeerCacheUsed(brc.cacheused); + }); + ltc.setPeerFull(brc.cacheused); }else { KLALBBlock sbk = new KLALBBlock(); sbk.cuid = brc.cuid; @@ -258,19 +330,18 @@ public class KLALBCore { synchronized (remotetcps) { for (Iterator iterator = remotetcps.iterator(); iterator.hasNext();) { RemoteTCPConnection remoteTCPConnection = (RemoteTCPConnection) iterator.next(); - Queuesq=remoteTCPConnection.getSendDeque(); - sq.removeIf((vb)->{ - return vb.number==nx&&vb.cuid==cuid; - }); + remoteTCPConnection.ackRemove(nx,cuid); + /*Queuesq=remoteTCPConnection.getSendDeque(); + */ } } } } } - private void sortRemoteTCPByPrediction() { + private void sortRemoteTCPByPrediction(int prio) { remotetcps.forEach((x)->{ - x.predictLatency(); + x.predictLatency(prio); }); Collections.sort(remotetcps, new Comparator() { @Override @@ -324,20 +395,19 @@ public class KLALBCore { submitDataBlock(ks,1); } public void submitDataBlock(KLALBBlock kb,int i) throws InterruptedException, IOException { + submitDataBlock(kb,1,5); + } + public void submitDataBlock(KLALBBlock kb,int i,int prio) throws InterruptedException, IOException { int ni=Math.min(i, remotetcps.size()); while(true) { if(remotetcps.isEmpty()) throw new IOException("发送错误"); synchronized (remotetcps) { - sortRemoteTCPByPrediction(); + sortRemoteTCPByPrediction(5); for (Iterator iterator = remotetcps.iterator(); iterator.hasNext();) { - RemoteTCPConnection remoteTCPConnection = (RemoteTCPConnection) iterator.next(); + RemoteTCPConnection rmt = (RemoteTCPConnection) iterator.next(); //if(remoteTCPConnection.getPredictedLatency() bdq=remoteTCPConnection.getSendDeque(); - bdq.add(kb); - synchronized (bdq) { - bdq.notifyAll(); - } + rmt.addBlock(kb, prio); ni--; if(ni<=0) return; @@ -352,11 +422,7 @@ public class KLALBCore { synchronized (remotetcps) { for (Iterator iterator = remotetcps.iterator(); iterator.hasNext();) { RemoteTCPConnection rmt = (RemoteTCPConnection) iterator.next(); - Queue sq = rmt.getNDSendDeque(); - sq.offer(kb); - synchronized (sq) { - sq.notifyAll(); - } + rmt.addBlock(kb, 0); } } } @@ -369,33 +435,39 @@ public class KLALBCore { return; } RemoteTCPConnection rmt = (RemoteTCPConnection) iterator.next(); - Queue sq = rmt.getNDSendDeque(); - sq.offer(kb); - synchronized (sq) { - sq.notifyAll(); - } + rmt.addBlock(kb, 0); coun++; } } } public void putDataBlock(LocalTCPConnection tc, KLALBBlock ks) throws InterruptedException, IOException { - while (true) { + /*while (true) { + if(!localtcps.containsKey(tc.getCuid())) + throw new IOException("本地连接已关闭"); if(remotetcps.isEmpty()) throw new IOException("发送错误"); boolean b; synchronized (tc.getOutputcache()) { - b=!tc.getOutputcache().isEmpty()&&(tc.getOutputcount()-tc.getOutputcache().get(0).number>cacheblocks); + b=!tc.getOutputcache().isEmpty()&&(tc.getOutputcount()-tc.getOutputcache().get(0).number>tc.getOutputcachesize()/tc.getBlocksize()); } if(!b) { break; } Thread.sleep(1); - } - /*while(tc.getOutputcache().size()>cacheblocks) { - Thread.sleep(1); }*/ + if(!localtcps.containsKey(tc.getCuid())) + throw new IOException("本地连接已关闭"); + if(remotetcps.isEmpty()) + throw new IOException("发送错误"); + while(tc.getOutputcache().size()>tc.getOutputcachesize()/tc.getBlocksize()) { + if(!localtcps.containsKey(tc.getCuid())) + throw new IOException("本地连接已关闭"); + if(remotetcps.isEmpty()) + throw new IOException("发送错误"); + Thread.sleep(1); + } ks.sendtime=System.nanoTime(); - ks.sendtimeForRTT=ks.sendtime; + // ks.sendtimeForRTT=ks.sendtime; //System.out.println(tc.getOutputcache().size()); /*if(ks.data==null) { submitDataBlock(ks); @@ -408,7 +480,9 @@ public class KLALBCore { }*/ submitDataBlock(ks); tc.getOutputcache().add(ks); - while( tc.getPeerCacheUsed()>2*cacheblocks) { + while( tc.isPeerFull()) { + if(!localtcps.containsKey(tc.getCuid())) + throw new IOException("本地连接已关闭"); if(remotetcps.isEmpty()) throw new IOException("发送错误"); Thread.sleep(1); @@ -435,7 +509,7 @@ public class KLALBCore { } }*/ if (timex > (1<Consts.UACK_TIME) { + if(curr-tc.uackt>Consts.UACK_TIME) { tc.uackt=curr; KLALBBlock ack=new KLALBBlock(); ack.cuid=tc.getCuid(); ack.number=Long.MIN_VALUE; - ack.cacheused=tc.getSendDeque().size(); + ack.cacheused=tc.getSendDeque().size()>tc.getInputcachesize()/tc.getBlocksize(); submitAckBlockNoDelay(ack,3); } /*if(tc.getSendDeque().size()Consts.KEEP_TIME) { @@ -470,8 +544,10 @@ public class KLALBCore { public KLALBBlock getDataBlock(LocalTCPConnection tc) throws InterruptedException, IOException { KLALBBlock recv=null; while(true) { + if(!localtcps.containsKey(tc.getCuid())) + throw new IOException("本地连接已关闭"); if(remotetcps.isEmpty()) - throw new IOException("发送错误"); + throw new IOException("接收错误"); recv=tc.getSendDeque().poll(); if(recv!=null) break; @@ -482,9 +558,76 @@ public class KLALBCore { public void waitOutput(LocalTCPConnection tc) throws InterruptedException, IOException { while(tc.getOutputcache().size()>0) { + if(!localtcps.containsKey(tc.getCuid())) + throw new IOException("本地连接已关闭"); if(remotetcps.isEmpty()) throw new IOException("发送错误"); Thread.sleep(10); } } + + public void closeLocal(LocalTCPConnection tc) { + tc.getOutputcache().clear(); + KLALBBlock sbk = new KLALBBlock(); + sbk.cuid = tc.getCuid(); + sbk.number = 0; + sbk.command = 3; + submitDataBlockNoDelay(sbk); + localtcps.remove(tc.getCuid()); + } + + public void openLocal(LocalTCPConnection tc) { + localtcps.put(tc.getCuid(), tc); + } + + public void connectLocal(LocalTCPConnection tc) throws InterruptedException, IOException { + + KLALBBlock sbk = new KLALBBlock(); + sbk.cuid = tc.getCuid(); + sbk.number = 0; + sbk.command = 2; + sbk.lservice = tc.getServiceElement().name; + sbk.lipport = tc.getServiceElement().ipport; + submitDataBlockNoDelay(sbk); + + while(!tc.isIssyk()) { + if(!localtcps.containsKey(tc.getCuid())) + throw new IOException("本地连接已关闭"); + if(tc.isIsrst()) { + throw new ConnectException("连接被拒绝"); + } + if(remotetcps.isEmpty()) + throw new IOException("接收错误"); + Thread.sleep(2); + } + } + + public void openRemote(RemoteTCPConnection tc) { + remotetcps.add(tc); + } + + public void closeRemote(RemoteTCPConnection tc) { + remotetcps.remove(tc); + List> l=tc.getSendDequeList(); + synchronized (l) { + for (int i = 0; i < l.size(); i++) { + Queue q=l.get(i); + int ix=i-1; + if(ix<0) { + ix=0; + } + int i2=ix; + q.forEach((kv)->{ + try { + submitDataBlock(kv,1,i2); + } catch (InterruptedException e) { + e.printStackTrace(); + } catch (IOException e) { + e.printStackTrace(); + } + }); + } + } + } + } diff --git a/src/org/kne/cloud/network/klalb/KLALBServer.java b/src/org/kne/cloud/network/klalb/KLALBServer.java index 5d46989..514d488 100644 --- a/src/org/kne/cloud/network/klalb/KLALBServer.java +++ b/src/org/kne/cloud/network/klalb/KLALBServer.java @@ -21,12 +21,14 @@ import org.kne.cloud.network.mport.ServiceElement; import org.kne.cloud.network.mport.ThreadTool; public class KLALBServer { - - Map whm=Collections.synchronizedMap( new WeakHashMap()); + Map whm=new ConcurrentHashMap<>(); + //Map whm=Collections.synchronizedMap( new WeakHashMap()); public KLALBServer(int port,Supplier gjso) throws IOException { TCPListener tcpl=new TCPListener(port); tcpl.setCon((s)->{ RemoteTCPConnection tcc=null; + KLALBController nx=null; + UUID uid=null; try { //s.setSoTimeout(10000); tcc=new RemoteTCPConnection(null,s); @@ -46,37 +48,50 @@ public class KLALBServer { Tunnel tll=new Tunnel(name,new IPPort( ip, portx)); tcc.setTunnel(tll); - UUID uid=new UUID(din.readLong(),din.readLong()); + uid=new UUID(din.readLong(),din.readLong()); tcc.handshake(10); - IOThreadManager nx=whm.computeIfAbsent(uid, (kuid)->{ - IOThreadManager iom=new IOThreadManager(); - iom.getCore().setAcceptSYN((b)->{ + + synchronized (whm) { + + nx=whm.computeIfAbsent(uid, (kuid)->{ + KLALBController iom=new KLALBController(); + iom.getCore().setAcceptSYN((b,bol)->{ try { STJson stj=new STJson(gjso.get()); if(!stj.checkSafety(b.lipport)) { System.out.println("未授权的请求:"+b.lipport); - return false; + bol.accept(false); + return; } - Socket soc=new Socket(b.lipport.getIp(),b.lipport.getPort()); - LocalTCPConnection ltc=new LocalTCPConnection(soc, b.cuid); - ThreadTool.makeVThreadIfSupport("局域网连接监视线程", ()->{ + + + LocalTCPConnection ltc=null; try { + Socket soc=new Socket(b.lipport.getIp(),b.lipport.getPort()); + ltc=new LocalTCPConnection(soc, b.cuid); + bol.accept(true); iom.handleLocal(ltc, false); + return; + } catch (IOException e) { + e.printStackTrace(); }finally { + if(ltc!=null) ltc.close(); } - }).start(); - return true; + bol.accept(false); + return ; } catch (IOException e) { System.out.println("连接本地服务"+b.lipport+"失败,请检查你的本地服务"); e.printStackTrace(); } - return false; + bol.accept(false); + return; }); return iom; }); + } /*if(whm.containsKey(uid)) { nx=whm.get(uid); }else { @@ -113,6 +128,11 @@ public class KLALBServer { }finally { if(tcc!=null) tcc.close(); + if(nx.getCore().getRemotetcps().isEmpty()) { + nx.close(); + if(uid!=null) + whm.remove(uid); + } } }); tcpl.open(); diff --git a/src/org/kne/cloud/network/klalb/LocalTCPConnection.java b/src/org/kne/cloud/network/klalb/LocalTCPConnection.java index c94b746..077058f 100644 --- a/src/org/kne/cloud/network/klalb/LocalTCPConnection.java +++ b/src/org/kne/cloud/network/klalb/LocalTCPConnection.java @@ -18,9 +18,14 @@ private int cuid; private ServiceElement serviceElement; -private Set inputcache = new TreeSet<>(); +//private Set inputcache = Collections.synchronizedSet( new TreeSet<>()); +private Listinputcache=new Vector<>(); private List outputcache = new Vector<>(); +private volatile int blocksize=8192; +private volatile int inputcachesize=100*blocksize; +private volatile int outputcachesize=100*blocksize; + private volatile long inputcount = 1; private volatile long outputcount = 1; public LocalTCPConnection(Socket s) throws IOException { @@ -33,6 +38,24 @@ private volatile long outputcount = 1; s.setKeepAlive(true); cuid=uid; } + public int getBlocksize() { + return blocksize; + } + public void setBlocksize(int blocksize) { + this.blocksize = blocksize; + } + public int getInputcachesize() { + return inputcachesize; + } + public void setInputcachesize(int inputcachesize) { + this.inputcachesize = inputcachesize; + } + public int getOutputcachesize() { + return outputcachesize; + } + public void setOutputcachesize(int outputcachesize) { + this.outputcachesize = outputcachesize; + } public int getCuid() { return cuid; } @@ -47,7 +70,7 @@ private volatile long outputcount = 1; } - public Set getInputcache() { + public List getInputcache() { return inputcache; } public List getOutputcache() { @@ -60,7 +83,7 @@ private volatile long outputcount = 1; return outputcount; } public void unpackBlock(KLALBBlock data) throws IOException { - if(data.number getSendDeque() { return sendDeque; } - private volatile int pcu=0; + //private volatile int pcu=0; private volatile Consumer hook; public volatile long uackt=System.nanoTime(); public volatile long ukeept=System.nanoTime(); - public void setPeerCacheUsed(int cacheused) { + /*public void setPeerCacheUsed(int cacheused) { pcu= cacheused; } public int getPeerCacheUsed() { return pcu; - } + }*/ public Consumer getRSTHook() { return hook; } public void setRSTHook(Consumer tc) { this.hook=tc; } + private volatile boolean pfl=false; + public void setPeerFull(boolean b) { + pfl=b; + } + public boolean isPeerFull() { + return pfl; + } + /*public int getSendDequeBytes() { + AtomicInteger a=new AtomicInteger(0); + getSendDeque().forEach((v)->{ + if(v.data!=null) + a.addAndGet(v.data.length); + }); + return a.get(); + }*/ + public void incInputcount() { + inputcount++; + } + + private volatile boolean issyk=false; + public boolean isIssyk() { + return issyk; + } + public void setIssyk(boolean issyk) { + this.issyk = issyk; + } + private volatile boolean isrst=false; + public boolean isIsrst() { + return isrst; + } + public void setIsrst(boolean isrst) { + this.isrst = isrst; + } } diff --git a/src/org/kne/cloud/network/klalb/RemoteTCPConnection.java b/src/org/kne/cloud/network/klalb/RemoteTCPConnection.java index 05d8195..b4e2899 100644 --- a/src/org/kne/cloud/network/klalb/RemoteTCPConnection.java +++ b/src/org/kne/cloud/network/klalb/RemoteTCPConnection.java @@ -10,12 +10,17 @@ import java.io.InputStream; import java.io.OutputStream; import java.io.StreamCorruptedException; import java.net.Socket; +import java.net.SocketException; import java.net.UnknownHostException; +import java.util.ArrayList; import java.util.Arrays; import java.util.Comparator; +import java.util.Iterator; +import java.util.List; import java.util.Queue; import java.util.Random; import java.util.UUID; +import java.util.Vector; import java.util.concurrent.ArrayBlockingQueue; import java.util.concurrent.BlockingDeque; import java.util.concurrent.BlockingQueue; @@ -124,13 +129,18 @@ public class RemoteTCPConnection extends TCPConnection { public double getPredictedLatency() { return latency; } - public void predictLatency() { + public void predictLatency(int prio) { AtomicLong data=new AtomicLong(0); - sendDeque.forEach((e)->{ + for (int i = 0; i <=prio; i++) { + Queue sendDequev=sendDequeList.get(i); + + sendDequev.forEach((e)->{ + data.addAndGet(12); if(e.data!=null) { data.addAndGet(e.data.length); } }); + } latency= delay/2000000.0+data.get()*1000.0/upsm; } @@ -143,7 +153,7 @@ public class RemoteTCPConnection extends TCPConnection { super(s); tunnel=t; initIOs(); - s.setSoTimeout(Consts.SO_TIMEOUT); + s.setSoTimeout(Consts.PRE_SO_TIMEOUT); } private long delay = Long.MAX_VALUE; @@ -209,11 +219,11 @@ public class RemoteTCPConnection extends TCPConnection { } } else if (data.number == 0) { dout.write(data.command); - if(data.command!=0&&data.command!=1) { + /*if(data.command!=0&&data.command!=1) { dout.writeLong(data.sn); }else { - } + }*/ switch (data.command) { case 0: case 1: @@ -227,7 +237,7 @@ public class RemoteTCPConnection extends TCPConnection { break; } } else { - dout.writeInt(data.cacheused); + dout.writeBoolean(data.cacheused); } if(isflush) @@ -262,11 +272,11 @@ public class RemoteTCPConnection extends TCPConnection { } } else if (klb.number == 0) { klb.command = din.read(); - if(klb.command!=0&&klb.command!=1) { + /*if(klb.command!=0&&klb.command!=1) { klb.sn = din.readLong(); }else { klb.sn=0; - } + }*/ switch (klb.command) { case 0: case 1: @@ -280,7 +290,7 @@ public class RemoteTCPConnection extends TCPConnection { break; } } else { - klb.cacheused = din.readInt(); + klb.cacheused = din.readBoolean(); } System.out.println("RECEIVE:" + klb); @@ -299,30 +309,36 @@ public class RemoteTCPConnection extends TCPConnection { } } - private Queue NDsendDeque = new ConcurrentLinkedQueue(); + + private List> sendDequeList =new Vector>(); + { + for(int i=0;i<11;i++) { + sendDequeList.add(new ConcurrentLinkedQueue()); + } + } + + public List> getSendDequeList() { + return sendDequeList; + } + private Object lock=new Object(); + public void addBlock(KLALBBlock blk,int prio) { + sendDequeList.get(prio).add(blk); + synchronized (lock) { + lock.notifyAll(); + } + } +/* private Queue NDsendDeque = new ConcurrentLinkedQueue(); public Queue getNDSendDeque() { return NDsendDeque; } - //private Queue sendDeque = new ConcurrentLinkedQueue(); - private Queue sendDeque = new PriorityBlockingQueue(2, new Comparator() { - - @Override - public int compare(KLALBBlock o1, KLALBBlock o2) { - if(o1.resend>o2.resend) { - return -1; - }else if(o1.resend sendDeque = new ConcurrentLinkedQueue(); public Queue getSendDeque() { return sendDeque; - } + }*/ + public void handshake(int x) throws IOException { for(int i=0;i> iterator = sendDequeList.iterator(); iterator.hasNext();) { + Queue queue = iterator.next(); + queue.removeIf((vb)->{ + return vb.number==number&&vb.cuid==cuid; + }); + } + } + + public boolean isEmpty() { + synchronized (sendDequeList) { + for (Iterator> iterator = sendDequeList.iterator(); iterator.hasNext();) { + Queue queue = (Queue) iterator.next(); + if(!queue.isEmpty()) { + return false; + } + } + } + return true; + } + + public KLALBBlock getNextBlock() { + synchronized (sendDequeList) { + for (Iterator> iterator = sendDequeList.iterator(); iterator.hasNext();) { + Queue queue = (Queue) iterator.next(); + KLALBBlock v=queue.poll(); + if(v!=null) + return v; + } + } + return null; + } + + public void setSoTimeout(int soTimeout) { + try { + connect.setSoTimeout(soTimeout); + } catch (SocketException e) { + //e.printStackTrace(); + } + } + } diff --git a/src/org/kne/cloud/network/klalb/TCPConnection.java b/src/org/kne/cloud/network/klalb/TCPConnection.java index 7e92a4f..d3a2ce0 100644 --- a/src/org/kne/cloud/network/klalb/TCPConnection.java +++ b/src/org/kne/cloud/network/klalb/TCPConnection.java @@ -63,7 +63,7 @@ public class TCPConnection { protected void initIO() throws IOException { dout = new DataOutputStream( connect.getOutputStream()); - din = new DataInputStream(connect.getInputStream()); + din = new DataInputStream( connect.getInputStream()); } public boolean isOpen() { diff --git a/src/org/kne/cloud/network/nathole/TCPNatHoleTeat.java b/src/org/kne/cloud/network/nathole/TCPNatHoleTeat.java new file mode 100644 index 0000000..c17e876 --- /dev/null +++ b/src/org/kne/cloud/network/nathole/TCPNatHoleTeat.java @@ -0,0 +1,67 @@ +package org.kne.cloud.network.nathole; + +import java.io.IOException; +import java.net.InetAddress; +import java.net.InetSocketAddress; +import java.net.ServerSocket; +import java.net.Socket; +import java.net.SocketTimeoutException; +import java.net.UnknownHostException; + +public class TCPNatHoleTeat { + public static void main(String[] args) throws IOException { + + x("183.198.152.240", 12187); + + } + + private static void x(String nat, int port) throws IOException { + for (int i = 10000; i < 20000; i++) { + System.out.println("try" + i); + Socket s = null; + ServerSocket srs = null; + try { + s = new Socket(); + s.bind(new InetSocketAddress(InetAddress.getByName("0.0.0.0"), i)); + try { + s.connect(new InetSocketAddress(InetAddress.getByName("1.1.1.1"), 443), 1); + }catch(SocketTimeoutException e) { + + } + s.close(); + + srs = new ServerSocket(); + srs.setReuseAddress(true); + srs.bind(new InetSocketAddress("0.0.0.0", i)); + ServerSocket srs2 = srs; + new Thread(() -> { + try { + while (true) { + Socket sa = srs2.accept(); + System.out.println(sa.toString()); + //sa.close(); + // System.exit(0); + } + } catch (IOException e) { + // e.printStackTrace(); + } + }).start(); + + /*Socket ste = new Socket(); + try { + ste.connect(new InetSocketAddress(nat, port), 1); + } catch (SocketTimeoutException e) { + + } + ste.close();*/ + } catch (IOException e) { + e.printStackTrace(); + } finally { + if (s != null) + s.close(); + /*if (srs != null) + srs.close();*/ + } + } + } +}