Queue object with strange behaviour??

[RobKinney1]

Here is another question we have been meaning to post for a long time. My
colleague is literally pulling chunks of hair out of his head right now.

We have a logging class. This class is mutlithreaded (as required by our
boss). So when we tell the class to log something, it puts the data into a
queue, and logs it to the disk when it has time later (low priority thread).

Its simple. As stated above, I throw data into the queue. The low priority
thread comes along, senses the queue has elements in it, pops them, and logs
them to disk.

For the most part it works fine. But sometimes entries will be radically
out of order. Some are even out of order by a placement of 100+ lines or
more!

I can reproduce this behavior in sample programs. I have a log file
integrity analyzer built as well that will tell you what lines are out of
order. And the results are bad.

We have played around with the creation of the queue object quite a bit.
And the results vary a little, but we cannot find that magical number
combination that works.

Ex: this.LoggingQueue = new Queue(2048, 10);

Before we just had this.LoggingQueue = new Queue();

Both are still yeilding terrible results.

Any suggestions? We can post more code if needed.

Thanks!

Rob K

[Marc Gravell]

First question: do you lock the queue (or some other agreed object) while
accessing it? [or mutex, or semaphore, or whatever...] Or is this the result
of chaotic access?

Marc

[RobKinney1]

Hello Marc. Thank you for replying. There is no locking or Mutexes in this
class. Here is a sample of where the queue empties itself:

private void LoggingWithThreads()
{

// run this thread while we have NOT terminated it AND there is
nothing more to be logged
// ... This will run (even if told to terminate) until it has
caught up on the logging.
while (!terminateThread || LoggingQueue.Count 0)
{
// purge the queue until there is nothing more to write,
then sleep and check all over again
// if there is anything to spit out to the log file
while (LoggingQueue.Count 0)
{
try
{
sw.WriteLine(LoggingQueue.Dequeue().ToString());
sw.Flush();
}
catch
{
// do nothing because we don't want to abort the
program just because it couldn't
// write 1 line of log code
}

}

Thread.Sleep(10); // give the processor a break :-}
}
} // end LoggingWithThreads ()
..........................
then we have:
public void LogData(String data)
{
......... // there is some other prep code before this next line .......
LoggingQueue.Enqueue(data); // enqueue data to be logged

} // end LogData ()
Maybe this is a bad way to do it. I am trying to avoid having the program
wait on disk access (since most everything is stored in memory tables) to
write a line of data to the log file.

Is there a cleaver way of doing this?

I don't think I want to lock the semaphore before and after the while loop
because then the program would have to wait until the queue empties to disk
before it can enqueue and proceed.

We are kinda at a loss what to do.

Thanks,

Rob K

"Marc Gravell" wrote:

First question: do you lock the queue (or some other agreed object) while
accessing it? [or mutex, or semaphore, or whatever...] Or is this the result
of chaotic access?

Marc

[Andy]

You'll have to syncronize access to the collection; its not a good idea
to have multiple threads trying to modify data in the same object.

Don't forget to syncronize access for the process taht's reading from
the Queue as well.

Unless the queue is an MS message queue..

RobKinney1 wrote:

Hello Marc. Thank you for replying. There is no locking or Mutexes in this
class. Here is a sample of where the queue empties itself:

private void LoggingWithThreads()
{

// run this thread while we have NOT terminated it AND there is
nothing more to be logged
// ... This will run (even if told to terminate) until it has
caught up on the logging.
while (!terminateThread || LoggingQueue.Count 0)
{
// purge the queue until there is nothing more to write,
then sleep and check all over again
// if there is anything to spit out to the log file
while (LoggingQueue.Count 0)
{
try
{
sw.WriteLine(LoggingQueue.Dequeue().ToString());
sw.Flush();
}
catch
{
// do nothing because we don't want to abort the
program just because it couldn't
// write 1 line of log code
}

}

Thread.Sleep(10); // give the processor a break :-}
}
} // end LoggingWithThreads ()
.........................
then we have:
public void LogData(String data)
{
........ // there is some other prep code before this next line .......
LoggingQueue.Enqueue(data); // enqueue data to be logged

} // end LogData ()
Maybe this is a bad way to do it. I am trying to avoid having the program
wait on disk access (since most everything is stored in memory tables) to
write a line of data to the log file.

Is there a cleaver way of doing this?

I don't think I want to lock the semaphore before and after the while loop
because then the program would have to wait until the queue empties to disk
before it can enqueue and proceed.

We are kinda at a loss what to do.

Thanks,

Rob K

"Marc Gravell" wrote:

First question: do you lock the queue (or some other agreed object) while
accessing it? [or mutex, or semaphore, or whatever...] Or is this the result
of chaotic access?

Marc

[Marc Gravell]

Then add some ;-p It really, *really* needs it.

Another advantage is that you can use Monitor.Pulse and Monitor.Wait to do
away with CPU hammering... i.e. your logging thread does something like
(don't quote me)

while(!terminate) {
T item;
lock(queue) {
if(queue.Count == 0) {
Monitor.Wait(queue); // release and re-aquire when Pulse()d
continue; // back to test condition... don't assume we got something
}
// count should be 0
item = queue.Dequeue();
}
// log item (outside of lock), then loop
}

the terminate should also Pulse in case we are Wait()ing, which we probably
will be ;-p

the add code would go something like:
lock(queue) {
queue.Enqueue(item);
if(queue.Count==1) { // queue was empty; listener probably Wait()ing
Monitor.Pulse(queue);
}
}

See also Jon Skeet's pages on threading (google will find it). The above is
off the top of my head and not guaranteed. At all ;-p

Marc

[RobKinney1]

Wow... After I implented this code, my log file integrity analyzer said there
were 0 errors and everything was logged!!

But I am not sure if I implented this right...:

private void LoggingWithThreads()
{

// run this thread while we have NOT terminated it AND there is
nothing more to be logged
// ... This will run (even if told to terminate) until it has
caught up on the logging.
while (!terminateThread || LoggingQueue.Count 0)
{
lock (LoggingQueue)
{
if (LoggingQueue.Count <= 0)
{
Monitor.Wait(LoggingQueue);
continue;
}
// purge the queue until there is nothing more to write,
then sleep and check all over again
// if there is anything to spit out to the log file
while (LoggingQueue.Count 0)
{
try
{

sw.WriteLine(LoggingQueue.Dequeue().ToString());
sw.Flush();

}
catch
{
// do nothing because we don't want to abort the
program just because it couldn't
// write 1 line of log code
}

} // end while()
} // end lock
//Thread.Sleep(10); // give the processor a break :-}
}
} // end LoggingWithThreads ()

..... and the enqueue is:
public void LogData(String data)
{
lock (LoggingQueue)
{
LoggingQueue.Enqueue(addToQueue); // enqueue data to
be logged

if (LoggingQueue.Count >= 1)
{
Monitor.Pulse(LoggingQueue);
}

} // end lock
} // end LogData ()

Is this correct? I don't quite understand how it is working, but somehow it
is.

It seems though that there is a potential, for example, 2000 lines to back
up... and then the main program will have to wait until the queue unloads to
disk. Are we looking at this correctly?

Thanks again Marc. We REALLY appreciate the help here.

Rob K

"Marc Gravell" wrote:

Then add some ;-p It really, *really* needs it.

Another advantage is that you can use Monitor.Pulse and Monitor.Wait to do
away with CPU hammering... i.e. your logging thread does something like
(don't quote me)

while(!terminate) {
T item;
lock(queue) {
if(queue.Count == 0) {
Monitor.Wait(queue); // release and re-aquire when Pulse()d
continue; // back to test condition... don't assume we got something
}
// count should be 0
item = queue.Dequeue();
}
// log item (outside of lock), then loop
}

the terminate should also Pulse in case we are Wait()ing, which we probably
will be ;-p

the add code would go something like:
lock(queue) {
queue.Enqueue(item);
if(queue.Count==1) { // queue was empty; listener probably Wait()ing
Monitor.Pulse(queue);
}
}

See also Jon Skeet's pages on threading (google will find it). The above is
off the top of my head and not guaranteed. At all ;-p

Marc

[Marc Gravell]

first... DO NOT check the Count whle you don't own the lock; this WILL
break... as the film goes "maybe not today, maybe not tomorrow, but
some day, soon, and for the rest of your life". That's the joy of
threading. Alternative suggested below.

// note: terminate should be a "volatile bool"
bool continue = true;
while(continue) {
T item;
lock(LoggingQueue) { // obtain exclusive access
if(LoggingQueue.Count==0) { // I wouldn't worry about -ve myself...
up to you
if(terminate) { // shut down requested
continue = false; // queue is empty, so exit loop
} else { // queue is empty; wait for work
Monitor.Wait(LoggingQueue); // relinquish lock, wait for a
kick, and reaquire
continue; // start loop again
}
//// etc as per my previous post, with logging near the bottom (outside
of the lock)

Note: I advise you to go a bit closer to my sample code; otherwise
(with your code below), if a queue has built up, it will need to log
all of it before more can be added. My way only processes one item per
lock, *and* it processes that item *outside* of the lock, which means
that adding new items is not bound by IO.

Also; the ==1 was for a reason; as it stands, you are doing unnecessary
Pulse()s; we know that we only Wait() if the queue is empty, i.e.
Count==0; this only happens if, after adding, the Count is 1. If the
Count is 7, we can say for sure that the logging thread is not
Wait()ing, but rather is simply patiently waiting on the lock that we
hold. No need for a Pulse().

Since the two threads are now isolated, I would also remove the
Sleep(), but again, that is up to you.

I can't really explain (in a short post) all of why this works. Threads
are tricky. You should really read Jon's pages end to end. Then read it
again ;-p

http://www.yoda.arachsys.com/csharp/threads/

Marc

[RobKinney1]

Marc,

This totally works! Perfectly! We were also having data loss in our logs
and this has totally fixed it including maintaining the correct order of the
log entries.

I have been reading Jon's page on threading. I am going to read more later
on this afternoon when I have more time. That is very important stuff as you
mentioned. I may have to go back in some of my other code and fix some
things.

On behalf of the IT team here, we thank you for taking the time to help us!!
This fix is like gold to us. Let me know if there is ever something I can
do for you.

Thanks,

Rob K

"Marc Gravell" wrote:

first... DO NOT check the Count whle you don't own the lock; this WILL
break... as the film goes "maybe not today, maybe not tomorrow, but
some day, soon, and for the rest of your life". That's the joy of
threading. Alternative suggested below.

// note: terminate should be a "volatile bool"
bool continue = true;
while(continue) {
T item;
lock(LoggingQueue) { // obtain exclusive access
if(LoggingQueue.Count==0) { // I wouldn't worry about -ve myself...
up to you
if(terminate) { // shut down requested
continue = false; // queue is empty, so exit loop
} else { // queue is empty; wait for work
Monitor.Wait(LoggingQueue); // relinquish lock, wait for a
kick, and reaquire
continue; // start loop again
}
//// etc as per my previous post, with logging near the bottom (outside
of the lock)

Note: I advise you to go a bit closer to my sample code; otherwise
(with your code below), if a queue has built up, it will need to log
all of it before more can be added. My way only processes one item per
lock, *and* it processes that item *outside* of the lock, which means
that adding new items is not bound by IO.

Also; the ==1 was for a reason; as it stands, you are doing unnecessary
Pulse()s; we know that we only Wait() if the queue is empty, i.e.
Count==0; this only happens if, after adding, the Count is 1. If the
Count is 7, we can say for sure that the logging thread is not
Wait()ing, but rather is simply patiently waiting on the lock that we
hold. No need for a Pulse().

Since the two threads are now isolated, I would also remove the
Sleep(), but again, that is up to you.

I can't really explain (in a short post) all of why this works. Threads
are tricky. You should really read Jon's pages end to end. Then read it
again ;-p

http://www.yoda.arachsys.com/csharp/threads/

Marc