Issues with concurrent reads from two serial ports via RxTx in a java application

Question

Some Background On My Issue

I have an application that reads from two serial ports. These serial ports are tapping into messages between two pieces of hardware that are custom made by the company I work for. One serial port is tapping into the messages from hardware A to hardware B, while the other carries the messages in the other direction. I need to read, parse, and display these messages in order to help debug issues that we have been having with apparently dropped and duplicate messages. More than anything else, my application must display the messages as they are sent (i.e., in the order that they are sent and the number of times they are sent), and with the correct timestamp.

My application is written in java, and interfaces with the serial ports via RxTx. I have been running my application on a 64-bit Windows 7 PC.

The vast majority of the messages that are sent (in both directions) are requests and ACKs (acknowledgements of those requests). Each request message contains a sequence number and the corresponding ACK will contain the same sequence number. When the system is not busy, my program seems to work near perfectly. However, when the messages start coming very quickly, I experience issues with my program.

The problem that I am experiencing is that it seems that my application is reading multiple messages from one serial port and then reading from the other one, rather than reading from both near simultaneously.

The core of my application uses 5 threads. Two of these are readers that only read a byte from the port and then place that byte in a queue. Each of these queues is shared with another thread that parses these bytes into a message. (So there are two processing threads, one for each reader). Then, both of these processing threads places these parsed messages into the another queue. A fifth thread that manages the displaying of these messages then takes messages from this queue.

(Unmentioned above is my GUI, which runs through the EDT).

It seems to me like the problem lies in (or possibly before) the reader threads, and it appears to have something to do with concurrency or something like sharing CPU timeslices. I just don't know if this holdup is occurring at the OS level (the operating system reading from one port than the other), at the RxTx level, or in my code.

My reader thread's code is at the bottom of this question.

My Reasoning:

1. I affiliate each byte with a timestamp that includes a microsecond approximation (which is obtained via a technique adapted from Jason Smith's answer to another question). When I've see that I am reading multiple messages from one port (port A) and then from the other port (port B), I have noticed that the timestamp (and hence the time of reading) of the first byte of the first message read from port B occurred after the timestamp of the first byte of any of the messages read from port A. In other words, it appears that the port A reader thread never yielded to the port B reader thread, despite calling Thread.yield() after reading each byte (as can be seen in the reader thread code at the bottom of this question). To illustrate, I have seen sequences of messages like this:

   [From Hardware A, To Hardware B] [Timestamp: 0 us] Request for 0x01
   [From Hardware A, To Hardware B] [Timestamp: 60 us] ACK for 0x02
   [From Hardware A, To Hardware B] [Timestamp: 120 us] ACK for 0x03
   [From Hardware B, To Hardware A] [Timestamp: 180 us] ACK for 0x01
   [From Hardware B, To Hardware A] [Timestamp: 240 us] Request for 0x02
   [From Hardware B, To Hardware A] [Timestamp: 300 us] Request for 0x03
   

   But I would expect these messages to arrive like this:

   [From Hardware A, To Hardware B] [Timestamp: 0 us] Request for 0x01
   [From Hardware B, To Hardware A] [Timestamp: 60 us] ACK for 0x01
   [From Hardware B, To Hardware A] [Timestamp: 120 us] Request for 0x02
   [From Hardware A, To Hardware B] [Timestamp: 180 us] ACK for 0x02
   [From Hardware B, To Hardware A] [Timestamp: 240 us] Request for 0x03
   [From Hardware A, To Hardware B] [Timestamp: 300 us] ACK for 0x03  
   

2. I know that this issue lies in my code rather than in the other pieces of hardware because I am seeing ACKs that have been timestamped as having arrived before the requests to which the ACK is responding. This is programmatically impossible, because the hardware that is sending the ACK would not know the sequence number or type of message to send if it had not received the request prior to sending the ACK.

3. I have decreased the time required by the reader thread to place a byte that has been read in the queue to about 1.5 uSec (based on the average benchmark after reading and queuing 500,000 bytes). The fastest I have seen messages coming in is around 1 message per 50 to 60 uSec, so if the two reader threads were yielding to each other after reading a byte, I believe that the timestamps of the first byte read from messages on each serial port to be fairly accurate.

The Bottom Line

The solutions that have occurred to me have included changing the external library I use to communicate with serial ports, somehow fixing my code (if the issue lies in it), running my application on PC with a linux-based OS, or somehow re-writing at least the reading component of my application in a lower-level language than java.

Each of these solutions could be the solution I need, or they could be a dead-end and a huge waste of time. Hopefully one of you has dealt with this issue before or sees an issue that I haven't so I can reduce the amount of time I waste trying to deal with this problem!

My Reader's Logic

My reader is a private inner class. The following four constants are defined in its containing class:

public static final long BYTE_MASK = 0x7F80000000000000L;
public static final int BYTE_SHIFT = 55;
public static final long TIMESTAMP_MASK = 0x007FFFFFFFFFFFFFL;
public static final long TIMESTAMP_SHIFT = 0;

The creation of the SerialPorts and of their corresponding InputStreams.
COMM_PORT_ID is an array of the CommPortIdentifier objects for my desired COM ports.
serialPorts is an array of the SerialPort objects for my desired COM ports.
commInput is an array of the InputStream objects for my desired COM ports.
All of this occurs in the containing class.

for (int i = 0; i < 2; i++)
{
      serialPorts[i] = (SerialPort) COMM_PORT_ID[i].open("SERIAL CONNECTION", 2000);
      serialPorts[i].setSerialPortParams(BAUD_RATE, DATA_BITS, STOP_BITS, PARITY);
      serialPorts[i].setFlowControlMode(SerialPort.FLOWCONTROL_NONE);
      serialPorts[i].enableReceiveTimeout(1000);
      serialPorts[i].setInputBufferSize(3);
      serialPorts[i].disableReceiveThreshold();
      serialPorts[i].notifyOnDataAvailable(false);
      serialPorts[i].notifyOnFramingError(true);
      serialPorts[i].notifyOnOutputEmpty(false);
      serialPorts[i].notifyOnOverrunError(true);
      commInput[i] = serialPorts[i].getInputStream();
}

The creation of the SerialPortReader objects and that starting of the threads (also occurs in the containing class):

comOneReader = new SerialPortReader(commInput[0],
                                    comOneInputByteQueue);
comTwoReader = new SerialPortReader(commInput[1],
                                    comTwoInputByteQueue);

Thread comOne = new Thread(comOneReader, "COM1 Reader");
Thread comTwo = new Thread(comTwoReader, "COM2 Reader");

comOne.setPriority(Thread.MAX_PRIORITY);
comTwo.setPriority(Thread.MAX_PRIORITY);

comOne.start();
comTwo.start();

The reader itself:

private class SerialPortReader implements Runnable
{
    private final long startSystemTime;
    private final long startNano;
    private final InputStream input;
    private AtomicBoolean reading; // used to stop and start the reading of bytes
    private final LinkedTransferQueue<Long> toProcessorByteQueue;
    private long holder;

    public SerialPortReader(InputStream input,
                            LinkedTransferQueue<Long> toProcessorByteQueue)
    {
        this.input = input;
        this.toProcessorByteQueue = toProcessorByteQueue;
        reading = new AtomicBoolean();
        startSystemTime = System.currentTimeMillis();
        startNano = System.nanoTime();
    }

    @Override
    public void run()
    {
        long in = -1;
        reading.set(true);
        while (reading.get())
        {
            try
            {

                in = -1;

                // Read the next byte from the serial port, blocking until one is available
                if ((in = input.read()) != -1)
                {   
                    // I store the byte and a timestamp (in microseconds) in a long
                    // Skipping bit 63 (the sign bit), I will store the byte from the serial read
                    // in bits 62 - 55.  Then the remainder (bits 54-0) will be the timestamp                  
                    holder = ((in & 0xFF)) << BYTE_SHIFT;
                    holder |= ((startSystemTime * 1000L) + ((System.nanoTime() - startNano) / 1000L)) & TIMESTAMP_MASK;

                    // Add the long to the queue
                    toProcessorByteQueue.add(holder);                      
                }

                Thread.yield();
            }
            catch (IOException ioe)
            {
                // Deal with Exception
            }
        }
    }

    public void stop()
    {
        reading.set(false);
    }

}

Sorry for such a long question, but hopefully I was able to answer any questions you might have!

Answer 1

Been a long time since I wrote serial port code :-)

Where did commInput[0] and [1] come from? Unless you are very particular about how how you created those input streams, there's bound to be a lot of buffering between your java program and the actual hardware. It's likely to happen in the java libraries, it's likely to happen in the WinDoze user-mode libraries, and it's probably happening in the device driver. It probably will be even worse than that if your serial port device is connected through a USB port.

I don't remember what options are available to eliminate some of that buffering, but you may not be able to do it through Java. In the worst case, and depending on how precisely you need to measure time, you might have to build your own device driver.

Quite frankly, if I were given your task, I would seriously consider programming a microcontroller to collect and timestamp the data, and then upload it to the PC for analysis.

---

FWIW: You may be going overboard with the threads. Why not parse the protocol in the same threads that read the bytes? Whats to gain by having the protocol parser threads read from a queue instead of reading directly from the device?

Also, you might want to reconsider the priorities. If memory is finite, and if the program has to run for a long time, then you'll want threads that pulling data off a queue to run at higher priorities than the threads putting stuff in. That bit of advice doesn't apply if you can afford to collect all the data first, and then analyze it later, but if you can afford to do that, then why not split it up into two separate programs? One to read the data into a file, and another to analyze the file.

---

Best of Luck!

Answer 2

You need set other kind event of notify on serialPorts[i].notifyO.... such as: serialPort.notifyOnRingIndicator(true)