5
\$\begingroup\$

Currently, the following code runs and produces the expected and desired output. However, I'm sure that there's much wrong with it.

This project queries the (flat) database, instantiates each row as an object, adds those objects to a queue, and makes the queue available to clients. Clients may then connect to the server, which pops objects from the queue and delivers that single object to the client. The client receives the object, the user updates it, and then it is sent back to the server. The server then updates the database accordingly.

So far as I can tell, this is functioning correctly, and allows for multiple clients.

However, I'm quite sure that the control flow is "odd", or at least sub-optimal.

How can I make the code clearer by using, for example, producer/client or another pattern or idiom?

WorkerRunnable:

package net.bounceme.dur.server.streams;

import net.bounceme.dur.jdbc.Title;
import java.io.IOException;
import java.io.ObjectInputStream;
import java.io.ObjectOutputStream;
import java.net.Socket;
import java.util.NoSuchElementException;
import java.util.logging.Level;
import java.util.logging.Logger;
import net.bounceme.dur.jdbc.Queries;

public class WorkerRunnable implements Runnable {

    private static final Logger log = Logger.getLogger(WorkerRunnable.class.getName());
    protected Socket socket = null;
    private RecordsQueueWrapper recordsQueue = null;
    private final Queries queries = new Queries();

    public WorkerRunnable(Socket clientSocket, RecordsQueueWrapper recordsQueue) {
        this.socket = clientSocket;
        this.recordsQueue = recordsQueue;
    }

    @Override
    public void run() {
        Title serverTitle = null;
        Title clientTitle = null;
        boolean queueEmpty = false;
        try (ObjectOutputStream objectOutputStream = new ObjectOutputStream(socket.getOutputStream());
                ObjectInputStream objectInputStream = new ObjectInputStream(socket.getInputStream())) {
            do {
                try {
                    serverTitle = recordsQueue.pop();
                    log.info(serverTitle.toString());
                    objectOutputStream.writeObject(serverTitle);
                    log.fine(serverTitle.getClass().getCanonicalName());
                    clientTitle = (Title) objectInputStream.readObject();
                    queries.updateTitle(clientTitle);
                } catch (NoSuchElementException nse) {
                    queueEmpty = true;
                }
            } while (!queueEmpty);
        } catch (java.io.EOFException eofe) {
            log.log(Level.WARNING, "why is there no record from client?\n", eofe);
        } catch (IOException | ClassNotFoundException ex) {
            Logger.getLogger(WorkerRunnable.class.getName()).log(Level.SEVERE, null, ex);
        }
    }
}

Server:

package net.bounceme.dur.server.streams;

import net.bounceme.dur.driver.MyProps;
import java.net.ServerSocket;
import java.net.Socket;
import java.io.IOException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.logging.Logger;

public class Server implements Runnable {

    private static final Logger log = Logger.getLogger(Server.class.getName());
    protected int serverPort = 8080;
    protected ServerSocket serverSocket = null;
    protected boolean isStopped = false;
    protected Thread runningThread = null;
    protected ExecutorService threadPool = Executors.newFixedThreadPool(10);
    private final RecordsQueueWrapper instance = RecordsQueueWrapper.getInstance();

    public Server(MyProps properties) {
        serverPort = properties.getServerPort();
    }

    @Override
    public void run() {
        synchronized (this) {
            this.runningThread = Thread.currentThread();
        }
        openServerSocket();
        while (!isStopped()) {
            Socket clientSocket = null;
            try {
                clientSocket = this.serverSocket.accept();
            } catch (IOException e) {
                if (isStopped()) {
                    log.warning("stopped server");
                    return;
                }
                throw new RuntimeException("Error accepting client connection", e);
            }
            this.threadPool.execute(new WorkerRunnable(clientSocket, instance));
        }
        this.threadPool.shutdown();
        log.info("server stopped");
    }

    private synchronized boolean isStopped() {
        return this.isStopped;
    }

    public synchronized void stop() {
        this.isStopped = true;
        try {
            this.serverSocket.close();
        } catch (IOException e) {
            throw new RuntimeException("Error closing server", e);
        }
    }

    private void openServerSocket() {
        try {
            this.serverSocket = new ServerSocket(this.serverPort);
        } catch (IOException e) {
            throw new RuntimeException("Cannot open port 8080", e);
        }
    }
}

I'm just looking for the single biggest, glaring opportunity of what to fix with this structure and flow; pseudo-code is more than sufficient.

\$\endgroup\$
4
\$\begingroup\$

The two main issues I see are not really code flow related. One is that you don't really have a protocol to communicate with your clients (as in, reporting success or failure, interpreting commands, et cetera).

The other is that your server may be prone to stalling. Your serving loop looks like this:

  • Accept connection
  • Send object to client
  • Wait for an edited record ← possible issue
  • Write down edited record
  • Close connection

Assuming that users manually edit the records you serve them, you have the risk that someone opens a connection, finds out she's out of coffee, and heads off for ten minutes (or lunch for an hour). That leaves a connection stuck until she returns or it times out.

If we move the user interaction to be offline, we limit chances of hogging your worker pool:

  • Accept connection
  • Read edited record from client, if any
  • Write down edited record
  • Send new record to client
  • Close connection

This does require a change in protocol, because now you have to recognise two scenarios:

  1. Your user has finished editing a record and sends it over.
  2. Your user is freshly starting and wants a new record to edit.

We would already need a change in protocol, because there are a number of messages that we may want to send to the client:

  • I'm out of records.
  • You sent me a record that's not in my database.
  • You sent me a query I don't understand.
  • ...

So let's introduce a kind of payload container that will sort this out for us:

/** Base class representing a response to a query. */
class Response implements java.io.Serializable {
  /** Indicates success or failure. */
  int code = CODE_SUCCESS; // consider re-using HTTP codes
  /** Informative message to go with the code. */
  String message = "";
  Title record;
}

class RecordQuery implements java.io.Serializable {
  /** Record the client has modified and wants saved.  May be null. */
  Title edited;
  /** True if the client wants another record to edit, false otherwise. */
  boolean sendMeMore;
}

// WorkerRunnable.run()
public void run() {
  try (ObjectOutputStream objectOutputStream = new ObjectOutputStream(socket.getOutputStream());
          ObjectInputStream objectInputStream = new ObjectInputStream(socket.getInputStream())) {
      final Object input;
      try {
        input = objectInputStream.readObject();
      } catch (ClassNotFoundException ex) {
        log.log(Level.FINE, "Unknown class in request", ex);
        objectOutputStream.writeObject(new Response(Response.CODE_UNKNOWN_REQUEST, "Huh?"));
        return;
      }
      if ( input instanceof RecordQuery ) {
        final RecordQuery query = (RecordQuery) input;
        if ( query.edited != null ) {
          queries.updateTitle(query.edited);
        }
        if ( query.sendMeMore ) {
          final Title nextTitle = queries.poll(); // null on empty
          if ( nextTitle != null ) {
            objectOutputStream.writeObject(new Response(nextTitle));
          } else {
            objectOutputStream.writeObject(new Response(Response.CODE_NO_MORE_RECORDS));
          }
        }
      } else {
        objectOutputStream.writeObject(new Response(Response.CODE_UNKNOWN_REQUEST, "Huh?"));
      }
  } catch (IOException ex) {
      log.log(Level.SEVERE, null, ex);
  }
}

But now WorkerRunnable.run() has multiple responsibilities. It parses input, determines a reply, and sends that reply. We can filter some things out:

public void run() {
  try (ObjectOutputStream objectOutputStream = new ObjectOutputStream(socket.getOutputStream());
          ObjectInputStream objectInputStream = new ObjectInputStream(socket.getInputStream())) {
      Response response = new Response(Response.CODE_UNKNOWN_REQUEST, "Huh?");
      Object input;
      try {
        input = objectInputStream.readObject();
      } catch (ClassNotFoundException ex) {
        log.log(Level.FINE, "Unknown class in request", ex);
        objectOutputStream.writeObject(response);
        return;
      }
      response = getReplyTo(input);
      objectOutputStream.writeObject(response);
  } catch (IOException ex) {
    log.log(Level.SEVERE, null, ex);
  }
}

protected Response getReplyTo(Object request) {
  if ( !(input instanceof RecordQuery) ) {
    return new Response(Response.CODE_UNKNOWN_REQUEST, "Huh?");
  }

  final RecordQuery query = (RecordQuery) input;
  if ( query.edited != null ) {
    queries.updateTitle(query.edited);
  }
  if ( query.sendMeMore ) {
    final Title nextTitle = queries.poll(); // null on empty
    if ( nextTitle != null ) {
      return new Response(nextTitle);
    } else {
      return new Response(Response.CODE_NO_MORE_RECORDS);
    }
  }
}

If inventing/implementing your own protocol seems a bit much (and it usually is), you could consider using an HTTP library or a general I/O framework like Apache MINA. But we wouldn't do the tag "reinventing the wheel" justice if we didn't go all the way, eh?

\$\endgroup\$
3
\$\begingroup\$

The one thing that is "glaring" to me is using the NoSuchElementException as part of the default control flow. Although this is valid and works (considering the multithreaded nature of the program) I believe that in most cases using an exception as the main control for exit of a do/while loop is not the most efficient way.

In the WorkerRunnable I would move the NoSuchElementException to the outer try/catch and modify the while statement to be something like:

while (recordsQueue.size() > 0);

If your RecordsQueueWrapper does not have a size() or length() sort of method you could instead use:

while (recordsQueue.peek() != null);

This makes you have more maintainable and understandable code, yet also ensures that an exception is still caught in the case of a race-condition.

Hope this helps!

\$\endgroup\$
1
\$\begingroup\$

As noted by JvR - the fact that the worker threads block on the client is a Big Red Flag.

It looks to me as though the state of the server needs a bit more work. You've set up a boolean isRunning as part of your shutdown protocol; access to it seems to be properly synchronized.

But my guess is that you've really got more than two states in play. For instance, if the server is no longer accepting connections, but clients are still connected, does that really count as "not running" ? If your controller code needs to be able to tell that the worker threads are still going, then you are going to need something finer grained than "on/off".

You may also want to visit the shutdown logic in Server.run() -- if the server throws a runtime exception because something strange has happened in the socket, are you sure that you want the client threads to continue running?

\$\endgroup\$
  • \$\begingroup\$ I'm trying to switch over to Netty, but yes. I wanted the server to recover from all errors, but that's not a good idea. \$\endgroup\$ – Thufir Jul 22 '14 at 0:19

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.