package com.planet_ink.coffee_web.util;

import java.util.Collection;
import java.util.HashMap;
import java.util.Iterator;
import java.util.LinkedList;
import java.util.concurrent.*;
import java.util.logging.Logger;

/*
   Copyright 2012-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.
*/

/**
 * Thread executor with a waiting queue plus one extra feature -- a passive system for
 * timing out threads that have been active longer than we want.
 * This system kicks in whenever a thread execute attempt is rejected by the underlying
 * executor due to the queue being full.  In that case, it will look through the active
 * threads to find one that can be timed out, and will interrupt it -- as if that will work. :/
 * @author Bo Zimmerman
 *
 */
public class CWThreadExecutor extends ThreadPoolExecutor
{
	protected HashMap<Runnable,RunWrap>active 		= new HashMap<Runnable,RunWrap>();
	protected long  				 	 timeoutMillis;
	protected CWThreadFactory   	 	 threadFactory;
	protected String				 	 poolName 		= "Pool";
	protected volatile long 		 	 lastRejectTime = 0;
	protected volatile int  		 	 rejectCount 	= 0;
	protected final Logger				 logger;

	protected static class CMLinkedBlockingQueue<E> extends ArrayBlockingQueue<E>{
		private static final long serialVersionUID = -4357809818979881831L;
		public CWThreadExecutor executor = null;

		public CMLinkedBlockingQueue(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 CWThreadExecutor(final String poolName,
							final CWConfig config,
							final int corePoolSize, final int maximumPoolSize,
							final long keepAliveTime, final TimeUnit unit,
							final long timeoutSecs, final int queueSize)
	{
		super(corePoolSize, maximumPoolSize, keepAliveTime, unit, new CMLinkedBlockingQueue<Runnable>(queueSize));
		((CMLinkedBlockingQueue<Runnable>)getQueue()).executor=this;
		timeoutMillis=timeoutSecs * 1000L;
		this.poolName=poolName;
		threadFactory=new CWThreadFactory(poolName, config);
		setThreadFactory(threadFactory);
		this.logger=config.getLogger();
		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)
		{
			try
			{
				active.put(r,new RunWrap(r,t));
			}
			catch(final Throwable e)
			{
				logger.throwing("", "", e);
			}
		}
	}

	@Override
	protected void afterExecute(final Runnable r, final Throwable t)
	{
		synchronized(active)
		{
			try
			{
				active.remove(r);
			}
			catch(final Throwable e)
			{
				logger.throwing("", "", e);
			}
		}
	}

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

	@Override
	public void execute(final Runnable r)
	{
		try
		{
			if(this.getQueue().contains(r))
				return;
			super.execute(r);

			// an optomization for logging purposes.  When my smtp server gets hit
			// by a spam-bot, the log fills up my hard drive.  this helps prevent that.
			if((rejectCount>0)&&(System.currentTimeMillis()-lastRejectTime)>5000)
			{
				logger.warning(rejectCount+" Pool_"+poolName+": Threads rejected.");
				rejectCount=0;
			}
		}
		catch(final RejectedExecutionException e)
		{
			// a thread is rejected only when the queue is filled.  Look for the blockages!
			final Collection<RunWrap> runsKilled = getTimeoutOutRuns(1);
			for(final RunWrap runnable : runsKilled)
				logger.severe("Pool_"+poolName+": Old(er) Runnable killed: "+runnable.toString());
			lastRejectTime=System.currentTimeMillis();
			rejectCount++;
		}
	}

	/**
	 * Scans the list of active running threads/runnables for ones that have
	 * been active longer than permitted.  It will attempt to kill those, returning
	 * those threads which were timed out.  There's also the funny feature that
	 * it will kill only the number you ask for, starting with the oldest offenders
	 * and working towards the youngest.
	 * @param maxToKill the maximum number of threads to kill
	 * @return the runnable/thread combo that was killed.
	 */
	public Collection<RunWrap> getTimeoutOutRuns(final int maxToKill)
	{
		final LinkedList<RunWrap> timedOut=new LinkedList<RunWrap>();
		if(timeoutMillis<=0) return timedOut;
		final LinkedList<Thread> killedOut=new LinkedList<Thread>();
		synchronized(active)
		{
			try
			{
				for(final Iterator<RunWrap> i = active.values().iterator();i.hasNext();)
				{
					final RunWrap runnable = i.next();
					if(runnable.activeTimeMillis() > timeoutMillis)
					{
						if(timedOut.size() >= maxToKill)
						{
							RunWrap leastWorstOffender=null;
							for(final RunWrap r : timedOut)
							{
								if((leastWorstOffender != null)
								&&(r.activeTimeMillis() < leastWorstOffender.activeTimeMillis()))
									leastWorstOffender=r;
							}
							if(leastWorstOffender!=null)
							{
								if(runnable.activeTimeMillis() < leastWorstOffender.activeTimeMillis())
									continue;
								else
									timedOut.remove(leastWorstOffender);
							}
						}
						timedOut.add(runnable);
						final Thread thread = runnable.getThread();
						killedOut.add(thread);
						i.remove();
					}
				}
			}
			catch(final Exception e)
			{ /**/ }
		}
		while(killedOut.size()>0)
		{
			final Thread thread = killedOut.remove();
			try
			{
				thread.interrupt();
				//CMLib.killThread(t,100,3);
			}
			catch(final Exception e)
			{ /**/ }
		}
		return timedOut;
	}
}
