2
\$\begingroup\$

I have around 60 sockets, 20 business threads and 1 pinger timer thread and I want to make sure each of those thread works on different socket every time so I don't want to share same socket between two threads at all.

In my SocketManager class, I have a background thread (pinger timer thread) which runs every 60 seconds and calls updateLiveSockets() method. In the updateLiveSockets() method, I iterate all the available sockets I have and then start pinging them one by one by calling send method of SendToQueue class and basis on the response I mark them as live or dead. In the updateLiveSockets() method, I always need to iterate all the sockets and ping them to check whether they are live or dead.

Now all the business threads will call getNextSocket() method of the SocketManager class concurrently to get the next live available socket to send the business message on that socket. And business threads are sending data very fast, just FYI. So I have two types of messages which I am sending on a socket:

  • One is ping message on a socket. This is only sent from pinger timer thread calling updateLiveSockets() method in SocketManager class.
  • Other is business message on a socket. This is done in SendToQueue class and being sent by all the business thread.

So if pinger timer thread is pinging a socket to check whether they are live or not then no other business thread should use that socket. Similarly if business thread is using a socket to send data on it, then pinger thread should not ping that socket at all. And this applies to all the socket. But I need to make sure that in updateLiveSockets method, we are pinging all the available sockets whenever my background thread starts so that we can figure out which socket is live or dead.

SocketManager class:

public class SocketManager {
  private static final Random random = new Random();
  private final ScheduledExecutorService scheduler = Executors.newSingleThreadScheduledExecutor();
  private final Map<Datacenters, List<SocketHolder>> liveSocketsByDatacenter =
      new ConcurrentHashMap<>();
  private final ZContext ctx = new ZContext();

  // ...

  private SocketManager() {
    connectToZMQSockets();
    scheduler.scheduleAtFixedRate(new Runnable() {
      public void run() {
        updateLiveSockets();
      }
    }, 60, 60, TimeUnit.SECONDS);
  }

  // during startup, making a connection and populate once
  private void connectToZMQSockets() {
    Map<Datacenters, List<String>> socketsByDatacenter = Utils.SERVERS;
    for (Map.Entry<Datacenters, List<String>> entry : socketsByDatacenter.entrySet()) {
      List<SocketHolder> addedColoSockets = connect(entry.getValue(), ZMQ.PUSH);
      liveSocketsByDatacenter.put(entry.getKey(), addedColoSockets);
    }
  }

  private List<SocketHolder> connect(List<String> paddes, int socketType) {
    List<SocketHolder> socketList = new ArrayList<>();
    // ....
    return socketList;
  }

  // this method will be called by multiple threads concurrently to get the next live socket
  // is there any concurrency or thread safety issue or race condition here?
  public Optional<SocketHolder> getNextSocket() {
    for (Datacenters dc : Datacenters.getOrderedDatacenters()) {
      Optional<SocketHolder> liveSocket = getLiveSocket(liveSocketsByDatacenter.get(dc));
      if (liveSocket.isPresent()) {
        return liveSocket;
      }
    }
    return Optional.absent();
  }

  private Optional<SocketHolder> getLiveSocket(final List<SocketHolder> listOfEndPoints) {
    if (!listOfEndPoints.isEmpty()) {
      // The list of live sockets
      List<SocketHolder> liveOnly = new ArrayList<>(listOfEndPoints.size());
      for (SocketHolder obj : listOfEndPoints) {
        if (obj.isLive()) {
          liveOnly.add(obj);
        }
      }
      if (!liveOnly.isEmpty()) {
        // The list is not empty so we shuffle it an return the first element
        return Optional.of(liveOnly.get(random.nextInt(liveOnly.size()))); // just pick one
      }
    }
    return Optional.absent();
  }

  // runs every 60 seconds to ping all the available socket to make sure whether they are alive or not
  private void updateLiveSockets() {
    Map<Datacenters, List<String>> socketsByDatacenter = Utils.SERVERS;

    for (Map.Entry<Datacenters, List<String>> entry : socketsByDatacenter.entrySet()) {
      List<SocketHolder> liveSockets = liveSocketsByDatacenter.get(entry.getKey());
      List<SocketHolder> liveUpdatedSockets = new ArrayList<>();
      for (SocketHolder liveSocket : liveSockets) {
        Socket socket = liveSocket.getSocket();
        String endpoint = liveSocket.getEndpoint();
        Map<byte[], byte[]> holder = populateMap();
        Message message = new Message(holder, Partition.COMMAND);

        // using socket as its own lock
        synchronized (socket) {
            // pinging to see whether a socket is live or not
            boolean isLive = SendToQueue.getInstance().send(message.getAddress(), message.getEncodedRecords(), socket);
            SocketHolder zmq = new SocketHolder(socket, liveSocket.getContext(), endpoint, isLive);
            liveUpdatedSockets.add(zmq);
        }
      }
      liveSocketsByDatacenter.put(entry.getKey(), Collections.unmodifiableList(liveUpdatedSockets));
    }
  }
}

SendToQueue class:

  // this method will be called by multiple business threads (around 20) concurrently to send the data
  public boolean sendAsync(final long address, final byte[] encodedRecords) {
    PendingMessage m = new PendingMessage(address, encodedRecords, true);
    cache.put(address, m);
    return doSendAsync(m);
  }

  private boolean doSendAsync(final PendingMessage pendingMessage) {
    Optional<SocketHolder> liveSocket = SocketManager.getInstance().getNextSocket();
    if (!liveSocket.isPresent()) {
      return false;
    }       
    ZMsg msg = new ZMsg();
    msg.add(pendingMessage.getEncodedRecords());
    try {
      // send data on a socket
      // this returns instantly which doesn't mean that socket is live or dead
      // so we can't rely on this boolean to check whether socket is live or dead.
      // using socket as its own lock
      synchronized (socket) {
        return msg.send(liveSocket.get().getSocket());
      }
    } finally {
      msg.destroy();
    }
  }

  public boolean send(final long address, final byte[] encodedRecords, final Socket socket) {
    PendingMessage m = new PendingMessage(address, encodedRecords, socket, false);
    cache.put(address, m);
    try {
      if (doSendAsync(m, socket)) {
        return m.waitForAck();
      }
      return false;
    } finally {
      cache.invalidate(address);
    }
  }

Now as you can see above that I am sharing same socket between two threads. It seems getNextSocket() in SocketManager class could return a 0MQ socket to Business Thread A. Concurrently, the pinger timer thread may access the same 0MQ socket to ping it. In this case Business Thread A and the timer thread are mutating the same 0MQ socket, which can lead to problems. So I am trying to find a way so that I can prevent different threads from sending data to the same socket at the same time and mucking up my data.

One solution I can think of is using synchronization on a socket while sending the data but if many threads uses the same socket, resources aren't well utilized. Moreover, if msg.send(socket); is blocked (technically it shouldn't) all threads waiting for this socket are blocked. So I guess there might be a better way to ensure that every thread uses a different single live socket at the same time instead of synchronization on a particular socket.

In my above answer, I am synchronizing on the socket both from pinger timer thread and business thread but it may not be the optimal solution as I mentioned above. As my code is working fine so opting for code review to see if there is any better way to do this.

Note: I am working with Java 7.

\$\endgroup\$
2
\$\begingroup\$

I see several issues here belonging to style, structure, and functionality. Also, the code shown is partial. I'd like to see what that cache thing is doing (if anything at all).


1 - Style

1.1 - Interfaces and contracts

I don't see any interface here. It'd help to set interfaces first, then build the class to implement. By doing it in that order, you'll see a cleaner design emerge.

SocketManager.java

Let's take the SocketManager. It could implement the Provider<Socket> interface. This carries th pont across. Also if some object just wants to get Sockets, and not manage them, you can downcast to Provider and pass that. Every Object should anly see what they need (separation of concerns)

SendToQueue.java

Let's look at the SendToQueue class. First of all, its name is a verb. A class name should be a Noun. A class name carries what the object is. A method name carries what the Object does. Ok let's look at what it does. It can sendAsynca message to any Socket or send (still asynchronously!) a message to a specific Socket. In general I didn't see any Queue functionality. This class looks more like a MessageSender.It has the nice properties of being (supposedly) asynchronous, and also can select a socket itself if the User doesn't want to. We can call it an AsynchronousMessageSender. Then you can just rename your methods send and sendToSocket.

1.3 - Constructor for SocketManager

If you're going to have a single constructor, then try to have all your initializations in it to make the order of instancing more obvious.

In the same vein, moving some init code to connectToZMQSocket isn't too great, particularly when no other code will ever call that one again. However extracting the connect method is probably a good call.

Looking at that method, having a local reference to Utils.SERVERS does nothing for you except write more code. use the true value instead, particularly as it looks like an enum, which will never get overwritten. It was good for me because otherwise I wouldn't have understood that for Utils.SERVERS was a map of datacenter to their sockets. But that just tells me that this enum should be called Datacenters.MAP_TO_SOCKETS instead! In genreral, when I see 'XyzUtils' I smell bad code.

1.4 - Package-private methods

SendToQueue has two public methods. But who cares about sending data to a particular socket? Only the SocketManager does (to check liveliness). Normal business message senders should not even have visibility to that. So hide it, make it package-private (assuming SocketManager and SendToQueue are in the same package).


2 - Structure

2.1 - Datacenters

You seem to have a priority between Datacenters because you're polling them in a particular order given by Datacenters.getOrderedDatacenters(). However you do not have any priority between Sockets of a same Datacenter, as evidenced by your use of Random. That looks very similar to the contract provided by the PriorityBlockingQueue, using poll and put. If you set your SocketHolder to be Comparable using the ordering based on their Datacenter, then you don't need to implement any of this priority thing yourself, and you can stop having to store a Map<String, List<>> all over the place.

2.2 - Singleton pattern

The SocketManager is a singleton, I believe. I'd like to verify that its instance is static final. Id' also like to see the class final, or even better use an enum. This way no need for getInstance() shenanigans, and you get safe singletons for free, backed by Java's nature.

In addition it is weird to mix static final variables like random with final instance variables like all others, when you know you only have one instance.

SendToQueue is also a singleton. I wonder what this brings to the table. For the SocketManager, it makes sense in order to have only one executor. But for the sender? I don't even see any state to it. It could be made uninstantiable, and all static method. Which hints at there being very little need for this class to be separate. It has a cache, but I'll come back to that.

2.3 - The Cache

You put data in, then (sometimes) invalidate it. You never even look at what's inside. It looks useless. Caches are hard to do right. Thread-safe caches are hell. You know what they say of premature optimization. I say drop it, fix your design, then look at it again later (maybe).

2.4 Liveliness

This may be a mistake in updateLiveSockets:

        boolean isLive = SendToQueue.getInstance().send(message.getAddress(), message.getEncodedRecords(), socket);
        SocketHolder zmq = new SocketHolder(socket, liveSocket.getContext(), endpoint, isLive);
        liveUpdatedSockets.add(zmq);

You're never using isLive to check your result. Then Socket ends up in liveUpdatedSockets regardless. Maybe this is more correct:

        boolean isLive = SendToQueue.getInstance().send(message.getAddress(), message.getEncodedRecords(), socket);
        if(isLive) {
          SocketHolder zmq = new SocketHolder(socket, liveSocket.getContext(), endpoint, true);
          liveUpdatedSockets.add(zmq);
        }

2.5 Nullcheck

This makes me nervous:

    return msg.send(liveSocket.get().getSocket());

I don't know what a SocketHolder returns with get() if the socket is dead or missing, but if it can be null then this is bad.


3 - Functionality

3.1 - Return early

Your socket selection in getLiveSocket is wasteful, and potentially wrong, thread-safety-wise. You iterate over all sockets first (calling isLive all N of them, which I assume might be slow-ish), then pick one live at random. By the time your return it, it may not be live anymore.

Instead you could randomize first, then return early (side note there is no reson to check if input list is empty, also never call an object obj, also don't share a Random between threads, it might lead to contention from its Javadoc):

private Optional<SocketHolder> getLiveSocket(final List<SocketHolder> listOfEndPoints) {
  List<SocketHolder> randomizedSockets = new ArrayList<>(listOfEndPoints);
  Collection.shuffle(randomizedSockets , new Random()); // super-fast
  for (SocketHolder socket: randomizedSockets ) {
    if (socket.isLive()) {
      return Optional.of(socket); // Early return
    }
  }
  return Optional.absent();
}

This doesn't fix that it may return dead Sockets, and also can return the same socket for different threads. But it's simpler (and faster).

3.2 - Fail fast

Detect dead sockets, and update the list when you find one. In getLiveSocket for example, you encounter dead sockets in a list of what you thought were live Sockets, but you do nothing about it. How selfish! now every other Thread has to find this out the hard way as well! Just change this:

if (socket.isLive()) {
  return Optional.of(socket); // Early return
} else {
  liveSockets.remove(socket); // Not this will be much easier to do if 
                              // all sockets are in a single PriorityQueue as I mentioned
                              // Otherwise you need to know its datacenter
                              // (why does the socket not know its DC btw?)
  deadSockets.add(socket); // Now you can revive those in priority in your update thread
}

3.3 - Thread safety

You are rather safe. No two threads can use the same Socket because of the Synchronize sections. So they're never mutating it at the same time. However you're not free of deadlocks, and you may very well end up with many threads waiting for the sync on the same Socket, when others are available.

Indeed your design does allows several Threads to obtain the same Socket handle. Your synchronized bit is just preventing them from using it at the same time, forcing them to wait for each other instead. It would be a better idea to synchronize the Socket request, instead of (or in addition to) the Socket usage. To do this you can have Objects acquire a lock on a Socket, then release the lock when they're done. I suggest using Java's automatic resource management for this, particularly for the SocketHolder object, to enforce the lock release.

Basically the sendAsync should do this:

  • SendToQueue: a thread requests to send a message
  • SendToQueue: the thread requests a SocketHolder to SocketManager (getNextSocket)
  • SocketManager: polls a SocketHolder from its queue. PriorityBlockingQueue's poll is thread safe, so only one thread has this Socket). You should use a timeout to remove deadlocks.
  • SocketManager: SocketHolder is put in a busy list in manager's side, to keep track of it
  • SendToQueue: if SocketHolder is not present, return failure (no resource available)
  • SendToQueue: creates a message
  • SendToQueue: synchronizes on the Socket (to be sure, but only he should have it)
  • SendToQueue: sends a message on the Socket
  • SendToQueue: returns a succcess status code
  • SendToQueue (automatic finally block): the SocketHolder releases itself back to the SocketManager
  • SocketManager : the SocketHolder reintegrates the Queue

The liveliness checker can acquire locks on individual SocketHolders in the same fashion (also with a timeout).

This code uses the same synchronization on the socket during usage, but it also makes sure that threads all get different sockets. Also uses timeout to prevent deadlocks.


Edit: Sample code

I'd like to change more, like make the Sockets smarter, but I don't have their source.

Assuming Socket object has compare() method implemented and which ranks sockets from best Datacenters first. Maybe order them by dc priority first, then by time since last usage?

SocketManager.java :

public class SocketManager implements Runnable{

  private static final ScheduledExecutorService scheduler = Executors.newSingleThreadScheduledExecutor();
  private static final Map<Datacenters, List<SocketHolder>> liveSocketsByDatacenter = new ConcurrentHashMap<>();
  private static final ZContext ctx = new ZContext();
  private static final PriorityBlockingQueue<Socket> liveSocketQueue = new PriorityBlockingQueue();
  private static final List<Socket> deadSockets = Collections.synchronizedList(new ArrayList());

  static {
    for (Map.Entry<Datacenters, List<String>> entry : Datacenters.MAP_TO_SOCKETS.entrySet()) {
      List<SocketHolder> colocatedSockets = connect(entry.getValue(), ZMQ.PUSH);
      liveSocketsByDatacenter.put(entry.getKey(), colocatedSockets);
      for(Socket socket: colocatedSockets){
        liveSocketQueue.add(colocatedSockets);
      }
    }
    scheduler.scheduleAtFixedRate(new SocketManager(), 60, 60, TimeUnit.SECONDS); // single instance of SocketManager here
  }

  private static List<SocketHolder> connect(List<String> paddes, int socketType) {
    List<SocketHolder> socketList = new ArrayList<>();
    // ....
    return socketList;
  }

  // runs every 60 seconds to ping all the available socket to make sure whether they are alive or not
  private void run() {
    // Check that the 'live' sockets are indeed live, otherwise put in dead list
    for(List<SocketHolder> socketsInDatacenter : liveSocketsByDatacenter.values()){
      for (SocketHolder questionableSocket : socketsInDatacenter) {
        if(liveSocketQueue.remove(questionableSocket)) { // Queue was alive
        if(!SendToQueue.sendPing(questionableSocket) {
          deadSockets.add(questionableSocket); // Queue is now pronounced dead
        }
      }
    }
    // Now revive all dead sockets
    Socket revivingSocket = null;
    do(!deadSockets.isEmpty()){
      Socket revivingSocket = deadSockets.poll();
      /// ??? I don't know how you revive sockets ???
      revivingSocket.revive(); // magic defibrillator activated !
      liveSocketQueue.add(revivingSocket); // Back from the dead
    } while(revivingSocket != null);
  }

  // this method will be called by multiple business threads (around 20) concurrently to send the data
  public static boolean send(final long address, final byte[] encodedRecords) {
    PendingMessage m = new PendingMessage(address, encodedRecords, true);
    SocketHolder liveSocket = liveSocketQueue.poll(); // Removing the socket, only *this* thread can use it now!
    if (liveSocket == null) {
      return false; // No socket available!
    }
    try {
      // using socket as its own lock
      synchronized (liveSocket) {
      return doSendAsync(m, liveSocket);
      }
    } finally {
      liveSocketQueue.add(liveSocket); // Reintegrating the socket to the live list after usage
      // In fact I would like to wait for ack to reintegrate it. If no ack, maybe put it in deadSockets list?
    }
  }

  private static boolean doSendAsync(final PendingMessage pendingMessage, Socket socket) {
    ZMsg msg = new ZMsg();
    msg.add(pendingMessage.getEncodedRecords());
    try {
      return msg.send(liveSocket.getSocket()); // This is weird. Socket.getSocket? they are Russian doll sockets now ? ^^
      // Also this is wasteful. If we receive 'false', doesn't that mean the socket is dead? then we could put it in 'dead'
      // Just because "true doesn't mean is alive", doesn't imply "false doesn"t mean it's dead". It does.
      // I advise you to use a Future to wait for ack  while returning early to original Thread.
    } finally {
      msg.destroy();
    }
  }

  // Send ping, wait for ack, returns accurate sending status
  private static boolean sendPing(final Socket socket) {
    String endpoint = socket.getEndpoint();
    Map<byte[], byte[]> holder = populateMap();
    Message message = new Message(holder, Partition.COMMAND); // I don't understand this, i left it in
    // using socket as its own lock
    synchronized (socket) {
      final long address = message.getAddress(); // I don't understand this, i left it in
      final byte[] encodedRecords = message.getEncodedRecords();  // I don't understand this, i left it in
      PendingMessage m = new PendingMessage(address, encodedRecords, socket, false); // I don't understand this, i left it in
      if (doSendAsync(m, socket)) {
        return m.waitForAck();
      } else {
        return false;
      }
    }
  }
}
\$\endgroup\$
  • \$\begingroup\$ Thanks a lot for detailed code review. It helped me a lot. There are few things I want to discuss with you to make sure you understood what I was thinking but it looks to me there might be some confusion so I created this chat room to discuss about the same. \$\endgroup\$ – david Dec 23 '17 at 20:46

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.