title: Do not share same socket between two threads at the same time
body: <p>I have around 60 sockets and 20 threads and I want to make sure each thread works on different socket everytime so I don't want to share same socket between two threads at all.</p>
  
  <p>In my <code>SocketManager</code> class, I have a background thread which runs every 60 seconds and calls <code>updateLiveSockets()</code> method. In the <code>updateLiveSockets()</code> method, I iterate all the sockets I have and then start pinging them one by one by calling <code>send</code> method of <code>SendToQueue</code> class and basis on the response I mark them as live or dead. In the <code>updateLiveSockets()</code> method, I always need to iterate all the sockets and ping them to check whether they are live or dead.</p>
  
  <p>Now all the reader threads will call <code>getNextSocket()</code> method of <code>SocketManager</code> class concurrently to get the next live available socket to send the business message on that socket. So I have two types of messages which I am sending on a socket:</p>
  
  <ul>
  <li>One is <code>ping</code> message on a socket. This is only sent from timer thread calling <code>updateLiveSockets()</code> method in <code>SocketManager</code> class.</li>
  <li>Other is <code>business</code> message on a socket. This is done in <code>SendToQueue</code> class.</li>
  </ul>
  
  <p>So if pinger 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. And this applies to all the socket. But I need to make sure that in <code>updateLiveSockets</code> 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.</p>
  
  <p>Below is my <code>SocketManager</code> class:</p>
  
  <pre><code>public class SocketManager {
    private static final Random random = new Random();
    private final ScheduledExecutorService scheduler = Executors.newSingleThreadScheduledExecutor();
    private final Map&lt;Datacenters, List&lt;SocketHolder&gt;&gt; liveSocketsByDatacenter =
        new ConcurrentHashMap&lt;&gt;();
    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&lt;Datacenters, List&lt;String&gt;&gt; socketsByDatacenter = Utils.SERVERS;
      for (Map.Entry&lt;Datacenters, List&lt;String&gt;&gt; entry : socketsByDatacenter.entrySet()) {
        List&lt;SocketHolder&gt; addedColoSockets = connect(entry.getValue(), ZMQ.PUSH);
        liveSocketsByDatacenter.put(entry.getKey(), addedColoSockets);
      }
    }
  
    private List&lt;SocketHolder&gt; connect(List&lt;String&gt; paddes, int socketType) {
      List&lt;SocketHolder&gt; socketList = new ArrayList&lt;&gt;();
      // ....
      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&lt;SocketHolder&gt; getNextSocket() {
      for (Datacenters dc : Datacenters.getOrderedDatacenters()) {
        Optional&lt;SocketHolder&gt; liveSocket = getLiveSocket(liveSocketsByDatacenter.get(dc));
        if (liveSocket.isPresent()) {
          return liveSocket;
        }
      }
      return Optional.absent();
    }
  
    private Optional&lt;SocketHolder&gt; getLiveSocket(final List&lt;SocketHolder&gt; listOfEndPoints) {
      if (!listOfEndPoints.isEmpty()) {
        // The list of live sockets
        List&lt;SocketHolder&gt; liveOnly = new ArrayList&lt;&gt;(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&lt;Datacenters, List&lt;String&gt;&gt; socketsByDatacenter = Utils.SERVERS;
  
      for (Map.Entry&lt;Datacenters, List&lt;String&gt;&gt; entry : socketsByDatacenter.entrySet()) {
        List&lt;SocketHolder&gt; liveSockets = liveSocketsByDatacenter.get(entry.getKey());
        List&lt;SocketHolder&gt; liveUpdatedSockets = new ArrayList&lt;&gt;();
        for (SocketHolder liveSocket : liveSockets) {
          Socket socket = liveSocket.getSocket();
          String endpoint = liveSocket.getEndpoint();
          Map&lt;byte[], byte[]&gt; holder = populateMap();
          Message message = new Message(holder, Partition.COMMAND);
  
          // 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));
      }
    }
  }
  </code></pre>
  
  <p>And here is my <code>SendToQueue</code> class:</p>
  
  <pre><code>  // this method will be called by multiple reader 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&lt;SocketHolder&gt; liveSocket = SocketManager.getInstance().getNextSocket();
      if (!liveSocket.isPresent()) {
        // log error
        return false;
      }       
      ZMsg msg = new ZMsg();
      msg.add(pendingMessage.getEncodedRecords());
      try {
        // send data on a socket LINE A
        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);
      }
    }
  </code></pre>
  
  <p><strong>Problem Statement</strong></p>
  
  <p>Now as you can see above that I am sharing same socket between two threads. It seems <code>getNextSocket()</code> in <code>SocketManager</code> class could return a <code>0MQ socket</code> to <code>Thread A</code>. Concurrently, the <code>timer thread</code> may access the same <code>0MQ socket</code> to ping it. In this case <code>Thread A</code> and the <code>timer thread</code> are mutating the same <code>0MQ socket</code>, 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.</p>
  
  <p>One solution I can think of is using <code>synchronization</code> on a socket while sending the data but if many threads uses the same socket, resources aren't well utilized. Moreover If <code>msg.send(socket);</code> 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.</p>