Add a coalescing writer (#1537)

Motivation:

Creating frames in HTTP/2 isn't free: they incur one allocation for the
frame payload and another for the data buffer. In addition to this
HTTP/2 currently creates a promise per write, so we also have an
additional write promise. On top of the allocations we have the repeated
work for processing extra frames.

We can recover much of this cost by coalescing writes before a flush.
For streaming RPCs this can have a fairly significant impact.

Modifications:

- Add a coalescing length prefixed writer: serialized messages are
  appended and length prefixing (and compression) is done lazily on
  calls to `next()`.
- Add an internal 'one-or-many' queue which is a FIFO queue backed by
  either a single element on the stack or a deque. This avoids
  allocating a queue on the heap for the length prefixed writer when
  there is only one message appended before flushing.

Result:

Nothing changes yet, but we have a writer in place which we can adopt
later.

Sources/GRPC/CoalescingLengthPrefixedMessageWriter.swift

@@ -0,0 +1,357 @@
+/*
+ * Copyright 2022, gRPC Authors All rights reserved.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+import DequeModule
+import NIOCore
+
+internal struct CoalescingLengthPrefixedMessageWriter {
+  /// Length of the gRPC message header (1 compression byte, 4 bytes for the length).
+  static let metadataLength = 5
+
+  /// Message size above which we emit two buffers: one containing the header and one with the
+  /// actual message bytes. At or below the limit we copy the message into a new buffer containing
+  /// both the header and the message.
+  ///
+  /// Using two buffers avoids expensive copies of large messages. For smaller messages the copy
+  /// is cheaper than the additional allocations and overhead required to send an extra HTTP/2 DATA
+  /// frame.
+  ///
+  /// The value of 16k was chosen empirically. We subtract the length of the message header
+  /// as `ByteBuffer` reserve capacity in powers of two and want to avoid overallocating.
+  static let singleBufferSizeLimit = 16384 - Self.metadataLength
+
+  /// The compression algorithm to use, if one should be used.
+  private let compression: CompressionAlgorithm?
+  /// Any compressor associated with the compression algorithm.
+  private let compressor: Zlib.Deflate?
+
+  /// Whether the compression message flag should be set.
+  private var supportsCompression: Bool {
+    return self.compression != nil
+  }
+
+  /// A scratch buffer that we encode messages into: if the buffer isn't held elsewhere then we
+  /// can avoid having to allocate a new one.
+  private var scratch: ByteBuffer
+
+  /// Outbound buffers waiting to be written.
+  private var pending: OneOrManyQueue<Pending>
+
+  private struct Pending {
+    var buffer: ByteBuffer
+    var promise: EventLoopPromise<Void>?
+    var compress: Bool
+
+    init(buffer: ByteBuffer, compress: Bool, promise: EventLoopPromise<Void>?) {
+      self.buffer = buffer
+      self.promise = promise
+      self.compress = compress
+    }
+
+    var isSmallEnoughToCoalesce: Bool {
+      let limit = CoalescingLengthPrefixedMessageWriter.singleBufferSizeLimit
+      return self.buffer.readableBytes <= limit
+    }
+
+    var shouldCoalesce: Bool {
+      return self.isSmallEnoughToCoalesce || self.compress
+    }
+  }
+
+  private enum State {
+    // Coalescing pending messages.
+    case coalescing
+    // Emitting a non-coalesced message; the header has been written, the body
+    // needs to be written next.
+    case emittingLargeFrame(ByteBuffer, EventLoopPromise<Void>?)
+  }
+
+  private var state: State
+
+  init(compression: CompressionAlgorithm? = nil, allocator: ByteBufferAllocator) {
+    self.compression = compression
+    self.scratch = allocator.buffer(capacity: 0)
+    self.state = .coalescing
+    self.pending = .init()
+
+    switch self.compression?.algorithm {
+    case .none, .some(.identity):
+      self.compressor = nil
+    case .some(.deflate):
+      self.compressor = Zlib.Deflate(format: .deflate)
+    case .some(.gzip):
+      self.compressor = Zlib.Deflate(format: .gzip)
+    }
+  }
+
+  /// Append a serialized message buffer to the writer.
+  mutating func append(buffer: ByteBuffer, compress: Bool, promise: EventLoopPromise<Void>?) {
+    let pending = Pending(
+      buffer: buffer,
+      compress: compress && self.supportsCompression,
+      promise: promise
+    )
+
+    self.pending.append(pending)
+  }
+
+  /// Return a tuple of the next buffer to write and its associated write promise.
+  mutating func next() -> (Result<ByteBuffer, Error>, EventLoopPromise<Void>?)? {
+    switch self.state {
+    case .coalescing:
+      // Nothing pending: exit early.
+      if self.pending.isEmpty {
+        return nil
+      }
+
+      // First up we need to work out how many elements we're going to pop off the front
+      // and coalesce.
+      //
+      // At the same time we'll compute how much capacity we'll need in the buffer and cascade
+      // their promises.
+      var messagesToCoalesce = 0
+      var requiredCapacity = 0
+      var promise: EventLoopPromise<Void>?
+
+      for element in self.pending {
+        if !element.shouldCoalesce {
+          break
+        }
+
+        messagesToCoalesce &+= 1
+        requiredCapacity += element.buffer.readableBytes + Self.metadataLength
+        if let existing = promise {
+          existing.futureResult.cascade(to: element.promise)
+        } else {
+          promise = element.promise
+        }
+      }
+
+      if messagesToCoalesce == 0 {
+        // Nothing to coalesce; this means the first element should be emitted with its header in
+        // a separate buffer. Note: the force unwrap is okay here: we early exit if `self.pending`
+        // is empty.
+        let pending = self.pending.pop()!
+
+        // Set the scratch buffer to just be a message header then store the message bytes.
+        self.scratch.clear(minimumCapacity: Self.metadataLength)
+        self.scratch.writeMultipleIntegers(UInt8(0), UInt32(pending.buffer.readableBytes))
+        self.state = .emittingLargeFrame(pending.buffer, pending.promise)
+        return (.success(self.scratch), nil)
+      } else {
+        self.scratch.clear(minimumCapacity: requiredCapacity)
+
+        // Drop and encode the messages.
+        while messagesToCoalesce > 0, let next = self.pending.pop() {
+          messagesToCoalesce &-= 1
+          do {
+            try self.encode(next.buffer, compress: next.compress)
+          } catch {
+            return (.failure(error), promise)
+          }
+        }
+
+        return (.success(self.scratch), promise)
+      }
+
+    case let .emittingLargeFrame(buffer, promise):
+      // We just emitted the header, now emit the body.
+      self.state = .coalescing
+      return (.success(buffer), promise)
+    }
+  }
+
+  private mutating func encode(_ buffer: ByteBuffer, compress: Bool) throws {
+    if let compressor = self.compressor, compress {
+      try self.encode(buffer, compressor: compressor)
+    } else {
+      try self.encode(buffer)
+    }
+  }
+
+  private mutating func encode(_ buffer: ByteBuffer, compressor: Zlib.Deflate) throws {
+    // Set the compression byte.
+    self.scratch.writeInteger(UInt8(1))
+    // Set the length to zero; we'll write the actual value in a moment.
+    let payloadSizeIndex = self.scratch.writerIndex
+    self.scratch.writeInteger(UInt32(0))
+
+    let bytesWritten: Int
+    do {
+      var buffer = buffer
+      bytesWritten = try compressor.deflate(&buffer, into: &self.scratch)
+    } catch {
+      throw error
+    }
+
+    self.scratch.setInteger(UInt32(bytesWritten), at: payloadSizeIndex)
+
+    // Finally, the compression context should be reset between messages.
+    compressor.reset()
+  }
+
+  private mutating func encode(_ buffer: ByteBuffer) throws {
+    self.scratch.writeMultipleIntegers(UInt8(0), UInt32(buffer.readableBytes))
+    self.scratch.writeImmutableBuffer(buffer)
+  }
+}
+
+/// A FIFO-queue which allows for a single to be stored on the stack and defers to a
+/// heap-implementation if further elements are added.
+///
+/// This is useful when optimising for unary streams where avoiding the cost of a heap
+/// allocation is desirable.
+internal struct OneOrManyQueue<Element>: Collection {
+  private var backing: Backing
+
+  private enum Backing: Collection {
+    case none
+    case one(Element)
+    case many(Deque<Element>)
+
+    var startIndex: Int {
+      switch self {
+      case .none, .one:
+        return 0
+      case let .many(elements):
+        return elements.startIndex
+      }
+    }
+
+    var endIndex: Int {
+      switch self {
+      case .none:
+        return 0
+      case .one:
+        return 1
+      case let .many(elements):
+        return elements.endIndex
+      }
+    }
+
+    subscript(index: Int) -> Element {
+      switch self {
+      case .none:
+        fatalError("Invalid index")
+      case let .one(element):
+        assert(index == 0)
+        return element
+      case let .many(elements):
+        return elements[index]
+      }
+    }
+
+    func index(after index: Int) -> Int {
+      switch self {
+      case .none:
+        return 0
+      case .one:
+        return 1
+      case let .many(elements):
+        return elements.index(after: index)
+      }
+    }
+
+    var count: Int {
+      switch self {
+      case .none:
+        return 0
+      case .one:
+        return 1
+      case let .many(elements):
+        return elements.count
+      }
+    }
+
+    var isEmpty: Bool {
+      switch self {
+      case .none:
+        return true
+      case .one:
+        return false
+      case let .many(elements):
+        return elements.isEmpty
+      }
+    }
+
+    mutating func append(_ element: Element) {
+      switch self {
+      case .none:
+        self = .one(element)
+      case let .one(one):
+        var elements = Deque<Element>()
+        elements.reserveCapacity(16)
+        elements.append(one)
+        elements.append(element)
+        self = .many(elements)
+      case var .many(elements):
+        self = .none
+        elements.append(element)
+        self = .many(elements)
+      }
+    }
+
+    mutating func pop() -> Element? {
+      switch self {
+      case .none:
+        return nil
+      case let .one(element):
+        self = .none
+        return element
+      case var .many(many):
+        self = .none
+        let element = many.popFirst()
+        self = .many(many)
+        return element
+      }
+    }
+  }
+
+  init() {
+    self.backing = .none
+  }
+
+  var isEmpty: Bool {
+    return self.backing.isEmpty
+  }
+
+  var count: Int {
+    return self.backing.count
+  }
+
+  var startIndex: Int {
+    return self.backing.startIndex
+  }
+
+  var endIndex: Int {
+    return self.backing.endIndex
+  }
+
+  subscript(index: Int) -> Element {
+    return self.backing[index]
+  }
+
+  func index(after index: Int) -> Int {
+    return self.backing.index(after: index)
+  }
+
+  mutating func append(_ element: Element) {
+    self.backing.append(element)
+  }
+
+  mutating func pop() -> Element? {
+    return self.backing.pop()
+  }
+}

Tests/GRPCTests/CoalescingLengthPrefixedMessageWriterTests.swift

@@ -0,0 +1,221 @@
+/*
+ * Copyright 2022, gRPC Authors All rights reserved.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+@testable import GRPC
+import NIOCore
+import NIOEmbedded
+import XCTest
+
+internal final class CoalescingLengthPrefixedMessageWriterTests: GRPCTestCase {
+  private let loop = EmbeddedEventLoop()
+
+  private func makeWriter(
+    compression: CompressionAlgorithm? = .none
+  ) -> CoalescingLengthPrefixedMessageWriter {
+    return .init(compression: compression, allocator: .init())
+  }
+
+  private func testSingleSmallWrite(withPromise: Bool) throws {
+    var writer = self.makeWriter()
+
+    let promise = withPromise ? self.loop.makePromise(of: Void.self) : nil
+    writer.append(buffer: .smallEnoughToCoalesce, compress: false, promise: promise)
+
+    let (result, maybePromise) = try XCTUnwrap(writer.next())
+    try result.assertValue { buffer in
+      var buffer = buffer
+      let (compressed, length) = try XCTUnwrap(buffer.readMessageHeader())
+      XCTAssertFalse(compressed)
+      XCTAssertEqual(length, UInt32(ByteBuffer.smallEnoughToCoalesce.readableBytes))
+      XCTAssertEqual(buffer.readSlice(length: Int(length)), .smallEnoughToCoalesce)
+      XCTAssertEqual(buffer.readableBytes, 0)
+    }
+
+    // No more bufers.
+    XCTAssertNil(writer.next())
+
+    if withPromise {
+      XCTAssertNotNil(maybePromise)
+    } else {
+      XCTAssertNil(maybePromise)
+    }
+
+    // Don't leak the promise.
+    maybePromise?.succeed(())
+  }
+
+  private func testMultipleSmallWrites(withPromise: Bool) throws {
+    var writer = self.makeWriter()
+    let messages = 100
+
+    for _ in 0 ..< messages {
+      let promise = withPromise ? self.loop.makePromise(of: Void.self) : nil
+      writer.append(buffer: .smallEnoughToCoalesce, compress: false, promise: promise)
+    }
+
+    let (result, maybePromise) = try XCTUnwrap(writer.next())
+    try result.assertValue { buffer in
+      var buffer = buffer
+
+      // Read all the messages.
+      for _ in 0 ..< messages {
+        let (compressed, length) = try XCTUnwrap(buffer.readMessageHeader())
+        XCTAssertFalse(compressed)
+        XCTAssertEqual(length, UInt32(ByteBuffer.smallEnoughToCoalesce.readableBytes))
+        XCTAssertEqual(buffer.readSlice(length: Int(length)), .smallEnoughToCoalesce)
+      }
+
+      XCTAssertEqual(buffer.readableBytes, 0)
+    }
+
+    // No more bufers.
+    XCTAssertNil(writer.next())
+
+    if withPromise {
+      XCTAssertNotNil(maybePromise)
+    } else {
+      XCTAssertNil(maybePromise)
+    }
+
+    // Don't leak the promise.
+    maybePromise?.succeed(())
+  }
+
+  func testSingleSmallWriteWithPromise() throws {
+    try self.testSingleSmallWrite(withPromise: true)
+  }
+
+  func testSingleSmallWriteWithoutPromise() throws {
+    try self.testSingleSmallWrite(withPromise: false)
+  }
+
+  func testMultipleSmallWriteWithPromise() throws {
+    try self.testMultipleSmallWrites(withPromise: true)
+  }
+
+  func testMultipleSmallWriteWithoutPromise() throws {
+    try self.testMultipleSmallWrites(withPromise: false)
+  }
+
+  func testSingleLargeMessage() throws {
+    var writer = self.makeWriter()
+    writer.append(buffer: .tooBigToCoalesce, compress: false, promise: nil)
+
+    let (result1, promise1) = try XCTUnwrap(writer.next())
+    XCTAssertNil(promise1)
+    try result1.assertValue { buffer in
+      var buffer = buffer
+      let (compress, length) = try XCTUnwrap(buffer.readMessageHeader())
+      XCTAssertFalse(compress)
+      XCTAssertEqual(Int(length), ByteBuffer.tooBigToCoalesce.readableBytes)
+      XCTAssertEqual(buffer.readableBytes, 0)
+    }
+
+    let (result2, promise2) = try XCTUnwrap(writer.next())
+    XCTAssertNil(promise2)
+    result2.assertValue { buffer in
+      XCTAssertEqual(buffer, .tooBigToCoalesce)
+    }
+
+    XCTAssertNil(writer.next())
+  }
+
+  func testMessagesBeforeLargeAreCoalesced() throws {
+    var writer = self.makeWriter()
+    // First two should be coalesced. The third should be split as two buffers.
+    writer.append(buffer: .smallEnoughToCoalesce, compress: false, promise: nil)
+    writer.append(buffer: .smallEnoughToCoalesce, compress: false, promise: nil)
+    writer.append(buffer: .tooBigToCoalesce, compress: false, promise: nil)
+
+    let (result1, _) = try XCTUnwrap(writer.next())
+    try result1.assertValue { buffer in
+      var buffer = buffer
+      for _ in 0 ..< 2 {
+        let (compress, length) = try XCTUnwrap(buffer.readMessageHeader())
+        XCTAssertFalse(compress)
+        XCTAssertEqual(Int(length), ByteBuffer.smallEnoughToCoalesce.readableBytes)
+        XCTAssertEqual(buffer.readSlice(length: Int(length)), .smallEnoughToCoalesce)
+      }
+      XCTAssertEqual(buffer.readableBytes, 0)
+    }
+
+    let (result2, _) = try XCTUnwrap(writer.next())
+    try result2.assertValue { buffer in
+      var buffer = buffer
+      let (compress, length) = try XCTUnwrap(buffer.readMessageHeader())
+      XCTAssertFalse(compress)
+      XCTAssertEqual(Int(length), ByteBuffer.tooBigToCoalesce.readableBytes)
+      XCTAssertEqual(buffer.readableBytes, 0)
+    }
+
+    let (result3, _) = try XCTUnwrap(writer.next())
+    result3.assertValue { buffer in
+      XCTAssertEqual(buffer, .tooBigToCoalesce)
+    }
+
+    XCTAssertNil(writer.next())
+  }
+
+  func testCompressedMessagesAreAlwaysCoalesced() throws {
+    var writer = self.makeWriter(compression: .gzip)
+    writer.append(buffer: .smallEnoughToCoalesce, compress: false, promise: nil)
+    writer.append(buffer: .tooBigToCoalesce, compress: true, promise: nil)
+
+    let (result, _) = try XCTUnwrap(writer.next())
+    try result.assertValue { buffer in
+      var buffer = buffer
+
+      let (compress1, length1) = try XCTUnwrap(buffer.readMessageHeader())
+      XCTAssertFalse(compress1)
+      XCTAssertEqual(Int(length1), ByteBuffer.smallEnoughToCoalesce.readableBytes)
+      XCTAssertEqual(buffer.readSlice(length: Int(length1)), .smallEnoughToCoalesce)
+
+      let (compress2, length2) = try XCTUnwrap(buffer.readMessageHeader())
+      XCTAssertTrue(compress2)
+      // Can't assert the length or the content, only that the length must be equal
+      // to the number of remaining bytes.
+      XCTAssertEqual(Int(length2), buffer.readableBytes)
+    }
+
+    XCTAssertNil(writer.next())
+  }
+}
+
+extension Result {
+  func assertValue(_ body: (Success) throws -> Void) rethrows {
+    switch self {
+    case let .success(success):
+      try body(success)
+    case let .failure(error):
+      XCTFail("Unexpected failure: \(error)")
+    }
+  }
+}
+
+extension ByteBuffer {
+  fileprivate static let smallEnoughToCoalesce = Self(repeating: 42, count: 128)
+  fileprivate static let tooBigToCoalesce = Self(
+    repeating: 42,
+    count: CoalescingLengthPrefixedMessageWriter.singleBufferSizeLimit + 1
+  )
+
+  mutating func readMessageHeader() -> (Bool, UInt32)? {
+    if let (compressed, length) = self.readMultipleIntegers(as: (UInt8, UInt32).self) {
+      return (compressed != 0, length)
+    } else {
+      return nil
+    }
+  }
+}

Tests/GRPCTests/OneOrManyQueueTests.swift

@@ -0,0 +1,138 @@
+/*
+ * Copyright 2022, gRPC Authors All rights reserved.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+@testable import GRPC
+import XCTest
+
+internal final class OneOrManyQueueTests: GRPCTestCase {
+  func testIsEmpty() {
+    XCTAssertTrue(OneOrManyQueue<Int>().isEmpty)
+  }
+
+  func testIsEmptyManyBacked() {
+    XCTAssertTrue(OneOrManyQueue<Int>.manyBacked.isEmpty)
+  }
+
+  func testCount() {
+    var queue = OneOrManyQueue<Int>()
+    XCTAssertEqual(queue.count, 0)
+    queue.append(1)
+    XCTAssertEqual(queue.count, 1)
+  }
+
+  func testCountManyBacked() {
+    var manyBacked = OneOrManyQueue<Int>.manyBacked
+    XCTAssertEqual(manyBacked.count, 0)
+    for i in 1 ... 100 {
+      manyBacked.append(1)
+      XCTAssertEqual(manyBacked.count, i)
+    }
+  }
+
+  func testAppendAndPop() {
+    var queue = OneOrManyQueue<Int>()
+    XCTAssertNil(queue.pop())
+
+    queue.append(1)
+    XCTAssertEqual(queue.count, 1)
+    XCTAssertEqual(queue.pop(), 1)
+
+    XCTAssertNil(queue.pop())
+    XCTAssertEqual(queue.count, 0)
+    XCTAssertTrue(queue.isEmpty)
+  }
+
+  func testAppendAndPopManyBacked() {
+    var manyBacked = OneOrManyQueue<Int>.manyBacked
+    XCTAssertNil(manyBacked.pop())
+
+    manyBacked.append(1)
+    XCTAssertEqual(manyBacked.count, 1)
+    manyBacked.append(2)
+    XCTAssertEqual(manyBacked.count, 2)
+
+    XCTAssertEqual(manyBacked.pop(), 1)
+    XCTAssertEqual(manyBacked.count, 1)
+
+    XCTAssertEqual(manyBacked.pop(), 2)
+    XCTAssertEqual(manyBacked.count, 0)
+
+    XCTAssertNil(manyBacked.pop())
+    XCTAssertTrue(manyBacked.isEmpty)
+  }
+
+  func testIndexes() {
+    var queue = OneOrManyQueue<Int>()
+    XCTAssertEqual(queue.startIndex, 0)
+    XCTAssertEqual(queue.endIndex, 0)
+
+    // Non-empty.
+    queue.append(1)
+    XCTAssertEqual(queue.startIndex, 0)
+    XCTAssertEqual(queue.endIndex, 1)
+  }
+
+  func testIndexesManyBacked() {
+    var queue = OneOrManyQueue<Int>.manyBacked
+    XCTAssertEqual(queue.startIndex, 0)
+    XCTAssertEqual(queue.endIndex, 0)
+
+    for i in 1 ... 100 {
+      queue.append(i)
+      XCTAssertEqual(queue.startIndex, 0)
+      XCTAssertEqual(queue.endIndex, i)
+    }
+  }
+
+  func testIndexAfter() {
+    var queue = OneOrManyQueue<Int>()
+    XCTAssertEqual(queue.startIndex, queue.endIndex)
+    XCTAssertEqual(queue.index(after: queue.startIndex), queue.endIndex)
+
+    queue.append(1)
+    XCTAssertNotEqual(queue.startIndex, queue.endIndex)
+    XCTAssertEqual(queue.index(after: queue.startIndex), queue.endIndex)
+  }
+
+  func testSubscript() throws {
+    var queue = OneOrManyQueue<Int>()
+    queue.append(42)
+    let index = try XCTUnwrap(queue.firstIndex(of: 42))
+    XCTAssertEqual(queue[index], 42)
+  }
+
+  func testSubscriptManyBacked() throws {
+    var queue = OneOrManyQueue<Int>.manyBacked
+    for i in 0 ... 100 {
+      queue.append(i)
+    }
+
+    for i in 0 ... 100 {
+      XCTAssertEqual(queue[i], i)
+    }
+  }
+}
+
+extension OneOrManyQueue where Element == Int {
+  static var manyBacked: Self {
+    var queue = OneOrManyQueue()
+    // Append and pop to move to the 'many' backing.
+    queue.append(1)
+    queue.append(2)
+    XCTAssertEqual(queue.pop(), 1)
+    XCTAssertEqual(queue.pop(), 2)
+    return queue
+  }
+}