Index: /java/net/sf/l2j/gameserver/Connection.java =================================================================== --- /java/net/sf/l2j/gameserver/Connection.java (revision 1327) +++ /java/net/sf/l2j/gameserver/Connection.java (working copy) @@ -66,11 +66,14 @@ int writeTimeStamp; /** The client that owns this connection */ final ClientThread _client; + + private SelectorThread _selectorthread; - public Connection(ClientThread client, Socket socket, byte[] cryptKey) + public Connection(ClientThread client, Socket socket, byte[] cryptKey, SelectorThread selectorT) { _client = client; _csocket = socket; + _selectorthread = selectorT; _chanell = _csocket.getChannel(); _receivedMsgQueue = new ConcurrentLinkedQueue(); _sendMsgQueue = new BasePacketQueue(); @@ -88,7 +91,7 @@ * Notifies all threads, that wait() on this Connection */ public synchronized void addReceivedMsg(ByteBuffer buf) { - if (Config.ASSERT) assert Thread.currentThread() == SelectorThread.instance; + if (Config.ASSERT) assert Thread.currentThread() == _selectorthread; ClientBasePacket pkt = PacketHandler.handlePacket(buf, _client); if (pkt != null) { if (pkt.getPriority() == TaskPriority.PR_URGENT) @@ -107,7 +110,7 @@ */ synchronized ClientBasePacket getNextReceivedMsg() { - if (Config.ASSERT) assert Thread.currentThread() != SelectorThread.instance; + if (Config.ASSERT) assert Thread.currentThread() != _selectorthread; if (_receivedMsgQueue.isEmpty()) return null; return _receivedMsgQueue.remove(); @@ -135,13 +138,13 @@ } WrappedMessage msg = new WrappedMessage(data, this); - SelectorThread.instance.sendMessage(msg); + _selectorthread.sendMessage(msg); } public void sendPacket(ServerBasePacket bp) { bp = (ServerBasePacket)bp.setConnection(this); - SelectorThread.instance.sendMessage(bp); + _selectorthread.sendMessage(bp); } public void activateCryptKey() @@ -171,14 +174,23 @@ */ public void close() { - try { - if (_csocket != null && _csocket.isConnected()) + try + { + if (_csocket != null && !_csocket.isClosed()) + { _csocket.close(); - } catch (IOException e) {} - try { + _selectorthread.decreaseConnectionNumber(); + } + } + catch (IOException e) + { } + try + { if (_client.getActiveChar() != null) + { _client.onDisconnect(); - } catch (Throwable t) {_log.log(Level.WARNING, "", t);} + } + } catch (Throwable t) {_log.log(Level.WARNING, "", t);} } /** @@ -186,10 +198,15 @@ */ public void onlyClose() { - try { - if (_csocket != null && _csocket.isConnected()) + try + { + if (_csocket != null && !_csocket.isClosed()) + { _csocket.close(); - } catch (IOException e) {} + _selectorthread.decreaseConnectionNumber(); + } + } + catch (IOException e) {} } /** @@ -278,4 +295,12 @@ return number; } + + /** + * @return + */ + public boolean isSocketOpen() + { + return !_csocket.isClosed(); + } } Index: /java/net/sf/l2j/gameserver/ClientThread.java =================================================================== --- /java/net/sf/l2j/gameserver/ClientThread.java (revision 1327) +++ /java/net/sf/l2j/gameserver/ClientThread.java (working copy) @@ -45,6 +45,8 @@ { protected static Logger _log = Logger.getLogger(ClientThread.class.getName()); + private SelectorThread _selectorThread; + public static int devCharId; class AutoSaveTask implements Runnable @@ -84,16 +86,17 @@ private ArrayList _charSlotMapping = new ArrayList(); - public static ClientThread create(Socket socket) + public static ClientThread create(Socket socket,SelectorThread selectorThread) { - final ClientThread client = new ClientThread(socket); + final ClientThread client = new ClientThread(socket,selectorThread); return client; } - public ClientThread(Socket socket) + public ClientThread(Socket socket,SelectorThread selectorThread) // { _sessionId = 0x12345678; - _connection = new Connection(this, socket, _cryptkey); + _connection = new Connection(this, socket, _cryptkey, selectorThread); + _selectorThread = selectorThread; _autoSaveInDB = ClientScheduler.getInstance().scheduleLowAtFixedRate( new AutoSaveTask(), 5*60*1000L, 15*60*1000L ); @@ -112,11 +115,12 @@ // notify the world about our disconnect player.deleteMe(); - try { + try + { saveCharToDisk(player); - } catch (Exception e2) { /* ignore any problems here */ } + } + catch (Exception e2) { /* ignore any problems here */ } } - _connection.close(); } catch (Exception e1) @@ -185,16 +189,16 @@ statement.execute(); statement.close(); - statement = con.prepareStatement("DELETE FROM character_macroses WHERE char_obj_id=?"); - statement.setInt(1, objid); - statement.execute(); - statement.close(); - + statement = con.prepareStatement("DELETE FROM character_macroses WHERE char_obj_id=?"); + statement.setInt(1, objid); + statement.execute(); + statement.close(); + statement = con.prepareStatement("DELETE pets FROM pets, items WHERE pets.item_obj_id=items.object_id AND items.owner_id=?"); statement.setInt(1, objid); statement.execute(); statement.close(); - + statement = con.prepareStatement("DELETE FROM items WHERE owner_id=?"); statement.setInt(1, objid); statement.execute(); @@ -210,16 +214,16 @@ statement.execute(); statement.close(); - statement = con.prepareStatement("DELETE FROM character_recipebook WHERE char_id=?"); - statement.setInt(1, objid); - statement.execute(); - statement.close(); - + statement = con.prepareStatement("DELETE FROM character_recipebook WHERE char_id=?"); + statement.setInt(1, objid); + statement.execute(); + statement.close(); + statement = con.prepareStatement("DELETE FROM merchant_lease WHERE player_id=?"); statement.setInt(1, objid); statement.execute(); statement.close(); - + statement = con.prepareStatement("DELETE FROM character_friends WHERE char_id=?"); statement.setInt(1, objid); statement.execute(); @@ -229,11 +233,11 @@ statement.setInt(1, objid); statement.execute(); statement.close(); - + } catch (Exception e) { - _log.log(Level.WARNING, "data error on delete char:", e); + _log.log(Level.WARNING, "data error on delete char:", e); } finally { @@ -260,10 +264,10 @@ restoreRecipeBook(character); L2MerchantInstance.playerEnter(character); - // preinit some values for each login - character.setRunning(); // running is default - character.standUp(); // standing is default - + // preinit some values for each login + character.setRunning(); // running is default + character.standUp(); // standing is default + character.updateStats(); character.updateKarma(1); character.setOnlineStatus(true); @@ -305,7 +309,7 @@ } catch (Exception e) { - _log.log(Level.WARNING, "could not store skills:", e); + _log.log(Level.WARNING, "could not store skills:", e); } finally { Index: /java/net/sf/l2j/gameserver/SelectorThread.java =================================================================== --- /java/net/sf/l2j/gameserver/SelectorThread.java (revision 1327) +++ /java/net/sf/l2j/gameserver/SelectorThread.java (working copy) @@ -19,9 +19,9 @@ package net.sf.l2j.gameserver; import java.io.IOException; -import java.net.InetSocketAddress; import java.nio.ByteBuffer; import java.nio.ByteOrder; +import java.nio.channels.CancelledKeyException; import java.nio.channels.SelectionKey; import java.nio.channels.Selector; import java.nio.channels.ServerSocketChannel; @@ -93,17 +93,28 @@ /** The selector */ private Selector _selector; - private final String _hostname; - private final int _port; - /** push counter, currently counts messages in putbound queue, * but better it count size of outbound queue... */ private int msgCounter; + + private int _connections = 0; + + private boolean _isLocked = false; // :p // just for testing HashMap clients = new HashMap(); + public boolean isLocked() + { + return _isLocked; + } + + public void setLockedState(boolean isLocked) + { + _isLocked = isLocked; + } + class DisconnectionTask implements Runnable { private Connection _connection; @@ -116,13 +127,22 @@ closeClient(_connection,true); } } + + public int getConnectionNumber() + { + return _connections; + } + + public void decreaseConnectionNumber() + { + _connections--; + _log.warning("Decreased connexion number("+_connections+")"); + } - public SelectorThread(String hostname, int port) { - super("NIO Selector"); + public SelectorThread() { + super("NIO Selector "+SelectorThreadManager.getInstance().getNextId()); if (Config.ASSERT) assert instance == null; instance = this; - this._hostname = hostname; - this._port = port; this.setDaemon(true); this.setPriority(Thread.NORM_PRIORITY+1); // write buffers (HeapByteBuffer is much faster for our perposes) @@ -189,22 +209,6 @@ try { // create selector _selector = Selector.open(); - // create a test server socket chanell - ServerSocketChannel ssc = ServerSocketChannel.open(); - InetSocketAddress isa; - if ("*".equals(_hostname)) - { - isa = new InetSocketAddress(_port); - _log.config("GameServer listening on all available IPs on Port "+_port); - } - else - { - isa = new InetSocketAddress(_hostname, _port); - _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 (;;) { // check if we have messages to pack and send @@ -217,53 +221,53 @@ // check for shutdown if (isInterrupted()) { - try { _selector.close(); } catch (Throwable t) { _log.log(Level.INFO, "", t); } + 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); + int numKeys = _selector.select(50L); // check for shutdown if (isInterrupted()) { - try { _selector.close(); } catch (Throwable t) { _log.log(Level.INFO, "", t); } + try { _selector.close(); } catch (Throwable t) { _log.log(Level.INFO, "", t); } return; } if (numKeys == 0) continue; // iterate over selected keys - Iterator it = _selector.selectedKeys().iterator(); + Iterator it = _selector.selectedKeys().iterator(); while (it.hasNext()) { - SelectionKey sk = it.next(); + SelectionKey sk = it.next(); + synchronized(sk) + { it.remove(); if (!sk.isValid()) continue; // Obtain the interest of the key - int readyOps = sk.readyOps(); - switch (readyOps) { - case SelectionKey.OP_WRITE: - writeData(sk); - break; - case SelectionKey.OP_READ: - readData(sk); - break; - case SelectionKey.OP_WRITE | SelectionKey.OP_READ: - writeData(sk); - readData(sk); - break; - case SelectionKey.OP_ACCEPT: - acceptConnection(sk); - break; - default: - System.err.println("Impossible readyOps="+readyOps); - _log.severe("Impossible readyOps="+readyOps); - sk.cancel(); - break; - } + int readyOps = sk.readyOps(); + switch (readyOps) { + case SelectionKey.OP_WRITE: + writeData(sk); + break; + case SelectionKey.OP_READ: + readData(sk); + break; + case SelectionKey.OP_WRITE | SelectionKey.OP_READ: + writeData(sk); + readData(sk); + break; + default: + System.err.println("Impossible readyOps="+readyOps); + _log.severe("Impossible readyOps="+readyOps); + sk.cancel(); + break; + } + } } } } catch (Throwable t) { - _log.log(Level.SEVERE, "", t); + _log.log(Level.SEVERE, "", t); System.exit(1); } } @@ -307,11 +311,11 @@ releaseBuffer(b); } sk.interestOps(sk.interestOps() & ~SelectionKey.OP_WRITE); - } catch (IOException e) { - _log.info("Error on network write, player disconnected?"); - closeClient(con); + } catch (IOException e) { + _log.info("Error on network write, player disconnected?"); + closeClient(con); } catch (Throwable t) { - _log.log(Level.INFO, "", t); + _log.log(Level.INFO, "", t); closeClient(con); } } @@ -403,7 +407,7 @@ _log.info(e.toString()); closeClient(con); } catch (Throwable t) { - _log.log(Level.INFO, "", t); + _log.log(Level.INFO, "", t); closeClient(con); } } @@ -415,11 +419,11 @@ * @param buf buffer with packet * @param sz size of the packet */ - private void parse(Connection con, ByteBuffer buf) - throws Throwable - { - try - { + private void parse(Connection con, ByteBuffer buf) + throws Throwable + { + try + { int sz = buf.getShort() & 0xFFFF; sz -= 2; ByteBuffer b = (ByteBuffer)buf.slice().limit(sz); @@ -435,12 +439,12 @@ con.readBuffer = null; throw e; } - catch (Throwable t) { - _log.log(Level.SEVERE, "", t); - releaseBuffer(buf); - con.readBuffer = null; - throw t; - } + catch (Throwable t) { + _log.log(Level.SEVERE, "", t); + releaseBuffer(buf); + con.readBuffer = null; + throw t; + } } /** Pack (encode) a message into a network buffer. @@ -458,7 +462,7 @@ return false; } catch (Exception e) { closeClient(msg.getConnection()); - return false; + return false; } if (Config.ASSERT) assert sharedWriteBuffer.position() == 0; if (Config.ASSERT) assert sharedWriteBuffer.limit() >= 3; @@ -466,21 +470,31 @@ } /** Accepts connection, creates a new chanell, client, etc */ - private void acceptConnection(SelectionKey sk) { - try { - if (sk.isAcceptable()) { - SocketChannel c = ((ServerSocketChannel) sk.channel()).accept(); - if (c != null) { - c.configureBlocking(false); - c.register(sk.selector(), SelectionKey.OP_READ|SelectionKey.OP_WRITE); - SelectionKey sk2 = c.keyFor(sk.selector()); - ClientThread client = ClientThread.create(c.socket()); - sk2.attach(client.getConnection()); - } - } - } catch (Throwable t) { - _log.log(Level.INFO, "", t); - } + public void acceptConnection(SelectionKey sk) { + synchronized (_selector) + { + try + { + if (sk.isAcceptable()) + { + SocketChannel c = ((ServerSocketChannel) sk.channel()).accept(); + if (c != null) + { + c.configureBlocking(false); + c.register(_selector, SelectionKey.OP_READ|SelectionKey.OP_WRITE); + SelectionKey sk2 = c.keyFor(_selector); + ClientThread client = ClientThread.create(c.socket(),this); + sk2.attach(client.getConnection()); + _connections++; + _log.warning("Connexion accepted! (connexions: "+_connections+")"); + } + } + } + catch (Throwable t) + { + _log.log(Level.INFO, "", t); + } + } } /** Send a message to client. @@ -523,7 +537,7 @@ private void processOutboudQueue() { if (sendMsgQueue.isEmpty()) return; - Iterator iter = sendMsgQueue.iterator(); + Iterator iter = sendMsgQueue.iterator(); while (iter.hasNext()) { ServerBasePacket msg = (ServerBasePacket)iter.next(); @@ -604,7 +618,11 @@ } } } - try { c.close(); } catch (Exception dummy) {} + if(!c.isSocketOpen()) + { + _connections--; + } + try { c.close(); } catch (Exception dummy) {} } private void closeClient(Connection c) { @@ -628,6 +646,18 @@ return; } } + if(!c.isSocketOpen()) + { + _connections--; + } try { c.close(); } catch (Exception dummy) {} } + + /** + * @return + */ + public int outboundQueueSize() + { + return sendMsgQueue.size(); + } } Index: /java/net/sf/l2j/gameserver/handler/admincommandhandlers/AdminTest.java =================================================================== --- /java/net/sf/l2j/gameserver/handler/admincommandhandlers/AdminTest.java (revision 1327) +++ /java/net/sf/l2j/gameserver/handler/admincommandhandlers/AdminTest.java (working copy) @@ -30,6 +30,7 @@ import net.sf.l2j.Config; import net.sf.l2j.gameserver.ClientScheduler; +import net.sf.l2j.gameserver.SelectorThreadManager; import net.sf.l2j.gameserver.Universe; import net.sf.l2j.gameserver.handler.IAdminCommandHandler; import net.sf.l2j.gameserver.idfactory.IdFactory; @@ -61,9 +62,12 @@ { if (command.equals("admin_stats")) { - String[] out = ClientScheduler.getInstance().getStats(); + for (String line : ClientScheduler.getInstance().getStats()) + { + activeChar.sendMessage(line); + } - for (String line : out) + for (String line : SelectorThreadManager.getInstance().getStats()) { activeChar.sendMessage(line); } Index: /java/net/sf/l2j/gameserver/GameServer.java =================================================================== --- /java/net/sf/l2j/gameserver/GameServer.java (revision 1327) +++ /java/net/sf/l2j/gameserver/GameServer.java (working copy) @@ -132,7 +132,7 @@ public class GameServer { private static final Logger _log = Logger.getLogger(GameServer.class.getName()); - private final SelectorThread _selectorThread; + private final SelectorThreadManager _selectorThread; private final TradeController _tradeController; private final SkillTable _skillTable; private final SkillTreeTable _skillTreeTable; @@ -189,7 +189,7 @@ return _loginController; } - public SelectorThread getSelectorThread() + public SelectorThreadManager getSelectorThread() { return _selectorThread; } @@ -218,7 +218,7 @@ } ClientScheduler.getInstance(); - _selectorThread = new SelectorThread(Config.GAMESERVER_HOSTNAME, Config.PORT_GAME); + _selectorThread = new SelectorThreadManager(Config.GAMESERVER_HOSTNAME, Config.PORT_GAME); _selectorThread.start(); _log.config("Maximum Numbers of Connected Players: " + Config.MAXIMUM_ONLINE_USERS);