Skip to content

Commit

Permalink
Queue
Browse files Browse the repository at this point in the history
- Added an `exitReason` and an `alive` (set to `true` on construction)
- Calling `shutdownQueue(ErrorType)` will set the exit reason, will also set the aliveness to `false` and wake up ALL `dequeue()`'s blocking
- `dequeue()` first check in wakeup routine duty cycle is to check if we are alive
  • Loading branch information
deavmi committed Nov 26, 2023
1 parent 198cb52 commit 713c102
Showing 1 changed file with 36 additions and 1 deletion.
37 changes: 36 additions & 1 deletion source/tristanable/queue/queue.d
Original file line number Diff line number Diff line change
Expand Up @@ -62,6 +62,13 @@ public class Queue
*/
private Duration wakeInterval;

/**
* Reason for a `dequeue()`
* to have failed
*/
private ErrorType exitReason;
private bool alive;

/**
* Constructs a new Queue and immediately sets up the notification
* sub-system for the calling thread (the thread constructing this
Expand All @@ -85,6 +92,9 @@ public class Queue

/* Set the slumber interval */
this.wakeInterval = dur!("msecs")(50); // TODO: Decide on value

/* Set status to alive */
this.alive = true;
}

/**
Expand Down Expand Up @@ -157,6 +167,25 @@ public class Queue
}
}


public void shutdownQueue(ErrorType reason)
{
// Set running state and reason
this.alive = false;
this.exitReason = reason;

// Wakeup sleeping dequeue()

// Lock the mutex
this.mutex.lock();

// Awake all condition variable sleepers
this.signal.notifyAll();

// Unlock the mutex
this.mutex.unlock();
}

// TODO: Make a version of this which can time out

/**
Expand Down Expand Up @@ -188,6 +217,13 @@ public class Queue
/* Block till we dequeue a message successfully */
while(dequeuedMessage is null)
{
/* Check if this queue is still alive */
if(!this.alive)
{
// Throw an exception to unblock the calling `dequeue()`
throw new TristanableException(this.exitReason);
}

scope(exit)
{
// Unlock the mutex
Expand All @@ -207,7 +243,6 @@ public class Queue
throw new TristanableException(ErrorType.DEQUEUE_FAILED);
}


/* Lock the item queue */
queueLock.lock();

Expand Down

0 comments on commit 713c102

Please sign in to comment.