A New API: msgSendMultipleRequest() - Design for Multicasting

Chapter 3 Design Methodology of Software Element

3.6 Design for Multicasting

3.6.4 A New API: msgSendMultipleRequest()

3.6.3 Comparison

In network programming, I/O can be classified as synchronous I/O or asynchronous I/O. “

msgSendRequestSync()

” acts as synchronous network I/O; the caller blocks until the response returns. “

msgSendRequest()

” acts as asynchronous network I/O;

the caller does not block and wait for the response. In network programming, asynchronous network I/O is generally more efficient than synchronous network I/O.

However, for the multicasting issue in HAVi, we think the synchronous design is better for two reasons.

First, the synchronous design in 3.6.1 is faster than the asynchronous design in 3.6.2 . In asynchronous mode, the caller has to wait until the acknowledgement returns, which may take some time. Since the caller sends asynchronous messages in turn, the delay could accumulate to a long time. In synchronous mode, every thread is responsible for one target and they are waiting concurrently. As long as the thread pooling is used to eliminate the thread initialization time, the synchronous design is faster than the asynchronous design.

Second, the synchronous design is straightforward and easier to implement. The most convenient way for application developers is to implement the multicasting in a single method. In the asynchronous design, the responses will be returned in the callback function, so the multicasting cannot be completed in a single method. And the developers have to implement a request/response table in the application to pass the responses between the threads. This makes the implementation more complicated. As for the synchronous design, all the implementation can be put in a single method. The threads can be provided by the HAVi system. All the application developers have to do is invoke the method.

3.6.4 A New API: msgSendMultipleRequest()

Since many software elements need to send messages to multiple targets, the design should be implemented in a single method of the SoftwareElement class. The function prototype is similar to msgSendRequestSync(), making it easier to use.

Example 3-6: The function prototype of the msgSendMultipleRequest() API

37 public class SoftwareElement {

public void msgSendMultipleRequest(

SEID[] destSeid,

OperationCode[] opCode, int timeout,

HaviByteArrayOutputStream[] bufferIn, HaviByteArrayInputStream[] bufferOut, StatusHolder[] returnCode);

}

Chapter 4 Implementation

4.1 Environment

4.1.1 HAVi stack

We implemented a HAVi stack. For platform independence, an abstraction layer was implemented to deal with Linux specific libraries and Java Native Interface [17]. The abstraction layer was written in C and Java. The rest, the system components and the applications, were all written in Java. Thus, our HAVi stack could be easily ported to other platform with a new abstraction layer. Since all system components were written in Java, this HAVi stack is aimed at FAV devices because IAV have no Java runtime environment.

Figure 4-1: Our HAVi stack

CMM 1394

HAVi Java API

Application Application

Messaging System

DCM Event

Manager Registry

DCM Manager

Libraw1394 Linux Kernel Abstraction Layer

Implemented in Java

Implemented

in C

We implemented all system components except Stream Manager and Resource Manager. The following is the description of our HAVi stack. The abstraction Layer is platform dependent, providing IEEE 1394 service for MS and CMM, using Libraw1394 [18], a Linux C library, to access IEEE 1394 bus. Our CMM, MS, Registry, and Event Manager were mostly implemented as the HAVi specification defines. As for DCM Manager, only the DCM installation service was implemented.

We implemented a DCM for the DV and implemented an application to control the DV.

4.1.2 Demonstration

Two personal computers and a DV (Digital Video Camcorder) were used to build a HAVi home network. One computer was simulated as a HAVi set-top box (FAV); the other simulated a HAVi TV. The DV represented a LAV. The computers were running on Linux and equipped with IEEE 1394 FireWire cards.

Figure 4-2: A HAVi network composed of two computers and a DV

Our HAVi stack was installed in both computers. The demonstration worked as the following. When the DV was plugged into the network, the set-top box detected it and downloaded the DCM for the DV. Then, the HAVi TV could control the DV via the DCM. The images were transferred isochronously from the DV and showed on the HAVi TV.

Set Top Box

Digital Video Camcorder

FAV (PC) FAV (PC) LAV

Internet

The DV acted as a LAV device, so no DCM was embedded in it. We assumed the set-top box and the DV were from the same device maker. When the DV was plugged into the network, the set-top box would receive a NEW_DEVICE event, recognize the device, and download the DCM from the Internet. The DCM Manager installed the DCM and then the DCM registered itself in the Registry. The set-top box communicated with the DV via Libavc1394 [19], a Linux C library for the AV/C (Audio/Video Control) Digital Interface Command Set [20]. The application in the HAVi TV queried the Registry for DV and obtained the SEID of the DCM. Users could control the DV via the UI of the application. Functions included play, stop, forward, and reverse.

4.1.3 Messaging System

The Messaging System was implemented in Thread-Per-Message with thread pooling.

The pool had five threads, meaning five callback functions could be invoked at once.

We did not choose Thread-Per-SoftwareElement because it may create too many threads. And the threads may just wait there and do nothing if no requests received.

4.2 Implementation of

msgSendMultipleRequest()

The implementation of

msgSendMultipleRequest()

method is shown in Example 4-1.

msgSendMultipleRequest()

is used to send HAVi messages to multiple software elements, and this method can be added to

SoftwareElement

class. A

Worker

class is used to send to a message.

Example 4-1: Implementation of msgSendMultipleRequest()

1: public class SoftwareElement extends HaviListener { 2: ####public void msgSendMultipleRequest(

3: ############SEID[] destSeid,

4: ############OperationCode[] opCode, 5: ############int timeout,

6: ############HaviByteArrayOutputStream[] bufferIn, 7: ############HaviByteArrayInputStream[] bufferOut, 8: ############StatusHolder[] returnCode) {

9: ########int n = destSeid.length;

10: ########for(int i = 0; i < n; i++) {

11: ############new Worker(mySe, destSeid[i], opCode[i], timeout, 12: ############bufferIn[i], bufferOut[i], returnCode[i]).start();

13: ########}

14: ########long startTime = System.currentTimeMillis();

15: ########boolean allReceivedFlag = false;

16: ########while((!allReceivedFlag) &&

17: ############ ((System.currentTimeMillis()-startTime) < timeout)) { 18: ############allReceivedFlag = true;

19: ############// to determine if having received all responses 20: ############for(int i = 0; i < n; i++) {

21: ################if(returnCode[i].getValue() == null) { 22: ####################allReceivedFlag = false;

23: ####################break;

24: ################}

25: ############} // for

26: ############if(!allReceivedFlag) { 27: ################try {

28: ####################Thread.currentThread().sleep(1000);

29: ################} catch(InterruptedException e) { 30: ####################e.printStackTrace();

31: ################}

32: ############}

33: ########}

34: ####} // end of method

35: ####class Worker extends Thread { 36: ########private SoftwareElement mySe;

37: ########private SEID destSeid;

38: ########private OperationCode opCode;

39: ########private int timeout;

40: ########private HaviByteArrayOutputStream bufferIn;

41: ########private HaviByteArrayInputStream bufferOut;

42: ########private StatusHolder returnCode;

43: ########Worker(SoftwareElement mySe,

42 44: ############ SEID destSeid,

45: ############ OperationCode opCode, 46: ############ int timeout,

47: ############ HaviByteArrayOutputStream bufferIn, 48: ############ HaviByteArrayInputStream bufferOut, 49: ############ StatusHolder returnCode) {

50: ############this.mySe = mySe;

51: ############this.destSeid = destSeid;

52: ############this.opCode = opCode;

53: ############this.timeout = timeout;

54: ############this.bufferIn = bufferIn;

55: ############this.bufferOut = bufferOut;

56: ############this.returnCode = returnCode;

57: ########}

58: ########public void run() { 59: ############try {

60: ################mySe.msgSendRequestSync(destSeid, opCode, timeout, 61: ########################################bufferIn, bufferOut, returnCode);

62: ############} catch(Exception e) {

The parameters (lines 3-8) are similar to those in

msgSendRequestSync()

except they are arrays. Worker threads are created or obtained to send the messages (line 11).

The while loop (lines 16-17) checks if a timeout occurs or all responses have been received. If not, the current thread sleeps for 1000ms (line 28). The

^Worker

class (line 35) is used to send a message using

msgSendRequestSync()

(line 60).

4.3 An Example: A HAVi Application

The Application class, shown in Example 4-2, implements the software element architecture proposed in Chapter 3. The application has two services: to print a string

“Hello World” in the console or to perform addition. If the application receives a

request, a worker thread is obtained to handle the request and the callback function can return quickly. The worker thread will handle the request and send the response.

This application has the following advantages. First, the callback function always returns quickly, so the caller software element does not block. Second, the application can handle multiple requests simultaneously. Third, there is no deadlock. Fourth, the interrupt() method in the ThreadPool class can interrupt all worker threads gracefully.

That is, the request being processed can be cancelled.

Example 4-2: Application.java--an application that implements the design methodology

1: import java.util.*;

2: import ntu.havi.constants.*;

3: import ntu.havi.system.*;

4: import ntu.havi.types.*;

5: ####

6: interface ConstApplication {

7: ####final short apiCode = (short)0x8000;

8: ####final byte HelloWorldId = (byte)0x80; // print "Hello World"

9: ####final byte AdditionId = (byte)0x81; // perform addition 10: }

11: ####

12: public class Application extends HaviListener { 13: ####private SoftwareElement mySe;

14: ####private ThreadPool pool;

15: ####

16: ####public Application () throws Exception { 17: ########mySe = new SoftwareElement(this);

18: ########// create a thread pool with three threads 19: ########pool = new ThreadPool(3);

20: ########// register this application 21: ########RegistryLocalClient registry = new

22: #################################### RegistryLocalClient(mySe);

23: ########Attribute[] att = new Attribute[0];

24: ########HaviByteArrayOutputStream hbaos = new 25: ################HaviByteArrayOutputStream();

44 26: ########hbaos.reset();

27: ########hbaos.writeHaviString("Hello World");

28: ########att[0] = new Attribute(

29: ################ConstAttributeName.ATT_DEVICE_MANUF, hbaos);

30: ########registry.registerElementSync(3000, mySe.getSeid(), att);

31: ####}

32: ####public void receiveMsg(byte protocolType, 33: ######################## SEID sourceId, 34: ######################## SEID destId, 35: ######################## Status state,

36: ######################## HaviByteArrayInputStream payload) { 37: ########boolean haveReplied = false; //

38: ########OperationCode opCode;

39: ########byte controlFlags;

40: ########int transactionId;

41: ########if(haveReplied) { 42: ############return false;

43: ########}

44: ########if(state.getErrorCode() != ConstGeneralErrorCode.SUCCESS) { 45: ############// Messaging System problem, ignore

46: ############return false;

47: ########}

48: ########if(protocolType != ConstProtocolType.HAVI_RMI) { 49: ############// incoming message is not a HAVi RMI service 50: ############return false;

51: ########}

52: ########try {

53: ############opCode = new OperationCode(payload);

54: ############controlFlags = payload.readByte();

55: ############transactionId = payload.readInt();

56: ############if((controlFlags & 0x01) == 1) {

57: ################// incoming message is a response, ignore 58: ################return false;

59: ############}

60: ############if(opCode.getApiCode() != ConstApplication.apiCode) { 61: ################// unknown API

62: ################return false;

45 63: ############}

64: ############switch(opCode.getOperationId()) { 65: ################case ConstApplication.HelloWorldId:

66: ################case ConstApplication.AdditionId:

67: ####################// obtain a worker thread to handle the message 68: ####################// pass the parameters to the message handler 69: ####################Worker worker = pool.getWorker();

70: ####################worker.start(sourceId, payload, opCode, 71: ############################ transactionId);

72: ####################return true;

73: ################default: return false; // unknow operation id 74: ############}

75: ########} catch(Exception e) { 76: ############e.printStackTrace();

77: ########}

78: ########return true; // the callback thread can return quickly 79: ####}

80: ####

81: ####class ThreadPool {

82: ########private Vector idleWorkers;

83: ########private Vector busyWorkers;

84: ########ThreadPool(int numWorkers) { 85: ############idleWorkers = new Vector();

86: ############busyWorkers = new Vector();

87: ############for(int i = 0; i < numWorkers; i++) {

88: ################idleWorkers.add(new Worker(idleWorkers, busyWorkers, 89: ################true));

90: ############}

91: ########}

92: ########Worker getWorker() {

93: ############synchronized(idleWorkers) { 94: ################if(idleWorkers.size() > 0) {

95: ####################Worker w = (Worker)idleWorkers.remove(0);

96: ####################busyWorkers.add(w);

97: ####################return w;

98: ################}

99: ############}

100: ############Worker w = new Worker(null, busyWorkers, false);

46 101: ############busyWorkers.add(w);

102: ############return w;

103: ########}

104: ########void interrupt() {

105: ############synchronized(busyWorkers) {

106: ################for(int i = 0; i < busyWorkers.size(); i++)

113: ########private Vector idleWorkers;

114: ########private Vector busyWorkers;

115: ########private Thread internalThread;

116: ########private volatile boolean noStopRequested;

117: ####

118: ########private SEID sourceId;

119: ########private HaviByteArrayInputStream payload;

120: ########private OperationCode opCode;

121: ########private int transactionId;

122: ####

123: ########Worker(Vector idleWorkers, Vector busyWorkers, 124: ############ boolean noStopRequested) {

125: ############this.idleWorkers = idleWorkers;

126: ############this.busyWorkers = busyWorkers;

127: ############this.noStopRequested = noStopRequested;

128: ############Runnable r = new Runnable() {

137: ############internalThread = new Thread(r);

138: ############internalThread.start();

47 139: ########}

140: ########void start(SEID sourceId,

141: ################ HaviByteArrayInputStream payload, 142: ################ OperationCode opCode,

143: ################ int transactionId) { 144: ############this.sourceId = sourceId;

145: ############this.payload = payload;

146: ############this.opCode = opCode;

147: ############this.transactionId = transactionId;

148: ############this.notify();

149: ########}

150: ########private void run() { 151: ############do {

152: ################try {

153: ####################if(idleWorkers != null) 154: ########################idleWorkers.add(this);

155: ####################wait();

156: ####################handleRequest();

157: ################} catch (InterruptedException e) { 158: ####################e.printStackTrace();

159: ################} finally {

160: ####################busyWorkers.remove(this);

161: ################}

162: ############} while(noStopRequested);

163: ########}

164: ########// interrupt the thread 165: ########void interrupt() {

166: ############internalThread.interrupt();

167: ########}

168: ########// interrupt the thread and request it to stop 169: ########void stop() {

170: ############noStopRequested = false;

171: ############internalThread.interrupt();

172: ########}

173: ########private void handleRequest() { 174: ############try {

175: ################if(opCode.getOperationId() ==

176: ################ ConstApplication.HelloWorldId) {

48 177:

178: ####################System.out.println("Hello World");

179: ####################HaviByteArrayOutputStream buffer = 180: ############################new HaviByteArrayOutputStream();

181: ####################Status returnCode = new Status(opCode.getApiCode(), 182: ####################################ConstGeneralErrorCode.SUCCESS);

183: ####################// send the response

184: ####################mySe.msgSendResponse(sourceId, opCode,

185: ############################ConstTransferMode.SIMPLE, returnCode, 186: ############################buffer, transactionId);

187: ################ return true;

188: ################}#else if(opCode.getOperationId() ==

189: ################ ConstApplication.AdditionId) {

190: ####################// read two integers from the payload 191: ####################int a = payload.readInt();

192: ####################int b = payload.readInt();

193: ####################HaviByteArrayOutputStream buffer = 194: ############################new HaviByteArrayOutputStream();

195: ####################// perform the addition and 196: ####################// store the result in the buffer 197: ####################buffer.writeInt(a+b);

198: ####################Status returnCode = new Status(opCode.getApiCode(), 199: ####################################ConstGeneralErrorCode.SUCCESS);

200: ####################// send the response

201: ####################mySe.msgSendResponse(sourceId, opCode,

202: ############################ConstTransferMode.RELIABLE, returnCode, 203: ############################buffer, transactionId);

204: ################}

205: ############} catch(InterruptedException e) { 206: ################Thread.currentThread().interrupt(); 212: } // class Application

This program has an ConstApplication interface, an Application class, a ThreadPool class, and a Worker class.

Interface

ConstApplication

defines the API code for the appilcation and the operation codes for two different services (lines 6-10).

In the constructor of

Application

class, a

SoftwareElement

object is created to represent this application (line 17); A thread pool with three threads is created (line 19); And the application is registered in Registry (lines 21-30). The request is handled in

msgReceive()

. At first, the message is checked if it is a request for the application (lines 41-63). Then, a worker thread will be obtained (line 69) and started (line 70).

The

^ThreadPool

class has two member variables (lines 82-83),

idleWorkers

and

busyWorkers

IdleWorkers

is a Vector containing idle worker theads;

BusyWorkers

is a Vector containing busy worker threads, the threads handling requests. In the constructor, worker threads are created and put into the pool (line 88).

The getWorker() method first checks if there is a idle worker thread in idleWorkers (line 94). If yes, the idle thread is returned. If not, a new worker thread is created and returned (lines 94-102). Note that the additional worker threads will not added into the pool. The number of worker threads in the pool is fixed. The

interrupt()

method is used to interrupt all the busy worker threads. The previous requests will be cancelled and the threads will be back in the wait state.

As for the

^Worker

class, the boolean variable

noStopRequested

(line 116) stands for whether there is a stop request. In the constructor, a thread is created and started (lines 137-138). This thread will execute the run() method (line 129) immediately.

Then, the worker thread executes in a while loop in the run() method all its life time (lines 151-162). Mostly the worker thread is in a wait state (line 155). If the worker thread is notified by

^start()

(line 148), it returns from

^wait()

and executes

handlRequest()

method. Before a worker enters wait state, it adds itself into the idleWorkers and makes itself available (line 154); after a worker thread finishes handling a request, it remove itself from the busyWorkers (line 160). In the

handleRequest()

, two integers read from the payload are added and written into the buffer. Then, the result is sent back by

msgSendResponse

().

Chapter 5 Conclusion and Future Work

5.1 Conclusion

In this report, four design issues about software elements are discussed and the solutions are provided. The first design issue is blocking time. Every software element should adopt the design in 3.3 so that the HAVi network can be efficient. The second is multiple requests. The software elements that may receive lots requests at the same time should adopt the design in 3.4 . The third is synchronization. All software elements should adopt the design in 3.5 to avoid deadlock. The fourth design issue is multicasting. The software elements need to multicast can use the design in 3.6 The main contribution of this report is to serve as a design guide for developers. We encountered those design issues while we designed and implemented our HAVi stack.

We believe that the design methodology can help developers design and implement software elements that are efficient and flexible.

Besides, the design in 3.6 could be useful especially for those systems providing asynchronous messaging via callback functions. A HAVi network is a message-based distributed system. In other distributed system, there may also be a need to send messages to multiple destinations.

How Messaging System dispatches messages affects the design of software elements.

In order to guarantee the portability of DCM and HAVi application, we suggested that the next version of HAVi specification should indicate that whether the callback function of a software element would be invoked concurrently. We also suggested that the callback function of a software element should not be invoked by Messaging System concurrently. Thus, software elements can process the messages in sequences.

If software elements want to handle the requests concurrently, worker threads have to be created and the software elements should deal with the synchronization problems.

5.2 Future Work

Although our programs have been tested in a simulated HAVi network composed by three computers, they have not tested with real HAVi-compliant devices. In the future, our HAVi stack and applications should be tested in a real HAVi home network. There may be more devices; the network topology may be more complicated; the distance among them may be longer. Then, the actual performance of our implementation can be measured.

References

[1] Frederic Feit et al, The Home Network Revisited: Which LAN Technologies Will Bring the Network Home?, http://www.itpapers.com, November, 1999.

[2] Nick Pidgeon, How Ethernet Works,

http://computer.howstuffworks.com/ethernet.htm.

[3] IEEE Std 1394-1995, Standard for a High Performance Serial Bus.

[4] The Home Phoneline Networking Alliance, http://www.homepna.org/.

[5] The HomePlug Powerline Alliance, http://www.homeplug.org/.

[6] IEEE P802.11, The Working Group for Wireless LANs, http://grouper.ieee.org/groups/802/11/.

[7] The Official Bluetooth® Wireless Info Site, http://www.bluetooth.com/.

[8] The HAVi Specification Version 1.1, http://www.havi.org/.

[9] The Jini Community, http://www.jini.org/.

[10] The UPnP^TM Forum, http://upnp.org/.

[11] The Source for Java Technology, http://java.sun.com/.

[12] Java Remote Method Invocation,

http://java.sun.com/products/jdk/rmi/index.html.

[13] The HAVi Organization, http://www.havi.org/.

[14] Object Management Group, IDL to Java Language Mapping, v1.2, http://www.omg.org, August, 2000.

[15] HAVi Java API 1.1, http://www.havi.org/, May, 2001.

[16] Hyde, Paul. Java Thread Programming. St., Indianapolis, Indiana:Sams Publishing, 200

[17] Sun Microsystems, Inc, Archive: Java Platform 1.1 API and Documentation, http://java.sun.com/products/jdk/1.1/docs/api/packages.html.

[18] Sun Microsystems, Inc, Java Native Interface Guides, http://java.sun.com/j2se/1.4.1/docs/guide/jni/.

[19] Andreas Bombe, Technical Documents of libraw1394, http://www.linux1394.org/doc/libraw1394/book1.html. 2001 [20] Project: GNU/Linux 1394 AV/C Library,

http://sourceforge.net/projects/libavc1394.

[21] 1394 Trade Association, “AV/C Digital Interface Command Set General Specification, Version 2.0.1”, http://www.1394TA.org/, 1998

Publication

Conference Paper:

Kuo-Wei Hsu, Chuen-Liang Chen, Wu-Cheng Li, and Ting-Ying Yu, “A Message Delivery Mechanism for HAVi Network”, Internet and Multimedia Systems and Applications (IMSA), 2003

在文檔中 EHOME: 電子家庭雛型之設計與實作(3/3]─子計畫二:嵌入式 JAVA 及 JINI 系統之設計與實作 (頁 45-0)