Add foldMRec

Resolves typelevel#1030, though it may be desirable change the default
implementation to be lazy in the future. This can be done with `FreeT`
but as of now I'm not sure how to do it without.

core/src/main/scala/cats/Foldable.scala

@@ -1,5 +1,6 @@
 package cats
 
+import cats.data.Xor
 import scala.collection.mutable
 import simulacrum.typeclass
 
@@ -77,11 +78,27 @@ import simulacrum.typeclass
     foldLeft(fa, B.empty)((b, a) => B.combine(b, f(a)))
 
   /**
-   * Left associative monadic folding on `F`.
+   * Left associative monadic fold on `F`.
+   *
+   * The default implementation of this is based on `foldLeft`, and thus will
+   * always fold across the entire structure. When `G` has a `MonadRec` instance,
+   * the [[foldMRec]] variation may be preferred, as it may be able to
+   * short-circuit the fold.
    */
   def foldM[G[_], A, B](fa: F[A], z: B)(f: (B, A) => G[B])(implicit G: Monad[G]): G[B] =
     foldLeft(fa, G.pure(z))((gb, a) => G.flatMap(gb)(f(_, a)))
 
+  /**
+   * Left associative monadic fold on `F`.
+   *
+   * This is similar to [[foldM]] (and the default implementation is
+   * identical), but certain structures are able to implement this in such a
+   * way that folds can be short-circuited (not traverse the entirety of the
+   * structure), depending on the `G` result produced at a given step.
+   */
+  def foldMRec[G[_], A, B](fa: F[A], z: B)(f: (B, A) => G[B])(implicit G: MonadRec[G]): G[B] =
+    foldM(fa, z)(f)
+
   /**
    * Traverse `F[A]` using `Applicative[G]`.
    *
@@ -306,4 +323,12 @@ object Foldable {
       Eval.defer(if (it.hasNext) f(it.next, loop()) else lb)
     loop()
   }
+
+  def iterableFoldMRec[M[_], A, B](fa: Iterable[A], z: B)(f: (B, A) => M[B])(implicit M: MonadRec[M]): M[B] = {
+    val go: ((B, Iterable[A])) => M[(B, Iterable[A]) Xor B] = { case (b, fa) =>
+      if (fa.isEmpty) M.pure(Xor.right(b))
+      else M.map(f(b, fa.head))(b1 => Xor.left((b1, fa.tail)))
+    }
+    M.tailRecM((z, fa))(go)
+  }
 }

core/src/main/scala/cats/std/list.scala

@@ -75,6 +75,9 @@ trait ListInstances extends cats.kernel.std.ListInstances {
         fa.forall(p)
 
       override def isEmpty[A](fa: List[A]): Boolean = fa.isEmpty
+
+      override def foldMRec[G[_], A, B](fa: List[A], z: B)(f: (B, A) => G[B])(implicit G: MonadRec[G]): G[B] =
+        Foldable.iterableFoldMRec(fa, z)(f)
     }
 
   implicit def listShow[A:Show]: Show[List[A]] =

core/src/main/scala/cats/std/map.scala

@@ -46,5 +46,8 @@ trait MapInstances extends cats.kernel.std.MapInstances {
         Foldable.iterateRight(fa.values.iterator, lb)(f)
 
       override def isEmpty[A](fa: Map[K, A]): Boolean = fa.isEmpty
+
+      override def foldMRec[G[_], A, B](fa: Map[K, A], z: B)(f: (B, A) => G[B])(implicit G: MonadRec[G]): G[B] =
+        Foldable.iterableFoldMRec(fa.values, z)(f)
     }
 }

core/src/main/scala/cats/std/set.scala

@@ -25,6 +25,11 @@ trait SetInstances extends cats.kernel.std.SetInstances {
         fa.forall(p)
 
       override def isEmpty[A](fa: Set[A]): Boolean = fa.isEmpty
+
+      override def foldMRec[G[_], A, B](fa: Set[A], z: B)(f: (B, A) => G[B])(implicit G: MonadRec[G]): G[B] =
+        // Repeatedly calling .tail on a large set is really slow, so we
+        // convert to a stream first.
+        Foldable.iterableFoldMRec(fa.toStream, z)(f)
     }
 
   implicit def setShow[A:Show]: Show[Set[A]] = new Show[Set[A]] {

core/src/main/scala/cats/std/stream.scala

@@ -53,6 +53,9 @@ trait StreamInstances extends cats.kernel.std.StreamInstances {
         fa.forall(p)
 
       override def isEmpty[A](fa: Stream[A]): Boolean = fa.isEmpty
+
+      override def foldMRec[G[_], A, B](fa: Stream[A], z: B)(f: (B, A) => G[B])(implicit G: MonadRec[G]): G[B] =
+        Foldable.iterableFoldMRec(fa, z)(f)
     }
 
   implicit def streamShow[A: Show]: Show[Stream[A]] =

core/src/main/scala/cats/std/vector.scala

@@ -53,6 +53,9 @@ trait VectorInstances extends cats.kernel.std.VectorInstances {
         fa.exists(p)
 
       override def isEmpty[A](fa: Vector[A]): Boolean = fa.isEmpty
+
+      override def foldMRec[G[_], A, B](fa: Vector[A], z: B)(f: (B, A) => G[B])(implicit G: MonadRec[G]): G[B] =
+        Foldable.iterableFoldMRec(fa, z)(f)
     }
 
   implicit def vectorShow[A:Show]: Show[Vector[A]] =

tests/src/test/scala/cats/tests/FoldableTests.scala

@@ -78,14 +78,36 @@ class FoldableTestsAdditional extends CatsSuite {
     larger.value should === (large.map(_ + 1))
   }
 
-  test("Foldable[List].foldM stack safety") {
-    def nonzero(acc: Long, x: Long): Option[Long] =
+  def checkFoldMStackSafety[F[_]](fromRange: Range => F[Int])(implicit F: Foldable[F]): Unit = {
+    def nonzero(acc: Long, x: Int): Option[Long] =
       if (x == 0) None else Some(acc + x)
 
-    val n = 100000L
-    val expected = n*(n+1)/2
-    val actual = Foldable[List].foldM((1L to n).toList, 0L)(nonzero)
-    assert(actual.get == expected)
+    val n = 100000
+    val expected = n.toLong*(n.toLong+1)/2
+    val foldMResult = F.foldM(fromRange(1 to n), 0L)(nonzero)
+    assert(foldMResult.get == expected)
+    val foldMRecResult = F.foldMRec(fromRange(1 to n), 0L)(nonzero)
+    assert(foldMRecResult.get == expected)
+  }
+
+  test("Foldable[List].foldM stack safety") {
+    checkFoldMStackSafety[List](_.toList)
+  }
+
+  test("Foldable[Stream].foldM stack safety") {
+    checkFoldMStackSafety[Stream](_.toStream)
+  }
+
+  test("Foldable[Vector].foldM stack safety") {
+    checkFoldMStackSafety[Vector](_.toVector)
+  }
+
+  test("Foldable[Set].foldM stack safety") {
+    checkFoldMStackSafety[Set](_.toSet)
+  }
+
+  test("Foldable[Map[String, ?]].foldM stack safety") {
+    checkFoldMStackSafety[Map[String, ?]](_.map(x => x.toString -> x).toMap)
   }
 
   test("Foldable[Stream]") {
@@ -112,6 +134,9 @@ class FoldableTestsAdditional extends CatsSuite {
     // test trampolining
     val large = Stream((1 to 10000): _*)
     assert(contains(large, 10000).value)
+
+    // test laziness of foldMRec
+    dangerous.foldMRec(0)((acc, a) => if (a < 2) Some(acc + a) else None) should === (None)
   }
 }