KNEMC/KLALB
1
2
Fork 1
Code Issues Pull Requests Actions Packages Projects Releases 1 Wiki Activity

正式版V1.0

Browse Source
This commit is contained in:
Administrator
2023-01-07 07:04:09 +08:00
parent ea97e3891f
commit a9a4ad7c21
7 changed files with 65 additions and 60 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/org/kne/cloud/network/klalb/Consts.java
+1 -1
View File
@@ -3,7 +3,7 @@ package org.kne.cloud.network.klalb;
public class Consts {
public static final int BLOCKSIZE=8192;
public static final long PINGTIMENS=1000000000L;
public static final int SO_TIMEOUT = 20000;
public static final int SO_TIMEOUT = 10000;
public static final long SN_KEEP = 60000000000L;
public static final int MAX_RESEND = 100;
public static final long UACK_TIME = 10000000000L;
src/org/kne/cloud/network/klalb/IOThreadManager.java
+1 -1
View File
@@ -22,7 +22,7 @@ import org.kne.cloud.network.mport.ServiceElement;
import org.kne.cloud.network.mport.ThreadTool;
public class IOThreadManager {
private KLALBCore klc = new KLALBCore(2000);
private KLALBCore klc = new KLALBCore(5000);
private volatile boolean isopen = true;
public IOThreadManager() {
src/org/kne/cloud/network/klalb/KLALBClientGUI.java
+1 -1
View File
@@ -247,7 +247,7 @@ public class KLALBClientGUI extends XFrame {
}
});
panel_1.add(btnNewButton_1, BorderLayout.EAST);
setTitle("KNE云网络负载均衡客户端");
setTitle("KNE云网络负载均衡客户端V1.0");
setSize(600, 370);
setLocationRelativeTo(null);
src/org/kne/cloud/network/klalb/KLALBCore.java
+32 -32
View File
@@ -109,7 +109,7 @@ public class KLALBCore {
pdb.pingtime=System.nanoTime();
tc.sendBlock(pdb);
}
if(!tc.getNDSendDeque().isEmpty()||!tc.getSendDeque().isEmpty()||!tc.getReSendDeque().isEmpty())
if(!tc.getNDSendDeque().isEmpty()||!tc.getSendDeque().isEmpty())
break;
synchronized (tc.getSendDeque()) {
tc.getSendDeque().wait(10);
@@ -122,16 +122,13 @@ public class KLALBCore {
}
tc.sendBlock(k0);
}else {
KLALBBlock k1=tc.getReSendDeque().poll();
if(k1!=null) {
tc.sendBlock(k1);
}else {
KLALBBlock k=tc.getSendDeque().poll();
if(k!=null) {
//if(localtcps.containsKey(k.cuid)||k.cuid.equals(ZERO_UUID))
tc.sendBlock(k);
}
}
}
}
public void remoteReceive(RemoteTCPConnection tc) throws IOException {
@@ -224,7 +221,7 @@ public class KLALBCore {
ack.cuid=brc.cuid;
ack.number=-brc.number;
ack.cacheused=ltcs.getSendDeque().size();
submitAckBlockNoDelay(ack,3);
submitAckBlockNoDelay(ack,5);
}
@@ -246,6 +243,16 @@ public class KLALBCore {
}
}
}
synchronized (remotetcps) {
for (Iterator<RemoteTCPConnection> iterator = remotetcps.iterator(); iterator.hasNext();) {
RemoteTCPConnection remoteTCPConnection = (RemoteTCPConnection) iterator.next();
Queue<KLALBBlock>sq=remoteTCPConnection.getSendDeque();
sq.removeIf((vb)->{
return vb.number==nx;
});
}
}
/*ltc.getOutputcache().removeIf((b) -> {
return b.number == nx;
});*/
@@ -265,7 +272,7 @@ public class KLALBCore {
for (Iterator<RemoteTCPConnection> iterator = remotetcps.iterator(); iterator.hasNext();) {
RemoteTCPConnection remoteTCPConnection = (RemoteTCPConnection) iterator.next();
Queue<KLALBBlock> bdq=remoteTCPConnection.getSendDeque();
if(bdq.size()<1) {
if(bdq.size()<2) {
bdq.add(kb);
synchronized (bdq) {
bdq.notifyAll();
@@ -279,25 +286,6 @@ public class KLALBCore {
Thread.sleep(1);
}
}
public void submitReDataBlock(KLALBBlock kb) throws InterruptedException {
while(true) {
synchronized (remotetcps) {
for (Iterator<RemoteTCPConnection> iterator = remotetcps.iterator(); iterator.hasNext();) {
RemoteTCPConnection remoteTCPConnection = (RemoteTCPConnection) iterator.next();
Queue<KLALBBlock> bdq=remoteTCPConnection.getReSendDeque();
if(bdq.size()<1) {
bdq.add(kb);
synchronized (bdq) {
bdq.notifyAll();
}
return;
}
}
}
Thread.sleep(1);
}
}
public void submitDataBlockNoDelay(KLALBBlock kb) {
synchronized (remotetcps) {
@@ -368,17 +356,29 @@ public class KLALBCore {
synchronized( l) {
for (int i = 0; i < l.size(); i++) {
KLALBBlock block=l.get(i);
long timex = (System.nanoTime() - block.sendtime) / 1000000;
if ((i==0&&timex>100)||timex > (RTO/1000000)*(1<<Math.min(8,block.resend))+2000*i) {
submitReDataBlock(block);
long timex = System.nanoTime() - block.sendtime;
if(i<10) {
if(timex > Math.min(RTO,1000000000L )*(1<<Math.min(8,block.resend))+200*i) {
submitDataBlock(block);
block.sendtime = System.nanoTime();
System.out.println("超时重传："+block+" 位置："+i);
block.resend++;
if(block.resend>=Consts.MAX_RESEND) {
l.remove(i);
i--;
}
}
}
/*if ((i<=10&&timex>500)||timex > (RTO/1000000)*(1<<Math.min(8,block.resend))+2000*i) {
submitDataBlock(block);
block.sendtime = System.nanoTime();
System.out.println("超时重传："+block+" 位置："+i);
block.resend++;
if(block.resend>=Consts.MAX_RESEND) {
l.remove(i);
i--;
}
System.out.println("超时重传："+block+" 位置："+i);
}
}*/
}
}
long curr=System.nanoTime();
src/org/kne/cloud/network/klalb/KLALBSM.java
+1 -1
View File
@@ -17,7 +17,7 @@ import org.kne.cloud.network.mport.ServiceElement;
public class KLALBSM {
public static void main(String[] args) throws IOException {
System.out.println("KNE云负载均衡调度软件服务端V0.1");
System.out.println("KNE云负载均衡调度软件服务端V1.0");
ServerPropties sp=new ServerPropties();
int remp=sp.getPort();
src/org/kne/cloud/network/klalb/RemoteTCPConnection.java
+22 -6
View File
@@ -11,6 +11,7 @@ import java.io.StreamCorruptedException;
import java.net.Socket;
import java.net.UnknownHostException;
import java.util.Arrays;
import java.util.Comparator;
import java.util.Queue;
import java.util.Random;
import java.util.UUID;
@@ -19,6 +20,7 @@ import java.util.concurrent.BlockingDeque;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.LinkedBlockingDeque;
import java.util.concurrent.PriorityBlockingQueue;
import java.util.zip.DataFormatException;
import java.util.zip.Deflater;
import java.util.zip.GZIPInputStream;
@@ -243,17 +245,31 @@ public class RemoteTCPConnection extends TCPConnection {
}
}
private Queue<KLALBBlock> NDsendDeque = new ConcurrentLinkedQueue<>();
private Queue<KLALBBlock> NDsendDeque = new ConcurrentLinkedQueue<KLALBBlock>();
public Queue<KLALBBlock> getNDSendDeque() {
return NDsendDeque;
}
private Queue<KLALBBlock> resendDeque = new ConcurrentLinkedQueue<>();
public Queue<KLALBBlock> getReSendDeque() {
return sendDeque;
}
private Queue<KLALBBlock> sendDeque = new ConcurrentLinkedQueue<>();
private Queue<KLALBBlock> sendDeque = new PriorityBlockingQueue<KLALBBlock>(2, new Comparator<KLALBBlock>() {
@Override
public int compare(KLALBBlock o1, KLALBBlock o2) {
if(o1.resend>o2.resend) {
return -1;
}else if(o1.resend<o2.resend) {
return 1;
}else {
if(o1.number>o2.number) {
return 1;
}else if(o1.number<o2.number) {
return -1;
}else {
return 0;
}
}
}
});
public Queue<KLALBBlock> getSendDeque() {
return sendDeque;