package org.kne.cloud.network.congestion;

import java.io.Closeable;
import java.net.SocketException;
import java.util.Collection;
import java.util.Iterator;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.atomic.AtomicLong;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.Lock;
import java.util.function.Consumer;

import org.kne.cloud.network.NetworkPacket;
import org.kne.cloud.network.ThreadTool;
import org.kne.cloud.network.klalb.SendItem;
import org.kne.concurrent.ThreadParker;

public class SendPacketSlidingWindow<K, V extends NetworkPacket> implements Closeable, AutoCloseable {
	private ThreadParker tp=new ThreadParker();
	
	private Map<K, SendItem<V>> sendMap = new ConcurrentHashMap<>(2048);
	private AtomicLong sendWindowUsed=new AtomicLong(0);
	//private LongAdder sendWindowUsed = new LongAdder();

	private int headerCalibrate = 0;

	public void setHeaderCalibrate(int headerCalibrate) {
		this.headerCalibrate = headerCalibrate;
	}

	private CongestionAlgorithm algorithm;

	private Consumer<V> resendConsumer;
	private Consumer<V> closingConsumer;
	private ResendChecker checker = new ResendChecker();

	private Lock lock;

	private Condition condition;
	private boolean removeAtResend;
	private volatile long windowSize =0;
	private volatile boolean closed = false;

	public SendPacketSlidingWindow(CongestionAlgorithm algorithm, long windowSize) {
		this(algorithm, windowSize, 0, true);
	}

	public SendPacketSlidingWindow(CongestionAlgorithm algorithm, long windowSize, int headerCalibrate) {
		this(algorithm, windowSize, headerCalibrate, true);
	}

	public SendPacketSlidingWindow(CongestionAlgorithm algorithm, long windowSize, int headerCalibrate,
			boolean removeAtResend) {
		this.headerCalibrate = headerCalibrate;
		this.removeAtResend = removeAtResend;
		this.algorithm = algorithm;
		this.windowSize=windowSize;
		Thread t = ThreadTool.makeVThread("重路由计时器线程", checker);
		t.start();
	}

	public long getWindowSize() {
		return windowSize;
	}

	public void setWindowSize(long windowSize) {
		this.windowSize=windowSize;
	}

	public Consumer<V> getResendConsumer() {
		return resendConsumer;
	}

	public void setResendConsumer(Consumer<V> resendConsumer) {
		this.resendConsumer = resendConsumer;
	}

	public Consumer<V> getClosingConsumer() {
		return closingConsumer;
	}

	public void setClosingConsumer(Consumer<V> closingConsumer) {
		this.closingConsumer = closingConsumer;
	}

	public long getSendWindowUsed() {
		return sendWindowUsed.get();
	}

	public int getHeaderCalibrate() {
		return headerCalibrate;
	}

	public CongestionAlgorithm getAlgorithm() {
		return algorithm;
	}

	public boolean isCongestion( double scale) {
		boolean congress = sendWindowUsed.get() > windowSize * scale;
		return congress;

	}

	public boolean isCongestion() {
		boolean congress = sendWindowUsed.get() > windowSize ;
		return congress;
	}
	
	public void waitForAvaliable() throws SocketException {
		while(isCongestion()) {
			if(closed) {
				throw new SocketException("Send window closed!");
			}
			tp.parkNanos(1000000L);
		}
	}
	public void putBlocked(K sequence, V packet) throws SocketException {
		waitForAvaliable();
		put(sequence,packet);
	}
	
	public void put(K sequence, V packet)  {
		SendItem<V> newitem = new SendItem<V>(packet,headerCalibrate);
		SendItem<V> old = sendMap.put(sequence, newitem);
		long dx=0;
		if (old != null) { 
			dx=-old.getPacketLength() ;
		}
		long newWindow =sendWindowUsed.addAndGet(newitem.getPacketLength() +dx);
		algorithm.setCurrentWindowUsed(newWindow);
	}
	public V ack(K sequence) {
		return ack(sequence,false);
	}
	
	public V ack(K sequence,boolean ecn) {
		SendItem<V> ipv6;
		if ((ipv6 = sendMap.remove(sequence)) != null) {
			long newWindow =sendWindowUsed.addAndGet(-ipv6.getPacketLength() );
			algorithm.setCurrentWindowUsed(newWindow);
			long RTTC = System.nanoTime() - ipv6.getSendtime();
			algorithm.putAck(ipv6.getPacketLength(), RTTC, ecn);
			tp.unpark();
			// System.out.println("remove:"+aseq.getSequence()+" "+sendMap.size());
			Lock lockx = lock;
			if (lockx != null) {
				lockx.lock();
				try {
					Condition cds = condition;
					if (cds != null) {
						cds.signalAll();
					}
				} finally {
					lockx.unlock();
				}
			}
			return ipv6.getPacket();
		} else {
			// System.out.println("miss:"+aseq.getSequence()+" "+sendMap.size());
			return null;
		}
	}

	public V ack(K sequence,boolean ecn,long OWDdown) {
		SendItem<V> ipv6;
		if ((ipv6 = sendMap.remove(sequence)) != null) {
			long newWindow =sendWindowUsed.addAndGet(-ipv6.getPacketLength() );
			algorithm.setCurrentWindowUsed(newWindow);
			long RTTC = System.nanoTime() - ipv6.getSendtime();
			algorithm.putAck(ipv6.getPacketLength(), RTTC,OWDdown, ecn);
			tp.unpark();
			// System.out.println("remove:"+aseq.getSequence()+" "+sendMap.size());
			Lock lockx = lock;
			if (lockx != null) {
				lockx.lock();
				try {
					Condition cds = condition;
					if (cds != null) {
						cds.signalAll();
					}
				} finally {
					lockx.unlock();
				}
			}
			return ipv6.getPacket();
		} else {
			// System.out.println("miss:"+aseq.getSequence()+" "+sendMap.size());
			return null;
		}
	}
	
	private class ResendChecker implements Runnable {

		@Override
		public void run() {
			while (!closed) {

				Collection<SendItem<V>> vals = sendMap.values();
				for (Iterator<SendItem<V>> iterator = vals.iterator(); iterator.hasNext();) {
					SendItem<V> sitm = (SendItem<V>) iterator.next();
					long current = System.nanoTime();
					if (current - sitm.getSendtime() > algorithm.getRTO()) {
						if (removeAtResend) {
							iterator.remove();
							long newWindow = sendWindowUsed.addAndGet((int) -sitm.getPacketLength() );
							algorithm.setCurrentWindowUsed(newWindow);
							tp.unpark();
						} else {
							sitm.setSendtime(current);
						}
						V pktr = sitm.getPacket();
						if (resendConsumer != null)
							resendConsumer.accept(pktr);
					}
				}

				try {
					Thread.sleep(2);
				} catch (InterruptedException e) {
					e.printStackTrace();
				}
			}
			// 清理资源
			if (closingConsumer != null) {
				for (SendItem<V> item : sendMap.values()) {
					closingConsumer.accept(item.getPacket());
				}
			}
			sendMap.clear();
			sendWindowUsed.set(0);
		}
	}

	public boolean isClosed() {
		return closed;
	}

	@Override
	public void close() {
		closed = true;
	}

	public void setCongressCondition(Lock lock2, Condition condition2) {
		this.lock = lock2;
		this.condition = condition2;
	}

	public Map<K, SendItem<V>> getSendmap() {
		return sendMap;
	}

	public int getWindowPacketCount() {
		return sendMap.size();
	}

	public boolean isEmpty() {
		return sendMap.isEmpty();
	}

	@Override
	public String toString() {
		return "SendPacketSlidingWindow [sendWindowUsed=" + sendWindowUsed + ", windowSize=" + windowSize + ", closed="
				+ closed + "]";
	}

	public void waitForEmpty() throws SocketException {
		while(!isEmpty()) {
			if(isClosed()) {
				throw new SocketException("Send window closed!");
			}
			tp.parkNanos(1000000L);
		}
	}




}
