rendered paste bodyIndex: /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<ClientBasePacket>();
_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<Integer> _charSlotMapping = new ArrayList<Integer>();
- 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<SelectionKey> it = _selector.selectedKeys().iterator();
+ Iterator<SelectionKey> 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<BasePacket> iter = sendMsgQueue.iterator();
+ Iterator<BasePacket> 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);