Re: Note on PushbackInputStream

[Dalibor Topic]

Am Dienstag, 21. August 2001 20:33 schrieben Sie:
> >>>>> "Dalibor" == Dalibor Topic <address@hidden> writes:
>
> Dalibor>     public void unread(int oneByte) throws IOException {
> Dalibor>       synchronized(this) {
> Dalibor>          notifyAll();
> Dalibor>          super.unread(oneByte);
>
> I think you want to call super.unread() before notifyAll().


I originally thought I could agree. The way I wrote it originally, the 
unreading thread would compete for CPU time with all the notified threads, 
which would be trying to acquire a monitor that hasn't been released by the 
unreading thread yet.

But after thinking about it, I came to the case where super.unread() throws 
an IOException. If you call super.unread() before notifyAll() , then no 
thread gets notified in the case of an IOException. All other threads that 
would like to be notified hang until some other thread does an unread that 
doesn't throw an IOException. If the IOException was "Stream closed", then 
tough luck, the waiting threads will hang forever.

So another alternative would be to do something like this:

public void unread(int oneByte) throws IOException {
   synchronized(this) {
       IOException caught = null;

       try {
         super.unread(oneByte);
       }
       catch (IOException e) {
          caught = e;
        }

        notifyAll();
        if (caught != null) {
          throw caught;
        }
   }
}

which is very ugly in my opinion. Instead, I think that using notify() 
instead of notifyAll() is a better cure for the problem. I'll explain further 
down what I mean.

> Dalibor> this solution avoids polling, and only reads when there is
> Dalibor> something available for reading. It relies on being notified
> Dalibor> that there is something available.
>
> Yes.  It is possible to do this.  However, there is an additional
> constraint you're ignoring, which is that PushbackInputStream must
> rely only on the API presented by InputStream.  That is why this bug
> is inherent in the design -- you can't do it without either extending
> the API (not allowed) or introducing a new thread (hugely expensive).


I agree that PushbackInputStream must rely on the InputStream API. Yet if 
operations over InputStreams should be thread safe, then they must internally 
use some mechanism to prevent each other from corrupting the object. The most 
likely/lightweight solution to that synchronization problem is to use the 
facilities already provided by Java. And if you decide to synchronize, the 
easiest way to do it is to synchronize on "this". 

On the other hand, if you don't care about thread-safety, then you would be 
lost anyway if several threads tried to access your InputStream at the same 
time, so nothing is lost here. My code just burns some cycles on getting and 
releasing the monitor it wouldn't really need in that particular case, but I 
don't think that outweighs the benefits of thread-safety.

That's why I think the synchronization related part of my code should work.

Regarding the wait and notification: I know that I cross the boundary of the 
API. I see two solutions to this: either make the call to wait limited to 
MAX_WAIT milliseconds, or use the unlimited waiting/notification mechanism 
consistently throughout your code.

The first solution should solve the problem, by making sure that a thread 
stuck waiting will eventually wake up, and check if there is something 
available. It also has two minor issues: you can have a "cascading bursts of 
activity" effect, if several threads go into wait at approximately the same 
time, and there is nothing to read for a while. They would all wake up every 
MAX_WAIT milliseconds, fight for the lock, and then wait again. You could 
address this effect by using random wait times within a certain span, but I'm 
getting sidetracked here. The other issue is that the value of MAX_WAIT 
should be chosen carefully.

The second solution solves the problem, too, however it does so only for the 
library. If you encounter code which does not follow the convention [like my 
polling loop attempt at thread safety] , you can be stuck waiting forever, 
since the sub-stream does not notify you. Of course, you could put it in the 
docs as a cute way to extend InputStreams while retaining thread safety etc. 
but that doesn't really help with the code that's already there.

I think a mix of the strategies (waiting up to some maximum time, 
implementation of the mechanism throughout the library) would be both 
conforming to the API as well as rather lightweight. But I might have 
misinterpreted your remark, or speculated too wildly. :-)

Finally, I'd like to revise what I said about notifyAll() in one of my 
previous mails. I thought that notifyAll() was necessary in order to make 
sure that if there are several threads waiting to read, some bytes are 
unread, and the first thread that gets to read consumes just a part of the 
unread bytes, then the other threads get a chance to read. After thinking 
about Tom's remark regarding notifyAll(), I think that would cause 
unnecessary "stampedes" of competing threads. So I'd like to propose a 
better way of notification.

An InputStream's methods can be essentially divided into methods that change 
its state (like read), and those that don't (like available). The only thing 
that matters for threads waiting to read from an InputStream is whether they 
can finally get some bytes to read after a state-modifying method was 
executed. But if they get to read something, they modify the state of the 
InputStream as well. So if instead of notifying all waiting threads you 
notify just one, and that one propagates the notification only if it gets to 
change the state, you can save a lot.

What you get are methods like this:

    public int read() throws IOException {
         synchronized(this) {
            synchronized(in) {
                 while(0 == available()) {
                    try {                       
                         wait(MAX_TIME);
                    }
                    catch (InterruptedException e) {
                         // ignore
                    }
                 }

                 // something is available for reading,
                 // and we've got both locks.

                 // the state is going to change, so
                 // let's notify another waiting thread about it.

                 notify();

                 // this read 
                 // should not block at all.
                 return super.read();
            }
         }
    }

    public void unread(int oneByte) throws IOException {
         synchronized(this) {
            // the state is going to change, so
            // let's notify another waiting thread about it.

            notify();
            super.unread(oneByte);
         }
    }

and so on for other state-modifying methods. The only exception is close, it 
needs to notify all waiting threads, since any notified thread is going to 
get an IOException and exit the read method before having a chance to notify 
another thread.

Let's say there are X threads waiting on an InputStream.

1. situation: there is nothing left to read:
- with notify a single thread wakes up, realizes there is nothing to do and 
goes to wait again.
- with notifyAll, X threads wake up, fight for the locks, each one realizes 
there is nothing to read, and goes to sleep.

2. situation: there is something left to read:
- with notify a single thread wakes up, reads his share, notifies another 
one, and so on until there is nothing left to read (or no thread is waiting).
-with notifyAll X threads wake up, fight for the locks, some get to read, 
others don't, and every one of them has to figure out there is nothing left 
for him, and go to sleep. (Or all threads have read and no thread is waiting).

Thanks for the time and patience, I'd like to hear your comments. Your input 
has been very valuable and I've learned a lot during this discussion.

Dalibor Topic

_________________________________________________________
Do You Yahoo!?
Get your free @yahoo.com address at http://mail.yahoo.com