2
\$\begingroup\$

I am using below class to send data to our messaging queue by using socket either in a synchronous way or asynchronous way as shown below. It depends on requirement whether I want to call synchronous or asynchronous method to send data on a socket. Most of the times we will send data through aysnc way but sometimes I may need to send data through sync way.

  • sendAsync - It sends data asynchronously without any timeout. After sending (on LINE A) it adds to cache bucket so that if acknowledgement is not received then it will retry again from the background thread which is started in SendToQueue constructor only.
  • send - It sends data synchronously on a socket. It internally calls doSendAsync method and then sleep for a particular timeout period and if acknowledgement is not received then it removes from cache bucket so that we don't retry again.

So the only difference between those two above methods is - For async case, I need to retry at all cost if acknowledgement is not received but for sync I don't need to retry at all and that's why I am storing more state in a PendingMessage class.

ResponsePoller is a class which receives the acknowledgement for the data that was sent to our messaging queue and then calls handleAckReceived method below to remove the address so that we don't retry after receiving the acknowledgement.

public class SendToQueue {
  private final ExecutorService cleanupExecutor = Executors.newFixedThreadPool(5);
  private final ScheduledExecutorService executorService = Executors.newScheduledThreadPool(3);

  private final Cache<Long, PendingMessage> cache =
      CacheBuilder
          .newBuilder()
          .maximumSize(1000000)
          .concurrencyLevel(100)
          .removalListener(
              RemovalListeners.asynchronous(new LoggingRemovalListener(), cleanupExecutor)).build();

  private static class PendingMessage {
    private final long _address;
    private final byte[] _encodedRecords;
    private final Socket _socket;
    private final boolean _retryEnabled;
    private final Object _monitor = new Object();
    private long _sendTimeMillis;
    private boolean _acknowledged;

    public PendingMessage(long address, byte[] encodedRecords, Socket socket, boolean retryEnabled) {
      _address = address;
      _sendTimeMillis = System.currentTimeMillis();
      _encodedRecords = encodedRecords;
      _socket = socket;
      _retryEnabled = retryEnabled;
    }

    public synchronized boolean hasExpired() {
      return System.currentTimeMillis() - _sendTimeMillis > 500L;
    }

    public void markResent() {
      _sendTimeMillis = System.currentTimeMillis();
    }

    public synchronized boolean shouldRetry() {
      return _retryEnabled && !_acknowledged;
    }

    public synchronized boolean waitForAck() {
      try {
        _monitor.wait(500L);
        return _acknowledged;
      } catch (InterruptedException e) {
        return false;
      }
    }

    public synchronized void ackReceived() {
      _acknowledged = true;
      _monitor.notifyAll();
    }

    public long getAddress() {
      return _address;
    }

    public byte[] getEncodedRecords() {
      return _encodedRecords;
    }

    public Socket getSocket() {
      return _socket;
    }
  }

  private static class Holder {
    private static final SendToQueue INSTANCE = new SendToQueue();
  }

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

  private void handleRetries() {
    List<PendingMessage> messages = new ArrayList<>(cache.asMap().values());
    for (PendingMessage m : messages) {
      if (m.hasExpired()) {
        if (m.shouldRetry()) {
          m.markResent();
          doSendAsync(m, m.getSocket());
        } else {
          // Or leave the message and let send remove it
          cache.invalidate(m.getAddress());
        }
      }
    }
  }

  private SendToQueue() {
    executorService.submit(new ResponsePoller()); // another thread which receives acknowledgement
                                                  // and then delete entry from the cache
                                                  // accordingly.
    executorService.scheduleAtFixedRate(new Runnable() {
      @Override
      public void run() {
        handleRetries();
      }
    }, 0, 1, TimeUnit.SECONDS);
  }

  public boolean sendAsync(final long address, final byte[] encodedRecords, final Socket socket) {
    PendingMessage m = new PendingMessage(address, encodedRecords, socket, true);
    cache.put(address, m);
    return doSendAsync(m, socket);
  }

  private boolean doSendAsync(final PendingMessage pendingMessage, final Socket socket) {
    ZMsg msg = new ZMsg();
    msg.add(pendingMessage.getEncodedRecords());
    try {
      // send data on a socket LINE A
      return msg.send(socket);
    } 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 {
      // Alternatively (checks that address points to m):
      // cache.asMap().remove(address, m);
      cache.invalidate(address);
    }
  }

  public void handleAckReceived(final long address) {
    PendingMessage m = cache.getIfPresent(address);
    if (m != null) {
      m.ackReceived();
      cache.invalidate(address);
    }
  }
}

Are there any potential issues or any race conditions in my above code? Is there any better or efficient way to do the same thing?

  • So for "async" case, I am trying to avoid all the race conditions so that I don't send same data twice if acknowledgement is not received.
  • Is there any other race conditions when messages are retried as well?
  • For "sync" case, I don't want to retry at all and I am using wait + notify concept here so that I don't block the thread if response comes back quickly.

I am using Guava Cache here instead of standard ConcurrentHashMap because if for whatever reason, I cannot send data on a socket (let's say socket is bad or dead) then I don't want my cache to keep growing and ultimately run out of memory so that is why I am using guava cache which can start dropping records if it reaches the limit and my application will keep running.

Any better way to implement what I am doing or get rid of synchronization altogether? And maybe we can create an interface for this class?

\$\endgroup\$
  • 1
    \$\begingroup\$ It's hard to determine since you haven't discussed how you will interact with each method and from which threads. Are you using a boss/worker model? Are you using in/out threads to split up work? Etc. Which methods are being called by which threads? \$\endgroup\$ – Breeze Dec 7 '17 at 0:09
  • \$\begingroup\$ @xTrollxDudex I did mentioned that I will be calling only two methods sendAsync or send method. Multiple reader threads will call either of these two methods to send data and then everything is self explanatory if you follow the code I think. \$\endgroup\$ – david Dec 7 '17 at 0:52
  • 2
    \$\begingroup\$ One thing to think about is what happens if the system terminates abruptly while you're in the middle of attempting a retry or waiting for acknowledgement. Does the data need to be retried even if the system has terminated? You'll need some kind of disk file/journaling system if so. I think Java Message Queues implements this kind of functionality. \$\endgroup\$ – markspace Dec 19 '17 at 20:21
1
\$\begingroup\$

First pass reading:

public class SendToQueue

You're naming a class after an action. This is a pretty surefire indicator that something is wrong with the design. Let's see what happens next.

  private final ExecutorService cleanupExecutor = Executors.newFixedThreadPool(5);

Why does a cleanupExecutor need 5 threads? That seems like an awful lot for threads that are only used at the end of a class' lifecycle... At least that's what the name implies...

  private final Cache<Long, PendingMessage> cache =
      CacheBuilder
          .newBuilder()
          .maximumSize(1000000)
          .concurrencyLevel(100)
          .removalListener(
              RemovalListeners.asynchronous(new LoggingRemovalListener(), cleanupExecutor)).build();

I'd format this a little differently:

  private final Cache<Long, PendingMessage> cache = CacheBuilder.newBuilder()
        .maximumSize(1_000_000)
        .concurrencyLevel(100)
        .removalListener(RemovalListeners.asynchronous(new LoggingRemovalListener(), cleanupExecutor))
        .build()

YMMV, it looks a little cleaner to me, though. Also it explains why cleanupExecutor has so many threads. Doesn't explain why it's not named loggingService, but eh.

  private static class PendingMessage {
    private final long _address;
    private final byte[] _encodedRecords;
    private final Socket _socket;
    private final boolean _retryEnabled;
    private final Object _monitor = new Object();
    private long _sendTimeMillis;
    private boolean _acknowledged;

Wow... This wants a separate file. If you need to restrict the visibility of that class, you might make it default-visible, but this is beyond the scope of an inner class.

Also I dislike that you're using the convention of prefixing member fields with _. It's just not something that you need with a modern IDE IME. Neither does it terribly help readability, or even discoverability of available fields.

  return System.currentTimeMillis() - _sendTimeMillis > 500L;

In both the code above and below, 500L is a Magic Number

    _monitor.wait(500L);
  private static class Holder {
    private static final SendToQueue INSTANCE = new SendToQueue();
  }

Ugh. A singleton. Really? I mean ... nothing against you or the style of using Singletons. But singletons suck and in 99.999% of the cases I've seen here on CodeReview they just were not used in a place where you should use them.

You shouldn't force a class into a Singleton scope, unless absolutely necessary. If you think it's absolutely necessary, consider first why you actually want to enforce a "only once in the world ever". As I see it, most of the time you actually want to move away from the basic concept of instances, and instead have something akin to first-class functions. This ties back into the classname as well. OOP just sucks at certain things, and singletons are not the correct solution for this specific problem.

That problem is much cleaner (and easier) solved using static methods on a non-creatable class (aka. one with exclusively private constructors). And before you try the "Lazy initialization" argument: The JVM already does exactly that for static members. No need for fancy singleton implementations using an internal enum ;)

    executorService.scheduleAtFixedRate(new Runnable() {
      @Override
      public void run() {
        handleRetries();
      }
    }, 0, 1, TimeUnit.SECONDS);

On Java 8 this should be:

    executorService.scheduleAtFixedRate(this::handleRetries, 0, 1, TimeUnit.SECONDS);

Multithreading and communication correctness:

You'll want to guard against IllegalMonitorStateException of your _monitor... to do that you need to synchronize on the monitor like so:

synchronized (_monitor) {
    _monitor.wait(ACK_TIMEOUT_MILLIS);
    // or alternatively
    _monitor.notifyAll();
}

Since you then synchronize on _monitor anyways, you should drop the synchronized modifier from waitForAck and ackReceived. Note that marking an instance method as synchronized is the same as synchronize(this). That could bite you in the backside when trying to acknowledge messages.

An additional weirdness I see in your implementation is that you expect sendAsync to return the value from the first invocation of msg.send(socket). In general that'd be surprising to me. I want sendAsync to instantly return with that name. This is "terrible" from the PoV of high-throughput code that just wants to hand off finished stuff to get on with the next item. Being forced to wait for a round of I/O can make a difference there.

Finally I don't see any obvious issues with multithreading itself. One thing that is possibly dangerous is the issue that your MessageQueue could respond after the 500ms timeout that it received the "first" message, after you already resent it once. This is not really "fixable" though.

\$\endgroup\$
  • \$\begingroup\$ Thanks for detailed explanation. Since you noticed synchronization can bite for the cases of acknowledging messages? Do you think is there any better way to solve this problem which don't involve synchronization? \$\endgroup\$ – david Dec 19 '17 at 23:01
  • \$\begingroup\$ Also _acknowledged should be volatile here? I was just thinking on it so wanted to make sure on that? \$\endgroup\$ – david Dec 20 '17 at 17:07
  • \$\begingroup\$ @david since all access to _acknowledged is synchronized anyways, you don't need volatile. Synchronized gives stronger guarantees already \$\endgroup\$ – Vogel612 Dec 20 '17 at 17:19
  • \$\begingroup\$ Actually what I meant was removing synchronized keyword from shouldRetry method and adding synchronized keyword in markResent method? In that case we can make _acknowledged volatile. No? And we will also remove synchronized keyword from waitForAck and ackReceived method and put instead on the monitor itself. \$\endgroup\$ – david Dec 20 '17 at 17:47

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.