6
\$\begingroup\$

I implemented a multithreaded TCP client in java for use in an android application. The purpose is to implement the TCPClient interface to allow another thread to post strings to the server and read responses without needing to deal with any of the networking. It doesn't use any android specific code except for the logging.

I tested the code and it appears to work, but I have a few concerns. Ideally someone with more experience with multithreading or tcp client/server applications could help me out with best practices and some of the rather arbitrary decisions I made. Here are my concerns:

  • Number of failures to determine to "give up" on connecting/reconnecting is arbitrary.
  • Time to wait between retries is also arbitrary. The implementation as shown here does not support changing the socket timeout, which is IMO much too long by default. I have modified this in my actual code but don't want to modify the sample code I have posted.
  • Appending newline to strings that are sent to server. Probably should check for newline first, or require that strings passed into sendMessage() end in newline.
  • Program architecture. Is this interface/class missing something important?

Those are my main concerns but general feedback is welcome.

TCPClient.java

public interface TCPClient
{
    String getNextMessage();
    boolean sendMessage(String a_sMessage);
}

StandardTCPClient.java

import android.util.Log;
import java.io.BufferedReader;
import java.io.InputStreamReader;
import java.io.PrintWriter;
import java.net.Socket;
import java.net.URL;
import java.util.concurrent.LinkedBlockingQueue;

public class StandardTCPClient implements TCPClient, Runnable
{
    private static final String LOG_TAG = "StandardTCPClient";
    private static final int NUM_RETRIES = 3;

    private URL m_connection;
    private LinkedBlockingQueue<String> m_receivedQueue;
    private LinkedBlockingQueue<String> m_sendQueue;
    private volatile boolean m_isRunning;

    public StandardTCPClient(URL a_connection)
    {
        if(a_connection == null)
            throw new IllegalArgumentException("Connection cannot be null");
        if(a_connection.getHost() == null)
            throw new IllegalArgumentException("Connection host cannot be null");
        if(a_connection.getPort() == 0)
            throw new IllegalArgumentException("Connection port cannot be 0");
        m_receivedQueue = new LinkedBlockingQueue<String>();
        m_sendQueue = new LinkedBlockingQueue<String>();
        m_connection = a_connection;
    }

    //returns null if there are no new messages
    @Override
    public String getNextMessage()
    {
        return m_receivedQueue.poll();
    }

    @Override
    public boolean sendMessage(String a_sMessage)
    {
        return m_sendQueue.offer(a_sMessage);
    }

    @Override
    public void run()
    {
        int connectionFailures = 0;
        while(connectionFailures < NUM_RETRIES)
        {
            try (
                    Socket socket = new Socket(m_connection.getHost(),
                            m_connection.getPort());
                    // from server
                    BufferedReader inputStream = new BufferedReader(new InputStreamReader(socket.getInputStream()));
                    // to server
                    PrintWriter outputStream = new PrintWriter(socket.getOutputStream(), true);
            )
            {
                connectionFailures = 0;
                StandardTCPClient client = this;
                Thread inputReader = new Thread()
                {
                    public void run()
                    {
                        try
                        {
                            while (!interrupted() && client.isRunning())
                                m_receivedQueue.put(inputStream.readLine());
                        }
                        catch (java.io.IOException ex)
                        {
                            Log.w(LOG_TAG, "IO Exception: ", ex);
                        }
                        catch (InterruptedException ex)
                        {
                            Log.i(LOG_TAG, "Receiving thread interrupted.");
                            Thread.currentThread().interrupt();
                        }
                        catch(Exception ex)
                        {
                            Log.e(LOG_TAG, "Unexpected exception in receiving thread:", ex);
                        }
                        client.setRunning(false);
                        synchronized (client)
                        {
                            client.notifyAll();
                        }
                    }
                };
                Thread outputWriter = new Thread()
                {
                    public void run()
                    {
                        try
                        {
                            while (!interrupted() && client.isRunning())
                                outputStream.printf(m_sendQueue.take() + "\n");
                        }
                        catch (InterruptedException ex)
                        {
                            Log.i(LOG_TAG, "sending thread interrupted.");
                            Thread.currentThread().interrupt();
                        }
                        catch(Exception ex)
                        {
                            Log.e(LOG_TAG, "Unexpected exception in sending thread:", ex);
                        }
                        client.setRunning(false);
                        synchronized (client)
                        {
                            client.notifyAll();
                        }
                    }
                };

                setRunning(true);
                inputReader.start();
                outputWriter.start();
                //wait for the threads to finish, they never should
                synchronized (client)
                {
                    while (isRunning())
                        wait();
                }

                // its not clear to us why the threads are stopping.
                // force them to stop so we can recreate
                inputReader.interrupt();
                outputWriter.interrupt();

                // Annoyingly, thread.interrupt() does not stop a blocking buffered stream
                // readline() io call. Close the socket to force the blocking call to end.
                socket.close();
                // wait for them to stop then attempt to recreate.
                inputReader.join();
                outputWriter.join();
            } catch (java.io.IOException ex)
            {
                setRunning(false);
                Log.w(LOG_TAG, "IO Exception:", ex);
                connectionFailures++;
            }
            catch (InterruptedException ex)
            {
                setRunning(false);
                Log.e(LOG_TAG, "Thread interrupted");
                Thread.currentThread().interrupt();
                return;
            }
            //wait 1 second then attempt to reconnect
            try
            {
                Thread.sleep(1000);
            }
            catch (InterruptedException ex)
            {
                Log.e(LOG_TAG, "retry loop interrupted");
                Thread.currentThread().interrupt();
                return;
            }
        }
    }

    public boolean isRunning()
    {
        return m_isRunning;
    }

    synchronized private void setRunning(boolean running)
    {
        m_isRunning = running;
    }
}
\$\endgroup\$
3
\$\begingroup\$

It might sound a bit petty but you might check for all exceptions that can be thrown from Socket (or at least a one-for-all-exception: RuntimeException)...

Socket throws:

  • (done) IOException - if an I/O error occurs when creating the socket.
  • (open) SecurityException - if a security manager exists and its checkConnect method doesn't allow the operation.
  • (partly done) IllegalArgumentException - if the port parameter is outside the specified range of valid port values, which is between 0 and 65535, inclusive.
  • (done) NullPointerException - if address is null.

I think especially the port is dangerous, because you thought you had already checked that. The Exception can be be thrown if you use the simple constructor from URL URL url = new URL("http://www.example.com/docs/resource1.html");. Then the method getPort() will return -1 (see URL API). Applying such a URL to your StandardTCPClient will throw an IllegalArgumentException since port is not in range 0 ... 65535 and your application will throw an unhandled RuntimeException - maybe you check better for if(a_connection.getPort() <= 0)...?

Another problem might be a simple compilation error:

  • In the constructor you set m_pool = a_connection; but I guess it would rather be m_connection = a_connection;, right?
\$\endgroup\$
  • 1
    \$\begingroup\$ Good points about URL.getPort() and the SecurityException. You are correct about the compilation error, which I will fix in the original post. I don't want to mark this as answer because I don't want to discourage future feedback on the multithreading aspect of the code. \$\endgroup\$ – Final Hazard Jun 1 '18 at 3:00
0
\$\begingroup\$

Unfortunately, I am not experienced enough to give you advice on the specific aspects you mentioned in your question, but maybe you still find some of the following points helpful.

  • You don't need the local variable client in the method StandardTCPClient.run(), because you can always refer to the enclosing instance of StandardTCPClient from within an inner classes with StandardTCPClient.this: For example:

    client.setRunning(false);
    synchronized (client)
    {
        client.notifyAll();
    }
    

    can be re-written to:

    StandardTCPClient.this.setRunning(false);
    synchronized (StandardTCPClient.this)
    {
        StandardTCPClient.this.notifyAll();
    }
    

    True, this is slightly more verbose, but on the other hand, it is instantly obvious what is being referenced, as opposed to when using a local variable which could reference anything. Theoretically, you could even omit the explicit reference to StandardTCPClient.this in the first line, because an inner class can refer to members of its enclosing class directly:

    setRunning(false);
    synchronized (StandardTCPClient.this)
    {
        StandardTCPClient.this.notifyAll();
    }
    

    With the invocation of notifyAll(), however, the explicit reference to StandardTCPClient.this is necessary, because otherwise, notifyAll() would be invoked on the instance of the anonymous Thread subclass rather than on the enclosing instance of StandardTCPClient (this is known as "shadowing", see also here). In the first line, this problem does not arise because Thread does not have any methods with the name setRunning (at least as of yet, so it might be better to keep the explicit reference to StandardTCPClient.this anyway in case Thread will ever contain a method named setRunning in the future).

  • Also, I think it would greatly improve readability if you made the two anonymous Thread subclasses for inputReader and outputWriter non-anonymous inner classes, because with all the try and catch blocks and the anonymous classes, the code is a huge mountain of nested code. For example, you could refactor the code like this:

    public class StandardTCPClient implements TCPClient, Runnable
    {
        //...
    
        private class InputReaderThread extends Thread {
            private final BufferedReader inputStream;
    
            private InputReaderThread(BufferedReader inputStream)
            {
                this.inputStream = inputStream;
            }
    
            @Override
            public void run()
            {
                //...
            }
        }
    
        private class OutputWriterThread extends Thread {
            private final PrintWriter outputStream;
    
            private OutputWriterThread(PrintWriter outputStream)
            {
                this.outputStream = outputStream;
            }
    
            @Override
            public void run()
            {
                //...
            }
        }
    
        @Override
        public void run()
        {
            int connectionFailures = 0;
            while(connectionFailures < NUM_RETRIES)
            {
                try (
                        Socket socket = new Socket(m_connection.getHost(),
                                m_connection.getPort());
                        // from server
                        BufferedReader inputStream = new BufferedReader(new InputStreamReader(socket.getInputStream()));
                        // to server
                        PrintWriter outputStream = new PrintWriter(socket.getOutputStream(), true);
                )
                {
                    connectionFailures = 0;
                    Thread inputReader = new InputReaderThread(inputStream);
                    Thread outputWriter = new OutputWriterThread(outputStream);
                    //...
                }
                //...
            }
        }
    

    This makes it easier to discern the action happening in the main thread from the action happening in the two IO threads.

  • It might be more elegant to open the socket's streams in the two corresponding IO threads rather than in the main thread, because by opening these streams in the main thread, you also make it the main thread's responsibility to close them. But it is not the main thread that uses these streams, it's the two IO threads, and by opening the streams in the main thread, the two IO threads depend on the main thread for the streams to remain open while the IO threads operate on the streams.

    Of course, opening the streams in the IO threads would mean that the main thread will no longer be able to observe failed attempts to open these streams, and, as a consequence, to distinguish stream opening failures from exceptions that occur after the streams have been opened. So in your current code, if the creation of the socket succeeds but the opening of one of the streams fails for whatever reason, connectionFailures would be incremented without being reset to 0, whereas if the streams are opened in the IO threads, connectionFailures would be reset to 0 before the IO threads fail to open the streams.

    I'm not sure whether this is desirable or not. Also, I have no idea how realistic it is for the socket creation to succeed but the stream creation to fail, maybe someone else can come along and provide some insight on this.

  • The name of the field m_isRunning (as well as the names of the corresponding get/set methods) is a bit misleading. The way you use it, it is not really an indicator of whether the client is running, but rather a way for the two IO threads to communicate to the main thread and to each other whether they themselves are running, so that the other IO thread can terminate as well, either by itself, or by being interrupted from the main thread. The best alternative I could come up with is IOThreadsAreRunning, maybe you could think of something better.

  • The code might actually be easier to handle if you didn't re-create the IO threads every time a connection failure happens, but only update the socket (or stream) instance they operate on. For example, you could make the socket an instance variable of StandardTCPClient, which can be accessed by both threads, and when one thread encounters an IOException, it can request that the socket be re-created by the main thread by setting a flag. To prevent the socket from being updated and the flag being reset after an IO thread encounters an IOException but before that thread can set the flag, I made each IO thread store its own copy of the socket, so that it can compare this copy to the current socket of the TCP client:

    public class StandardTCPClient implements TCPClient, Runnable {
    
        private Socket m_socket;
        private boolean m_socketIsBroken;
    
        private synchronized void updateSocket(Socket socket) {
            m_socket = socket;
            m_socketIsBroken = false;
        }
    
        private class InputReaderThread extends Thread {
            @Override
            public void run() {
                try {
                    while (!Thread.interrupted()) {
                        Socket socket;
                        synchronized (StandardTCPClient.this) {
                            /*
                            so that the IO thread can be started before
                            a socket is created, since it will be running
                            continuously and not be re-created every time a
                            new socket is created
                             */
                            while (m_socket == null) {
                                StandardTCPClient.this.wait();
                            }
                            socket = m_socket;
                        }
                        try (BufferedReader inputStream = new BufferedReader(new InputStreamReader(socket.getInputStream()))) {
                            while (!Thread.currentThread().isInterrupted()) {
                                m_receivedQueue.put(inputStream.readLine());
                            }
                        } catch (java.io.IOException ex) {
                            Log.w(LOG_TAG, "IO Exception: ", ex);
                            synchronized (StandardTCPClient.this) {
                                /*
                                in case the other IO thread has already complained and
                                the socket has been updated in the meantime
                                 */
                                if (socket == m_socket) {
                                    m_socketIsBroken = true;
                                    StandardTCPClient.this.notifyAll();
                                    while (socket == m_socket) {
                                        StandardTCPClient.this.wait();
                                    }
                                }
                            }
                        }
                    }
                } catch (InterruptedException ex) { //return
                    Log.i(LOG_TAG, "Receiving thread interrupted.");
                }
            }
        }
    
        private class OutputWriterThread extends Thread {
            /*
            analogous to InputReaderThread – in fact, there is a lot of code duplication
            here, maybe this can be mitigated by creating a common superclass
             */
        }
    
        //...
    
        @Override
        public void run() {
            InputReaderThread inputReader = new InputReaderThread();
            OutputWriterThread outputWriter = new OutputWriterThread();
    
            inputReader.start();
            outputWriter.start();
    
            int connectionFailures = 0;
    
            try {
                while (connectionFailures < NUM_RETRIES) {
                    try (
                            Socket socket = new Socket(
                                    m_connection.getHost(),
                                    m_connection.getPort())
                    ) {
                        connectionFailures = 0;
                        synchronized (this) {
                            updateSocket(socket);
                            notifyAll();
                            while (!m_socketIsBroken) {
                                wait();
                            }
                            socket.close();
                        }
                    } catch (java.io.IOException ex) {
                        Log.w(LOG_TAG, "IO Exception:", ex);
                        connectionFailures++;
                    }
    
                    //wait 1 second then attempt to reconnect
                    try {
                        Thread.sleep(1000);
                    } catch (InterruptedException ex) {
                        Log.e(LOG_TAG, "retry loop interrupted");
                        Thread.currentThread().interrupt();
                        return;
                    }
                }
    
                /*
                interrupting the threads now should terminate them for good, since
                they cannot be blocked in an IO operation anymore, with the socket being closed
                 */
                inputReader.interrupt();
                outputWriter.interrupt();
    
                inputReader.join();
                outputWriter.join();
            } catch (InterruptedException ex) { //return
                Log.e(LOG_TAG, "Thread interrupted");
            }
        }
    }
    

    For the sake of simplicity, I omitted the catch blocks for unexpected RuntimeExceptions (at least, that's what your catch(Exception) blocks amounted to, since all checked exceptions had already been caught before them).

\$\endgroup\$

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service, privacy policy and cookie policy

Not the answer you're looking for? Browse other questions tagged or ask your own question.