Managing socket pool per datacenter

Question

I have a list of sockets per datacenter (in general, I have around three datacenter and each datacenter having 20 sockets, so in total I have around 60 sockets). Now I have 20 business threads which want to send data on a socket. I need to make sure each thread works on a different socket so I created a pool of sockets for each datacenter where I poll socket from ArrayBlockingQueue. I also have a timer pinger thread which checks whether dead sockets are live or not. So queue variable will hold all live sockets and deadSockets will hold all dead sockets.

SocketPool class:

public class SocketPool implements AutoCloseable {
  private final BlockingQueue<Socket> queue;
  private final Queue<Socket> deadSockets = new ConcurrentLinkedQueue<>();
  private final ScheduledFuture<?> scheduledFuture;

  public SocketPool(Set<Socket> sockets, ScheduledExecutorService scheduledExecutorService) {
      queue = new ArrayBlockingQueue<>(sockets.size());
      queue.addAll(sockets);
      scheduledFuture = scheduledExecutorService.scheduleAtFixedRate(this::recheckDeadSockets, 60, 60, TimeUnit.SECONDS);
  }

  public void reportDead(final Socket socket) {
    deadSockets.add(socket);
  }

  public void releaseSocket(final Socket socket) {
    queue.add(socket);
  }

  public Optional<Socket> getSocket() {
    return Optional.fromNullable(queue.poll());
  }

  private void recheckDeadSockets() {
    for (int i = 0; i < deadSockets.size(); i++) {
      Socket socket = deadSockets.poll();
      if (checkAlive(socket)) {
        queue.add(socket);
      } else {
        deadSockets.add(socket);
      }
    }
  }

  private boolean checkAlive(Socket socket) {
    // check those dead socket and see if they are live now by pinging them
    // if they are live, then put them back into `queue` otherwise put in `deadSockets`
    return true;
  }

  @Override
  public void close() throws Exception {
    scheduledFuture.cancel(true);
  }
}

Here is my SocketManager class which manages dc and its pool of sockets. In the constructor I initialize a Map of Datacenter to its SocketPool so in total I will have three entries in it because I have three dcs.

public class SocketManager {
  private final ScheduledExecutorService scheduler = Executors.newSingleThreadScheduledExecutor();
  private final Map<Datacenters, SocketPool> liveSocketsByDatacenter = new ConcurrentHashMap<>();

  // Lazy Loaded Singleton Pattern
  private static class Holder {
    private static final SocketPoolManager INSTANCE = new SocketPoolManager();
  }

  public static SocketManager getInstance() {
    return Holder.INSTANCE;
  }

  // during startup, making a connection and populate once
  private SocketManager() {
    Map<Datacenters, ImmutableList<String>> socketsByDatacenter = Utils.SERVERS;
    for (Entry<Datacenters, ImmutableList<String>> entry : socketsByDatacenter.entrySet()) {
      Set<Socket> sockets = connect(entry.getValue(), ZMQ.PUSH);
      liveSocketsByDatacenter.put(entry.getKey(), new SocketPool(sockets, scheduler));
    }
  }

  // all my business thread concurrently will call this method to get the different live socket
  public Optional<Socket> useSocket() {
    for (Datacenters dc : Datacenters.getOrderedDatacenters()) {
      SocketPool socketPool = liveSocketsByDatacenter.get(dc);
      return socketPool.getSocket();
    }
  }

  // called from business threads to report dead socket
  public void reportDead(final Datacenters dc, final Socket socket) {
    SocketPool socketPool = liveSocketsByDatacenter.get(dc);
    socketPool.reportDead(socket);
  }

  // called from business threads to release the socket back in queue
  public void releaseSocket(final Datacenters dc, final Socket socket) {
    SocketPool socketPool = liveSocketsByDatacenter.get(dc);
    socketPool.releaseSocket(socket);
  }
}

All the business thread will call useSocket method of SocketManager class to get the live socket. In general, business threads needs a way to get live socket, release socket and report dead sockets in an atomic and thread safe way without any race condition or any concurrent issues.

Overall flow:

• All the business threads will get the unique live socket (making sure no two thread get the same socket) by calling useSocket method to send data on it.
• Once they are done with it, they will release it back to the queue by calling releaseSocket for corresponding datacenter which means all other business thread can start using them if they want.
• If any of the socket is dead (meaning if we didn't receive the acknowledgment back), then they will add them to deadSockets queue for corresponding datacenter by calling reportDead method.
• Then my timer thread in SocketPool class will ping all the dead sockets to check whether they are live now or they are still dead and basis on that I will put them back in respective queue.

As you can see now, I have same method (reportDead, releaseSocket) duplicated across both of these classes. I may not have designed these two classes properly basis on OOPS standard so wanted to see how we can structure these two classes properly to achieve the above thing. What is the best way to do this where I can make sure each thread works on different live socket and once done, release it back to the queue and also report back the dead sockets if any? Also, are there any thread safety issues or any race conditions?

I also need to make sure once the sockets are released back to queue or have been put to deadSockets, all the threads should see the state immediately. I don't want any business thread sending data on a socket which has been marked as dead or vice versa. I want to see if there is any better way to do the same thing.

Note: I am mainly working with Java 7.

Answer 1

As you can see now, I have same method (reportDead, releaseSocket) duplicated across both of these classes.

This wouldn't bother me much as the methods are very short, but I don't get why they both need to be public. Whatever you do, I'd prefer to see only one class (with the other being hidden), but maybe that's just my wrong understanding.

reportDead and releaseSocket are pretty the same method, so you could use

releaseSocket(Socket socket, boolean isDead)

or better

releaseSocket(Socket socket, DeadOrAlive state)

Just an idea.

Don't let your idea sprinkle final in the argument list randomly.

---

recheckDeadSockets is wrong. The number of sockets in the queue may grow anytime, which is no big deal. After some program change, it may also shrink giving you an NPE. The standard way is

while (true) { Socket socket = deadSockets.poll(); if (socket == null) { break; } }

Alternatively, you could always check a single socket and set the rate accordingly.

---

I guess, your Datacenters is actually a single data center, so the plural name is pretty misleading.

---

The loop in useSocket makes no sense to me as it always gets exited prematurely.

---

public void reportDead(Datacenters dc, Socket socket) - Why do I have to pass both arguments, when a socket always belong to a single data center?

How can I find out the data center???

I guess, you should hide the existence of multiple data centers.

All in all I didn't spend much time with it and may be missing a lot, but it feels like a half-baked solution. It may work as is, but it probably offers more methods than you really want to use, exposing things you actually don't want to expose.