package com.planet_ink.coffee_mud.core.threads;

import java.util.Collection;
import java.util.Enumeration;
import java.util.HashMap;
import java.util.LinkedList;
import java.util.Map;
import java.util.concurrent.*;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicInteger;

import com.planet_ink.coffee_mud.Common.interfaces.Session;
import com.planet_ink.coffee_mud.core.CMLib;
import com.planet_ink.coffee_mud.core.Log;
import com.planet_ink.coffee_mud.core.collections.Pair;
import com.planet_ink.coffee_mud.core.collections.PairSVector;
import com.planet_ink.coffee_mud.core.collections.SLinkedList;
import com.planet_ink.coffee_mud.core.collections.STreeMap;
import com.planet_ink.coffee_mud.core.collections.STreeSet;
import com.planet_ink.coffee_mud.core.collections.UniqueEntryBlockingQueue;
import com.planet_ink.coffee_mud.core.interfaces.TickableGroup;
/*
   Copyright 2010-2020 Bo Zimmerman

   Licensed under the Apache License, Version 2.0 (the "License");
   you may not use this file except in compliance with the License.
   You may obtain a copy of the License at

	   http://www.apache.org/licenses/LICENSE-2.0

   Unless required by applicable law or agreed to in writing, software
   distributed under the License is distributed on an "AS IS" BASIS,
   WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
   See the License for the specific language governing permissions and
   limitations under the License.
*/
public class CMThreadPoolExecutor extends ThreadPoolExecutor
{
	protected Map<Runnable,Thread>	active = new HashMap<Runnable,Thread>();
	protected long  				timeoutMillis;
	protected CMThreadFactory		threadFactory;
	protected int   				queueSize = 0;
	protected String				poolName = "Pool";
	protected volatile long 		lastRejectTime = 0;
	protected volatile int  		rejectCount = 0;

	protected static class CMArrayBlockingQueue<E> extends ArrayBlockingQueue<E>{
		private static final long serialVersionUID = -4557809818979881831L;
		public CMThreadPoolExecutor executor = null;

		public CMArrayBlockingQueue(final int capacity)
		{
			super(capacity);
		}

		@Override
		public boolean offer(final E o)
		{
			final int allWorkingThreads = executor.getActiveCount() + super.size();
			return (allWorkingThreads < executor.getPoolSize()) && super.offer(o);
		}
	}

	public CMThreadPoolExecutor(final String poolName,
								final int corePoolSize, final int maximumPoolSize,
								final long keepAliveTime, final TimeUnit unit,
								final long timeoutMins, final int queueSize)
	{
		super(corePoolSize, maximumPoolSize, keepAliveTime, unit, new CMArrayBlockingQueue<Runnable>(queueSize));
		((CMArrayBlockingQueue<Runnable>)this.getQueue()).executor=this;
		timeoutMillis=timeoutMins * 60 * 1000;
		this.poolName=poolName;
		threadFactory=new CMThreadFactory(poolName);
		setThreadFactory(threadFactory);
		this.queueSize=queueSize;
		setRejectedExecutionHandler(new RejectedExecutionHandler()
		{
			@Override
			public void rejectedExecution(final Runnable r, final ThreadPoolExecutor executor)
			{
				try
				{
					executor.getQueue().put(r);
				}
				catch (final InterruptedException e)
				{
					throw new RejectedExecutionException(e);
				}
			}
		});
	}

	@Override
	protected void beforeExecute(final Thread t, final Runnable r)
	{
		synchronized(active)
		{
			if(t instanceof CMFactoryThread)
				((CMFactoryThread)t).setRunnable(r);
			active.put(r,t);
		}
	}

	@Override
	protected void afterExecute(final Runnable r, final Throwable t)
	{
		synchronized(active)
		{
			final Thread th=active.get(r);
			if(th instanceof CMFactoryThread)
				((CMFactoryThread)th).setRunnable(null);
			active.remove(r);
		}
	}

	@Override
	public int getActiveCount()
	{
		return active.size();
	}

	public boolean isActive(final Runnable r)
	{
		return active.containsKey(r);
	}

	public boolean isActiveOrQueued(final Runnable r)
	{
		return active.containsKey(r) || getQueue().contains(r);
	}

	@Override
	public void execute(final Runnable r)
	{
		try
		{
			if(this.getQueue().contains(r))
				return;
			super.execute(r);
			if((rejectCount>0)&&(System.currentTimeMillis()-lastRejectTime)>5000)
			{
				Log.warnOut(rejectCount+" Pool_"+poolName,"Threads rejected.");
				rejectCount=0;
			}
		}
		catch(final RejectedExecutionException e)
		{
			if(r instanceof CMRunnable)
			{
				final Collection<CMRunnable> runsKilled = getTimeoutOutRuns(1);
				for(final CMRunnable runnable : runsKilled)
				{
					if(runnable instanceof Session)
					{
						final Session S=(Session)runnable;
						final StringBuilder sessionInfo=new StringBuilder("");
						sessionInfo.append("status="+S.getStatus()+" ");
						sessionInfo.append("active="+S.activeTimeMillis()+" ");
						sessionInfo.append("online="+S.getMillisOnline()+" ");
						sessionInfo.append("lastloop="+(System.currentTimeMillis()-S.getInputLoopTime())+" ");
						sessionInfo.append("addr="+S.getAddress()+" ");
						sessionInfo.append("mob="+((S.mob()==null)?"null":S.mob().Name()));
						Log.errOut("Pool_"+poolName,"Timed-Out Runnable: "+sessionInfo.toString());
					}
					else
					if(runnable instanceof TickableGroup)
					{
						final TickableGroup G=(TickableGroup)runnable;
						Log.errOut("Pool_"+poolName,"Timed-Out Runnable: "+G.getName()+"-"+G.getStatus()+"\n\r");
					}
					else
						Log.errOut("Pool_"+poolName,"Timed-Out Runnable: "+runnable.toString());
				}
			}
			lastRejectTime=System.currentTimeMillis();
			rejectCount++;
		}
	}

	public Collection<CMRunnable> getTimeoutOutRuns(final int maxToKill)
	{
		final LinkedList<CMRunnable> timedOut=new LinkedList<CMRunnable>();
		if(timeoutMillis<=0)
			return timedOut;
		final LinkedList<Thread> killedOut=new LinkedList<Thread>();
		synchronized(active)
		{
			try
			{
				for (final Runnable runnable : active.keySet())
				{
					if(runnable instanceof CMRunnable)
					{
						final CMRunnable cmRunnable=(CMRunnable)runnable;
						final Thread thread=active.get(runnable);
						if(cmRunnable.activeTimeMillis() > timeoutMillis)
						{
							if(timedOut.size() >= maxToKill)
							{
								CMRunnable leastWorstOffender=null;
								for(final CMRunnable r : timedOut)
								{
									if((leastWorstOffender != null)
									&&(r.activeTimeMillis() < leastWorstOffender.activeTimeMillis()))
										leastWorstOffender=r;
								}
								if(leastWorstOffender!=null)
								{
									if(cmRunnable.activeTimeMillis() < leastWorstOffender.activeTimeMillis())
										continue;
									else
										timedOut.remove(leastWorstOffender);
								}
							}
							timedOut.add(cmRunnable);
							killedOut.add(thread);
						}
					}
				}
			}
			catch(final Exception e)
			{
			}
		}
		try
		{
			while(killedOut.size()>0)
			{
				final Thread t = killedOut.remove();
				active.remove(t);
				CMLib.killThread(t,100,3);
			}
		}
		catch(final Exception e)
		{
		}
		return timedOut;
	}
}
