synchronized design problem

[Jeff]

I'd appreciate some expert opinions on a design problem. I have many input
threads, but one processing thread so I need to demultiplex the input. The
solution I have currently is flawed. I'd like to add to the processing
queue while the processing thread processes, but I can't figure out the
synchronization syntax.

Here's a simplified version of my current code. The demultiplexing is done
in the Bridge class. Thanks in advance for any suggestions.

public class MainClass {

static LinkedList rawInputList = new LinkedList();

static LinkedList processingList = new LinkedList();

/**
* This Bridge class is intended to allow the processing thread to
always run. When the processing thread is ready for new data, it releases
the
* synchronization The flaw in this implementation is that the
InputThreads wait for the rawInputList synchronization to end in order to
enqueue
*/
class Bridge extends Thread {

public void run() {

while (true) {

synchronized (rawInputList) {

try {
rawInputList.wait();
} catch (InterruptedException e) {
System.out.println("Exception: " + e + " while
waiting on rawInputList");
}

synchronized (processingList) {
while (rawInputList.size() > 0) {
// copy the rawInput to the processing list
processingList.addAll(rawInputList);
rawInputList.clear();
processingList.notify();

} // while rawInputList > 0
} // sync processingList
} // sync rawInputList

} // while true
} // run
} // Bridge

/** there are many Input threads */
class InputThread extends Thread {
public void run() {
while (true) {
Object input = new Object();
synchronized (rawInputList) {

rawInputList.addLast(input);
rawInputList.notify();
}
}
}
}

/** there is only one processing thread due to some data constraints.
Hope to change that in later versions */
class Processor extends Thread {
public void run() {
while (true) {
synchronized (processingList) {

try {
processingList.wait();
} catch (InterruptedException e) {
System.out.println("Exception: " + e + " while
waiting on rawInputList");
}

while (processingList.size() > 0) {
Object toProcess = processingList.removeFirst();
/** lots of processing code here */
}
}
}
}
}

}

 

[Babu Kalakrishnan]

I'd appreciate some expert opinions on a design problem. I have many input
threads, but one processing thread so I need to demultiplex the input. The
solution I have currently is flawed. I'd like to add to the processing
queue while the processing thread processes, but I can't figure out the
synchronization syntax.

See notes below :

Here's a simplified version of my current code. The demultiplexing is done
in the Bridge class. Thanks in advance for any suggestions.

public class MainClass {

static LinkedList rawInputList = new LinkedList();

static LinkedList processingList = new LinkedList();

/**
* This Bridge class is intended to allow the processing thread to
always run. When the processing thread is ready for new data, it releases
the
* synchronization The flaw in this implementation is that the
InputThreads wait for the rawInputList synchronization to end in order to
enqueue
*/
class Bridge extends Thread {

public void run() {

while (true) {

synchronized (rawInputList) {

try {
rawInputList.wait();
} catch (InterruptedException e) {
System.out.println("Exception: " + e + " while
waiting on rawInputList");
}

synchronized (processingList) {
while (rawInputList.size() > 0) {
// copy the rawInput to the processing list
processingList.addAll(rawInputList);
rawInputList.clear();
processingList.notify();

} // while rawInputList > 0
} // sync processingList
} // sync rawInputList

Slight problem above. It is possible that an Input thread inserts something into
the rawInputList at this point - ie. after you've come out of the rawInputList
synchronized block but before you reacquire the lock on it in the next
iteration. In such a case you'll go into a wait state with data present in the
list. A better construct would be (I've omitted the InterruptedException
handling code for clarity)

while (true)
{
synchronized(rawInputList)
{
while (rawInputList.isempty())
{
rawInputList.wait();
// Exception handling code required
}

synchronized(processingList)
{
processingList.addAll(rawInputList);
rawInputList.clear();
processingList.notify();
}

}
}

[snip]

}

/** there is only one processing thread due to some data constraints.
Hope to change that in later versions */
class Processor extends Thread {
public void run() {
while (true) {
synchronized (processingList) {

try {
processingList.wait();
} catch (InterruptedException e) {
System.out.println("Exception: " + e + " while
waiting on rawInputList");
}

while (processingList.size() > 0) {
Object toProcess = processingList.removeFirst();
/** lots of processing code here */
}
}
}
}
}

}

Since you don't want the lock to be held while processing is going on, you
should do something like :

Object toProcess;
synchronized(processingList)
{
while (processingList.isEmpty())
{
processingList.wait();
}
toProcess = processingList.removeFirst();
}

// Processing code using the Object "toProcess" goes here - outside the
// synchroinized block.


BK

 

[Jeff]

Vast improvement - Thanks!

I'd appreciate some expert opinions on a design problem. I have many input
threads, but one processing thread so I need to demultiplex the input. The
solution I have currently is flawed. I'd like to add to the processing
queue while the processing thread processes, but I can't figure out the
synchronization syntax.

See notes below :

Here's a simplified version of my current code. The demultiplexing is done
in the Bridge class. Thanks in advance for any suggestions.

public class MainClass {

static LinkedList rawInputList = new LinkedList();

static LinkedList processingList = new LinkedList();

/**
* This Bridge class is intended to allow the processing thread to
always run. When the processing thread is ready for new data, it releases
the
* synchronization The flaw in this implementation is that the
InputThreads wait for the rawInputList synchronization to end in order to
enqueue
*/
class Bridge extends Thread {

public void run() {

while (true) {

synchronized (rawInputList) {

try {
rawInputList.wait();
} catch (InterruptedException e) {
System.out.println("Exception: " + e + " while
waiting on rawInputList");
}

synchronized (processingList) {
while (rawInputList.size() > 0) {
// copy the rawInput to the processing list
processingList.addAll(rawInputList);
rawInputList.clear();
processingList.notify();

} // while rawInputList > 0
} // sync processingList
} // sync rawInputList

Slight problem above. It is possible that an Input thread inserts something into
the rawInputList at this point - ie. after you've come out of the rawInputList
synchronized block but before you reacquire the lock on it in the next
iteration. In such a case you'll go into a wait state with data present in the
list. A better construct would be (I've omitted the InterruptedException
handling code for clarity)

while (true)
{
synchronized(rawInputList)
{
while (rawInputList.isempty())
{
rawInputList.wait();
// Exception handling code required
}

synchronized(processingList)
{
processingList.addAll(rawInputList);
rawInputList.clear();
processingList.notify();
}

}
}

[snip]

}

/** there is only one processing thread due to some data constraints.
Hope to change that in later versions */
class Processor extends Thread {
public void run() {
while (true) {
synchronized (processingList) {

try {
processingList.wait();
} catch (InterruptedException e) {
System.out.println("Exception: " + e + " while
waiting on rawInputList");
}

while (processingList.size() > 0) {
Object toProcess = processingList.removeFirst();
/** lots of processing code here */
}
}
}
}
}

}

Since you don't want the lock to be held while processing is going on, you
should do something like :

Object toProcess;
synchronized(processingList)
{
while (processingList.isEmpty())
{
processingList.wait();
}
toProcess = processingList.removeFirst();
}

// Processing code using the Object "toProcess" goes here - outside the
// synchroinized block.


BK