Structural `dropRepeat` mechanism and unexpected repeats

[bbarker]

This is more of a discussion rather than a direct issue, but as there is a fair amount to say I bring it up here instead of gitter.

The basic idea is that Rx.dropRepeat is fine for values but not fine for objects in general.

The problem might be avoided with a call to Rx.foldp; here is a real example that isn't (too) long:

  val currentApiUri: Rx[URI] = (checkApiUri(localApiUri)
    |@| apiUriApp.app._2.flatMap(uri => checkApiUri(uri.toString))
    ).map{
    case (Some(lUri), Some(sUri)) => sUri
    case (Some(lUri), None) => lUri
    case (None, Some(sUri)) => sUri
    case (None, None) => defaultUri
  }.foldp(defaultUri){(curUriObj, newUriObj) =>
    if (curUriObj.toString == newUriObj.toString) curUriObj
    else newUriObj
  }.dropRepeats.map{ capiUri =>
    // DEBUG
    println(s"current API URI is $capiUri; hash = ${System.identityHashCode(capiUri)}")
    capiUri
  }

The problem with foldp here is that curUriObj doesn't come from the current value of currentApiUri itself, but rather from its input sources:

(checkApiUri(localApiUri)
    |@| apiUriApp.app._2.flatMap(uri => checkApiUri(uri.toString))
    )

In our case, since checkApiUri creates and returns a new URI object every time it is called, foldp always updates. A relevant part of the web console looks like this for a particular route:

"current API URI is http://localhost:8081/ced2ar-rdb/api; hash = 25"  Dynamic.scala:78:70
"current API URI is http://localhost:8081/ced2ar-rdb/api; hash = 26"  Dynamic.scala:78:70
"current API URI is http://localhost:8081/ced2ar-rdb/api; hash = 27"  Dynamic.scala:78:70
"current API URI is http://localhost:8081/ced2ar-rdb/api; hash = 28"  Dynamic.scala:78:70

An alternative is to define a version of dropRepeats that maintains an internal Var so that Var.update can be called. However, this ends up being a bit tricky, but perhaps it can be simplified (especially if internalized to monadic-rx, I suspect):

  case class Comp[A](oldA: A, newA: A)
  def dropRepeat[A](init: A, rx: Rx[A], keep: Comp[A] => Boolean): Rx[A] = {
    object holder {
      val hvar = Var(init)
      val cc: Cancelable = rx.impure.foreach(newVal => hvar.update(curVal =>
        if (keep(Comp(oldA = curVal, newA = newVal))) curVal else newVal
      ))
      override def finalize(): Unit = {
        try cc.cancel
        finally super.finalize()
      }
    }
    holder.hvar.dropRepeats
  }

  def keepUri(uris: Comp[URI]): Boolean = uris.oldA == uris.newA

  val currentApiSource: Rx[URI] = (checkApiUri(localApiUri)
    |@| apiUriApp.app._2.flatMap(uri => checkApiUri(uri.toString))
    ).map{
    case (Some(lUri), Some(sUri)) => sUri
    case (Some(lUri), None) => lUri
    case (None, Some(sUri)) => sUri
    case (None, None) => defaultUri
  }

  val currentApiUri: Rx[URI] = dropRepeat(defaultUri, currentApiSource, keepUri)
    .map{ capiUri =>
    // DEBUG
    println(s"current API URI is $capiUri; hash = ${System.identityHashCode(capiUri)}")
    capiUri
  }

In particular, I'm dubious about my use of finalize, but I hope that at least demonstrats what I hope is happening...

I'm also a bit surprised that, in this case, the different URI objects apparently evaluate as being unequal in case DropRep(self) => in rx.scala, though maybe this isn't what is happening. Nonetheless, Rx.dropRepeats is apparently not sufficient to prevent multiple emissions with the same URI (capiUri) but different hashcodes in the first example/println above.

When we test URI equivlance we get what we expect in the Scala repl:

scala> val uriA = new URI("https://www.ibm.com/us-en/")
uriA: java.net.URI = https://www.ibm.com/us-en/

scala> val uriB = new URI("https://www.ibm.com/us-en/")
uriB: java.net.URI = https://www.ibm.com/us-en/

scala> uriA == uriB
res0: Boolean = true

The equivalent web console output for our new dropRepeat function is :

"current API URI is http://localhost:8081/ced2ar-rdb/api; hash = 19" Dynamic.scala:78:70

However, this happens 9 times for the same route, and is even more surprising since it is being executed 9 times, even though each URI has the same URI string and hash code. In my code, all these definitions are in a top-level object, so I don't see multiple instance creation being a factor in causing repeated identical messages.

I'm at a bit of a loss. If nothing obvious comes up, I can try to convert some of this into unit test to see if that may help identifying (or at least sharing) the issue.

[fizzy33]

At first blush it looks like dropRepeats isn't working properly. noting that System.identityHashCode(capiUri) will return different values for different objects even if the objects are equal... If you re-ran adding capiUri.hashCode instead I suspect you would get the same hashcode. Noting this is similar dropRepeats issue I have occasionally seen but never been able to pin down.

[OlivierBlanvillain]

@bbarker Could this be solved by using an equality type class (cats.Eq) instead of == in dropRepeats? IIUC you could work around the issue by warping your problematic object into a class implementing whatever equality you need:

case class UriEq(value: URI) {
  override def equals(other: Object): Boolean = value.toString == other.toString
}

val rx: Rx[URI] = ...
rx.map(UriEq.apply).dropRepeats.map(_.value)

[bbarker]

@OlivierBlanvillain That would be nice for some things (very concise!), but I don't understand how it would help here: it appears == works as expected on java.net.URI (at least in the console example shown above).

[bbarker]

@fizzy33 right, I wanted to use the identity hash to see if new objects were actually being created. it is interesting to note that in the second example, even though we have the same identity hash, I see multiple firings.

[OlivierBlanvillain]

@bbarker Indeed it looks like the issue is not on equality itself, I read your text a bit too fast ^^'

I would be very surprised if the problem came from dropRepeats directly, the implementation is so simple... My best guess is that currentApiUri is used in the right hand side of a flatMap (the following does not hold: ra.flatMap(_ => rb.dropRepeats) =!= ra.flatMap(_ => rb).dropRepeats).

I think I would need a minimized version to get to the bottom of this.

[fizzy33]

To be more explicit. Are we sure that URI.equals() isn't broken on the js platform?

[OlivierBlanvillain]

Good suspicion, equality looks non trivial indeed

[bbarker]

@fizzy33 I admit I was wondering about that, but at least in this simple case == works as expected in scala.js: https://scalafiddle.io/sf/nnfr0VI/0 , but I'll investigate at bit more along those lines.

@OlivierBlanvillain Yeah, I also doubt it is dropRepeats causing the problem ... I looked at the simple implementation and thought the same. However, I do think it would be nice to have a dropRepeats that can inspect its current value using a specified "keep" function.

It makes sense to me that you'd want to dropRepeates after a flatMap, and not inside of flatMap, but don't see anything obvious popping up here (the only flatMap in my code is the one you see) - I'll see if I can make a minified example

[bbarker]

I found the (indirect) cause for redundancies using the second method in the original post (or at least most of the redundancies); I have several use cases along these lines that simply use Rx.map:

  implicit val port: Rx[Port] = currentApiUri.map{ curUri =>
    Option(curUri.getPort) match {
      case Some(prt) => if (prt < 0) Port(80) else Port(prt)
      case None => Port(defaultPort)
    }
  }

When I remove this, the log count went from 9 to 7 in my app.

So it seems just mapping again can cause an additional execution in the upstream Rx (currentApiUri), though my impression is that this value should be memoized in some form in the upstream Rx.

[OlivierBlanvillain]

So it seems just mapping again can cause an additional execution in the upstream Rx (currentApiUri), though my impression is that this value should be memoized in some form in the upstream Rx.

This is working as intended, there is no memoization implemented in the library whatsoever:

val source: Var[Int] = Var(0)

def printing[X](x: X): X = {
  println(x)
  x
}

val mapped = source.map(x => printing(x + 1)).map(x => printing(x + 2))

val cc1 = mapped.impure.foreach(_ => ())
val cc2 = mapped.impure.foreach(_ => ())
// 1
// 3
// 1
// 3

source := 1
// 2
// 4
// 2
// 4

source := 2
// 3
// 5
// 3
// 5

source := 3
// 4
// 6
// 4
// 6

cc1.cancel
cc2.cancel

If you add a 3rd mapped.impure.foreach(_ => ()), then you would get a 3rd version of every line...

[fizzy33]

@bbarker it was one of those check the obvious things first, I am fairly convinced it isn't that.

fwiw I have been trying repro as well with no success yet.

[bbarker]

For the sake of completeness here is a simple test, though i guess we've probably beaten this one to death by now:

    var count = 0
    val sourceVar = Var(0)
    val source: Rx[Int] = sourceVar.map{x => count +=1; x}
    val noshare1: Rx[Int] = source.map(identity)
    val noshare2: Rx[Int] = source.map(identity)
    val cc_ns1 = noshare1.impure.foreach(_ => ())
    val cc_ns2 = noshare2.impure.foreach(_ => ())
    assert(count == 2)
    sourceVar := 1
    assert(count == 4)
    cc_ns1.cancel
    cc_ns2.cancel

[fizzy33]

I think the point of good tests is to beat things to death :-)

[bbarker]

Actually, I may have spoke too soon (at least for myself) - I thought I ran the current form but I'm getting the second assert failing with count == 6, not count == 4.

[fizzy33]

fwiw I ran the @bbarker test and it completed without error i.e. both asserts passed. Running in chrome.

[bbarker]

@fizzy33 I realized my issue; I had added some asserts that use impure.value - these run foreach behind the scenes.