377 lines
9.2 KiB
Java
377 lines
9.2 KiB
Java
package org.kne.cloud.network.kltp;
|
|
|
|
import java.io.IOException;
|
|
import java.io.OutputStream;
|
|
import java.net.BindException;
|
|
import java.net.Inet6Address;
|
|
import java.net.SocketException;
|
|
import java.net.SocketTimeoutException;
|
|
import java.nio.ByteBuffer;
|
|
import java.nio.channels.WritableByteChannel;
|
|
|
|
import org.kne.cloud.network.klalb.KLALBScheduler;
|
|
import org.kne.concurrent.*;
|
|
import org.kne.cloud.network.NetworkPacket;
|
|
import org.kne.cloud.network.congestion.CongestionAlgorithm;
|
|
import org.kne.cloud.network.congestion.DCTCP2CongestionAlgorithm;
|
|
import org.kne.cloud.network.congestion.DCTCPCongestionAlgorithm;
|
|
import org.kne.cloud.network.congestion.SendPacketSlidingWindow;
|
|
import org.kne.cloud.network.ipv6.IPv6Address;
|
|
import org.kne.cloud.network.ipv6.IPv6Packet;
|
|
import org.kne.cloud.network.klalb.DATATPacket;
|
|
import org.kne.cloud.network.klalb.KLALBController;
|
|
|
|
import java.util.UUID;
|
|
import java.util.concurrent.ScheduledFuture;
|
|
import java.util.concurrent.TimeUnit;
|
|
import java.util.concurrent.atomic.AtomicLong;
|
|
import java.util.concurrent.atomic.AtomicReference;
|
|
import java.util.concurrent.locks.*;
|
|
public class KLTPOutputStream extends OutputStream implements KLTPPacketConsumer,WritableByteChannel{
|
|
private static final int HEADER_CALIBRATE = 150;
|
|
private KLALBController controller;
|
|
|
|
private IPv6Address remoteaddr;
|
|
public IPv6Address getRemoteAddress() {
|
|
return remoteaddr;
|
|
}
|
|
public KLTPOutputStream(KLALBController controller,IPv6Address remoteaddr,UUID uuid) throws BindException {
|
|
this.controller=controller;
|
|
this.streamUUID =uuid;
|
|
this.remoteaddr=remoteaddr;
|
|
if(controller.getConfigItem()!=null)
|
|
this.delaytime=controller.getConfigItem().getNagleDelayTime();
|
|
|
|
algorithm.setWindowControlConsumer((window)->{
|
|
sendMap.setWindowSize(window);
|
|
});
|
|
|
|
controller.getKLTPregister().registerSendStream(this);
|
|
}
|
|
private UUID streamUUID;
|
|
|
|
|
|
private int MTU=8192;
|
|
|
|
private CongestionAlgorithm algorithm=new DCTCPCongestionAlgorithm();
|
|
private SendPacketSlidingWindow<Long,KLTPPacket>sendMap=new SendPacketSlidingWindow<>(algorithm, 4*MTU,HEADER_CALIBRATE,false);
|
|
{
|
|
sendMap.setResendConsumer((dtp)->{
|
|
if(dtp.getSendCounter()>=50) {
|
|
System.err.println("send error!");
|
|
try {
|
|
close0(false);
|
|
} catch (IOException e) {
|
|
e.printStackTrace();
|
|
}
|
|
return;
|
|
}
|
|
|
|
controller.getIpv6Router().runPacketSendTask(()->{
|
|
long length= dtp.getTotalLength();
|
|
//socketRawMonitor.getUploadBandwidth().recordPacket(pidg.generate(), (int) length);
|
|
//System.out.println("第"+(dtp.getSendCounter()-1)+"次重传："+dtp);
|
|
return controller.createPacketToAddress( remoteaddr,0,dtp,1);
|
|
});
|
|
dtp.incSendCounter();
|
|
});
|
|
}
|
|
|
|
private AtomicLong outputcount=new AtomicLong( 0);
|
|
|
|
private KLTPPacket dataPack =null;
|
|
|
|
|
|
private long delaytime=2;
|
|
|
|
|
|
|
|
private Lock olock=new SpinLock();
|
|
|
|
private volatile boolean autoFlush=true;
|
|
@Override
|
|
public void write(int b) throws IOException {
|
|
if (sendMap.isClosed())
|
|
throw new SocketException("Socket is closed");
|
|
if(olock!=null)
|
|
olock.lock();
|
|
createDataPack();
|
|
try {
|
|
dataPack.getKLTPData().put((byte) b);
|
|
if (dataPack!=null&& dataPack.getKLTPData().hasRemaining()) {
|
|
if(autoFlush)
|
|
delayFlush();
|
|
}else {
|
|
flush0();
|
|
}
|
|
}finally {
|
|
if(olock!=null)
|
|
olock.unlock();
|
|
}
|
|
}
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
@Override
|
|
public void write(byte[] b, int off, int len) throws IOException {
|
|
write(ByteBuffer.wrap(b, off, len));
|
|
return ;
|
|
}
|
|
|
|
private int writeWithoutFlush(ByteBuffer src)throws IOException {
|
|
if (sendMap.isClosed())
|
|
throw new SocketException("Socket is closed");
|
|
int counter=0;
|
|
if(olock!=null)
|
|
olock.lock();
|
|
try {
|
|
|
|
while(src.hasRemaining()) {
|
|
createDataPack();
|
|
int min=Math.min(src.remaining(), dataPack.getKLTPData().remaining());
|
|
int olm=src.limit();
|
|
src.limit(src.position()+min);
|
|
dataPack.getKLTPData().put(src);
|
|
counter+=min;
|
|
src.limit(olm);
|
|
|
|
if (!dataPack.getKLTPData().hasRemaining()) {
|
|
flush0();
|
|
}
|
|
}
|
|
|
|
|
|
}finally {
|
|
if(olock!=null)
|
|
olock.unlock();
|
|
}
|
|
return counter;
|
|
}
|
|
|
|
@Override
|
|
public int write(ByteBuffer src) throws IOException {
|
|
if (sendMap.isClosed())
|
|
throw new SocketException("Socket is closed");
|
|
int counter=0;
|
|
if(olock!=null)
|
|
olock.lock();
|
|
try {
|
|
|
|
while(src.hasRemaining()) {
|
|
createDataPack();
|
|
int min=Math.min(src.remaining(), dataPack.getKLTPData().remaining());
|
|
int olm=src.limit();
|
|
src.limit(src.position()+min);
|
|
dataPack.getKLTPData().put(src);
|
|
counter+=min;
|
|
src.limit(olm);
|
|
|
|
if (!dataPack.getKLTPData().hasRemaining()) {
|
|
flush0();
|
|
}
|
|
}
|
|
|
|
if(dataPack!=null&& dataPack.getKLTPData().position()>0) {
|
|
if(autoFlush)
|
|
delayFlush();
|
|
}
|
|
}finally {
|
|
if(olock!=null)
|
|
olock.unlock();
|
|
}
|
|
return counter;
|
|
}
|
|
|
|
|
|
private void createDataPack() {
|
|
if(dataPack==null) {
|
|
long pl=outputcount.getPlain();
|
|
dataPack=new KLTPPacket(streamUUID,KLTPPacket.KLTP_TYPE_DATA,pl++, MTU);
|
|
outputcount.setPlain(pl);
|
|
//System.out.println("EOF:"+dataPack);
|
|
}
|
|
}
|
|
public void waitForAllAcknowledged(int timeout) throws IOException {
|
|
long start=System.nanoTime();
|
|
while(true){
|
|
if (sendMap.isClosed())
|
|
throw new SocketException("Socket is closed");
|
|
if(sendMap.isEmpty())
|
|
break;
|
|
if(timeout!=0&&(System.nanoTime()-start>timeout*1000000L))
|
|
throw new SocketTimeoutException("wait for acknowledged timout");
|
|
}
|
|
}
|
|
AtomicReference<IOException> ioe=new AtomicReference<>();
|
|
private volatile ScheduledFuture tt;
|
|
|
|
@Override
|
|
public void flush() throws IOException {
|
|
if(olock!=null)
|
|
olock.lock();
|
|
try {
|
|
delayFlush();
|
|
}finally {
|
|
if(olock!=null)
|
|
olock.unlock();
|
|
}
|
|
}
|
|
public void forceFlush() throws IOException{
|
|
if(olock!=null)
|
|
olock.lock();
|
|
try {
|
|
flush0();
|
|
}finally {
|
|
if(olock!=null)
|
|
olock.unlock();
|
|
}
|
|
}
|
|
public void delayFlush()throws IOException{
|
|
if(delaytime<=0) {
|
|
flush0();
|
|
}else {
|
|
if(tt==null) {
|
|
Runnable r= new Runnable() {
|
|
|
|
@Override
|
|
public void run() {
|
|
if(sendMap.isClosed())
|
|
tt.cancel(false);
|
|
try {
|
|
if(olock!=null)
|
|
olock.lock();
|
|
try {
|
|
flush0();
|
|
}finally {
|
|
if(olock!=null)
|
|
olock.unlock();
|
|
}
|
|
} catch (IOException e) {
|
|
ioe.set(e);
|
|
}
|
|
}
|
|
};
|
|
tt= KLALBScheduler.getScheduler().scheduleWithFixedDelay(r, delaytime, delaytime,TimeUnit.NANOSECONDS);
|
|
}
|
|
|
|
IOException ioex=ioe.get();
|
|
if(ioex!=null) {
|
|
ioex.fillInStackTrace();
|
|
throw ioex;
|
|
}
|
|
}
|
|
}
|
|
|
|
private void flush0() throws IOException {
|
|
KLTPPacket pack=dataPack;
|
|
if (pack!=null&&pack.getKLTPData().position() > 0) {
|
|
|
|
sendMap.waitForAvaliable();
|
|
|
|
controller.getIpv6Router().runPacketSendTask(()->{
|
|
pack.getKLTPData().flip();
|
|
pack.incSendCounter();
|
|
sendMap.put(pack.getSequence(),pack);
|
|
|
|
int size= pack.getKLTPData().limit();
|
|
//socketMonitor.getUploadBandwidth().recordPacket(pid, size);
|
|
//controller.getDatatMonitor().getUploadBandwidth().recordPacket(KLALBUtils.createGlobalUUID(), size);
|
|
|
|
//socketRawMonitor.getUploadBandwidth().recordPacket(pid, size);
|
|
return controller.createPacketToAddress(remoteaddr,0,pack);
|
|
|
|
});
|
|
dataPack=null;
|
|
}
|
|
}
|
|
@Override
|
|
public void close() throws IOException {
|
|
close0(true);
|
|
}
|
|
private void close0(boolean grace) throws IOException {
|
|
try {
|
|
if(sendMap.isClosed())
|
|
return;
|
|
if(grace) {
|
|
if(olock!=null)
|
|
olock.lock();
|
|
try {
|
|
flush0();
|
|
}finally {
|
|
if(olock!=null)
|
|
olock.unlock();
|
|
}
|
|
|
|
long pl=outputcount.getPlain();
|
|
KLTPPacket pack=new KLTPPacket(streamUUID,KLTPPacket.KLTP_TYPE_DATAFIN, pl++ ,MTU);
|
|
outputcount.setPlain(pl);
|
|
pack.getKLTPData(). flip();
|
|
pack.incSendCounter();
|
|
controller.getIpv6Router().runPacketSendTask (()->{
|
|
return controller.createPacketToAddress(remoteaddr,0,pack);
|
|
});
|
|
sendMap.put(pack.getSequence(),pack);
|
|
// System.out.println("EOF send:"+pack);
|
|
}
|
|
|
|
sendMap.close();
|
|
}finally {
|
|
if(tt!=null)
|
|
tt.cancel(false);
|
|
controller.getKLTPregister().unregisterSendStream(this);
|
|
}
|
|
}
|
|
@Override
|
|
public boolean isOpen() {
|
|
return !sendMap.isClosed();
|
|
}
|
|
public boolean isAutoFlush() {
|
|
return autoFlush;
|
|
}
|
|
public void setAutoFlush(boolean b) {
|
|
autoFlush=b;
|
|
}
|
|
public long write(ByteBuffer[] srcs, int offset, int length) throws IOException {
|
|
long lth=0;
|
|
for(int i=offset;i<offset+length;i++) {
|
|
lth+=writeWithoutFlush(srcs[i]);
|
|
}
|
|
if(dataPack!=null&& dataPack.getKLTPData().position()>0) {
|
|
if(autoFlush)
|
|
flush();
|
|
}
|
|
return lth;
|
|
}
|
|
|
|
|
|
@Override
|
|
public void accept(IPv6Packet u) {
|
|
if(u.getPayload() instanceof KLTPPacket) {
|
|
KLTPPacket kltp=(KLTPPacket) u.getPayload();
|
|
switch(kltp.getType()) {
|
|
case KLTPPacket.KLTP_TYPE_ACK:
|
|
sendMap.ack(kltp.getSequence(), kltp.isCE());
|
|
|
|
break;
|
|
}
|
|
}
|
|
}
|
|
|
|
@Override
|
|
public UUID getStreamUUID() {
|
|
return streamUUID;
|
|
}
|
|
public boolean isClosed() {
|
|
return sendMap.isClosed();
|
|
}
|
|
@Override
|
|
public String toString() {
|
|
return "KLTPOutputStream [streamUUID=" + streamUUID + ", sendMap=" + sendMap + "]";
|
|
}
|
|
|
|
} |