Hello,

There is a task ( Task )

There is a queue ( Queue , blocking priority queue)

There is a queue manager ( QueueManager )

There is a queue handler ( QueueHandler )

QueueManager - singleton.

Task , among other things, has such a characteristic as the desired execution time ( desired_at_ts ). This field indicates the time not earlier than which this task should be performed. This field participates in the comparator, on its basis the priority of the task in the queue is calculated.

QueueManager initializes the queue, provides methods put() / take() and others that are not important at the moment.

QueueHandler implements the Runnable interface. QueueHandler processes the queue in an infinite loop.

  • QueueHandler checks if the time specified in desired_at_ts
  • if the time has come, it performs the task.
  • if the time has not come, QueueHandler should put the task back
  • the cycle repeats.

Now what I want to know:

If the desired_at_ts time desired_at_ts not come, QueueHandler puts the task back into the queue. Since the priority queue - it is clear that this task will get to the top of the queue; and at the next cycle it will be received from the queue. Waiting for the time to come

 time = desired_at_ts - now_ts
can be implemented using the wait(long ms) method. But since the QueueManager can add a new task to the queue at any time, the new task can have desired_at_ts closer to the current time than the task that has just been returned to the queue. The QueueHandler thread must be awakened at the moment when the queue comparator runs and the task is put in its place in priority. Questions:

  1. Where should the code be placed that wakes the thread? Which queue / comparator method or something else?
  2. How to organize such a lock? I think it's more convenient to use ReentrantLock and Condition from it. But in this case the objects of these classes must exist in a single copy and will have to be transferred from class to class. This is normal?
  3. If there are comments / criticisms to the algorithm I described, I will be happy to discuss them.
  • I would just make a wrapper over a queue that does not issue an element until await has expired on any synchronizer (this await is reset by adding a new element to the wrapper) - etki
  • 2
    See the implementation of jucDelayQueue in the JDK. There, one of the waiting threads receives a wake-up signal when it is added, if the new item is at the head of the queue, and when it exits from take , if the queue is not empty. Waiting and waking up threads occur in the queue code, so ReentrantLock not transmitted anywhere. - zRrr
  • ZRrr , and how to deal with the priority when using jucDelayQueue? - jonua
  • 2
    And, in this case, the delay will be the very priority. What you need - jonua

1 answer 1

From the comment by @zRrr :

See the implementation of jucDelayQueue in the JDK. There, one of the waiting threads receives a wake-up signal when it is added, if the new item is at the head of the queue, and when it exits from take , if the queue is not empty. Waiting and waking up threads occur in the queue code, so ReentrantLock not transmitted anywhere.