12.8 Low-Level Communication Optimizations

There are a number of optimizations you can make to the low-level communications infrastructure. These optimizations can be difficult to implement, and it is usually easier to buy these types of optimizations than to build them.

12.8.1 Compression

Where the distributed application is transferring large amounts of data over a network, the communications layer can be optimized to support compression of the data transfers. In order to minimize compression overhead for small data transfers, the compression mechanism should have a filter size below which compression is not used for data packets.

The JDK documentation includes an extended example of installing a compression layer in the RMI communications layer (the main documentation index page leads to RMI documentation under the "Enterprise Features" heading). The following code illustrates a simple example of adding compression into a communications layer. The bold type shows the extra code required:

void writeTransfer(byte[  ] transferbuffer, int offset, int len)
{
  if (len <= 0)
    return;
  int newlen = compress(transferbuffer, offset, len);
  communicationSocket.write(len);
  communicationSocket.write(newlen);
  communicationSocket.write(transferbuffer, offset, newlen);
  communicationSocket.flush(  );
}
   
byte[  ] readTransfer(  )
  throws IOException
{
  int len = communicationSocket.read(  );
  if (len <= 0)
    throw new IOException("blah blah");
  int newlen = communicationSocket.read(  );
  if (newlen <= 0)
    throw new IOException("blah blah");
  int readlen = 0;
  byte[  ] transferbuffer = new byte[len];
  int n;
  while(readlen < newlen)
  {
    //n = communicationSocket.read(transferbuffer, readlen, len-readlen);
    n = communicationSocket.read(transferbuffer, readlen, newlen-readlen);
    if (n >= 0)
      readlen += n;
    else
      throw new IOException("blah blah again");
  }
  int decompresslen = decompress(transferbuffer, 0, newlen);
  if (decompresslen != len)
    throw new IOException("blah blah decompression");
  return transferbuffer;
}

12.8.2 Caching

Caching at the low-level communications layer is unusual and often a fallback position where the use of the communications layer is spread too widely within the application to retrofit low-level caching in the application itself. But caching is generally one of the best techniques for speeding up client/server applications and should be used whenever possible, so you could consider low-level caching when caching cannot be added directly to the application. Caching at the low-level communications layer cannot be achieved generically. The following code illustrates an example of adding the simplest low-level caching in the communications layer. The bold type shows the extra code required:

void writeTransfer(byte[  ] transferbuffer, int offset, int len)
{
  if (len <= 0)
    return;
  //check if we can cache this code
  CacheObject cacheObj = isCachable(transferbuffer, offset, len);
  if (cacheObj != null)
  {
    //Assume this is simple non-interleaved writes, so we can simply
    //set this cache obj as the cache to be read. The isCachable(  )
    //method must have filled in the cache, so it may include a
    //remote transfer if this is the first time we cached this object.
    LastCache = cacheObj;
    return;
  }
  else
  {
    cacheObj = null;
    realWriteTransfer(transferbuffer, offset, len);
  }
}
   
void realWriteTransfer(byte[  ] transferbuffer, int offset, int len)
{
  communicationSocket.write(len);
  communicationSocket.write(transferbuffer, offset, len);
  communicationSocket.flush(  );
}
   
byte[  ] readTransfer(  )
  throws IOException
{
  if (LastCache != null)
  {
    byte[  ] transferbuffer = LastCache.transferBuffer(  );
    LastCache = null;
    return transferbuffer;
  }
  int len = communicationSocket.read(  );
  if (len <= 0)
    throw new IOException("blah blah");
  int readlen = 0;
  byte[  ] transferbuffer = new byte[len];
  int n;
  while(readlen < newlen)
  {
    n = communicationSocket.read(transferbuffer, readlen, len-readlen);
    if (n >= 0)
      readlen += n;
    else
      throw new IOException("blah blah again");
  }
  return transferbuffer;
}

12.8.3 Transfer Batching

Batching can be useful when your performance analysis indicates there are too many network transfers occurring. The standard batching technique uses two cutoff values: a timeout and a data limit. The technique is to catch and hold all data transfers at the batching level (just above the real communication-transfer level) and send all data transfers together in one transfer. The batched transfer is triggered either when the timeout is reached or when the data limit (which is normally the batch buffer size) is exceeded. Most message-queuing systems support this type of batching. The following code illustrates a simple example of adding batching to the communications layer. The bold type shows the extra code required:

//method synchronized since there will be another thread
//which sends the batched transfer if the timeout is reached
void synchronized writeTransfer(byte[  ] transferbuffer, int offset, int len)
{
  if (len <= 0)
    return;
  if (len >= batch.length - 4 - batchSize)
  {
    //batch is too full to take this chunk, so send off the last lot
    realWriteTransfer(batchbuffer, 0, batchSize);
    batchSize = 0;
    lastSend = System.currentTimeMillis(  );
  }
  addIntToBatch(len);
  System.arraycopy(transferbuffer, offset, batchBuffer, batchSize, len);
  batchSize += len;
}
   
void realWriteTransfer(byte[  ] transferbuffer, int offset, int len)
{
  communicationSocket.write(len);
  communicationSocket.write(transferbuffer, offset, len);
  communicationSocket.flush(  );
}
   
//batch timeout thread method
void run(  )
{
  int elapsedTime;
  for(;;)
  {
    synchronized(this)
    {
      elapsedTime = System.currentTimeMillis(  ) - lastSend;
      if ((elapsedTime >= timeout) && (batchSize > 0))
      {
        realWriteTransfer(batchbuffer, 0, batchSize);
        batchSize = 0;
        lastSend = System.currentTimeMillis(  );
      }
    }
    try{Thread.sleep(timeout - elapsedTime);}catch(InterruptedException e){  }
  }
}
   
realReadTransfer(  )
  throws IOException
{
  //Don't socket read until the buffer has been completely used
  if (readBatchBufferlen - readBatchBufferOffset > 0)
    return;
   
  //otherwise read in the next batched communication
  readBatchBufferOffset = 0;
  int readBatchBufferlen = communicationSocket.read(  );
  if (readBatchBufferlen <= 0)
    throw new IOException("blah blah");
  int readlen = 0;
  byte[  ] readBatchBuffer = new byte[readBatchBufferlen];
  int n;
  while(readlen < readBatchBufferlen)
  {
    n = communicationSocket.read(readBatchBuffer, readlen,
                                 readBatchBufferlen-readlen);
    if (n >= 0)
      readlen += n;
    else
      throw new IOException("blah blah again");
  }
}
   
byte[  ] readTransfer(  )
  throws IOException
{
  realReadTransfer(  );
  int len = readIntFromBatch(  );
  if (len <= 0)
    throw new IOException("blah blah");
  byte[  ] transferbuffer = new byte[len];
  System.arraycopy(readBatchBuffer, readBatchBufferOffset,
                         transferBuffer, 0, len);
  readBatchBufferOffset += len;
  return transferbuffer;
}

12.8.4 Multiplexing

Multiplexing is a technique where you combine multiple pseudo-connections into one real connection, intertwining the actual data transfers so that they use the same communications pipe. This reduces the cost of having many communications pipes (which can incur a heavy system load) and is especially useful when you would otherwise be opening and closing connections a lot: repeatedly opening connections can cause long delays in responses. Multiplexing can be managed in a similar way to the transfer-batching example in the previous section.