Server keeps running out of memory

Hi dere everyone,

I've got a little issue, I'm working on a server network and in order for the servers to communicate I pass messaged between them using the following class:

Code:java

package net.invasioncraft.api.network;
 
import java.io.DataInputStream;
import java.io.DataOutputStream;
import java.io.IOException;
import java.lang.reflect.Method;
import java.net.Socket;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
 
import org.bukkit.Bukkit;
 
import net.invasioncraft.api.icAPI;
import net.invasioncraft.api.config.VirtualConfigFile;
import net.invasioncraft.api.network.events.DatabridgeCloseEvent;
import net.invasioncraft.api.network.events.DatabridgePackageReceiveEvent;
import net.invasioncraft.api.network.events.DatabridgePackageSendEvent;
import net.invasioncraft.api.network.events.DatabridgeCloseEvent.Reason;
import net.invasioncraft.api.network.events.DatabridgePingEvent;
import net.invasioncraft.api.network.packages.ConfigPackage;
import net.invasioncraft.api.network.packages.StringPackage;
import net.invasioncraft.api.network.packages.icPackage;
 
public class DataBridge extends Thread{
 
    private static String leaveMessage = "_socketClosed_",
            pingMessage = "_socketPing_",
            pingConfirmMessage = "_socketConfirmPing_",
            data = "_socketData_";
    private static char split = '|';
 
    private static List<DataBridge> list = new ArrayList<DataBridge>();
    private static List<Class<? extends icPackage>> packageTypes = new ArrayList<Class<? extends icPackage>>();
 
    private Socket socket;
    private DataInputStream input;
    private DataOutputStream output;
    private BridgeType type;
 
    private boolean reading;
    private Map<String, Integer> ping;
 
    public DataBridge(Socket socket, BridgeType type) throws IOException{
        this.socket = socket;
 
        input = new DataInputStream(socket.getInputStream());
        output = new DataOutputStream(socket.getOutputStream());
 
        this.type = type;
 
        reading = true;
        ping = new HashMap<String, Integer>();
 
        list.add(this);
        this.start();
    }
 
    public void run(){
        while(true){
            if(reading){
                try{
                    String message = input.readUTF();
                    if(message == null)return;
 
                    final String[] messages = icAPI.stringUtil.split(message, split);
 
                    if(messages.length <= 0)return;
                    String prefix = messages[0];
 
                    if(prefix.equals(leaveMessage)){
                        DatabridgeCloseEvent event = new DatabridgeCloseEvent(this, Reason.REMOTE_SHUTDOWN);
                        Bukkit.getPluginManager().callEvent(event);
                        icAPI.console.info("Received leave-message from remote-address " + socket.getRemoteSocketAddress());
 
                        reading = false;
 
                        list.remove(this);
                        input.close();
                        output.close();
                        socket.close();
                    }
                    else if(prefix.equals(pingMessage)){
                        sendRaw(pingConfirmMessage + split + messages[1]);
                    }
                    else if(prefix.equals(pingConfirmMessage)){
                        if(!ping.containsKey(messages[1]))return;
                        rp(messages[1], ping.get(messages[1]));
                    }else{
                        String[] msgs = new String[messages.length-2];
                        String typeId = messages[1];
 
                        for(int n = 0; n < msgs.length; n ++){
                            msgs[n] = messages[n+2];
                        }
 
                        if(prefix.equals(data)){
                            icPackage icp = null;
                            for(Class<? extends icPackage> cp : packageTypes){
                                if(icp != null)break;
 
                                try{
                                    String s = (String)pid(cp);
 
                                    if(s.equals(typeId)){
                                        Method a = cp.getDeclaredMethod("convert", new Class<?>[]{String[].class});
                                        a.setAccessible(true);
                                        icp = (icPackage)a.invoke(null, (Object)msgs);
                                    }
                                }catch(Exception e){
                                    e.printStackTrace();
                                }
                            }
 
                            if(icp == null){
                                icAPI.console.warn("Failed to get packageType! Given value: '" + typeId + "'");
                                icAPI.console.warn("Not responding to message ...");
                            }else{
                                DatabridgePackageReceiveEvent event = new DatabridgePackageReceiveEvent(this, icp);
                                Bukkit.getPluginManager().callEvent(event);
                            }
                        }else{
                            icAPI.console.warn("==================================================");
                            icAPI.console.warn("Invalid messageType registered at " + getClass().getPackage() + getClass().getName());
                            icAPI.console.warn("Registered messageType: " + prefix);
                            icAPI.console.warn("Not responding to message ...");
                            icAPI.console.warn("==================================================");
                        }
                    }                   
                }catch(IOException e){
                }catch(Exception e){
                    e.printStackTrace();
                }
            }
        }
    }
 
    /**
    * Closes the dataBridge
    */
    public void close(){
        close(Reason.LOCAL_SHUTDOWN);
    }
    public void close(Reason reason){
        try{
            icAPI.console.info("Ending connection with address: " + socket.getRemoteSocketAddress());
            icAPI.console.info("Reason: " + reason.toString().toLowerCase());
            output.writeUTF(leaveMessage);
 
            reading = false;
 
            DatabridgeCloseEvent event = new DatabridgeCloseEvent(this, reason);
            Bukkit.getPluginManager().callEvent(event);
 
            list.remove(this);
            input.close();
            output.close();
            socket.close();
        }
        catch(IOException e){
            icAPI.console.error("Failed to close dataBridge to remote address " + socket.getRemoteSocketAddress() + " !");
            e.printStackTrace();
        }
    }
 
    /**
    * Returns the dataInputStream
    * @return
    */
    public DataInputStream getDataInputStream(){
        return input;
    }
 
    /**
    * Returns the dataOutputStream
    * @return
    */
    public DataOutputStream getDataOutputStream(){
        return output;
    }
 
    /**
    * Returns the socket
    * @return
    */
    public Socket getSocket(){
        return socket;
    }
 
    /**
    * Returns the bridgeType
    * @return
    */
    public BridgeType getType(){
        return type;
    }
 
    /**
    * Sends a raw message
    */
    private boolean sendRaw(String msg){
        try{
            output.writeUTF(msg);
            return true;
        }catch(Exception e){
            return false;
        }
    }
 
    /**
    * Sends a package
    * @param package
    * @return
    */
    public boolean send(icPackage icpackage){
        DatabridgePackageSendEvent event = new DatabridgePackageSendEvent(this, icpackage);
        Bukkit.getPluginManager().callEvent(event);
 
        if(event.isCancelled())return false;
 
        return sendRaw(data + split + icpackage.getTypeId() + split + icpackage.toSendableString());
    }
    public boolean send(String message){
        return send(new StringPackage(message));
    }
    public boolean send(String[] messages){
        return send(new StringPackage(messages));
    }
    public boolean send(String title, VirtualConfigFile virtualConfigFile){
        return send(new ConfigPackage(title, virtualConfigFile));
    }
 
    /**
    * Returns the character used to spit
    * message-parts
    * @return
    */
    public static char getSplit(){
        return split;
    }
 
    /**
    * Registers a new icPackage type
    * @param icpackage
    */
    public static boolean registerNewPackage(Class<? extends icPackage> icClass){       
        String s = pid(icClass);
 
        if(s == null)return false;
 
        for(Class<? extends icPackage> icp : packageTypes){
            if(pid(icp).equals(s))return false;
        }
 
        packageTypes.add(icClass);
        return true;
    }
 
    public static void d(icAPI plugin){
        List<DataBridge> l = list;
        for(DataBridge db : l){
            db.close(Reason.API_SHUTDOWN);
        }
    }
 
    public static void e(icAPI plugin){
        icAPI.scheduler.scheduleSyncRepeatingTask(new Runnable(){
            public void run(){
                for(DataBridge db : list){
                    db.dp();
                }
            }
        }, 5*20, 10*20);
        icAPI.scheduler.scheduleSyncRepeatingTask(new Runnable(){
            public void run(){
                List<DataBridge> l = list;
                for(DataBridge db : l){
                    db.ap();
                }
            }
        }, 1, 1);
    }
 
    private static Integer currentPingId = null;   
    private static String generatePingId(){
        if(currentPingId == null)currentPingId = 0;
        currentPingId = currentPingId + 1;
        return ""+currentPingId;
    }
 
    private void dp(){
        String id = generatePingId();
        ping.put(id, 0);
        sendRaw(pingMessage + split + id);
    }
 
    private void ap(){
        //icAPI.scheduler.scheduleSyncDelayedTask(new Runnable(){
            //public void run(){
                Map<String, Integer> edit = new HashMap<String, Integer>();
                for(String s : ping.keySet()){
                    if(ping.get(s) >= 10*20){
                        close(Reason.TIMED_OUT);
                        return;
                    }else{
                        edit.put(s, ping.get(s)+1);
                    }
                }
                for(String s : edit.keySet()){
                    ping.put(s, edit.get(s));
                }
            //}
        //}, 1);
    }
 
    private void rp(String id, double t){
        ping.remove(id);
 
        DatabridgePingEvent event = new DatabridgePingEvent(this, ((t/20)*100), id);
        Bukkit.getPluginManager().callEvent(event);
    }
 
    @Deprecated
    private static String pid(Class<? extends icPackage> c){
        try{
            Method a = c.getDeclaredMethod("getPackageTypeId");
            a.setAccessible(true);
            return (String)a.invoke(null);
        }catch(Exception e){
            e.printStackTrace();
            return null;
        }
    }
}
 

However after a while the server keeps running out of memory, any ideas what may be causing this?

Ps. it only happens on the serverSocket and NOT on the client. Here's the code I use only for servers:

Code:java

package net.invasioncraft.api.network;
 
import java.io.IOException;
import java.net.ServerSocket;
import java.net.SocketTimeoutException;
import java.util.ArrayList;
import java.util.List;
 
import org.bukkit.Bukkit;
 
import net.invasioncraft.api.icAPI;
import net.invasioncraft.api.network.events.DatabridgeCreateEvent;
 
public class VirtualDataServer extends Thread{
 
    private static List<VirtualDataServer> list = new ArrayList<VirtualDataServer>();
 
    /**
    * DataServers are used to allow connections from a client
    */
 
    private int port;
    private ServerSocket serverSocket;
    private boolean pending;
 
    public VirtualDataServer(int port) throws IOException{
        this(port, 10000);
    }
    public VirtualDataServer(int port, int soTimeout) throws IOException{
        this.port = port;
        serverSocket = new ServerSocket(port);
        serverSocket.setSoTimeout(soTimeout);
        pending = true;
 
        list.add(this);
    }
 
    /**
    * Returns the local port
    * @return
    */
    public int getPort(){
        return port;
    }
 
    /**
    * Returns the serverSocket
    * @return
    */
    public ServerSocket getServerSocket(){
        return serverSocket;
    }
 
    @Override
    @SuppressWarnings("deprecation")
    public void destroy(){
        try{
            super.destroy();
            list.remove(this);
        }catch(Exception e){}
    }
 
    public void run(){
        while(pending){
            try{
                DataBridge db = new DataBridge(serverSocket.accept(), BridgeType.SERVER);
                icAPI.console.info("Succesfully connected with client-address: " + db.getSocket().getRemoteSocketAddress());
 
                DatabridgeCreateEvent event = new DatabridgeCreateEvent(db);
                Bukkit.getPluginManager().callEvent(event);
 
            }catch(SocketTimeoutException e){
            }catch(IOException e){
                e.printStackTrace();
                break;
            }
        }
    }
 
    public static void d(icAPI plugin){
        List<VirtualDataServer> l = list;
        for(VirtualDataServer vds : l){
            vds.destroy();
        }
    }
}
 

Thanks for your time!

 

jeussa Try closing the ServerSocket onDisable? I really have no idea are you sure its your plugin causing it?

 

I'm actually using this as an API. The older version did not have this issue however it was very unstable so I redid the whole thing. Right now all that needs to be fixed is the server running out of memory.

Ow btw, almost forgot to add:
the reason it's running out of memory is because of one of the threads, not sure which one tho.

 

jeussa
A little tip, when creating a new thread don't use "while (true)" instead create a variable (e.g. Boolean running = true) then use "while ( running ) " then, when you want the thread to stop just make the variable false. If you want to know why you should do this instead of "Thread.stop()/Thread.destroy()" then read this by Sun.

Edit: You could also do "while ( ! Thread.currentThread().isInterrupted() )" and then later call the "Thread.interrupt()" method. Similarly you could use the "while (true)" and then when you want to get rid of the thread call the "Thread.interrupt()" method.

 

TGRHavoc I agree...almost. Use Thread.interrupted() so that the interrupted flag gets cleared. If you're checking for interruption in the current thread you should have a good reason not to use this method.

jeussa There are some things I would like to point out. You're swallowing exceptions. Don't do that. You're only hiding useful information which, by chance, might help you debug this problem. I would also suggest you wrap the try blocks around the entire while loop. If something goes wrong you don't always want the connection to keep going, do you? I suppose it depends on the use case.

EDIT: To figure out which thread runs out of memory, name them. That way the name of the thread will be displayed in the stack trace.

 

Code:

 while(pending){
try{
DataBridge db = new DataBridge(serverSoc

This seems evil to me. Where do you set the "pending" varibale to FALSE?

 

With the code I provided, not. However I slightly worked on an update which sets pending to false when the databridge is closed. However the server is still running out of memory after some time.

 

jeussa

Im not entirely sure but why do you have a static List of all VirtualServers plus it isn't volatile?, volatile keyword means that it can be modified by other threads.

make sure those in that List got removed, also double test it with contains() in case for some reason ??? the VirtualServer you try to lookup in contains() is differentated than the one you would expect.

a good example which I would like to show is why for example a List with a ItemStack could cause a memory leak, I know this may has nothing todo with ItemStack but maybe it still has todo with your own objects so please try reflect onto this:

Code:

 
public class Test {
        private List<ItemStack> items = new ArrayList<ItemStack>();
 
        public Test() {
              ItemStack item = new ItemStack(Material.GLOWSTONE, 1);
              items.add(item);
       }
 
       public boolean compair() {
               ItemStack item = new ItemStack(Material.GLOWSTONE);
               item.setAmount(2);
               return items.contains(item); //this will return false
       }
}

now if you see I placed a ItemStack into the items list through the constructor.

however when I would use the compair() method it will returns false even though the ItemStack looks smilliar.
why? because since there is difference in the amount per stack this would not work and contains() will return false, now lets say this would happen to your VirtualServer List then it would be possible those elements never will be removed due a change in the object it self which triggers that equals sents a other result than you would/could expect.

also the while(true) loop isn't that evil aslong its ran on a other thread plus theres a way to stop it again (I would preferly use a long sleep till it throws a exception or the deprecated stop() but yet ive to lookup why it is deprecated).

but what I don't understand is why should there be more 'servers' than just one who handles all the requests?

also please post the stacktrace of the permgen error sometimes it could give alot of information what could happen to.

 

xize

That's not at all what volatile does, but nice try. We aren't saying the while(true) loop is evil, he was saying you need to check for interrupts, and yea the multiple servers thing is a good point.

jeussa

As for your code, it doesn't seem have very good object-oriented design at all. Why does each class have a static reference to a List of all instances of that class? That's normally where a manager comes in. Furthermore, the design doesn't look very thread-safe at all, and will likely error out with all the different ways you can violate the thread-safety of this design. Your constants should be static and final, otherwise you can inadvertently modify them, and you shouldn't be catching Exception.

As for why it's running out of memory, have you tried making sure that one of your awful lists isn't overfilling with useless sockets? Or perhaps that the DataBridges are never actually removed because the threads are on infinite uninterruptable loops? Try using a Java Profiler to see where your issue is.

 

My point was, that he is never setting "pending" to FALSE which means, as soon as a connection has been closed client-side, the loop will again run and create another dataBridge-object instead of just doing ".accept()" for the next incoming connection.
Depending on the traffic this may cause a rapid depletion of RAM, because the garbage collector is a pretty lazy dude by default.
I see no reason to throw the socket object away after each transfer.
It's just a guess. But the current implementation is surely not optimal.

 

CubieX

I wasn't disagreeing with you, I agree wholeheartedly.

 

The code you see here is something I made to be used inside an API. Multiple plugins that use the API may need different ports, which is why I'm listing all pending serverSockets (or in this case VirtualDataServers containing the sockets) rather then creating a single one.

Also I'm afraid I don't get any stacktraces but the message 'server stopped responding'

I've slightly been editing my code so as soon as I use the close() method on a databridge pending will be set to false and after that I attempt to destroy the thread. Unfortunately without any success.

Now my issue is only occuring on servers and not on clients, which would suggest the issue should be in the VirtualDataServer.java. However I'm still a nub when it comes to socket programming.