有效检查 Scala Stream 中的值是否存在（以功能方式）

Question

我遇到了this question on reddit - 我对此感到很困惑，但我不知道（而且似乎没有令人满意的答案）。我这里敢抄：

假设我们有一个流，其中每个元素取决于其前身，例如简单的伪随机序列，例如：

def neum(a:Int): Stream[Int] = Stream.iterate(a)(a => (a*a/100)%10000)

这是来自问题引用的exercise 的冯-诺依曼随机发生器

从给定值开始，我们想知道序列何时进入循环。我们可以很容易地以命令的形式做到这一点，使用 Set 来存储值：

// like in java
Set<Integer> values = new HashSet<>();
while (true) {
    int x = nextValueInSequence(x)
    if (values.contains(x)) {
        break;
    }
    values.add(x);
}

但是，对于 Scala，提出“功能性”解决方案很有趣。但是问题的作者似乎不知道如何在保持O(N) 时间复杂度的情况下实现这一点。我也是。那里唯一的评论看起来像是建议直接的O(N^2) 解决方案。

Answer 1

我相信，有一种算法具有 O(N log N) 时间复杂度（或许可以将其提高到 O(N)），并且 O(1) 总内存消耗。也就是说，我们不必记住大部分以前的数字。不过常数因子相当高。

此内存消耗不是使用Stream 计算的，而是使用由起始元素和递归公式定义的通用数字序列计算的。例如Iterator.iterate(start)(a => a * a / 100 % 10000)。 Stream 会记住以前的结果并有效地使其 O(N) 记忆。

假设序列在循环开始前有 P ≥ 0 个元素，并且循环中 L ≥ 1 个元素。例如，序列[2, 10, 13, 9, 11, 17, 11, 17, ...] 有 P = 4 和 L = 2。我们需要找到 P + L。

在算法中，我们必须遍历序列。我将当前位置称为“指针”。在数字序列中，指针仅表示数字。最初指针等于序列的起始元素，为了将指针向前移动 1 步，我们必须对其应用递归公式。

现在是算法：

1. 以两个指向序列开头的指针开始：“慢”和“快”。慢速指针一次移动 1 步，快速指针一次移动 2 步（即递归公式的 2 次应用）。

2. 最初指针是相等的。开始向前移动它们直到它们再次相等，并跟踪慢指针的步数。让我们命名指针再次相等的步数K₀。可以证明 P ≤ K₀ P+L 和K₀ = 0 (mod L)。

   这一步我们还要特别注意的情况，当P = 0：当指针变得相等时，如果它们也等于起始元素，我们应该设置 K₀ = 0，以便以后能够区分这种情况。

   这一步的时间复杂度是O(N)。

3. 现在指针肯定在序列的循环内。再次开始向前移动它们并跟踪慢速指针的步数，直到它们再次相等。这个步数就是序列的循环L的长度。 （也可以在这一步只移动慢指针，直到它回到相同的位置，但我会复用这个函数来移动两者，不会增加时间复杂度）

   这一步的时间复杂度是O(N)。

4. 现在我们要计算 P。我们可以注意到，如果在算法的第 2 步）中，我们不是从头开始“快速”指针，而是有一些移位 S：0 ≤ S L，那么结果就会不同：K_S = K₀ – S，如果 S ≤ K₀ – P；或 K_S = K₀ + L – S强>否则。所以，我们可以用二分查找最大位移S^*：0≤S^*L，其中 K_S^* = K₀ – S^*。然后我们可以找到P = K₀ - S^*，并返回P + L = K₀ – S^* + L

   这一步的时间复杂度为 O(N log N)，因为二分查找的每一步都需要 O(N)。

所以我们有一个算法，它在 O(N log N) 和 O(1) 内存中工作。这是一个代码示例：

case class Sequence[T](start: T)(f: T => T) {
  def next = Sequence(f(start))(f)
  def forward(steps: Int) =
    Sequence(Function.chain(List.fill(steps)(f))(start))(f)
}

object Sequence {
  def neum(a: Int) = Sequence(a)(a => a * a / 100 % 10000)

  def movesToEquality[T](
    slow: Sequence[T], fast: Sequence[T], count: Int = 1
  ): (Sequence[T], Int) = {
    val nextSlow = slow.next
    val nextFast = fast.forward(2)
    if (nextSlow == nextFast) (nextSlow, count)
    else movesToEquality(nextSlow, nextFast, count+1)
  }

  def findLoopStart[T](seq: Sequence[T]): Int = {
    val (inLoop, k0) = movesToEquality(seq, seq) match {
      case (c, k) if c == seq => (c, 0)
      case other => other
    }
    val (_, loopSize) = movesToEquality(inLoop, inLoop)

    def binarySearch(lo: Int, hi: Int): Int = {
      if (lo + 1 >= hi) lo
      else {
        val mid = (lo + hi) / 2
        if (movesToEquality(seq, seq.forward(mid))._2 == k0 - mid)
          binarySearch(mid, hi)
        else
          binarySearch(lo, mid)
      }
    }

    k0 - binarySearch(0, loopSize) + loopSize
  }
}

object Main extends App {
  println(Sequence.findLoopStart(Sequence.neum(1)))
  println(Sequence.findLoopStart(Sequence.neum(4100)))
  println(Sequence.findLoopStart(Sequence.neum(5761)))
}

Answer 2

这是在 O(N) 时间和 O(1) 空间中运行的 Kolmar 答案的简化版本。 它基本上是这样做的：

1. 使用快速\慢速策略查找任何重复的数字。它是可被实数周期整除且大于非循环前缀的最小整数。
2. 寻找真实时期。
3. 查找在句点之后重复的第一个元素，即循环的开始。
4. 返回非循环前缀部分和循环部分。

代码：

def cycleOf[T](seq: => Iterator[T]): (Iterator[T], Iterator[T]) = {
  def fast = seq.sliding(1, 2) map (_.head)
  val meet = seq zip fast drop 1 dropWhile { case (x, y) => x != y }
  val met = meet.next()
  val period = (meet indexOf met) + 1
  val start = seq drop period zip seq indexWhere { case (x, y) => x == y }
  (seq take start, seq.slice(start, start + period))
}

你可以试试

val (prefix, cycle) = cycleOf(neum(5761).iterator)

那么prefix.toList就是

列表（5761、1891、5758、1545、3870、9769、4333、7748、315、992、9840、 8256、1615、6082、9907、1486、2081、3305、9230、1929、7210、9841、 8452、4363、357、1274、6230、8129、806、6496、1980、9204、7136、9224、 821、6740、4276、2841、712、5069、6947、2608、8016、2562、5638、7870、 9369、7781、5439、5827、9539、9925、5056、5631、7081、1405、9740、 8676、2729、4474、166、275、756、5715、6612、7185、6242、9625、6406、 368、1354、8333、4388、2545、4770、7529、6858、321、1030、609、3708、 7492、1300、6900）

而cycle.toList 是

List(6100, 2100, 4100, 8100)

还请注意 SpiderPig 的建议：您可以在您的 neum 定义中简单地将 Stream 替换为 Iterator 以获得更节省内存的版本。

Answer 3

我确信有更好的方法来写这个，但这是我的第一次尝试：

def loop[A](xs: Stream[A]): Set[A] =
  xs.scanLeft(Set.empty[A])(_ + _).sliding(2)
    .find(_.map(_.size).toSet.size == 1).get.head

scala> neum(93).take(8).toList
res0: List[Int] = List(93, 86, 73, 53, 28, 7, 0, 0)

scala> loop(neum(93))
res1: Set[Int] = Set(0, 93, 28, 53, 73, 86, 7)

Answer 4

我相信可以在尾递归函数的帮助下使用Set 的解决方案：

@tailrec
def neumannCount(x: Int, m: Set[Int] = Set[Int]()): Int = {
    if (m.contains(x)) m.size else neumannCount(x * x / 100 % 10000, m + x)
}

该函数只是获取当前值和先前元素的集合。它检查 Set 中是否存在值，如果不存在 - 然后生成下一个元素和另一个添加当前元素的集合 - 并将它们传递给对同一函数的另一个调用。当最后遇到该值时 - 我们只返回 Set 的大小作为结果（因此该函数是尾递归的）。

我相信这在时间上应该是 O(1)，在空间上应该是 O(N)，这要归功于建立在彼此之上的不可变集合（如果我理解正确的话）。

Answer 5

必须是流吗？迭代器为此更快。 这里有两种不同的解决方案。两者都是功能性的，不会改变任何状态。

def neumann(seed: Int): Int = {
  Iterator.iterate(seed)(s => ((s * s)/100)%10000)
          .scanLeft(Set.empty[Int])((set, n) => if(set(n)) Set(-1) else set + n)
          .takeWhile(_ != Set(-1)).size - 1
}

def neumann(seed: Int): Int = {
  def search(s: Int, set: Set[Int], count: Int): Int = {
    if(set(s)) count
    else search(((s * s)/100)%10000, set + s, count + 1)
  }
  search(seed, Set.empty[Int], 0)
}