synchronized message queue operations in multi-thread

[david]

I have a requirement to have a queue to communicate among a set of
sender threads and a receiver thread.

I have decided to have a ConcurrentLinkedQueue object an a Semaphore
object for a maximum size to allow the threads to operate on the
queue.
I am planning to write my message queue class to support synchronized
(share) queue operations.


public class MyQueue
{
private final Semaphore sem ;
private final ConcurrentLinkedQueue<MyObject> updateQueue;
NotificationQueue()
{
sem = new Semaphore (QUEUE_SIZE,true);
updateQueue = new ConcurrentLinkedQueue<MyObject> ();
}

public boolean getObject(MyObject obj)
{

boolean flag = false;
obj = updateQueue.poll();

if(obj != null)
{
flag = true;
sem.release();
}

return ;
}

public boolean putObject(MyObject obj)
{
sem.tryAcquire();
boolean update = false;

update= updateQueue.put(obj);
if(!update)
sem.release();

return update;
}//end of the method
}//end of the class

In the above class, if the sender thread wants to pass an object, it
would call the 'putObject' method. In the putObject method, I will
call the 'tryAcquire' and then put the object in the queue. If the
operation is not successful, I will call the 'release' method.

And in the receiver thread, I will try to get the object, if the
object is not null, I will call the 'release' method.

Question i) Is the semaphore method calls (tryAcquire and release)
accurate in the context of the queue operations that I am trying to
do??


Question ii)
And I am also planning to have an infinite 'while' loop in the
receiver thread to check whether there is any object in the queue,
like as follows,
//in the receiver thread
....
MyObj obj = null;
boolean flag = false;
while(true)
{
flag = myQueue.getObject(obj)

//if there is object..
if (flag)
//do something...

}//end of while loop
....

Will this above receiver thread implementation give any performance
issue? Or is there any better solution for this?

Thanks for any suggestion.

 

[Martin Gregorie]

Question i) Is the semaphore method calls (tryAcquire and release)
accurate in the context of the queue operations that I am trying to do??

Looks OK. However, you need to reread the ConcurrrentLinkedQueue
javadocs: there is no put() method. You probably meant add().

Question ii)
And I am also planning to have an infinite 'while' loop in the receiver
thread to check whether there is any object in the queue, like as
follows,
//in the receiver thread
....
MyObj obj = null;
boolean flag = false;
while(true)
{
flag = myQueue.getObject(obj)

//if there is object..
if (flag)
//do something...

}//end of while loop
....

Will this above receiver thread implementation give any performance
issue? Or is there any better solution for this?

It certainly will cause problems, since it will spin if the queue is
empty. You definitely don't want that so you could do one of the
following:

1) use Semaphore.acquire() in place of Semaphore.tryAcquire() so the
consuming thread blocks if the queue is empty.

It can be forced to wake up, e.g. to tidily end the thread, by
interrupting the thread.

2) let the consuming thread sleep for a short (variable?) interval
before polling the queue again.

3) use a long sleep when the queue is empty and arrange for it to
be woken up when needed by interrupting the thread.

I'd use (1).

(3) boils down to another way of achieving (1). I only included it for
completeness and to make it clear that the sleep in (2) can also be
interrupted when its sleeping.

 

[david]

...
The requirement is to send messages from multiple clients to a single
server. So I decided to have a single receiver thread(which would have
the server connection) to process the messages placed in the queue and
send it across to the server.

This queue operations (adding elements by multiple threads and
removing elements by a single receiver thread) will be done by
multiple threads at the same time. So I thought LinkedBlockingQueue
might affect the performance if many threads do operations on the
queue simultaneously. (That was a blind thought. This is my first
experiment with java.util.concurrent classes. So I was kind of
confused on which feature to use in different scenarios.)

But now I feel that for the above case, LinkedBlockingQueue would be a
better choice.

That's absolutely true. But it would still annoy me to write code that
when it wakes up, it has to still check to see if there's something to do,
when the only reason it would be explicitly woken up is if there's
something to do.

Yes, I realize that in Java, we have "spurious wakens" and this is an
issue anyway in a lot of cases. But it still bugs me. I prefer an
API where I can tell a thread to block, and know for sure that it won't
unblock until the condition _I_ specified for it to unblock has occurred.
(And implicit in that preference is that the API doesn't force me to
"specify" conditions that aren't actually a requirement in the problem my
code is trying to solve).

Initially I decided to have something like this.
i) the receiver thread waits or blocks or sleeps, if the queue is
empty.
ii) And when any of the sending threads places the element in the
queue, it informs or notifies the receiver thread to wake up and
process the element in the queue.
iii)Even if multiple sending threads notifies at the same time, the
receiver thread should get the first notification and process queue
for all the elements, and should ignore the other notifications.

But I think the poll option in the LinkedBlockingQueue with a long
timeout will be fine for the requirement. Simple and meets the
requirement.

Thanks for the suggestions.