Take the 2-minute tour ×
Programmers Stack Exchange is a question and answer site for professional programmers interested in conceptual questions about software development. It's 100% free, no registration required.

So I've been asking the wrong questions for the last three days and after searching a lot I think I have an idea of what I need, my question is how to use timers with two threads writing and reading on one queue. In other words, I have one queue(or any other data structure) and two threads, producer and consumer, so the producer write values on the queue, with a time for each value(let's say 10 seconds) and the consumer read the queue, both are using lock mechanism, what I need is a timer function that check which value has passed the 10 seconds and remove it from the queue.

I know how to implement the producer/consumer threads, but I don't know where to put this, is it a third thread or what? and do I need to write the time of the value when entered the queue, so the timer function can check it, or just write 10 and the timer function will decrement 1 each second?

share|improve this question

3 Answers 3

up vote 0 down vote accepted

First, the structure you want in this case is probably a priority queue. You'll use the expiration time of each item to determine its priority in the queue.

In this case, the consumer thread waits for one of two things to happen: the time of the first item in the queue to come due, or a newly-inserted item ends up at the beginning of the queue.

share|improve this answer
    
Thank you both so much, what about this scenario, the producer lives in a thread that send to the network first and then write to the queue, the consumer lives in the receiver thread which will check the queue when packet is received from the network. So it will be like TCP stack, not sure how that works? but I won't need the retransmission queue, I only need to delete the timed out packet? –  poly May 24 '12 at 15:07
    
In this case, the items in the queue apparently expire in order, so you can use a normal queue instead of a priority queue. The receiving thread just sleeps until the expiration time for the item at the front of the queue, removes it (if it's still present) and repeats. –  Jerry Coffin May 24 '12 at 16:17
    
Thanks Jerry, Yeah I was thinking about that but the problem is that receiving thread works when there is a packet received from the network, so it's like this thread{ for(;;) receive_msg}. I'm not sure how to make the receiving thread check the queue, for example if there was no packet received then who is going to check the queue? –  poly May 24 '12 at 21:05

If I understand correctly, you need each item in the queue to expire after a certain amount of time, but you still need items which are not expired to be consumed with observing the FIFO rule.

In this case, you do not necessarily need a timer or a 3rd thread; simply leave the expired items in the queue, and on consume check if the item you get has expired or not (check current time vs. enqueue time + your timeout). If it hasn't, push it to consumer, if it has, dispose of it.

In case you need to dispose resources when items expire, or to raise events to observers, you can create an external data structure which does exactly this (the collector with a priority queue by expire time, as pointed out by Jerry Coffin), but you don't need to connect it in any direct way to your producer-consumer (other than synchronizing to make sure that you don't consume something that's being disposed of).

share|improve this answer

If the time to use is the time when it entered the queue, then you need no priority queue, just a simple FIFO. The consumer thread reads items until it finds one that is not outdated, then it reads that one as well.

Pseudo code:

typedef struct
{
  obj_t  obj;
  time_t time;
} queue_item_t;


result_t fifo_add (const obj_t* obj)
{
  queue_item_t qi;

  qi.obj  = *obj;
  qi.time = time(NULL);

  return add_to_fifo(&qi);
}


result_t fifo_remove (obj_t* returned_obj)
{
  queue_item_t qi;
  bool item_expired;

  do
  {
    if(remove_from_fifo(&qi) == EMPTY_QUEUE)
    {
      return EMPTY_QUEUE;
    }
    item_expired = expired(&qi);
  } while(item_expired);

  if(item_expired) // all items removed, all had expired
  {
    return EMPTY_QUEUE;
  }

  *returned_obj = qi.obj;
  return OK;
}

inline bool expired (const queue_item_t* ptr_qi)
{
  return ptr_qi->time < time(NULL) - TIMEOUT_CONSTANT;
}

Just pseudo, nothing compiled or tested. You can put the mutex guards inside the add and remove functions for optimal performance.

share|improve this answer

Your Answer

 
discard

By posting your answer, you agree to the privacy policy and terms of service.

Not the answer you're looking for? Browse other questions tagged or ask your own question.