java socket i/o with callback/non-blocking

[Shane Wealti]

I have the O'Reilly Book Java NIO (with the mouse on it) and I'm trying
to figure out how to do non-blocking socket i/o. Basically I want to
have some subroutine to be called whenever there is data available to
read on a socket object without having to poll it. I'm having trouble
understanding how to do this in Java and the NIO book isn't really
helpful in that regard. Right now the way I have my program written is
that I spawn a separate thread that has a while loop that checks the
socket for data and then goes to sleep for a while but that doesn't
seem like a very efficient implementation. I'm having trouble finding
much information on NIO, especially tutorials and code samples that
deal with this.

 

[Bjorn Borud]

["Shane Wealti" <[email protected]>]
|
| I have the O'Reilly Book Java NIO (with the mouse on it) and I'm trying
| to figure out how to do non-blocking socket i/o. Basically I want to
| have some subroutine to be called whenever there is data available to
| read on a socket object without having to poll it.

in additional to the other advice given here (to look at EmberIO) I'd
recommend reading up on the Reactor pattern:

http://www.cs.wustl.edu/~schmidt/PDF/reactor-siemens.pdf

-Bjørn

 

[iksrazal]

Take a look at the emberio sourceforge project for some source
examples/libraries - the project is seemingly now abandoned but it did
provide me a good starting point - when I was in the same boat of not
being able to get past the o'reilly book. The emberio author also wrote
some good articles on NIO.

Also, search this group as some good theoretical discussions have
appeared. Concerning threads, looking hard enough you will find some
Doug Lea / PooledExecutor tutorials/PDF's with NIO.

HTH,
iksrazal
http://www.braziloutsource.com/

 

[Shane Wealti]

Thanks for the info. Emberio looks like exactly what I need to point
me in the right direction.

 

[John C. Bollinger]

I have the O'Reilly Book Java NIO (with the mouse on it) and I'm trying
to figure out how to do non-blocking socket i/o. Basically I want to
have some subroutine to be called whenever there is data available to
read on a socket object without having to poll it. I'm having trouble
understanding how to do this in Java and the NIO book isn't really
helpful in that regard. Right now the way I have my program written is
that I spawn a separate thread that has a while loop that checks the
socket for data and then goes to sleep for a while but that doesn't
seem like a very efficient implementation. I'm having trouble finding
much information on NIO, especially tutorials and code samples that
deal with this.

The NIO mechanism for waiting on data without polling is the Selector.
You obtain a new instance via Selector.open(), register your
SocketChannel with it via the SocketChannel's register() method
(specifying SelectionKey.OP_READ for its interest operations), then
invoke one of the Selector's select() methods. When the method returns,
you can query its selected keys set to determine which channels are
ready for which operations, and do whatever you want to with the
results. Typically you would put all this in a loop.

Now, all that may be dramatic overkill if you only have one channel to
worry about. You still need to run it in its own thread (or at least,
it's much simpler to do so), and all it really saves you is guessing at
how long to wait for data. Selectors are really intended for handling
multiple channels with a single thread. Note also that the technique
described is blocking, not non-blocking, as indeed is the technique you
described even if the channel is in non-blocking mode. You can use a
Selector in a non-blocking way (see Selector.selectNow()), but I don't
yet see how that would help you in your particular situation.

For just one socket, you are probably much better off to dump all the
NIO stuff, and simply use blocking I/O in its own thread. (Even with
multiple sockets this is often a viable approach, with one thread per
socket.) What you are doing now is an inefficient simulation of that,
and the Selector-based scheme for a single channel is a somewhat less
inefficient simulation of it. Non-blocking I/O sounds good, but it
leaves you with questions about what the thread is to do when it's not
performing I/O and how to schedule the I/O attempts so as to not create
an unnecessary dataflow bottleneck. Naive uses may needlessly spin,
consuming CPU cycles that could have been devoted to more productive
purposes.

 

[Shane Wealti]

You're right, that sounds more like what I'm looking for.
So I'll have my i/o thread that will issue a read command which will
block (allowing other threads to run) while it is waiting for input.
When there are incoming bytes to read in the socket's buffer, the i/o
thread will automatically become unblocked/wake up and I can process
the bytes and eventually issue another read command. What if I need to
write to the socket while it is blocked on a read? Should I create
another thread for the writing (which means I have to worry about
deadlock/race conditions, right?) or can I somehow interrupt the
thread, abort the blocked read command and write instead? Am I on the
right track at all here?

 

[Bjorn Borud]

["Shane Wealti" <[email protected]>]
|
| What if I need to write to the socket while it is blocked on a read?
| Should I create another thread for the writing (which means I have
| to worry about deadlock/race conditions, right?)

in an application I've written I have an IO thread which takes care of
all the reading and "pending" writes. I'll come back to what I mean
by "pending writes".

the reading and writing happens at the same time. most of the time
the writing takes place in threads other than the IO thread. I have a
per connection object which encapsulates the connection and its
associated state information.

when I want to write I do the following: synchronize on a List
instance within the Connection object which is used for outbound data
and check if the list is empty. (this synchronization would be
functionally equivalent to a write-lock if that terminology helps).

if it isn't empty I append the buffer I wanted to write to the List and
ensure that the selector is informed that I am interested in OP_WRITE¹
and return. the buffer will be written when the receiving socket is
ready for more data (by the IO thread. asynchronously).

if the list is empty I attempt an opportunistic write() -- that is, I
write the buffer to the socket directly². if there's unwritten data
left in the buffer after the write() call I enqueue the buffer on the
outbound List of buffers, just like in the first case, and set
OP_WRITE¹ in the interest set.

if the list is empty I remove OP_WRITE¹ from the interest set and
return.

since synchronization is always performed on the outbound List of
buffers this will behave correctly as long as you don't split data
across several buffers and then submit them in order. there are
several ways to address this problem, but I'll leave that as an
exercise to the reader.

when the IO thread, which is responsible for doing readiness
select()ion, gets a connection wich is writable, it will call write()
which in turn will attempt to drain the outbound list.


(wow, this description was really a lot harder to follow that just
reading the code that implements it. I am trying to get permission to
publish the code though, but don't hold your breath. it isn't all
that complex).


¹) due to the way the locking hierarchy in Selector is arranged I have
to do this by enqueueing an object that represents a change request
for the interest set of the Selector used in my IO framework.
these changes are then applied before I call select(). if I didn't
do it this way changing the interest set would block if a select()
was in progress. (block until select() returned).

²) when opportunistic writes succeed they speed things up greatly.
in one application most of my writes (~98%) succeeded and were
completed during the "opportunistic" write.

-Bjørn

 

[Shane Wealti]

So, I think I've almost got it I just need to figure out how to combine
the reading and writing into one i/o thread. I'm having a little bit
of trouble wrapping my head around it. I know what I would do if I
just wanted the i/o thread to read:

private synchronized void waitForDataFromSocket() throws
InterruptedException {
byte[] dataRead = new byte[MAX_BUFFER_SIZE]
int bytesRead;

while(true) {
bytesRead = in.read(dataRead); // this call blocks the thread
until there is something to read from the InputStream wrapping the
Socket
ProcessReadData(data); // process the data
}
}

and if I just wanted it to write I would do something like this

private synchronized void waitForDataToWrite() {
while(!dataToWrite) {
wait();
WriteDataToSocketOutputStream(someData);
}
}

where dataToWrite is a boolean set to true every time a buffer has some
bytes that need to be written to the socket and the .notifyAll() method
is called (by some other object) to wake this thread up so that it
writes the data at which point it sets dataToWrite to false again and
waits.

What sort of while loop/notifying mechanism would I use if I wanted the
same thread to be able to do both reading and writing?