/*
* This program is free software: you can redistribute it and/or modify it under
* the terms of the GNU General Public License as published by the Free Software
* Foundation, either version 3 of the License, or (at your option) any later
* version.
*
* This program is distributed in the hope that it will be useful, but WITHOUT
* ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS
* FOR A PARTICULAR PURPOSE. See the GNU General Public License for more
* details.
*
* You should have received a copy of the GNU General Public License along with
* this program. If not, see .
*/
package net.sf.l2j.gameserver;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.RejectedExecutionException;
import java.util.concurrent.ScheduledFuture;
import java.util.concurrent.ScheduledThreadPoolExecutor;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.logging.Logger;
import javolution.text.TextBuilder;
import net.sf.l2j.Config;
import net.sf.l2j.gameserver.network.L2GameClient;
import org.mmocore.network.ReceivablePacket;
/**
*
This class is made to handle all the ThreadPools used in L2j.
* Scheduled Tasks can either be sent to a {@link #_generalScheduledThreadPool "general"} or {@link #_effectsScheduledThreadPool "effects"} {@link ScheduledThreadPoolExecutor ScheduledThreadPool}:
* The "effects" one is used for every effects (skills, hp/mp regen ...) while the "general" one is used for
* everything else that needs to be scheduled.
* There also is an {@link #_aiScheduledThreadPool "ai"} {@link ScheduledThreadPoolExecutor ScheduledThreadPool} used for AI Tasks.
* Tasks can be sent to {@link ScheduledThreadPoolExecutor ScheduledThreadPool} either with:
*
* - {@link #scheduleEffect(Runnable, long)} : for effects Tasks that needs to be executed only once.
* - {@link #scheduleGeneral(Runnable, long)} : for scheduled Tasks that needs to be executed once.
* - {@link #scheduleAi(Runnable, long)} : for AI Tasks that needs to be executed once
*
* or
*
* - {@link #scheduleEffectAtFixedRate(Runnable, long, long)(Runnable, long)} : for effects Tasks that needs to be executed periodicaly.
* - {@link #scheduleGeneralAtFixedRate(Runnable, long, long)(Runnable, long)} : for scheduled Tasks that needs to be executed periodicaly.
* - {@link #scheduleAiAtFixedRate(Runnable, long, long)(Runnable, long)} : for AI Tasks that needs to be executed periodicaly
*
*
* For all Tasks that should be executed with no delay asynchronously in a ThreadPool there also are usual {@link ThreadPoolExecutor ThreadPools}
* that can grow/shrink according to their load.:
*
* - {@link #_generalPacketsThreadPool GeneralPackets} where most packets handler are executed.
* - {@link #_ioPacketsThreadPool I/O Packets} where all the i/o packets are executed.
* - There will be an AI ThreadPool where AI events should be executed
* - A general ThreadPool where everything else that needs to run asynchronously with no delay should be executed ({@link net.sf.l2j.gameserver.model.actor.knownlist KnownList} updates, SQL updates/inserts...)?
*
*
* @author -Wooden-
*
*/
public class ThreadPoolManager
{
protected static final Logger _log = Logger.getLogger(ThreadPoolManager.class.getName());
private static ThreadPoolManager _instance;
public ScheduledThreadPoolExecutor _effectsScheduledThreadPool;
private ScheduledThreadPoolExecutor _generalScheduledThreadPool;
private ThreadPoolExecutor _generalPacketsThreadPool;
private ThreadPoolExecutor _ioPacketsThreadPool;
// will be really used in the next AI implementation.
private ThreadPoolExecutor _aiThreadPool;
private ThreadPoolExecutor _generalThreadPool;
// temp
private ScheduledThreadPoolExecutor _aiScheduledThreadPool;
/** temp workaround for VM issue */
private static final long MAX_DELAY = Long.MAX_VALUE/1000000/2;
private boolean _shutdown;
public static ThreadPoolManager getInstance()
{
if(_instance == null)
{
_instance = new ThreadPoolManager();
}
return _instance;
}
private ThreadPoolManager()
{
_effectsScheduledThreadPool = new ScheduledThreadPoolExecutor(Config.THREAD_P_EFFECTS, new PriorityThreadFactory("EffectsSTPool", Thread.NORM_PRIORITY));
_generalScheduledThreadPool = new ScheduledThreadPoolExecutor(Config.THREAD_P_GENERAL, new PriorityThreadFactory("GerenalSTPool", Thread.NORM_PRIORITY));
_ioPacketsThreadPool = new ThreadPoolExecutor(Config.IO_PACKET_THREAD_CORE_SIZE, Integer.MAX_VALUE,
5L, TimeUnit.SECONDS,
new LinkedBlockingQueue(),
new PriorityThreadFactory("I/O Packet Pool",Thread.NORM_PRIORITY+1));
_generalPacketsThreadPool = new ThreadPoolExecutor(Config.GENERAL_PACKET_THREAD_CORE_SIZE, Config.GENERAL_PACKET_THREAD_CORE_SIZE+2,
15L, TimeUnit.SECONDS,
new LinkedBlockingQueue(),
new PriorityThreadFactory("Normal Packet Pool",Thread.NORM_PRIORITY+1));
_generalThreadPool = new ThreadPoolExecutor(Config.GENERAL_THREAD_CORE_SIZE, Config.GENERAL_THREAD_CORE_SIZE+2,
5L, TimeUnit.SECONDS,
new LinkedBlockingQueue(),
new PriorityThreadFactory("General Pool",Thread.NORM_PRIORITY));
// will be really used in the next AI implementation.
_aiThreadPool = new ThreadPoolExecutor(1, Config.AI_MAX_THREAD,
10L, TimeUnit.SECONDS,
new LinkedBlockingQueue());
_aiScheduledThreadPool = new ScheduledThreadPoolExecutor(Config.AI_MAX_THREAD, new PriorityThreadFactory("AISTPool", Thread.NORM_PRIORITY));
}
public static long validateDelay(long delay)
{
if (delay < 0)
{
delay = 0;
}
else if (delay > MAX_DELAY)
{
delay = MAX_DELAY;
}
return delay;
}
public ScheduledFuture> scheduleEffect(Runnable r, long delay)
{
try
{
delay = ThreadPoolManager.validateDelay(delay);
return _effectsScheduledThreadPool.schedule(r, delay, TimeUnit.MILLISECONDS);
}
catch (RejectedExecutionException e)
{
return null;
}
}
public ScheduledFuture> scheduleEffectAtFixedRate(Runnable r, long initial, long delay)
{
try
{
delay = ThreadPoolManager.validateDelay(delay);
initial = ThreadPoolManager.validateDelay(initial);
return _effectsScheduledThreadPool.scheduleAtFixedRate(r, initial, delay, TimeUnit.MILLISECONDS);
}
catch (RejectedExecutionException e)
{
return null; /* shutdown, ignore */
}
}
public ScheduledFuture> scheduleGeneral(Runnable r, long delay)
{
try
{
delay = ThreadPoolManager.validateDelay(delay);
return _generalScheduledThreadPool.schedule(r, delay, TimeUnit.MILLISECONDS);
}
catch (RejectedExecutionException e)
{
return null; /* shutdown, ignore */
}
}
public ScheduledFuture> scheduleGeneralAtFixedRate(Runnable r, long initial, long delay)
{
try
{
delay = ThreadPoolManager.validateDelay(delay);
initial = ThreadPoolManager.validateDelay(initial);
return _generalScheduledThreadPool.scheduleAtFixedRate(r, initial, delay, TimeUnit.MILLISECONDS);
}
catch (RejectedExecutionException e)
{
return null; /* shutdown, ignore */
}
}
public ScheduledFuture> scheduleAi(Runnable r, long delay)
{
try
{
delay = ThreadPoolManager.validateDelay(delay);
return _aiScheduledThreadPool.schedule(r, delay, TimeUnit.MILLISECONDS);
}
catch (RejectedExecutionException e)
{
return null; /* shutdown, ignore */
}
}
public ScheduledFuture> scheduleAiAtFixedRate(Runnable r, long initial, long delay)
{
try
{
delay = ThreadPoolManager.validateDelay(delay);
initial = ThreadPoolManager.validateDelay(initial);
return _aiScheduledThreadPool.scheduleAtFixedRate(r, initial, delay, TimeUnit.MILLISECONDS);
}
catch (RejectedExecutionException e)
{
return null; /* shutdown, ignore */
}
}
public void executePacket(ReceivablePacket pkt)
{
_generalPacketsThreadPool.execute(pkt);
}
public void executeIOPacket(ReceivablePacket pkt)
{
_ioPacketsThreadPool.execute(pkt);
}
public void executeTask(Runnable r)
{
_generalThreadPool.execute(r);
}
public void executeAi(Runnable r)
{
_aiThreadPool.execute(r);
}
public String[] getStats()
{
return new String[] {
"STP:",
" + Effects:",
" |- ActiveThreads: "+_effectsScheduledThreadPool.getActiveCount(),
" |- getCorePoolSize: "+_effectsScheduledThreadPool.getCorePoolSize(),
" |- PoolSize: "+_effectsScheduledThreadPool.getPoolSize(),
" |- MaximumPoolSize: "+_effectsScheduledThreadPool.getMaximumPoolSize(),
" |- CompletedTasks: "+_effectsScheduledThreadPool.getCompletedTaskCount(),
" |- ScheduledTasks: "+(_effectsScheduledThreadPool.getTaskCount() - _effectsScheduledThreadPool.getCompletedTaskCount()),
" | -------",
" + General:",
" |- ActiveThreads: "+_generalScheduledThreadPool.getActiveCount(),
" |- getCorePoolSize: "+_generalScheduledThreadPool.getCorePoolSize(),
" |- PoolSize: "+_generalScheduledThreadPool.getPoolSize(),
" |- MaximumPoolSize: "+_generalScheduledThreadPool.getMaximumPoolSize(),
" |- CompletedTasks: "+_generalScheduledThreadPool.getCompletedTaskCount(),
" |- ScheduledTasks: "+(_generalScheduledThreadPool.getTaskCount() - _generalScheduledThreadPool.getCompletedTaskCount()),
" | -------",
" + AI:",
" |- ActiveThreads: "+_aiScheduledThreadPool.getActiveCount(),
" |- getCorePoolSize: "+_aiScheduledThreadPool.getCorePoolSize(),
" |- PoolSize: "+_aiScheduledThreadPool.getPoolSize(),
" |- MaximumPoolSize: "+_aiScheduledThreadPool.getMaximumPoolSize(),
" |- CompletedTasks: "+_aiScheduledThreadPool.getCompletedTaskCount(),
" |- ScheduledTasks: "+(_aiScheduledThreadPool.getTaskCount() - _aiScheduledThreadPool.getCompletedTaskCount()),
"TP:",
" + Packets:",
" |- ActiveThreads: "+_generalPacketsThreadPool.getActiveCount(),
" |- getCorePoolSize: "+_generalPacketsThreadPool.getCorePoolSize(),
" |- MaximumPoolSize: "+_generalPacketsThreadPool.getMaximumPoolSize(),
" |- LargestPoolSize: "+_generalPacketsThreadPool.getLargestPoolSize(),
" |- PoolSize: "+_generalPacketsThreadPool.getPoolSize(),
" |- CompletedTasks: "+_generalPacketsThreadPool.getCompletedTaskCount(),
" |- QueuedTasks: "+_generalPacketsThreadPool.getQueue().size(),
" | -------",
" + I/O Packets:",
" |- ActiveThreads: "+_ioPacketsThreadPool.getActiveCount(),
" |- getCorePoolSize: "+_ioPacketsThreadPool.getCorePoolSize(),
" |- MaximumPoolSize: "+_ioPacketsThreadPool.getMaximumPoolSize(),
" |- LargestPoolSize: "+_ioPacketsThreadPool.getLargestPoolSize(),
" |- PoolSize: "+_ioPacketsThreadPool.getPoolSize(),
" |- CompletedTasks: "+_ioPacketsThreadPool.getCompletedTaskCount(),
" |- QueuedTasks: "+_ioPacketsThreadPool.getQueue().size(),
" | -------",
" + General Tasks:",
" |- ActiveThreads: "+_generalThreadPool.getActiveCount(),
" |- getCorePoolSize: "+_generalThreadPool.getCorePoolSize(),
" |- MaximumPoolSize: "+_generalThreadPool.getMaximumPoolSize(),
" |- LargestPoolSize: "+_generalThreadPool.getLargestPoolSize(),
" |- PoolSize: "+_generalThreadPool.getPoolSize(),
" |- CompletedTasks: "+_generalThreadPool.getCompletedTaskCount(),
" |- QueuedTasks: "+_generalThreadPool.getQueue().size(),
" | -------",
" + AI:",
" |- Not Done"
};
}
private class PriorityThreadFactory implements ThreadFactory
{
private int _prio;
private String _name;
private AtomicInteger _threadNumber = new AtomicInteger(1);
private ThreadGroup _group;
public PriorityThreadFactory(String name, int prio)
{
_prio = prio;
_name = name;
_group = new ThreadGroup(_name);
}
/* (non-Javadoc)
* @see java.util.concurrent.ThreadFactory#newThread(java.lang.Runnable)
*/
public Thread newThread(Runnable r)
{
Thread t = new Thread(_group,r);
t.setName(_name+"-"+_threadNumber.getAndIncrement());
t.setPriority(_prio);
return t;
}
public ThreadGroup getGroup()
{
return _group;
}
}
/**
*
*/
public void shutdown()
{
_shutdown = true;
try
{
_effectsScheduledThreadPool.awaitTermination(1,TimeUnit.SECONDS);
_generalScheduledThreadPool.awaitTermination(1,TimeUnit.SECONDS);
_generalPacketsThreadPool.awaitTermination(1,TimeUnit.SECONDS);
_ioPacketsThreadPool.awaitTermination(1,TimeUnit.SECONDS);
_generalThreadPool.awaitTermination(1,TimeUnit.SECONDS);
_aiThreadPool.awaitTermination(1,TimeUnit.SECONDS);
_effectsScheduledThreadPool.shutdown();
_generalScheduledThreadPool.shutdown();
_generalPacketsThreadPool.shutdown();
_ioPacketsThreadPool.shutdown();
_generalThreadPool.shutdown();
_aiThreadPool.shutdown();
_log.info("All ThreadPools are now stoped");
}
catch (InterruptedException e)
{
// TODO Auto-generated catch block
e.printStackTrace();
}
}
public boolean isShutdown()
{
return _shutdown;
}
/**
*
*/
public void purge()
{
_effectsScheduledThreadPool.purge();
_generalScheduledThreadPool.purge();
_aiScheduledThreadPool.purge();
_ioPacketsThreadPool.purge();
_generalPacketsThreadPool.purge();
_generalThreadPool.purge();
_aiThreadPool.purge();
}
/**
*
*/
public String getPacketStats()
{
TextBuilder tb = new TextBuilder();
ThreadFactory tf = _generalPacketsThreadPool.getThreadFactory();
if (tf instanceof PriorityThreadFactory)
{
tb.append("General Packet Thread Pool:\r\n");
tb.append("Tasks in the queue: "+_generalPacketsThreadPool.getQueue().size()+"\r\n");
tb.append("Showing threads stack trace:\r\n");
PriorityThreadFactory ptf = (PriorityThreadFactory) tf;
int count = ptf.getGroup().activeCount();
Thread[] threads = new Thread[count+2];
ptf.getGroup().enumerate(threads);
tb.append("There should be "+count+" Threads\r\n");
for(Thread t : threads)
{
if(t == null)
continue;
tb.append(t.getName()+"\r\n");
for(StackTraceElement ste :t.getStackTrace())
{
tb.append(ste.toString());
tb.append("\r\n");
}
}
}
tb.append("Packet Tp stack traces printed.\r\n");
return tb.toString();
}
public String getIOPacketStats()
{
TextBuilder tb = new TextBuilder();
ThreadFactory tf = _ioPacketsThreadPool.getThreadFactory();
if (tf instanceof PriorityThreadFactory)
{
tb.append("I/O Packet Thread Pool:\r\n");
tb.append("Tasks in the queue: "+_ioPacketsThreadPool.getQueue().size()+"\r\n");
tb.append("Showing threads stack trace:\r\n");
PriorityThreadFactory ptf = (PriorityThreadFactory) tf;
int count = ptf.getGroup().activeCount();
Thread[] threads = new Thread[count+2];
ptf.getGroup().enumerate(threads);
tb.append("There should be "+count+" Threads\r\n");
for(Thread t : threads)
{
if(t == null)
continue;
tb.append(t.getName()+"\r\n");
for(StackTraceElement ste :t.getStackTrace())
{
tb.append(ste.toString());
tb.append("\r\n");
}
}
}
tb.append("Packet Tp stack traces printed.\r\n");
return tb.toString();
}
public String getGeneralStats()
{
TextBuilder tb = new TextBuilder();
ThreadFactory tf = _generalThreadPool.getThreadFactory();
if (tf instanceof PriorityThreadFactory)
{
tb.append("General Thread Pool:\r\n");
tb.append("Tasks in the queue: "+_generalThreadPool.getQueue().size()+"\r\n");
tb.append("Showing threads stack trace:\r\n");
PriorityThreadFactory ptf = (PriorityThreadFactory) tf;
int count = ptf.getGroup().activeCount();
Thread[] threads = new Thread[count+2];
ptf.getGroup().enumerate(threads);
tb.append("There should be "+count+" Threads\r\n");
for(Thread t : threads)
{
if(t == null)
continue;
tb.append(t.getName()+"\r\n");
for(StackTraceElement ste :t.getStackTrace())
{
tb.append(ste.toString());
tb.append("\r\n");
}
}
}
tb.append("Packet Tp stack traces printed.\r\n");
return tb.toString();
}
}