package org.kne.concurrent;

import java.util.Queue;
import java.util.concurrent.Executor;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.function.Consumer;

import org.jctools.queues.MpscArrayQueue;
import org.kne.cloud.network.klalb.PerformanceStrategy;

public class HighPerformanceExecutor2 implements Executor {
	
	private ThreadElement[] threads;

	public HighPerformanceExecutor2(int threadcount) {
		//this(threadcount,Executors.defaultThreadFactory());
		this(threadcount,Thread.ofVirtual().factory());
	}
	public HighPerformanceExecutor2(int threadcount,ThreadFactory th) {
		threads=new ThreadElement[threadcount];
		for (int i = 0; i < threadcount; i++) {
			ThreadElement t=	new ThreadElement();
		th.newThread(t).start();
		threads[i]=t;
		}
	}


	private static class ThreadElement implements  Runnable{
		private MpscArrayQueue<Runnable> queue=new MpscArrayQueue<Runnable>(2048);
		private AtomicInteger size=new AtomicInteger();
		
		private volatile ThreadParker parker=new ThreadParker();
		public Queue<Runnable> getQueue() {
			return queue;
		}
		public int size() {
			
			return queue.size();
		}
		public long prev=System.nanoTime();
		public boolean putTask(Runnable e) {
			boolean b=queue.offer(e);
			if(b) {
			//	size.incrementAndGet();
				long curr=System.nanoTime();
				if(curr-prev>1000L||queue.size()>=8) {
					prev=curr;
					parker.unpark();
				}
			}
			return b;

		}
		@Override
		public void run() {
			while(true) {
				Runnable r=queue.poll();
				if(r!=null) {
				//	size.decrementAndGet();
					try {
					r.run();
					}catch(Throwable e) {
						e.printStackTrace();
					}
					ThreadYieldCheckpoint.yieldCheckpoint(1000000L);
				}else {
					parker.parkNanos(1000000L);
					
				}
			}
		}
	}
	@Override
	public void execute(Runnable command) {
			if(execute0((x)->{command.run();},1000, PerformanceStrategy.MULTI_FILL)) {
				return;
			}
			System.out.println("loss!");
		//backup.execute(command);
			
			
	}
	
	private long vl=0;
	
	private boolean execute0(Consumer<Boolean> command, int limit, PerformanceStrategy strategy) {
		if(strategy.equals(PerformanceStrategy.MULTI_SCATTER)){
			long ord=vl++;
			ThreadElement te= threads[(int) (ord%threads.length)];
			boolean b=te.size()>limit;
			return te.putTask( ()->{command.accept(b);});
		}else {
			for (int i = 0; i < threads.length; i++) {
				ThreadElement te = threads[i];
				boolean b = te.size() > limit;
				if (!b) {
					if (te.putTask(() -> {
						command.accept(false);
					})) {
						return true;
					}
				}
			}

			for (int i = 0; i < threads.length; i++) {
				ThreadElement te = threads[i];
				if (te.putTask(() -> {
					command.accept(true);
				})) {
					return true;
				}
			}
			return false;
		}
	}
	
	public void executeWithCongestionReport(Consumer<Boolean> command) {
		if(execute0(command,1000, PerformanceStrategy.MULTI_FILL)) {
			return;
		}
		System.out.println("loss!");
	}
	public void executeWithCongestionReport(Consumer<Boolean> command, int limit, PerformanceStrategy strategy) {
		if(execute0(command,limit,strategy)) {
			return;
		}
		System.out.println("loss!");
			
		
	}
}