Blocking on calls to read/writeObject

[Mike Curwen]

Apologies if this is a newbie question. I've had very little call for Streams so far in my coding life. I've inherited some code that I'm trying to make some sense of.

Imagine two instances of an object. They are communicating over ethernet ports. Each object has an
Object RXTransmission(InputStream is, OutputStream os)
and
boolean TXTransmission(InputStream is, OutputStream os, Object o)

The point is to transmit the 'o' object from one place to the other.

The code uses CRC32 checking to determine if the object was successfully sent. And it appears that they do so in a hand-shaking manner.

"
TX:Here's the object.
RX: I have that object, I've calculated the checksum, I'm sending you the checksum.
TX:I have the checksum, and it matches the one I made before sending. I'm sending you a Boolean 'true'.
RX:I see that 'true' and so I'm convinced the object was received successfully.
"

Here's the real question.

Will calls to readObject() (on ObjectInputStream) block until there is an object to read on that stream? Or will they return immediately with 'null' ? I've read somewhere that stream in java are "blocking". Is this what is meant (and what NIO is supposed to 'fix')?

The answer to the first question must be yes, or else I'll have to scratch my head a bit more.

[Thomas Paul]

readObject() is a blocking call.
And so is writeObject().
[ April 03, 2003: Message edited by: Thomas Paul ]

[Mike Curwen]

So I'm trying to research where giants have tread (or so I suspect). How *does* one detect a blocked method? Not at all?

No, I can't use nio, I'm stuck with 1.1.8

I've got a hint from a rancher (Jim Yingst) on this link http://gee.cs.oswego.edu/dl/cpj/s3.1.html
(the Follow-up section at the top)

Here's a better one:
http://www.experts-exchange.com/Programming/Programming_Languages/Java/Q_20143709.html

The reason I'm enquiring further is that I've heard from co-workers that sometimes they have to go in and kill the "listener" (running on a company box here), or sometimes the "sender" (at a client's site) will stop working until they reboot the system next morning. I'm thinking that the blocked i/o could be to blame. It will just sit there until ... until you kill the JVM? It all seems so extreme, but I guess that's why everyone is so happy with nio ?

Apologies again if this is neophyte.

[Peter den Haan]

I'm not sure if this will work on such an early JVM, but on recent versions running java in a console and pressing Control-Break gives you a very interesting thread dump. If you're on a Unix platform, I'm not sure either; try kill -SIGQUIT.
- Peter
[ April 04, 2003: Message edited by: Peter den Haan ]

[Jim Yingst]

I've got a hint from a rancher (Jim Yingst) on this link http://gee.cs.oswego.edu/dl/cpj/s3.1.html
(the Follow-up section at the top)
Ah, my moment of fame. Unfortunately my contribution there was to point out that you can't use available() to prevent blocking if you also need to be able to detect an end-of-file. When you get to the end of file, available() returns 0, and you don't have any way of knowing if that means the file is at an end, or the next read() will block - until you actually attempt the read(), at which point you may block indefinitely - which is what we were trying to avoid in the first place. So you can't really use the method Doud Lea was originally suggesting.
I think that to implement some sort of timeout using pre-nio techniques, you basically need to have two threads - one to do the read, and one to wake up periodically to see if there's been any change. If there's no change after a set timeout period, the best way I know to stop the block is to close() the stream that's blocked. Be aware that this isn't guaranteed to work (unless you're using nio-derived streams) - in some situations, close() may simply block as well. Depends how the stream is set up, I think, and what platform and JVM you're using, probably. But it often works, and it's worth a try, especially if no one has a beter alternative.
[ April 03, 2003: Message edited by: Jim Yingst ]

[Mike Curwen]

What if my input stream is for communicating through a COM port (modem). I'm not reading a file, so I don't care about EOF.

If available() returns 0, then maybe I should sleep for 5 seconds and try again... And if it's STILL not available(), I can assume that "ok, that's all she wrote!"

If on the other hand, that is NOT all she wrote, then the client that is trying to write to the stream will block (or will it?)

Something like that?

[Jim Yingst]

Well, maybe. You're using a Socket then? Is the socket going to be permanently open? Is there some end-of-session message that tells you when you're done? Or do you just catch a SocketException if the socket is close from the other side while you're still waiting for more info?
Checking available() again after 5 sec may work, but it also seems like it will often introduce an unnecessary lag. Consider: A requests an object from B. There's a little latency in the network - maybe it takes .5 sec before A sees a response. If A checked availabe() immediately after the request, it returned 0. Now A is waiting for 5 sec, even though B's response started arriving after .5 sec. Perhaps if available() == 0, A should recheck every .5 sec or so, and then keep rechecking unless the total delay exceeds 5 sec.
Actually, I think I've got an old class that may be of help to you - let me see if I can find it.
If on the other hand, that is NOT all she wrote, then the client that is trying to write to the stream will block (or will it?)
No, if available() > 0, the stream will not block - it will return at least as many bytes as are available (unless the byte[] buffer you're storing them in is too small).

[Mike Curwen]

I'm not entirely sure how the package works. This is code I 'inherited' at work. The package import at the top is gnu.io , for which the javadocs are utterly in want of something more than just the method signature.

He has a comment beside that import that reads "for win32, replace this with javax.comm" (i'm going from memory, could it be javx.com ?)

I'm at home for the weekend, so I don't have the code with me (that uses this gnu.io package). But the thing is.. I'd have to make those suggested changes in the gnu.io package. I think the only thing we do is try to .open() a communication port. Once the SerialPortEvent is sent to our registered listener, we have to see if it's a "ring" (I assume someone is dialing in to our modem) or if it's the other thing.. someone sending data. So the blocking happens inside their package I think.

As for your class... I wouldn't mind seeing what you have. It will probably help clarify a few things re: sockets and i/o

[Jim Yingst]

Here's the old class I was working on. It's still somewhat rough, as I found a workaround to the original problem that motivated me (now forgotten). Still has debug print statements and the like. This has nothing to do with Sockets specifically - it's an attempt to make a FilterInputStream which can be chained to any other InputStream in order to provide timeout capability. However whether it works or not depends on whether the underlying stream can be closed without blocking. The main() method attempts to use a timeout on System.in, but unforturnately this doesn't work on my JDK/OS - close() simply blocks. The terminate() method shows other more aggressive things I was trying to see if they'd work. I'd say, give this a try to see if it works for whatever stream types you're dealing with. If close() blocks, or you need code you can trust on other unknown platforms as well, try something uing available() as previously discussed instead. That will probably be slower than TimeoutInputStream, but can be guaranteed much more easily.

import java.io.*;
 
/*
 *  Each TimeoutInputStream has an associated TimerThread which monitors
 *  whether it's gone too long without any activity.  All communication
 *  between the TimeoutInputStream and its TimerThread is synchronized
 *  using the TimerThread instance (held in instance variable "timer")
 *  as the monitor, rather than the TimeoutInputStream instance.
 *  This is done to ensure that no outside code can send notify() or
 *  interrupt() through the monitor.
 */
public class TimeoutInputStream extends FilterInputStream
{
  private TimerThread timer = new TimerThread();
 
  private int timeout; // number of milliseconds to wait before timing out
 
  private long lastActivityTime; // value read from System.currentTimeMillis() at last activity
 
  private boolean isReading;
  private boolean isOpen = true;
  private Thread readingThread;
 
  public TimeoutInputStream(InputStream inner, int timeout)
  {
    super(inner);
 
    if (timeout <= 0)
      throw new IllegalArgumentException("timeout = " + timeout);
 
    synchronized (timer)
    {
      this.timeout = timeout;
      timer.start();
    }
  }
  /**
   *  Users are strongly encouraged to use one of the other read() methods instead of this one 
   *  - this has a lot of performance overhead to read just one byte.  You might as well grab as 
   *  many bytes as you can at once.
   */ 
  public int read() throws IOException
  {
    startRead();
    int value = in.read();
    endRead();
    return value;
  }
 
  public int read(byte[] b) throws IOException
  {
    startRead();
    int value = in.read(b, 0, b.length);
    endRead();
    return value;
  }
 
  public int read(byte[] b, int off, int len) throws IOException
  {
    startRead();
    int value = in.read(b, off, len);
    endRead();
    return value;
  }
 
  private void startRead()
  {
    synchronized (timer)
    {
      lastActivityTime = System.currentTimeMillis();
      isReading = true;
      readingThread = Thread.currentThread();
      System.out.println("\tstartRead: " + lastActivityTime);
      timer.notify();
    }
  }
 
  private void endRead()
  {
    synchronized (timer)
    {
      lastActivityTime = System.currentTimeMillis();
      isReading = false;
      readingThread = null;
      System.out.println("\tendRead: " + lastActivityTime);
    }
  }
 
  public void close() throws IOException
  {
    super.close();
    synchronized (timer)
    {
      isOpen = false;
      timer.notifyAll();
    }
  }
 
  private class TimerThread extends Thread
  {
    public void run()
    {
      synchronized (this)
      {
        while (isOpen)
        {
          if (!isReading)
          {
            System.out.println("\twait for startRead");
            // wait until notified (indicating start of read, or a close)
            try { wait(); }
            catch (InterruptedException e) {}
          }
          else
          {
            long timeNow = System.currentTimeMillis();
            long alreadyWaited = timeNow - lastActivityTime;
            System.out.println("\ttiming: now: " + timeNow + "\twaited: " + alreadyWaited);
            if (alreadyWaited >= timeout)
            {
              terminate();
            }
            else
            {
              System.out.println("\twait now for " + (timeout - alreadyWaited));
              try { wait(timeout - alreadyWaited); }
              catch (InterruptedException e)  {}
            }
          }
        }
      }
    }
  }
 
  private void terminate()
  {
    // hope this interrupts the blocking read
    System.out.println("closing stream...");
    try { close(); }
    catch (IOException e) { e.printStackTrace(); }
    System.out.println("close() completed");
 
    /*
    try { Thread.sleep(2000); }
    catch (InterruptedException e) {}
 
    System.out.println("interrupt parent thread");
    readingThread.interrupt();
 
    try { Thread.sleep(2000); }
    catch (InterruptedException e) {}
 
    System.out.println("stop parent thread");
    readingThread.stop();
 
    try { close(); }
    catch (IOException e) { e.printStackTrace(); }
    */
  }
 
  public static void main(String[] args) throws IOException
  {
    System.out.println("type stuff - hit return to send (3 sec timeout)");
    InputStream in = new TimeoutInputStream(System.in, 3000);
    while (true)
    {
      int n = in.read();
      System.out.println(n);
    }
  }
}

[ April 05, 2003: Message edited by: Jim Yingst ]