6
\$\begingroup\$

I have a class in which I am populating a map liveSocketsByDatacenter from a single background thread every 30 seconds inside updateLiveSockets() method and then I have a method getNextSocket() which will be called by multiple reader threads to get a live socket available which uses the same map to get this info.

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

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

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

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

  private void connectToZMQSockets() {
    Map<Datacenters, ImmutableList<String>> socketsByDatacenter = Utils.SERVERS;
    // The map in which I put all the live sockets
    Map<Datacenters, List<SocketHolder>> updatedLiveSocketsByDatacenter = new HashMap<>();
    for (Map.Entry<Datacenters, ImmutableList<String>> entry : socketsByDatacenter.entrySet()) {
      List<SocketHolder> addedColoSockets = connect(entry.getKey(), entry.getValue(), ZMQ.PUSH);
      updatedLiveSocketsByDatacenter.put(entry.getKey(),
          Collections.unmodifiableList(addedColoSockets));
    }
    // Update the map content
    this.liveSocketsByDatacenter.set(Collections.unmodifiableMap(updatedLiveSocketsByDatacenter));
  }

  private List<SocketHolder> connect(Datacenters colo, List<String> addresses, int socketType) {
    List<SocketHolder> socketList = new ArrayList<>();
    for (String address : addresses) {
      try {
        Socket client = ctx.createSocket(socketType);
        // Set random identity to make tracing easier
        String identity = String.format("%04X-%04X", random.nextInt(), random.nextInt());
        client.setIdentity(identity.getBytes(ZMQ.CHARSET));
        client.setTCPKeepAlive(1);
        client.setSendTimeOut(7);
        client.setLinger(0);
        client.connect(address);

        SocketHolder zmq = new SocketHolder(client, ctx, address, true);
        socketList.add(zmq);
      } catch (Exception ex) {
        // log error
      }
    }
    return socketList;
  }

  // this method will be called by multiple threads to get the next live socket
  public Optional<SocketHolder> getNextSocket() {
    // For the sake of consistency make sure to use the same map instance
    // in the whole implementation of my method by getting my entries
    // from the local variable instead of the member variable
    Map<Datacenters, List<SocketHolder>> liveSocketsByDatacenter =
        this.liveSocketsByDatacenter.get();
    Optional<SocketHolder> liveSocket = Optional.absent();
    List<Datacenters> dcs = Datacenters.getOrderedDatacenters();
    for (Datacenters dc : dcs) {
      liveSocket = getLiveSocket(liveSocketsByDatacenter.get(dc));
      if (liveSocket.isPresent()) {
        break;
      }
    }
    return liveSocket;
  }

  private Optional<SocketHolder> getLiveSocket(final List<SocketHolder> listOfEndPoints) {
    if (!CollectionUtils.isEmpty(listOfEndPoints)) {
      // 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
        Collections.shuffle(liveOnly);
        return Optional.of(liveOnly.get(0));
      }
    }
    return Optional.absent();
  }

  // Added the modifier synchronized to prevent concurrent modification
  // it is needed because to build the new map we first need to get the
  // old one so both must be done atomically to prevent concistency issues
  private synchronized void updateLiveSockets() {
    Map<Datacenters, ImmutableList<String>> socketsByDatacenter = Utils.SERVERS;

    // Initialize my new map with the current map content
    Map<Datacenters, List<SocketHolder>> liveSocketsByDatacenter =
        new HashMap<>(this.liveSocketsByDatacenter.get());

    for (Entry<Datacenters, ImmutableList<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();

        boolean status = SendToSocket.getInstance().execute(3, holder, socket);
        boolean isLive = (status) ? true : false;
        SocketHolder zmq = new SocketHolder(socket, liveSocket.getContext(), endpoint, isLive);
        liveUpdatedSockets.add(zmq);
      }
      liveSocketsByDatacenter.put(entry.getKey(),
          Collections.unmodifiableList(liveUpdatedSockets));
    }
    this.liveSocketsByDatacenter.set(Collections.unmodifiableMap(liveSocketsByDatacenter));
  }
}

As you can see in my class:

  • From a single background thread which runs every 30 seconds, I populate liveSocketsByDatacenter map with all the live sockets in updateLiveSockets() method.
  • And then from multiple threads, I call the getNextSocket() method to give me a live socket available which uses a liveSocketsByDatacenter map to get the required information.

I have my code working fine without any issues and wanted to see if there is any better or more efficient way to write this. I also wanted to get an opinion on thread safety issues or any race conditions if any are there, but so far I haven't seen any. Maybe I don't need a synchronized modifier on my updateLiveSockets() method.

\$\endgroup\$
  • \$\begingroup\$ What is your definition of "efficient"? \$\endgroup\$ – Timothy Truckle Mar 10 '17 at 12:32
2
+25
\$\begingroup\$

In general, the concept you are trying to accomplish is appropriate for the task at hand, a lazy build of the live sockets, followed by a quick "switch" of what the clients can see.

Using an AtomicReference is also what I would do to accomplish the core thread-safe switch.

In general, the approach you take is a good one.

Some of the implementation details are not as ideal, or modern, as I would expect, and there are some tricks with Atomics that you're not using.... which would help.

Singletons

The singleton you have for your SocketManager is reasonable, but the best practice is to use an enum to solve it. Your code has:

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

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

but this should really be:

  // Lazy Loaded Singleton Pattern
  private enum Holder {
      INSTANCE;
      private final SocketManager manager = new SocketManager();
  }

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

There are a number of advantages to using an enum for the singleton pattern...

Joshua Bloch writes: "This approach is functionally equivalent to the public field approach, except that it is more concise, provides the serialization machinery for free, and provides an ironclad guarantee against multiple instantiation, even in the face of sophisticated serialization or reflection attacks. While this approach has yet to be widely adopted, a single-element enum type is the best way to implement a singleton."

Implementing getNextSocket()

Your code implementing this method has a number of 2nd-tier issues (there are no functionality/bug/edge-case issues I can see, just code-style and best-practice issues), some are hard to describe.

First up, just to remind myself, and others, this is a multi-thread entry point - this method can be called concurrently by many threads. Your implementation is such that all actions on the "live" socket map should be read-only, and thus there are no mutations to the map that we need to worry about... and we know that the update thread does not change THIS map, it creates a new map, and at some point swaps that new map in. So, in this method, as soon as we have retrieved the map from the atomic reference, we can do what we like (as long as it is read-only), and not have concurrency issues.

Right, your code correctly does all the above.... good. There's no problems I can see in that regard.

Also, I really like that you are returning an Optional result, it's a good strategy.

But.... let's look at some of the issues, starting with "early returns". It is common, and simpler in Java to have an early-return, or a nested return, than it is to have a "placeholder" variable. In addition, you can iterate on the ordered datacenters directly without the variable. Further, while I do appreciate that your variable names are very descriptive, you can find a compromise between too short, and too long. I would reduce the "verbose-ness" of variable names especially when they are private and in a small scope like a small function. Similarly I would remove this references as they are mostly redundant. In your code you have:

   Map<Datacenters, List<SocketHolder>> liveSocketsByDatacenter =
        this.liveSocketsByDatacenter.get();
    Optional<SocketHolder> liveSocket = Optional.absent();
    List<Datacenters> dcs = Datacenters.getOrderedDatacenters();
    for (Datacenters dc : dcs) {
      liveSocket = getLiveSocket(liveSocketsByDatacenter.get(dc));
      if (liveSocket.isPresent()) {
        break;
      }
    }
    return liveSocket;

but that would be simpler if you got rid of the liveSocket variable completely, and have some shorter variable names, and had just:

Map<Datacenters, List<SocketHolder>> liveByDC = liveSocketsByDatacenter.get();
for (Datacenters dc : Datacenters.getOrderedDatacenters()) {
  Optional<SocketHolder> socket = getLiveSocket(liveByDC.get(dc));
  if (socket.isPresent()) {
    return socket;
  }
}
return Optional.absent();

Implementing getLiveSocket()

Similar to the calling function, this one is accessed concurrently, and again, I like the use of Optional, etc.

Where this function has problems is primarily the way you copy and shuffle the data. It's overkill, and inefficient. You also have similar issues with overly long variable names, etc.

Oddly, in this method, you use the early-return concept really well.

What you should be using instead, in here, is the concept of a "Guard Clause". Instead of checking for a non-empty list (using CollectionsUtils - whatever that is - is overkill too) you should just check for an empty list and return.

Your method is:

  private Optional<SocketHolder> getLiveSocket(final List<SocketHolder> listOfEndPoints) {
    if (!CollectionUtils.isEmpty(listOfEndPoints)) {
      // 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
        Collections.shuffle(liveOnly);
        return Optional.of(liveOnly.get(0));
      }
    }
    return Optional.absent();
  }

but I think if you just selected a random member (using ThreadLocalRandom), instead of a full shuffle, you would be able to reduce it to:

  private Optional<SocketHolder> getLiveSocketX(final List<SocketHolder> endpoints) {
      if (endpoints == null || endpoints.isEmpty()) {
          return Optional.absent();
      }

      int rand = ThreadLocalRandom.current().nextInt(endpoints.size());
      return Optional.of(endpoints.get(rand));
  }

Atomic Updates

You have synchronized the updateLiveSockets method with the comment:

  // Added the modifier synchronized to prevent concurrent modification
  // it is needed because to build the new map we first need to get the
  // old one so both must be done atomically to prevent concistency issues

This makes it look like there's something you are guarding against, but that's not true .... nothing else is calling this method, and no other methods are synchronized, so there's no protection here at all.

instead, what you should do, is a checked set.... Your code is essentially:

// Initialize my new map with the current map content
Map<Datacenters, List<SocketHolder>> liveSocketsByDatacenter =
    new HashMap<>(this.liveSocketsByDatacenter.get());

// ...... change the liveSocketsByDatacenter

this.liveSocketsByDatacenter.set(Collections.unmodifiableMap(liveSocketsByDatacenter));

This code should instead be (stripping long variable names, and unmodifiable redundancy):

Map<Datacenters, List<SocketHolder>> original = this.liveSocketsByDatacenter.get();
Map<Datacenters, List<SocketHolder>> updated = new HashMap<>(original);

// .... update the changes to updated

if (!liveSocketsByDatacenter.compareAndSet(original, updated)) {
    // something else was running this method at the same time as us!
    // should never happen, but we just ignore our changes until
    // we are called again in 30 seconds
    // perhaps:
    System.out.println("Crap happened :( Ignoring our changes");
}

Note that the compare-and-set is also atomic - we make sure that nobody changed our source map while we were building updates to it.

In theory, the compareAndSet() will always return true because nothing can call the method concurrently (ScheduledExecutor prevents this too).

There's no need for the synchronization....

Update process Socket Details

Focusing on just the low-level details of your scheduled updates (excluding the atomic "switch" handled above). Your update mechanisms are intricately linked to your ZeroMQ, and other details. I am not an expert on those libraries, but there are a few things I can point out... "in general".

This makes me realize that your method should be split in to two... an atomic handler, and the actual update handler.

First up, all the ImmutableList and Collections.unmodifyableList stuff is not needed. You are not exposing any collections outside this code, and nobody in your code modifies things, so your code is redundant. get rid of things like that.

Next up, your use of Utils.SERVERS is overly complicated. I would use it inline, and just have the keySet(). There are other variables you create to "extract" the data from a map, etc. but that actually makes things more complicated.

This method allows some significant simplifications, for example:

  private synchronized void updateLiveSockets() {
    Map<Datacenters, ImmutableList<String>> socketsByDatacenter = Utils.SERVERS;

    // Initialize my new map with the current map content
    Map<Datacenters, List<SocketHolder>> liveSocketsByDatacenter =
        new HashMap<>(this.liveSocketsByDatacenter.get());

    for (Entry<Datacenters, ImmutableList<String>> entry : socketsByDatacenter.entrySet()) {

becomes:

  private synchronized void updateLiveSockets() {

    // Initialize my new map with the current map content
    Map<Datacenters, List<SocketHolder>> liveSocketsByDatacenter =
        new HashMap<>(this.liveSocketsByDatacenter.get());

    for (Datacenters dc : Utils.SERVERS.keySet()) {

There are a few other small changes in there that can be done to improve readability....

Updated Updater.

All told, I would take this code:

  // Added the modifier synchronized to prevent concurrent modification
  // it is needed because to build the new map we first need to get the
  // old one so both must be done atomically to prevent concistency issues
  private synchronized void updateLiveSockets() {
    Map<Datacenters, ImmutableList<String>> socketsByDatacenter = Utils.SERVERS;

    // Initialize my new map with the current map content
    Map<Datacenters, List<SocketHolder>> liveSocketsByDatacenter =
        new HashMap<>(this.liveSocketsByDatacenter.get());

    for (Entry<Datacenters, ImmutableList<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();

        boolean status = SendToSocket.getInstance().execute(3, holder, socket);
        boolean isLive = (status) ? true : false;
        SocketHolder zmq = new SocketHolder(socket, liveSocket.getContext(), endpoint, isLive);
        liveUpdatedSockets.add(zmq);
      }
      liveSocketsByDatacenter.put(entry.getKey(),
          Collections.unmodifiableList(liveUpdatedSockets));
    }
    this.liveSocketsByDatacenter.set(Collections.unmodifiableMap(liveSocketsByDatacenter));
  }

and reduce it to two functions, one for the atomicity, and the other for the business logic:

private void updateLiveSockets() {

    Map<Datacenters, List<SocketHolder>> orig = liveSocketsByDatacenter.get();
    Map<Datacenters, List<SocketHolder>> updated = updateSocketStatus(orig);
    if (!liveSocketsByDatacenter.compareAndSet(orig, updated)) {
        // report concurrency bug somehow.....
        System.err.println("Concurrency bug in scheduled updater");
    }
}

private Map<Datacenters, List<SocketHolder>> updateSocketStatus(Map<Datacenters, List<SocketHolder>> source) {
    // note, don't initialize with the source data!
    Map<Datacenters, List<SocketHolder>> updatedDCS = new HashMap<>();

    for (Datacenters dc : Utils.SERVERS.keySet()) {
        List<SocketHolder> updatedSockets = new ArrayList<>();
        for (SocketHolder sh : source.get(dc)) {
            Socket socket = sh.getSocket();
            boolean isLive = getInstance().execute(3, populateMap(), socket);
            SocketHolder zmq = new SocketHolder(socket, sh.getContext(), sh.getEndpoint(), isLive);
            updatedSockets.add(zmq);
        }
        updatedDCS.put(dc, updatedSockets);
    }
    return updatedDCS;
}
\$\endgroup\$
1
\$\begingroup\$

Comments

Just wanted to point out you made great, great comments.

Those explain implicit relationships (why you use local variable, why synchronise methods etc.), they do no useless hand-holding, are concise yet precise. Excellent job!

Counterpoint: I saw no Javadoc on your public methods. You were so close...


Start using Java 8

There are a number of places that would benefit from the newer goodies from Java 8 (assuming you're not restricted in Java versions?).

In getNextSocket() you're essentially filtering a list based on a condition, and trying to get the first match:

public Optional<SocketHolder> getNextSocket() {
    // For the sake of consistency make sure to use the same map instance
    // in the whole implementation of my method by getting my entries
    // from the local variable instead of the member variable
    Map<Datacenters, List<SocketHolder>> liveSocketsByDatacenter =
        this.liveSocketsByDatacenter.get();
        List<Datacenters> dcs = Datacenters.getOrderedDatacenters();
        return dcs.stream()
            .map(dc -> getLiveSocket(liveSocketsByDatacenter.get(dc)))
            .findFirst(); // This returns an Optional. And the whole process is lazy.
    }
}

Also in getLiveSocket(List<SocketHolder> listOfEndPoints) the same can be done:

// The list of live sockets
List<SocketHolder> liveOnly = listOfEndPoints.filter(socketHolder - > socketHolder.isLive())
                                             .collect(Collectors.toList());

By the way, in the same method, the shuffling is excessive. You're painfully shuffling (with multiple copies etc.) an entire array, when you just needed to get one instance? Do this instead:

private static final Random random = new Random();
(...)
return Optional.of(liveOnly.get(random.nextInt(liveOnly.size())));

The other methods would not benefit so much from Java 8: streaming a Map is not nearly as fun, and would be more verbose.

However I would strongly encourage you to try to break those double-for into a for and a method call to a for. In the same fashion, the for { try { ... } } would benefit from having its try block extracted to a method of its own, for readability.


Constants

This bit is slightly worrying in updateLiveSockets():

Map<Datacenters, ImmutableList<String>> socketsByDatacenter = Utils.SERVERS;

First of all you're using a Constants class. You should avoid those and prefer dedicated enums.

Especially here, you have a constant pointing to a list which I assume is immutable (?). If this is all known at compile-time, it'd be much tidier as an enum.

Finally the naming is odd:

  • The constant class's name (Utils) is ultra-broad. If using an enum this would be solved. Since those servers are DataCenters, you hsould make that constant a member of the DataCenter class.
  • The constant calls itself SERVERS which makes me expect a List. However it's a Map linking a DataCenter to a List of addresses. Why not call it DATACENTER_ADDRESSES?

Generating IDs

// Set random identity to make tracing easier
String identity = String.format("%04X-%04X", random.nextInt(), random.nextInt());

Try using Java's UUID class to generate unique IDs. Using the class with the correct name makes it obvious you're generating an ID. Using a standard class makes it reusable and you get the certainty of it being bug-free. Also this prevents reinventing the wheel. Also those have very good garantees.


Magic Values

There are a few magic values scattered accross the code, like those timeout(7), and the socket refresh frequency. Make those values explicitly named.


Synchronization

Not much to say, you went the route of immutable collections with an atomic reference at the root, so that's really solid for read-only operations.

I do not believe the sychronized keyword is necessary. You got a hint: the method is private, and only called every 30sec. But it won't hurt, either.

\$\endgroup\$
0
\$\begingroup\$

Consider concurrent collections over synchronization

If your primary worry is thread safety, since Java 1.5, there are collections you may utilize that, reliably, abstract this concern away. They are considered unconditionally thread-safe, containing sufficient synchronization that their instances do not need to explicitly handle any synchronizing, in your case you would use ConcurrentMap<Datacenters, ImmutableList<String>> socketsByDatacenter = new ConcurrentHashMap<>();

Upon further inspection, you effectively handle concurrent through immutability, the above is nonetheless worth noting for reference. Although, given the aforementioned concurrent collection or your schema synchronization only serves to unnecessarily slow down performance. Synchronizers lock a block and wait for completion but in either case it serves no purpose.

\$\endgroup\$

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.