// Do nothing.
    }

    taskFaker.advanceUntil(0.µs)
    assertThat(taskRunner.activeQueues()).containsExactly(redQueue, blueQueue)

    taskFaker.advanceUntil(100.µs)
    assertThat(taskRunner.activeQueues()).containsExactly(blueQueue)

    taskFaker.advanceUntil(200.µs)
    assertThat(taskRunner.activeQueues()).isEmpty()

    taskFaker.assertNoMoreTasks()

      }
    }
  }

  private fun ensureAllTaskQueuesIdle() {
    val entryTime = System.nanoTime()

    for (queue in TaskRunner.INSTANCE.activeQueues()) {
      // We wait at most 1 second, so we don't ever turn multiple lost threads into
      // a test timeout failure.
      val waitTime = (entryTime + 1_000_000_000L - System.nanoTime())

    factory.newConnection(pool, routeA1)
    // Racy causing flaky tests
    // assertThat(taskRunner.activeQueues()).isNotEmpty()
    assertThat(taskRunnerThreads).isNotEmpty()
    Thread.sleep(100)
    for (t in taskRunnerThreads) {
      t.interrupt()
    }
    Thread.sleep(100)
    assertThat(taskRunner.activeQueues()).isEmpty()
  }

  @Test fun connectionPreWarmingHttp1() {

      maxElements: Int,
    ) = error("unsupported")
  }

  /** Returns true if no tasks have been scheduled. This runs the coordinator for confirmation. */
  fun isIdle() = taskRunner.activeQueues().isEmpty()

  override fun close() {
    tasksExecutor.shutdownNow()
  }

  companion object {
    var instance = 0
  }

  }

  /**
   * Returns a snapshot of queues that currently have tasks scheduled. The task runner does not
   * necessarily track queues that have no tasks scheduled.
   */
  fun activeQueues(): List<TaskQueue> {
    this.withLock {
      return busyQueues + readyQueues
    }
  }

  fun cancelAll() {
    assertLockHeld()
    for (i in busyQueues.size - 1 downTo 0) {

    // Cause acceptConnections() to break out.
    serverSocket.closeQuietly()

    // Await shutdown.
    for (queue in taskRunner.activeQueues()) {
      if (!queue.idleLatch().await(5, TimeUnit.SECONDS)) {
        throw AssertionError("Gave up waiting for queue to shut down")
      }
    }
    taskRunnerBackend.shutdown()
  }

        .socket(peer.openSocket().asBufferedSocket(), "peer")
        .pushObserver(IGNORE)
        .build()
    connection.start(sendConnectionPreface = false)
    val queues = taskRunner.activeQueues()
    assertThat(queues).hasSize(1)
  }

  private fun data(byteCount: Int): Buffer = Buffer().write(ByteArray(byteCount))

  private fun assertStreamData(
    expected: String?,