Overview

	Chapter 15 - Breaking Out of a Blocked I/O State
	Java Thread Programming
	Paul Hyde


Closing a Stream to Break Out of the Blocked State


Although interrupt() and stop() are ignored by a thread blocked on read() , the thread can usually be unblocked by closing the stream with another thread.


The next series of examples presents three classes: CalcServer , CalcWorker , and CalcClient . These classes work with each other to demonstrate in a somewhat realistic client/server configuration the technique of closing a blocked stream with another thread.

CalcServer is started first and waits for socket connections on port 2001 . When a connection is received, CalcServer creates a new CalcWorker to handle the detailed communications with the client. In a separate VM on the same machine, CalcClient is launched and creates a connection to CalcServer . CalcClient makes one request of the server, and then leaves the line open . The server (through CalcWorker ) responds to the first request and then blocks on a socket read() waiting for an additional requesta request that never comes. After running for 15 seconds, the server shuts down and uses the stream closing technique to get the thread inside CalcWorker to break out of its blocked state.


Class CalcServer and Breaking Out of a Blocked accept()


The class CalcServer (see Listing 15.3) waits for socket connections from CalcClient . It passes the socket to a new instance of CalcWorker to manage the server side of the client session. When CalcServer is shut down, it uses another closing technique to break the internal thread out of its blocked state on the accept() method of ServerSocket .


Listing 15.3 CalcServer.javaListens for Connections from CalcClient


1: import java.io.*;

2: import java.net.*;

3: import java.util.*;

5: public class CalcServer extends Object {

6: private ServerSocket ss;

7: private List workerList;

9: private Thread internalThread;

10: private volatile boolean noStopRequested;

11:

12: public CalcServer(int port) throws IOException {

13: ss = new ServerSocket(port);

14: workerList = new LinkedList();

15:

16: noStopRequested = true;

17: Runnable r = new Runnable() {

18: public void run() {

19: try {

20: runWork();

21: } catch (Exception x) {

22: // in case ANY exception slips through

23: x.printStackTrace();

24: }

25: }

26: };

27:

28: internalThread = new Thread(r);

29: internalThread.start();

30: }

31:

32: private void runWork() {

33: System.out.println(

34: in CalcServer - ready to accept connections);

35:

36: while (noStopRequested) {

37: try {

38: System.out.println(

39: in CalcServer - about to block +

40: waiting for a new connection);

41: Socket sock = ss.accept();

42: System.out.println(

43: in CalcServer - received new connection);

44: workerList.add(new CalcWorker(sock));

45: } catch (IOException iox) {

46: if (noStopRequested) {

47: iox.printStackTrace();

48: }

49: }

50: }

51:

52: // stop all the workers that were created

53: System.out.println(in CalcServer - putting in a +

54: stop request to all the workers);

55: Iterator iter = workerList.iterator();

56: while (iter.hasNext()) {

57: CalcWorker worker = (CalcWorker) iter.next();

58: worker.stopRequest();

59: }

60:

61: System.out.println(in CalcServer - leaving runWork());

62: }

63:

64: public void stopRequest() {

65: System.out.println(

66: in CalcServer - entering stopRequest());

67: noStopRequested = false;

68: internalThread.interrupt();

69:

70: if (ss != null) {

71: try {

72: ss.close();

73: } catch (IOException x) {

74: // ignore

75: } finally {

76: ss = null;

77: }

78: }

79: }

80:

81: public boolean isAlive() {

82: return internalThread.isAlive();

83: }

84:

85: public static void main(String[] args) {

86: int port = 2001;

87:

88: try {

89: CalcServer server = new CalcServer(port);

90: Thread.sleep(15000);

91: server.stopRequest();

92: } catch (IOException x) {

93: x.printStackTrace();

94: } catch (InterruptedException x) {

95: // ignore

96: }

97: }

98: }

CalcServer is a self-running object, and 15 seconds after it is created, its stopRequest() method is called (lines 8991). CalcServer creates a ServerSocket to listen to port 2001 for client connections (line 13). In its runWork() method, the internal thread blocks on the accept() method of ServerSocket waiting for new socket connections (line 41). When a connection is received, the socket is passed to a new instance of CalcWorker , and a reference to this CalcWorker is added to workerList (line 44).

When stopRequest() is invoked, noStopRequested is set to false (line 67) and the internal thread is interrupted (line 68). Much like a blocked read() , the accept() method ignores interrupts. To get the accept() method to throw an exception, the ServerSocket is closed (lines 7078). Back in runWork() , the internal thread jumps down to the catch block (lines 4549) because of this IOException . A stack trace is printed only if a stop has not been requested (lines 4648) and in this case, no trace is printed because the closing of the ServerSocket caused the exception. The internal thread continues on and invokes stopRequest() on each of the CalcWorker objects (lines 5559).


Class CalcWorker and Breaking Out of a Blocked read()


The class CalcWorker (see Listing 15.4) handles the server-side portion of a client session. It creates streams off the socket and communicates with CalcClient over them. When it is shut down, it uses a closing technique to break the internal thread out of its blocked state on the read() method of InputStream .


Listing 15.4 CalcWorker.javaThe Server-Side Portion of a Client Session


1: import java.io.*;

2: import java.net.*;

4: public class CalcWorker extends Object {

5: private InputStream sockIn;

6: private OutputStream sockOut;

7: private DataInputStream dataIn;

8: private DataOutputStream dataOut;

10: private Thread internalThread;

11: private volatile boolean noStopRequested;

12:

13: public CalcWorker(Socket sock) throws IOException {

14: sockIn = sock.getInputStream();

15: sockOut = sock.getOutputStream();

16:

17: dataIn = new DataInputStream(

18: new BufferedInputStream(sockIn));

19: dataOut = new DataOutputStream(

20: new BufferedOutputStream(sockOut));

21:

22: noStopRequested = true;

23: Runnable r = new Runnable() {

24: public void run() {

25: try {

26: runWork();

27: } catch (Exception x) {

28: // in case ANY exception slips through

29: x.printStackTrace();

30: }

31: }

32: };

33:

34: internalThread = new Thread(r);

35: internalThread.start();

36: }

37:

38: private void runWork() {

39: while (noStopRequested) {

40: try {

41: System.out.println(in CalcWorker - about to +

42: block waiting to read a double);

43: double val = dataIn.readDouble();

44: System.out.println(

45: in CalcWorker - read a double!);

46: dataOut.writeDouble(Math.sqrt(val));

47: dataOut.flush();

48: } catch (IOException x) {

49: if (noStopRequested) {

50: x.printStackTrace();

51: stopRequest();

52: }

53: }

54: }

55:

56: // In real-world code, be sure to close other streams and

57: // the socket as part of the clean-up. Omitted here for

58: // brevity.

59:

60: System.out.println(in CalcWorker - leaving runWork());

61: }

62:

63: public void stopRequest() {

64: System.out.println(

65: in CalcWorker - entering stopRequest());

66: noStopRequested = false;

67: internalThread.interrupt();

68:

69: if (sockIn != null) {

70: try {

71: sockIn.close();

72: } catch (IOException iox) {

73: // ignore

74: } finally {

75: sockIn = null;

76: }

77: }

78:

79: System.out.println(

80: in CalcWorker - leaving stopRequest());

81: }

82:

83: public boolean isAlive() {

84: return internalThread.isAlive();

85: }

86: }


CalcWorker is a self-running object that spends most of its time with its internal thread blocked waiting to read a new request from the client. It keeps a reference to the raw InputStream retrieved from the socket for use in closing (line 14). It then wraps the raw streams with DataInputStream and DataOutputStream to get the desired functionality (lines 1720).


	Tip	When using the closing technique to try to get a blocked read() to unblock, always attempt to close the rawest stream possible. In this example, the rawest InputStream is the one returned by the getInputStream() method on Socket . This maximizes the chances that the stream will throw an exception.


Inside runWork() , the internal thread blocks waiting to read a double from the client (line 43). When the thread reads a double , it calculates the mathematical square root of the double and sends the resulting double back down to the client (lines 4647). The thread then goes back to its blocked state waiting to read more data from the client.

When stopRequest() is invoked, noStopRequested is set to false (line 66) and the internal thread is interrupted (line 67). A blocked read() does not respond to an interrupt, so the sockets input stream is closed (lines 6977) causing the blocked read() to throw an IOException . Back up in runWork() , the internal thread jumps down to the catch block (lines 4853). A stack trace is printed only if a stop has not been requested (lines 4952), and in this case, no trace is printed because the closing of the stream caused the exception.


	Caution	Invoking close() on an InputStream can sometimes block. This can occur if both read() and close() are synchronized . The thread blocked on the read() has the instance lock. Meanwhile, another thread is trying to invoke close() and blocks trying to get exclusive access to the lock. As of JDK 1.2, BufferedInputStream has added the synchronized modifier to its close() method (it wasnt present in 1.0 and 1.1). This is not indicated in the Javadoc, but is shown in the source code and through reflection. If this is the case, the closing technique will not work unless done on the underlying stream.


Class CalcClient


The class CalcClient (see Listing 15.5) is run in a different VM than CalcServer , but on the same machine. It creates a connection to the server and makes only one request. After that, it keeps the connection open, but does no further communication.


Listing 15.5 CalcClient.javaCode to Test CalcWorker


1: import java.io.*;

2: import java.net.*;

4: public class CalcClient extends Object {

5: public static void main(String[] args) {

6: String hostname = localhost;

7: int port = 2001;

9: try {

10: Socket sock = new Socket(hostname, port);

11:

12: DataInputStream in = new DataInputStream(

13: new BufferedInputStream(sock.getInputStream()));

14: DataOutputStream out = new DataOutputStream(

15: new BufferedOutputStream(sock.getOutputStream()));

16:

17: double val = 4.0;

18: out.writeDouble(val);

19: out.flush();

20:

21: double sqrt = in.readDouble();

22: System.out.println(sent up + val + , got back + sqrt);

23:

24: // Dont ever send another request, but stay alive in

25: // this eternally blocked state.

26: Object lock = new Object();

27: while (true) {

28: synchronized (lock) {

29: lock.wait();

30: }

31: }

32: } catch (Exception x) {

33: x.printStackTrace();

34: }

35: }

36: }

CalcClient is crudely written to simply show the minimal communication necessary. It creates a socket connection to the server, extracts the data streams, and wraps them to get a DataInputStream and a DataOutputStream (lines 1015). Then it writes a double over to the server and waits for a different double to be returned (lines 1721). After that, CalcClient keeps the socket connection up, but does not communicate with the server any more. Instead, it goes into an infinite wait state (lines 2631).


Output When Run


When CalcClient is run, it produces one line of output and waits to be killed :


sent up 4.0, got back 2.0


On the server side, CalcServer and CalcWorker produce the output shown in Listing 15.6. Notice that when the stop request comes in, no exceptions are printed and the worker and server shut down in an orderly manner.


Listing 15.6 Output from CalcServer and CalcWorker


1: in CalcServer - ready to accept connections

2: in CalcServer - about to block waiting for a new connection

3: in CalcServer - received new connection

4: in CalcServer - about to block waiting for a new connection

5: in CalcWorker - about to block waiting to read a double

6: in CalcWorker - read a double!

7: in CalcWorker - about to block waiting to read a double

8: in CalcServer - entering stopRequest()

9: in CalcServer - putting in a stop request to all the workers

10: in CalcWorker - entering stopRequest()

11: in CalcWorker - leaving stopRequest()

12: in CalcServer - leaving runWork()

13: in CalcWorker - leaving runWork()


Toc

Chapter 15 - Breaking Out of a Blocked I/O State