/*
* 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 2, 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, write to the Free Software
* Foundation, Inc., 59 Temple Place - Suite 330, Boston, MA
* 02111-1307, USA.
*
* http://www.gnu.org/copyleft/gpl.html
*/
package net.sf.l2j.gameserver;
import java.net.InetSocketAddress;
import java.nio.channels.SelectionKey;
import java.nio.channels.Selector;
import java.nio.channels.ServerSocketChannel;
import java.util.ArrayList;
import java.util.Collections;
import java.util.Comparator;
import java.util.Iterator;
import java.util.List;
import java.util.logging.Level;
import java.util.logging.Logger;
import net.sf.l2j.Config;
/**
* SelectorThreadManager
* This Thread will dispatch connections between the SelectorThreads
* @author -Wooden-
*
*/
public class SelectorThreadManager extends Thread
{
private static Logger _log = Logger.getLogger(SelectorThreadManager.class.getName());
/** The selector */
private Selector _selector;
/** push counter, currently counts messages in putbound queue,
* but better it count size of outbound queue...
*/
private int messageCounter;
private final String _hostname;
private final int _port;
private ArrayList<SelectorThread> _selectorThreads;
private int _threadCount;
public static int SELEC_INIT_NUMBER = 4;
public static int CONNECTIONS_PER_THREAD = 50;
private int _id = 0;
private static SelectorThreadManager _instance;
/**
* @return
*/
public static SelectorThreadManager getInstance()
{
return _instance;
}
public int getNextId()
{
_id++;
return _id;
}
SelectorThreadManager(String hostname, int port)
{
super("NIO SelectorMan");
_instance = this;
_threadCount = SELEC_INIT_NUMBER; //TODO:read this from config
_hostname = hostname;
_port = port;
_selectorThreads = new ArrayList<SelectorThread>(SELEC_INIT_NUMBER);
for(int i = 0; i < SELEC_INIT_NUMBER;i++)
{
_selectorThreads.add(new SelectorThread());
}
for(SelectorThread st : _selectorThreads)
{
st.start();
}
}
/** Main loop, see class description */
public void run() {
try {
// create selector
_selector = Selector.open();
// create a test server socket chanell
ServerSocketChannel ssc = ServerSocketChannel.open();
InetSocketAddress isa;
if ("*".equals(_hostname))
{
isa = new InetSocketAddress(Config.PORT_GAME);
_log.config("GameServer listening on all available IPs on Port "+_port);
}
else
{
isa = new InetSocketAddress(_hostname, Config.PORT_GAME);
_log.config("GameServer listening on IP: "+_hostname + " Port "+_port);
}
ssc.socket().bind(isa, 100);
ssc.configureBlocking(false);
ssc.register(_selector, SelectionKey.OP_ACCEPT);
// loop forever
for (;;) {
// reset counter
if (messageCounter == 0)
this.setPriority(Thread.NORM_PRIORITY); // idle
else
messageCounter = 0;
// check for shutdown
if (isInterrupted())
{
try { _selector.close(); } catch (Throwable t) { _log.log(Level.INFO, "", t); }
return;
}
// wait for read/write, timeout is to be on safe side,
// if waking up selector after a message push will fail
int numKeys = _selector.select(50L);
// check for shutdown
if (isInterrupted())
{
try { _selector.close(); } catch (Throwable t) { _log.log(Level.INFO, "", t); }
return;
}
if (numKeys == 0)
continue;
// iterate over selected keys
Iterator<SelectionKey> it = _selector.selectedKeys().iterator();
while (it.hasNext()) {
SelectionKey sk = it.next();
it.remove();
if (!sk.isValid())
continue;
// Obtain the interest of the key
int readyOps = sk.readyOps();
switch (readyOps)
{
case SelectionKey.OP_ACCEPT:
DispatchConnection(sk);
break;
default:
System.err.println("Manager: Impossible readyOps="+readyOps);
_log.severe("Manager: Impossible readyOps="+readyOps);
sk.cancel();
break;
}
}
}
} catch (Throwable t) {
_log.log(Level.SEVERE, "", t);
System.exit(1);
}
}
/**
* @param sk
*/
private void DispatchConnection(SelectionKey sk)
{
int minimumConnections = Integer.MAX_VALUE;
SelectorThread minimalThread = null;
int threadConnections = 0;
int totalConnections = 0;
int unlockedSelectorThreads = 0;
for(SelectorThread st : _selectorThreads)
{
threadConnections = st.getConnectionNumber();
totalConnections += threadConnections;
if(!st.isLocked())
{
if(threadConnections == 0)
{
st.acceptConnection(sk);
_log.info("Connection accepted by empty SelectorThread.");
return;
}
if(threadConnections < minimumConnections)
{
minimumConnections = threadConnections;
minimalThread = st;
}
unlockedSelectorThreads++;
}
}
if(minimalThread != null)
{
if(minimumConnections >= CONNECTIONS_PER_THREAD)
{
synchronized (_selectorThreads)
{
_selectorThreads.ensureCapacity(_selectorThreads.size()+1);
_selectorThreads.add(0, new SelectorThread());
_selectorThreads.get(0).start();
_selectorThreads.get(0).acceptConnection(sk);
}
_log.info("Connection added to newly created Thread.");
_threadCount++;
}
else
{
minimalThread.acceptConnection(sk);
_log.info("Connection added to SelectorThread with the least connections.");
}
}
else
{
_selectorThreads.get(0).acceptConnection(sk);
}
if((totalConnections / unlockedSelectorThreads < CONNECTIONS_PER_THREAD / 2) && unlockedSelectorThreads > SELEC_INIT_NUMBER)
{
SelectorCleanup();
}
}
private void SelectorCleanup()
{
_log.info("Beginning Cleanup.");
int total = 0;
int unlocked = 0;
int goal = CONNECTIONS_PER_THREAD / 2;
Collections.sort(_selectorThreads, threadComparator);
for(SelectorThread st : _selectorThreads)
{
total = st.getConnectionNumber();
if(!st.isLocked())
{
unlocked++;
}
}
int mean = total / unlocked;
_log.info("Unlocked "+unlocked+" Thread(s).");
_log.info("Connections: "+total+", Mean: "+mean+" and Target: "+goal);
while(mean > goal && unlocked > SELEC_INIT_NUMBER)
{
if(_selectorThreads.get(0).getConnectionNumber() == 0)
{
if(!_selectorThreads.get(0).isLocked())
{
unlocked--;
}
_selectorThreads.get(0).setLockedState(true);
_selectorThreads.get(0).interrupt();
_selectorThreads.remove(0);
_log.info("Deleted SelectorThread with 0 Connections.");
}
else
{
_selectorThreads.get(0).setLockedState(true);
_log.info("Locked a Thread.");
unlocked--;
}
mean = total / unlocked;
_log.info("After Iteration, Unlocked: "+unlocked+" Mean: "+mean+" and Target: "+goal);
}
_log.info("Cleanup finished.");
}
private static final Comparator<SelectorThread> threadComparator = new Comparator<SelectorThread>()
{
public int compare(SelectorThread st1, SelectorThread st2)
{
return (st1.getConnectionNumber() < st2.getConnectionNumber() ? -1 : st1.getConnectionNumber() == st2.getConnectionNumber() ? 0 : 1);
}
};
public List<String> getStats()
{
List<String> stats = new ArrayList<String>(2+_selectorThreads.size());
stats.add("SelectorThreadManager");
stats.add("Total Threads: "+_selectorThreads.size());
for(SelectorThread st : _selectorThreads)
{
if(st.isLocked())
{
stats.add("Thread "+st.getName()+" (locked):");
}
else
{
stats.add("Thread "+st.getName()+":");
}
stats.add("Connections: "+st.getConnectionNumber()+" OutBoundQueueSize: "+st.outboundQueueSize());
}
return stats;
}
}