wait/notify mechanics

[vpk]

I need to implement msg exchange mechanism, where multiple Requestors send
messages and expect responses with exactly same ID as sent request.
I looked into Doug Lea materials and tried to use concepts of Guarded Waits,
Monitors etc.
Unfortunately, I did not find complete working example, that does something
similar to my logic. I keep getting IllegalMonitorStateExceptions when
running prototype. I would
appreciate any help in this matter!

My setup is: 3 Requestors send MCMessages with unique IDs, they wrap msgs
into SyncElement object, place them into SyncBuffer, then send them to
Producer.
Then Requestor waits until msg will be replaced by some response msg.
Producer emulates communication connection, it replaces contents of each
message leaving ID as it was.
Producer puts new msg back into SyncElement and notifies Requestor (here
Exception happens!). Requestor wakes up and takes msg back from SyncBuffer.
This msg has same ID
as request, but new contents.

public class Requestor implements Runnable {
private SyncBuffer sb;
private int id;
private String name;
private long timeout;

public Requestor (SyncBuffer sb, int id, long timeout) {
this.sb = sb;
this.id = id;
this.name = "Requestor_"+id;
this.timeout = timeout;
}

public void run() {
System.out.println(name + " started");
MCMessage req = new MCMessage(id, name);
System.out.println(name + ": req=" + req);
Producer p = sb.getProducer();
System.out.println(name + ": p=" + p);
SyncElement se = sb.put(req);
System.out.println(name + " got SyncElement");
p.send(req);
System.out.println(name + " sent");
try {
synchronized (se) {
while (!se.getReplaced())
wait();
}
} catch (InterruptedException e) {
System.out.println(name + " interrupted");
System.exit(0);
}

System.out.println(name + " woke up");
MCMessage resp = sb.take(req.getSvcRequestID());
System.out.println(name + ": resp=" + resp);
}

public String toString() {
return name;
}
}

public class Producer implements Runnable {
private SyncBuffer sb;
private int id;
private String name;
private MCMessage msg;

public Producer (SyncBuffer sb, int id) {
this.sb = sb;
this.id = id;
this.name = "Producer_"+id;
this.msg = null;
}

public void send (MCMessage msg) {
this.msg = msg;
}

public void run() {
System.out.println(name + " started");
while (msg == null)
continue;
synchronized (this) {
System.out.println(name + ": received req=" + msg);
MCMessage resp = new MCMessage(msg.getSvcRequestID(), name);
System.out.println(name + ": resp=" + resp);
sb.replace(resp);
System.out.println(name + " msg replaced");
msg = null;
}
}

public String toString() {
return name;
}
}

import java.util.*;

public class SyncBuffer {
private Hashtable h;
file://parameters needed for test
private Producer prod;

public SyncBuffer () {
h = new Hashtable();
}

public SyncElement put (MCMessage msg) {
file://check if this ID already exist - throw Exception
SyncElement se = new SyncElement(msg);
h.put(new Integer(msg.getSvcRequestID()), se);
return se;
}

public MCMessage take (int svcRequestID) {
SyncElement se = (SyncElement) h.get(new Integer(svcRequestID));
h.remove(new Integer(svcRequestID));
return se.getMessage();
}

public void replace (MCMessage msg)
throws IllegalMonitorStateException, NoSuchElementException {
file://check if this ID does not exist - throw Exception
SyncElement se = (SyncElement) h.get(
new Integer(msg.getSvcRequestID()));
se.setMessage(msg);
}

public Producer getProducer() {return prod;}

public static void main(String args[]) {
SyncBuffer sb = new SyncBuffer();
Requestor[] requestors = new Requestor[3];
Producer prod = new Producer(sb, 1);

(new Thread(prod)).start();
sb.prod = prod;

for (int i = 0; i < 3; i++) {
requestors[i] = new Requestor(sb, i, 0);
(new Thread(requestors[i])).start();
}
}
}

public class SyncElement {
private MCMessage msg;
private Object monitor;
private boolean replaced;

public SyncElement (MCMessage msg) {
this.msg = msg;
monitor = new Object();
replaced = false;
}

public MCMessage getMessage() {return this.msg;}
public boolean getReplaced() {return this.replaced;}

public void setMessage(MCMessage msg) {
synchronized (this) {
this.msg = msg;
replaced = true;
notify();
}
}

public Object getMonitor() {return this.monitor;}
}

public class MCMessage {
public int id;
public String text;

public MCMessage (int id, String text) {
this.id = id;
this.text = text;
}

public int getSvcRequestID() {
return this.id;
}
public String getText() {
return this.text;
}

public String toString() {
return new String ("ID:"+id+"; TEXT:"+text);
}
}

Output:
D:\>java SyncBuffer
Producer_1 started
Requestor_0 started
Requestor_1 started
Requestor_2 started
Requestor_1: req=ID:1; TEXT:Requestor_1
Requestor_2: req=ID:2; TEXT:Requestor_2
Requestor_1: p=Producer_1
Requestor_2: p=Producer_1
Requestor_0: req=ID:0; TEXT:Requestor_0
Requestor_1 got SyncElement
Requestor_2 got SyncElement
Requestor_1 sent
Requestor_2 sent
Requestor_0: p=Producer_1
Requestor_0 got SyncElement
java.lang.IllegalMonitorStateException: current thread not owner
Requestor_0 sent
java.lang.IllegalMonitorStateException: current thread not owner
java.lang.IllegalMonitorStateException: current thread not owner
at java.lang.Object.wait(Object.java:307)
at java.lang.Object.wait(Object.java:307)
at java.lang.Object.wait(Object.java:307)
at Requestor.run(Compiled Code)
at Requestor.run(Compiled Code)
at java.lang.Thread.run(Thread.java:466)
at java.lang.Thread.run(Thread.java:466)
at Requestor.run(Compiled Code)
at java.lang.Thread.run(Thread.java:466)

[Daniel Sterling]

Your problem is that you are calling wait() instead of se.wait(). This
means you are waiting on the requester (this) instead of on the object
you've synchronized on (se).

Try that and let me know if it works,
Dan

"vpk" <v...@cisco.com> wrote in message
news:983079471.766342@sj-nntpcache-5...