diff --git a/src/org/kne/cloud/network/klalb/KLALBBlock.java b/src/org/kne/cloud/network/klalb/KLALBBlock.java index 522939e..0a8e56c 100644 --- a/src/org/kne/cloud/network/klalb/KLALBBlock.java +++ b/src/org/kne/cloud/network/klalb/KLALBBlock.java @@ -3,13 +3,13 @@ package org.kne.cloud.network.klalb; import java.util.Arrays; import java.util.Objects; -public class KLALBBlock implements Comparable{ +public class KLALBBlock { volatile byte[]data; volatile int size; volatile long number; - volatile long time=-1; + //volatile long time=-1; volatile Thread thread; public KLALBBlock(byte[] b, int size,long number) { data=b; @@ -42,16 +42,6 @@ public class KLALBBlock implements Comparable{ KLALBBlock other = (KLALBBlock) obj; return number == other.number; } - @Override - public int compareTo(KLALBBlock o) { - if(time>o.time) { - return 1; - }else if(time inputcache = Collections.synchronizedSet(new HashSet<>()); private List outputcache = Collections.synchronizedList(new ArrayList<>()); - private BlockingQueue kack = new LinkedBlockingQueue<>(); private volatile boolean inlocal=true; @@ -91,37 +90,32 @@ public class KLALBCore { public void sendDataBlock(TCPConnection out) throws IOException, InterruptedException { if (close) throw new InterruptedException(); - while (kack.isEmpty() && outputcache.isEmpty()) { + while ( outputcache.isEmpty()) { if (close) Thread.currentThread().interrupt(); Thread.sleep(1); } - KLALBBlock kx = kack.poll(); - if (kx != null) { - kx.time+=System.nanoTime(); - send0(out, kx); - } else { KLALBBlock ks = null; synchronized (outputcache) { for (int i = 0; i < outputcache.size(); i++) { KLALBBlock kd = outputcache.get(i); - if (kd.time == -1) { - kd.time = System.nanoTime(); + if (kd.thread == null) { + //kd.time = System.nanoTime(); kd.thread = Thread.currentThread(); ks = kd; } else { if (kd.thread.isAlive()) { - long timex = (System.nanoTime() - kd.time) / 1000000; + /*long timex = (System.nanoTime() - kd.time) / 1000000; if (timex > 10000) { System.out.println("超时重传："+kd); kd.time = System.nanoTime(); kd.thread = Thread.currentThread(); ks = kd; - } + }*/ } else { System.out.println("掉线重传："+kd); - kd.time = System.nanoTime(); + //kd.time = System.nanoTime(); kd.thread = Thread.currentThread(); ks = kd; } @@ -134,7 +128,6 @@ public class KLALBCore { if (ks != null) { send0(out, ks); // System.out.println(outputcache.size()); - } } } @@ -144,14 +137,17 @@ public class KLALBCore { throw new InterruptedException(); KLALBBlock x = receive0(in); if (x.number > 0) { - try { + KLALBBlock klb=new KLALBBlock(null, 0, -x.number); - klb.time=x.time-System.nanoTime(); - kack.put(klb); - } catch (InterruptedException e) { - e.printStackTrace(); - } - if (x.number >= inputcount) { + ThreadTool.makeVThreadIfSupport("ACK", ()->{ + try { + send0(in, klb); + } catch (IOException e) { + e.printStackTrace(); + } + }).start(); + + if (x.number >= inputcount) { while(inputcache.size()>(2*cacheblocks)&&inlocal) { if (close) @@ -169,19 +165,18 @@ public class KLALBCore { }); } } - + private void send0(TCPConnection tcp, KLALBBlock kd) throws IOException { DataOutputStream out=tcp.getDout(); synchronized (out) { out.writeLong(kd.number); if (kd.number > 0) { - out.writeLong(System.nanoTime()); out.writeInt(kd.size); out.write(kd.data, 0, kd.size); - }else { - out.writeLong(kd.time); } - out.flush(); + + + tcp.flush(); } System.out.println("SEND:" + kd); } @@ -191,17 +186,10 @@ public class KLALBCore { DataInputStream in=tcp.getDin(); synchronized (in) { kb.number = in.readLong(); - kb.time = in.readLong(); - - - if (kb.number > 0) { kb.size = in.readInt(); kb.data = new byte[kb.size]; in.readFully(kb.data); - }else{ - long dela=(System.nanoTime()-kb.time)/2000000; - //System.out.println(tcp.getTunnel()+" "+dela); } } System.out.println("RECEIVE:" + kb); diff --git a/src/org/kne/cloud/network/klalb/KLALBServer.java b/src/org/kne/cloud/network/klalb/KLALBServer.java index 5319d0e..5b58ae1 100644 --- a/src/org/kne/cloud/network/klalb/KLALBServer.java +++ b/src/org/kne/cloud/network/klalb/KLALBServer.java @@ -27,7 +27,7 @@ public class KLALBServer { nx=whm.get(uid); }else { nx=new KLALBServerProtocol(); - Socket soc=new Socket("192.168.1.233",8444); + Socket soc=new Socket("192.168.1.233",3389); nx.setOut(soc.getOutputStream()); nx.setIn(soc.getInputStream()); nx.startLocal(); diff --git a/src/org/kne/cloud/network/klalb/TCPConnection.java b/src/org/kne/cloud/network/klalb/TCPConnection.java index 986d757..c1f18b7 100644 --- a/src/org/kne/cloud/network/klalb/TCPConnection.java +++ b/src/org/kne/cloud/network/klalb/TCPConnection.java @@ -12,61 +12,94 @@ public class TCPConnection { private Tunnel tunnel; private Socket connect; -private DataInputStream din; -public Tunnel getTunnel() { - return tunnel; -} -private DataOutputStream dout; -private long delay; -public Socket getConnect() { - return connect; -} -public DataInputStream getDin() { - return din; -} -public DataOutputStream getDout() { - return dout; -} -public void close() { - // TODO 自动生成的方法存根 - try { - if(din!=null) - din.close(); - } catch (IOException e1) { - // TODO 自动生成的 catch 块 - e1.printStackTrace(); + private DataInputStream din; + + public Tunnel getTunnel() { + return tunnel; } - try { - if(dout!=null) - dout.close(); - } catch (IOException e1) { - // TODO 自动生成的 catch 块 - e1.printStackTrace(); + + private DataOutputStream dout; + private long delay; + + public Socket getConnect() { + return connect; } - try { - if(connect!=null) - connect.close(); - } catch (IOException e) { - // TODO 自动生成的 catch 块 - e.printStackTrace(); + + public DataInputStream getDin() { + return din; + } + + public DataOutputStream getDout() { + return dout; + } + + public void close() { + // TODO 自动生成的方法存根 + try { + if (din != null) + din.close(); + } catch (IOException e1) { + // TODO 自动生成的 catch 块 + e1.printStackTrace(); + } + try { + if (dout != null) + dout.close(); + } catch (IOException e1) { + // TODO 自动生成的 catch 块 + e1.printStackTrace(); + } + try { + if (connect != null) + connect.close(); + } catch (IOException e) { + // TODO 自动生成的 catch 块 + e.printStackTrace(); + } + connect = null; + } + + public long getDelay() { + return delay; + } + + public void setDelay(long delay) { + this.delay = delay; + } + + public TCPConnection(Tunnel t) throws UnknownHostException, IOException { + this(t, t.connectClientSocket()); + } + + public TCPConnection(Tunnel t, Socket soc) throws IOException { + connect = soc; + tunnel = t; + // connect.setSoTimeout(10000); + din = new DataInputStream(new BufferedInputStream(connect.getInputStream(), 65536)); + dout = new DataOutputStream(new BufferedOutputStream(connect.getOutputStream(), 65536)); + ThreadTool.makeVThreadIfSupport("FLUSH", () -> { + try { + while (true) { + Thread.sleep(100); + if(flush) { + synchronized (dout) { + dout.flush(); + + } + flush=false; + } + } + } catch (IOException e) { + e.printStackTrace(); + } catch (InterruptedException e) { + e.printStackTrace(); + } + }).start(); + } + + private volatile boolean flush = false; + + public void flush() throws IOException { + flush = true; } - connect=null; -} -public long getDelay() { - return delay; -} -public void setDelay(long delay) { - this.delay = delay; -} -public TCPConnection(Tunnel t) throws UnknownHostException, IOException { - this(t,t.connectClientSocket()); -} -public TCPConnection(Tunnel t,Socket soc) throws IOException { - connect=soc; - tunnel=t; - //connect.setSoTimeout(10000); - din=new DataInputStream(new BufferedInputStream( connect.getInputStream(),65536)); - dout=new DataOutputStream(new BufferedOutputStream (connect.getOutputStream(),65536)); - -} }