{"action":"created","issue":{"url":"https://api.github.com/repos/scalaz/scalaz-stream/issues/290","labels_url":"https://api.github.com/repos/scalaz/scalaz-stream/issues/290/labels{/name}","comments_url":"https://api.github.com/repos/scalaz/scalaz-stream/issues/290/comments","events_url":"https://api.github.com/repos/scalaz/scalaz-stream/issues/290/events","html_url":"https://github.com/scalaz/scalaz-stream/issues/290","id":51905840,"number":290,"title":"termination of a Process fails with collect and filter","user":{"login":"alissapajer","id":1088100,"avatar_url":"https://avatars.githubusercontent.com/u/1088100?v=3","gravatar_id":"","url":"https://api.github.com/users/alissapajer","html_url":"https://github.com/alissapajer","followers_url":"https://api.github.com/users/alissapajer/followers","following_url":"https://api.github.com/users/alissapajer/following{/other_user}","gists_url":"https://api.github.com/users/alissapajer/gists{/gist_id}","starred_url":"https://api.github.com/users/alissapajer/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/alissapajer/subscriptions","organizations_url":"https://api.github.com/users/alissapajer/orgs","repos_url":"https://api.github.com/users/alissapajer/repos","events_url":"https://api.github.com/users/alissapajer/events{/privacy}","received_events_url":"https://api.github.com/users/alissapajer/received_events","type":"User","site_admin":false},"labels":[],"state":"open","locked":false,"assignee":null,"milestone":null,"comments":47,"created_at":"2014-12-14T01:39:28Z","updated_at":"2015-02-03T09:49:34Z","closed_at":null,"body":"Hi folks,\r\n\r\nWhat follows is a minimized case of a `Process` constructed with `collect` that doesn't halt when instructed to do so with a `wye.interrupt`.\r\n\r\n```scala\r\nimport scalaz._\r\nimport scalaz.concurrent._\r\nimport scalaz.stream._\r\nimport scalaz.stream.async.mutable._\r\n\r\nclass Example {\r\n  val poison: Signal[Boolean] = async.signal[Boolean]\r\n\r\n  val original: Process[Task, Int] = Process.constant(42) //onComplete Process.eval_(Task.delay(println(s\"ORIGINAL COMPLETE\")))\r\n\r\n  val secondary: Process[Task, Int] = original collect {\r\n    case in if in < 0 => in\r\n  } //onComplete Process.eval_(Task.delay(println(s\"SECONDARY COMPLETE\")))\r\n\r\n  val output: Task[Seq[Int]] = {\r\n    val signal: Process[Task, Boolean] = poison.discrete\r\n    val result: Process[Task, Int] = (signal wye secondary)(wye.interrupt) //onComplete Process.eval_(Task.delay(println(s\"INTERRUPT COMPLETE\")))\r\n    result.runLog[Task, Int]\r\n  }\r\n\r\n  object Effects extends Thread {\r\n    override def run(): Unit = {\r\n      poison.set(true).run\r\n      println(\"POISON SET\")\r\n    }\r\n  }\r\n}\r\n\r\n// val example = new Example(); example.Effects.start(); example.output.run.size\r\n// for { i <- (1 to 100) } yield { val example = new Example(); example.Effects.start(); example.output.run.size }\r\n```\r\n\r\nWhen this example is run, the process sometimes halts (as expected), but more often than not it fails to halt. Here is a [thread dump](https://gist.github.com/alissapajer/c8237dd846cd522ea84a) taken while the process is spinning.\r\n\r\n\r\n****Ways the example can be modified so that the process still spins****\r\n\r\n(1) Include a value caught by the collect:\r\n```scala\r\nval original: Process[Task, Int] = Process(-12) ++ Process.constant(42)\r\n```\r\n(2) Use `filter` instead of `collect`: \r\n```scala\r\nval secondary: Process[Task, Int] = original filter { _ < 0 }\r\n```\r\n\r\n\r\n****Ways the example can be modified so that the process never spins****\r\n\r\n(1) Bypass the `collect` entirely with\r\n```scala\r\nval secondary = original\r\n```\r\n(2) Replace `<` with `>` in the `collect` `case` statement\r\n```scala\r\ncase in if in > 0 => in\r\n```\r\n(3) Explicitly keep everything that is collected using\r\n```scala\r\ncase in => in\r\n```\r\n(4) Use `filter` with `>` instead of collect: \r\n```scala\r\nval secondary: Process[Task, Int] = original filter { _ > 0 }\r\n```"},"comment":{"url":"https://api.github.com/repos/scalaz/scalaz-stream/issues/comments/72622016","html_url":"https://github.com/scalaz/scalaz-stream/issues/290#issuecomment-72622016","issue_url":"https://api.github.com/repos/scalaz/scalaz-stream/issues/290","id":72622016,"user":{"login":"pchlupacek","id":678560,"avatar_url":"https://avatars.githubusercontent.com/u/678560?v=3","gravatar_id":"","url":"https://api.github.com/users/pchlupacek","html_url":"https://github.com/pchlupacek","followers_url":"https://api.github.com/users/pchlupacek/followers","following_url":"https://api.github.com/users/pchlupacek/following{/other_user}","gists_url":"https://api.github.com/users/pchlupacek/gists{/gist_id}","starred_url":"https://api.github.com/users/pchlupacek/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/pchlupacek/subscriptions","organizations_url":"https://api.github.com/users/pchlupacek/orgs","repos_url":"https://api.github.com/users/pchlupacek/repos","events_url":"https://api.github.com/users/pchlupacek/events{/privacy}","received_events_url":"https://api.github.com/users/pchlupacek/received_events","type":"User","site_admin":false},"created_at":"2015-02-03T09:49:34Z","updated_at":"2015-02-03T09:49:34Z","body":"@alissapajer can you perhaps try with this step implementation? : \r\n\r\n```scala \r\n\r\n  final def step: HaltOrStep[F, O] = {\r\n    @tailrec\r\n    def go(cur: Process[F,O], stack: Vector[Cause => Trampoline[Process[F,O]]], cnt: Int) : HaltOrStep[F,O] = {\r\n      if (stack.nonEmpty) cur match {\r\n        case Halt(End) if cnt <=0  => Step(empty,Cont(stack))\r\n        case Halt(cause) => go(Try(stack.head(cause).run), stack.tail, cnt -1)\r\n        case Emit(os) if os.isEmpty => Step(empty,Cont(stack))\r\n        case emt@(Emit(os)) => Step(emt,Cont(stack))\r\n        case awt@Await(_,_) => Step(awt,Cont(stack))\r\n        case Append(h,st) => go(h, st fast_++ stack, cnt -1)\r\n      } else cur match {\r\n        case hlt@Halt(cause) => hlt\r\n        case emt@Emit(os) if (os.isEmpty) => halt0\r\n        case emt@Emit(os) => Step(emt,Cont(Vector.empty))\r\n        case awt@Await(_,_) => Step(awt,Cont(Vector.empty))\r\n        case Append(h,st) => go(h,st, cnt -1)\r\n      }\r\n    }\r\n    go(this,Vector.empty, 10)\r\n\r\n  }\r\n\r\n\r\n\r\n```"}}