Add priority queue

src/lib/DataStructures/PriorityQueue.vala

@@ -0,0 +1,191 @@
+using Invercargill;
+
+namespace Invercargill.DataStructures {
+
+    public class PriorityQueue<T> : Enumerable<T>, Queue<T> {
+        
+        public bool is_unblocked { get { return !is_blocking; } }
+        
+        private bool is_blocking = true;
+        private CompareDelegate<T> compare_func;
+        private Vector<PriorityQueueItem<T>*> items = new Vector<PriorityQueueItem<T>*>();
+
+        public PriorityQueue(owned CompareDelegate<T> compare_func) {
+            this.compare_func = (owned)compare_func;
+        }
+
+        public override int? peek_count() {
+            mutex.lock();
+            var count = (int)items.length;
+            mutex.unlock();
+            return count;
+        }
+
+        public override EnumerableInfo get_info() {
+            return new EnumerableInfo.infer_ultimate (this, EnumerableCategory.EXTERNAL);
+        }
+
+        public void unblock() {
+            mutex.lock();
+            is_blocking = false;
+            cond.broadcast();
+            mutex.unlock();
+        }
+
+        public void push(owned T item) {
+            mutex.lock();
+            PriorityQueueItem<T>* queue_item = new PriorityQueueItem<T>() {
+                item = (owned)item,
+            };
+
+            // Add to the end of the vector
+            items.add(queue_item);
+            
+            // Bubble up to maintain heap property
+            bubble_up((int)(items.length - 1));
+
+            cond.broadcast();
+            mutex.unlock();
+        }
+
+        public T pop() throws SequenceError {
+            mutex.lock();
+            while(is_blocking && items.length == 0){
+                cond.wait(mutex);
+            }
+
+            if(items.length == 0) {
+                mutex.unlock();
+                throw new SequenceError.NO_ELEMENTS("No elements in unblocked priority queue to pop");
+            }
+
+            // Get the highest priority item (at index 0)
+            var item = items.get(0);
+            var result = item->item;
+            
+            // Move the last item to the root
+            var last_item = items.get(items.length - 1);
+            items.set(0, last_item);
+            items.remove_at(items.length - 1);
+            
+            // Bubble down to maintain heap property
+            if(items.length > 0) {
+                bubble_down(0);
+            }
+            
+            delete item;
+            mutex.unlock();
+            return result;
+        }
+
+        public T peek() throws SequenceError {
+            mutex.lock();
+            if(items.length == 0) {
+                mutex.unlock();
+                throw new SequenceError.NO_ELEMENTS("No elements in priority queue");
+            }
+            
+            var item = items.get(0);
+            var result = item->item;
+            mutex.unlock();
+            return result;
+        }
+
+        private void bubble_up(int index) {
+            if(index == 0) return;
+            
+            int parent = (index - 1) / 2;
+            var current = items.get(index);
+            var parent_item = items.get(parent);
+            
+            if(compare_func(current->item, parent_item->item) < 0) {
+                // Swap with parent
+                items.set(index, parent_item);
+                items.set(parent, current);
+                bubble_up(parent);
+            }
+        }
+
+        private void bubble_down(int index) {
+            int left = 2 * index + 1;
+            int right = 2 * index + 2;
+            int smallest = index;
+            
+            if(left < items.length) {
+                var left_item = items.get(left);
+                var smallest_item = items.get(smallest);
+                if(compare_func(left_item->item, smallest_item->item) < 0) {
+                    smallest = left;
+                }
+            }
+            
+            if(right < items.length) {
+                var right_item = items.get(right);
+                var smallest_item = items.get(smallest);
+                if(compare_func(right_item->item, smallest_item->item) < 0) {
+                    smallest = right;
+                }
+            }
+            
+            if(smallest != index) {
+                // Swap with smallest child
+                var current = items.get(index);
+                var smallest_item = items.get(smallest);
+                items.set(index, smallest_item);
+                items.set(smallest, current);
+                bubble_down(smallest);
+            }
+        }
+
+        private bool has_next() {
+            mutex.lock();
+            while(is_blocking && items.length == 0){
+                cond.wait(mutex);
+            }
+
+            var state = items.length > 0;
+            mutex.unlock();
+            return state;
+        }
+
+        private T get_next() {
+            mutex.lock();
+            var item = items.get(0);
+            var result = item->item;
+            
+            // Move the last item to the root
+            var last_item = items.get(items.length - 1);
+            items.set(0, last_item);
+            items.remove_at(items.length - 1);
+            
+            // Bubble down to maintain heap property
+            if(items.length > 0) {
+                bubble_down(0);
+            }
+            
+            delete item;
+            mutex.unlock();
+            return result;
+        }
+
+        private Cond cond = Cond ();
+        private Mutex mutex = Mutex ();
+
+        public override Tracker<T> get_tracker () {
+            return new LambdaTracker<T>(
+                () => has_next(),
+                () => get_next());
+        }
+
+        ~PriorityQueue() {
+            for (uint i = 0; i < items.length; i++) {
+                delete items.get(i);
+            }
+        }
+
+        [Compact]
+        private class PriorityQueueItem<T> {
+            public T item;
+        }
+    }
+}

src/lib/meson.build

@@ -103,6 +103,7 @@ sources += files('Interfaces/Queue.vala')
 sources += files('DataStructures/Series.vala')
 sources += files('DataStructures/Fifo.vala')
 sources += files('DataStructures/Lifo.vala')
+sources += files('DataStructures/PriorityQueue.vala')
 sources += files('DataStructures/Vector.vala')
 sources += files('DataStructures/HashSet.vala')
 sources += files('DataStructures/Dictionary.vala')

src/tests/Integration/PriorityQueue.vala

@@ -0,0 +1,275 @@
+using Invercargill;
+using Invercargill.DataStructures;
+
+void priority_queue_tests() {
+
+    Test.add_func("/invercargill/priority_queue/basic_push_pop", () => {
+        var pq = new PriorityQueue<int>((a, b) => a - b); // Min-heap (smaller values have higher priority)
+        
+        // Push items
+        pq.push(5);
+        pq.push(1);
+        pq.push(3);
+        pq.push(2);
+        pq.push(4);
+        
+        // Pop in priority order (smallest first)
+        assert_cmpint(1, CompareOperator.EQ, pq.pop());
+        assert_cmpint(2, CompareOperator.EQ, pq.pop());
+        assert_cmpint(3, CompareOperator.EQ, pq.pop());
+        assert_cmpint(4, CompareOperator.EQ, pq.pop());
+        assert_cmpint(5, CompareOperator.EQ, pq.pop());
+    });
+
+    Test.add_func("/invercargill/priority_queue/max_heap", () => {
+        var pq = new PriorityQueue<int>((a, b) => b - a); // Max-heap (larger values have higher priority)
+        
+        // Push items
+        pq.push(5);
+        pq.push(1);
+        pq.push(3);
+        pq.push(2);
+        pq.push(4);
+        
+        // Pop in priority order (largest first)
+        assert_cmpint(5, CompareOperator.EQ, pq.pop());
+        assert_cmpint(4, CompareOperator.EQ, pq.pop());
+        assert_cmpint(3, CompareOperator.EQ, pq.pop());
+        assert_cmpint(2, CompareOperator.EQ, pq.pop());
+        assert_cmpint(1, CompareOperator.EQ, pq.pop());
+    });
+
+    Test.add_func("/invercargill/priority_queue/peek functionality", () => {
+        var pq = new PriorityQueue<int>((a, b) => a - b); // Min-heap
+        
+        pq.push(5);
+        pq.push(1);
+        pq.push(3);
+        
+        // Peek should return highest priority item without removing
+        var top = pq.peek();
+        assert_cmpint(1, CompareOperator.EQ, top);
+        
+        // Pop should still return 1
+        var popped = pq.pop();
+        assert_cmpint(1, CompareOperator.EQ, popped);
+        
+        // Peek now should return 3 (next highest priority)
+        var new_top = pq.peek();
+        assert_cmpint(3, CompareOperator.EQ, new_top);
+    });
+
+    Test.add_func("/invercargill/priority_queue/empty_priority_queue_behavior", () => {
+        // Test peek on empty priority queue
+        var pq1 = new PriorityQueue<string>((a, b) => a.collate(b));
+        try {
+            pq1.peek();
+            assert_not_reached();
+        }
+        catch (SequenceError e) {
+            assert_cmpstr("No elements in priority queue", CompareOperator.EQ, e.message);
+        }
+        
+        // Test pop on empty priority queue - need to unblock first
+        var pq2 = new PriorityQueue<string>((a, b) => a.collate(b));
+        pq2.unblock();
+        try {
+            pq2.pop();
+            assert_not_reached();
+        }
+        catch (SequenceError e) {
+            assert_cmpstr("No elements in unblocked priority queue to pop", CompareOperator.EQ, e.message);
+        }
+    });
+
+    Test.add_func("/invercargill/priority_queue/push_all", () => {
+        var pq = new PriorityQueue<int>((a, b) => a - b); // Min-heap
+        var items = new Series<int>();
+        items.add(5);
+        items.add(1);
+        items.add(3);
+        
+        // Push all items
+        pq.push_all(items);
+        
+        // Should pop in priority order
+        assert_cmpint(1, CompareOperator.EQ, pq.pop());
+        assert_cmpint(3, CompareOperator.EQ, pq.pop());
+        assert_cmpint(5, CompareOperator.EQ, pq.pop());
+    });
+
+    Test.add_func("/invercargill/priority_queue/enumerable_behavior", () => {
+        var pq = new PriorityQueue<int>((a, b) => a - b); // Min-heap
+        
+        pq.push(5);
+        pq.push(1);
+        pq.push(3);
+        pq.unblock();
+        
+        // Iteration should return items in priority order
+        var tracker = pq.get_tracker();
+        var items = new int[3];
+        var index = 0;
+        
+        while (tracker.has_next()) {
+            items[index++] = tracker.get_next();
+        }
+        
+        assert_cmpint(1, CompareOperator.EQ, items[0]);
+        assert_cmpint(3, CompareOperator.EQ, items[1]);
+        assert_cmpint(5, CompareOperator.EQ, items[2]);
+    });
+
+    Test.add_func("/invercargill/priority_queue/blocking_behavior", () => {
+        var pq = new PriorityQueue<string>((a, b) => a.collate(b));
+        bool thread_started = false;
+        bool thread_finished = false;
+        
+        // Start a thread that will wait for an item
+        Thread<void*> thread = new Thread<void*>("test-thread", () => {
+            thread_started = true;
+            var item = pq.pop();
+            assert_cmpstr("test_item", CompareOperator.EQ, item);
+            thread_finished = true;
+            return null;
+        });
+        
+        // Wait for thread to start and block
+        while (!thread_started) {
+            Thread.usleep(1000); // 1ms
+        }
+        // Give additional time for thread to block on pop()
+        Thread.usleep(50000); // 50ms
+        
+        // Push an item to unblock the thread
+        pq.push("test_item");
+        
+        // Wait for thread to finish
+        thread.join();
+        assert_true(thread_finished);
+    });
+
+    Test.add_func("/invercargill/priority_queue/unblock_behavior", () => {
+        var pq = new PriorityQueue<int>((a, b) => a - b);
+        bool thread_started = false;
+        bool exception_thrown = false;
+        
+        // Start a thread that will wait for an item
+        Thread<void*> thread = new Thread<void*>("test-thread", () => {
+            thread_started = true;
+            try {
+                var item = pq.pop(); // This should throw after unblock
+                assert_not_reached();
+            }
+            catch (SequenceError e) {
+                assert_cmpstr("No elements in unblocked priority queue to pop", CompareOperator.EQ, e.message);
+                exception_thrown = true;
+            }
+            return null;
+        });
+        
+        // Wait for thread to start and block
+        while (!thread_started) {
+            Thread.usleep(1000); // 1ms
+        }
+        // Give additional time for thread to block on pop()
+        Thread.usleep(50000); // 50ms
+        
+        // Unblock the priority queue
+        pq.unblock();
+        
+        // Wait for thread to finish
+        thread.join();
+        assert_true(exception_thrown);
+    });
+
+    Test.add_func("/invercargill/priority_queue/try_peek_and_try_pop", () => {
+        var pq = new PriorityQueue<string>((a, b) => a.collate(b));
+        
+        // Try operations on empty priority queue
+        string item;
+        assert_false(pq.try_peek(out item));
+        assert_null(item);
+        
+        string popped;
+        pq.unblock();
+        assert_false(pq.try_pop(out popped));
+        assert_null(popped);
+        
+        // Add an item
+        pq.push("test");
+        
+        // Try operations should now succeed
+        assert_true(pq.try_peek(out item));
+        assert_cmpstr("test", CompareOperator.EQ, item);
+        
+        assert_true(pq.try_pop(out popped));
+        assert_cmpstr("test", CompareOperator.EQ, popped);
+        
+        // Should be empty again
+        assert_false(pq.try_pop(out popped));
+        assert_null(popped);
+    });
+
+    Test.add_func("/invercargill/priority_queue/duplicate_values", () => {
+        var pq = new PriorityQueue<int>((a, b) => a - b); // Min-heap
+        
+        // Push items with duplicates
+        pq.push(3);
+        pq.push(1);
+        pq.push(3);
+        pq.push(2);
+        pq.push(1);
+        
+        // Pop in priority order
+        assert_cmpint(1, CompareOperator.EQ, pq.pop());
+        assert_cmpint(1, CompareOperator.EQ, pq.pop());
+        assert_cmpint(2, CompareOperator.EQ, pq.pop());
+        assert_cmpint(3, CompareOperator.EQ, pq.pop());
+        assert_cmpint(3, CompareOperator.EQ, pq.pop());
+    });
+
+    Test.add_func("/invercargill/priority_queue/peek_count", () => {
+        var pq = new PriorityQueue<int>((a, b) => a - b); // Min-heap
+        
+        // Initially empty
+        assert_cmpint(0, CompareOperator.EQ, pq.peek_count());
+        
+        // Add items
+        pq.push(3);
+        assert_cmpint(1, CompareOperator.EQ, pq.peek_count());
+        
+        pq.push(1);
+        assert_cmpint(2, CompareOperator.EQ, pq.peek_count());
+        
+        pq.push(2);
+        assert_cmpint(3, CompareOperator.EQ, pq.peek_count());
+        
+        // Remove items
+        pq.pop();
+        assert_cmpint(2, CompareOperator.EQ, pq.peek_count());
+        
+        pq.pop();
+        assert_cmpint(1, CompareOperator.EQ, pq.peek_count());
+        
+        pq.pop();
+        assert_cmpint(0, CompareOperator.EQ, pq.peek_count());
+    });
+
+    Test.add_func("/invercargill/priority_queue/string_priority", () => {
+        var pq = new PriorityQueue<string>((a, b) => a.collate(b)); // Alphabetical order
+        
+        // Push items
+        pq.push("zebra");
+        pq.push("apple");
+        pq.push("banana");
+        pq.push("cherry");
+        
+        // Pop in alphabetical order
+        assert_cmpstr("apple", CompareOperator.EQ, pq.pop());
+        assert_cmpstr("banana", CompareOperator.EQ, pq.pop());
+        assert_cmpstr("cherry", CompareOperator.EQ, pq.pop());
+        assert_cmpstr("zebra", CompareOperator.EQ, pq.pop());
+    });
+
+}

src/tests/TestRunner.vala

@@ -2,6 +2,14 @@
 public static int main(string[] args) {
 
     Test.init(ref args);
+    
+    bool run_speed_tests = false;
+    for (int i = 0; i < args.length; i++) {
+        if (args[i] == "--speed" || args[i] == "-s") {
+            run_speed_tests = true;
+            break;
+        }
+    }
 
     where_tests();
     select_tests();
@@ -25,16 +33,19 @@ public static int main(string[] args) {
     set_tests();
     fifo_tests();
     lifo_tests();
+    priority_queue_tests();
 
     var result = Test.run();
     
-    series_speed_test();
-    vector_speed_test();
-    buffer_speed_test();
-    set_speed_test();
-    dictionary_speed_test();
-    sorted_series_speed_test();
-    fifo_speed_test();
+    if (run_speed_tests) {
+        series_speed_test();
+        vector_speed_test();
+        buffer_speed_test();
+        set_speed_test();
+        dictionary_speed_test();
+        sorted_series_speed_test();
+        fifo_speed_test();
+    }
 
     return result;
 }

src/tests/meson.build

@@ -24,6 +24,7 @@ sources += files('Integration/OrderBy.vala')
 sources += files('Integration/Set.vala')
 sources += files('Integration/Fifo.vala')
 sources += files('Integration/Lifo.vala')
+sources += files('Integration/PriorityQueue.vala')
 
 sources += files('Speed/SpeedTest.vala')
 sources += files('Speed/Series.vala')