Graceful shutdown of multithreaded java server using executor service

I refactored a single-threaded server to be capable of multithreading and accepting multiple clients at the same time. To do so, I spawn a new ClientHandler thread for each new client and submit it to an ExecutorService. I want to initiate a server shutdown by entering a new line to System.In.

However, I am not able to shut down the server from within (using the shutdown method suggested in Oracle’s ExecutorService doc) – can somebody explain to me why? My Server is a Runnable and I put it into the same ThreadPool as my individual client threads – could this be the problem?

PS: This is a university project. I deliberately left out the names of the implemented interfaces and the request processing methods, as well as renamed the classes to prevent this from becoming the go-to solution for every lazy student in the future.

Server

public class Server extends Runnable {

  private final List<ClientHandler> activeHandlers = new ArrayList<>();
  private int port;
  private volatile boolean terminated = false;
  private ExecutorService service;


  @Override
  public void start(int port) throws ServerException {
      this.port = port;
      service = Executors.newCachedThreadPool();
      service.submit(this);
  }

  @Override
  public void shutdown() throws ServerException {
      System.out.println("Shutdown initiated.");
      this.terminated = true;
      PoolUtil.safeShutdown(service);
  }

  @Override
  public void run() {
      try (ServerSocket serverSocket = new ServerSocket(port)) {
          while (!terminated) {
              try {
                  Socket client = serverSocket.accept();
                  ClientHandler clientSocket = connect(client);
                  service.submit(clientSocket);
              } catch (IOException e) {
                  System.err.println("ERROR: Connection to client failed.");
              }
          }
      } catch (IOException e) {
          System.err.println("ERROR: Could not create a socket on port " + port);
      } finally {
          PoolUtil.safeShutdown(service);
      }
  }

  @Override
  public ClientHandler connect(Socket client) {
      ClientHandler clientHandler = new ClientHandler(client, this);
      activeHandlers.add(clientHandler);
      System.out.println("Registered new ClientHandler for " + client.getInetAddress().toString());
      return clientHandler;
  }

  @Override
  public void disconnect(ClientHandler clientHandler) {
      activeHandlers.remove(clientHandler);
      System.out.println("Client successfully disconnected.");
  }
}

ClientHandler

ublic class ClientHandler extends Runnable {
  private final Socket client;
  private final DirectoryServer server;
  private boolean terminated;
  private final Result result = new Result();

  public ClientHandler(Socket client, DirectoryServer server) {
      this.client = client;
      this.server = server;
      terminated = false;
  }

  @Override
  public void run() {
      try (client;
           ObjectOutputStream oos = new ObjectOutputStream(client.getOutputStream());
           ObjectInputStream ois = new ObjectInputStream(client.getInputStream())) {
          while (!terminated) {
              Object message = ois.readObject();
              if (message instanceof SomeRequest) {
                 // dostuff...
              } else if (message instanceof TerminateConnection) {
                  TerminateConnection termination = (TerminateConnection) message;
                  process(termination);
              } else {
                  System.err.println(
                          "ERROR: the received object of class "
                                  + message.getClass().toString()
                                  + "can not be processed."
                  );
              }
          }
      } catch (IOException e) {
          // FIXME: Error handling
          System.err.println("ERROR concerning client " + client.getInetAddress() + " -> " + e.getMessage());
      } catch (ClassNotFoundException e) {
          // FIXME: Error handling
          System.err.println("ERROR: the class of the received object unknown to server --> " + e.getMessage());
      }

  }


  @Override
  public void process(TerminateConnection terminateConnection) {
      this.terminated = true;
      server.disconnect(this);
  }
}

ServerMain

public class ServerMain {

    public static void main(String[] args) throws ServerException, IOException {
        Server server = new Server();
        server.start(1337);
        System.out.println("Server started. Press enter to terminate.");

        System.in.read();

        server.shutdown();
        System.out.println("Server is shut down...");
    }
}

PoolUtil.shutdown( )

public static void safeShutdown(ExecutorService threadPool){
        threadPool.shutdown();
        try {
            // Waits a minute for all tasks to terminate
            if (!threadPool.awaitTermination(60, TimeUnit.SECONDS)) {
                // Cancel all tasks that are still running after a minute
                threadPool.shutdownNow();
                // Waits another minute for all tasks to be cancelled
                if (!threadPool.awaitTermination(60, TimeUnit.SECONDS)) {
                    System.err.println("Service did not terminate!");
                }
            }
        } catch (InterruptedException e) {
            threadPool.shutdownNow();
            Thread.currentThread().interrupt();
        }
    }

Answer

The JavaDoc for the ExecutorService.shutdown() method states that it means “previously submitted tasks are executed, but no new tasks will be accepted”. Termination, however, will not happen until all tasks have completed. The Runnables for your tasks perform a blocking operation, serverSocket.accept() so you should not expect the awaitTermination method to return after a shutdown until enough requests come in after the shutdown to use up all of the tasks that are blocking. You could try using shutdownNow() instead of shutdown() so that it attempts to immediately cancel/interrupt all of the running tasks, which should hopefully unblock them.

Advertisement

Answer