Fix tryPush/tryPop semantics and DRY queue ops

Author: dyxushuai

src/queue.zig

@@ -30,13 +30,7 @@ pub fn Queue(
                 self.not_empty.wait(&self.mutex);
             }
             std.debug.assert(!self.isEmptyLH());
-            if (self.isFullLH()) {
-                // If we are full, wake up a push that might be
-                // waiting here.
-                self.not_full.signal();
-            }
-
-            return self.popLH();
+            return self.popAndSignalLH();
         }
 
         /// Push an item into the queue. Blocks until an item has been
@@ -48,41 +42,27 @@ pub fn Queue(
                 self.not_full.wait(&self.mutex);
             }
             std.debug.assert(!self.isFullLH());
-            const was_empty = self.isEmptyLH();
-
-            self.buf[self.mask(self.write_index)] = item;
-            self.write_index = self.mask2(self.write_index + 1);
-
-            // If we were empty, wake up a pop if it was waiting.
-            if (was_empty) {
-                self.not_empty.signal();
-            }
+            self.pushAndSignalLH(item);
         }
 
         /// Push an item into the queue. Returns true when the item
         /// was successfully placed in the queue, false if the queue
         /// was full.
         pub fn tryPush(self: *Self, item: T) bool {
             self.mutex.lock();
-            if (self.isFullLH()) {
-                self.mutex.unlock();
-                return false;
-            }
-            self.mutex.unlock();
-            self.push(item);
+            defer self.mutex.unlock();
+            if (self.isFullLH()) return false;
+            self.pushAndSignalLH(item);
             return true;
         }
 
         /// Pop an item from the queue. Returns null when no item is
         /// available.
         pub fn tryPop(self: *Self) ?T {
             self.mutex.lock();
-            if (self.isEmptyLH()) {
-                self.mutex.unlock();
-                return null;
-            }
-            self.mutex.unlock();
-            return self.pop();
+            defer self.mutex.unlock();
+            if (self.isEmptyLH()) return null;
+            return self.popAndSignalLH();
         }
 
         /// Poll the queue. This call blocks until events are in the queue
@@ -150,6 +130,24 @@ pub fn Queue(
             return index % (2 * self.buf.len);
         }
 
+        fn pushAndSignalLH(self: *Self, item: T) void {
+            const was_empty = self.isEmptyLH();
+            self.buf[self.mask(self.write_index)] = item;
+            self.write_index = self.mask2(self.write_index + 1);
+            if (was_empty) {
+                self.not_empty.signal();
+            }
+        }
+
+        fn popAndSignalLH(self: *Self) T {
+            const was_full = self.isFullLH();
+            const result = self.popLH();
+            if (was_full) {
+                self.not_full.signal();
+            }
+            return result;
+        }
+
         fn popLH(self: *Self) T {
             const result = self.buf[self.mask(self.read_index)];
             self.read_index = self.mask2(self.read_index + 1);