A standard for continuation-local variables?

[eirslett]

https://groups.google.com/forum/#!topic/rxjava/3iW-U6VZ_8c

I'm cross-posting this here, since I'm not quite sure where it belongs. One could think of it as a feature request. (Just close the issue if it shouldn't be on GitHub) The topic didn't get that much discussion on the mailing list...

Hey!

I've been working a lot with Twitter's Zipkin project, which makes heavy use of the "Local" construct: https://github.com/twitter/util/blob/master/util-core/src/main/scala/com/twitter/util/Local.scala
It's basically an improved ThreadLocal, but a key feature of a Local variable is that its state is passed through to other threads whenever you make Futures, and compose these Futures. That makes it easy to do debug tracing in reactive systems. (Especially Zipkin tracing!)

Scala has DynamicVariable, and I think it works in a similar way with Scala Futures, but I'm not 100 % sure.

On of the problems with RxJava, is that ThreadLocal state is not passed on between Observables, at least not intentionally. So you have to add a custom execution hook that is Zipkin-specific, and transfer the state between Observables.
Would it be a good idea to factor out a common construct for this type of situation, a ContinuationLocal maybe (or just modify Twitter's "Local", as it's already implemented, but without the Scala dependency and dependency on Twitter's libraries) - so that one could reliably expect its state to be passed on across threads, as long as it's within the same context?

Even better, it would be great to have a construct in the core Java library that is expected to be passed between ExecutorServices - it's a real pain when working with legacy code, and Trace debugging doesn't work because we lose the state inside e.g. a ForkJoinPool. One has to decorate every ExecutorService in the application with a custom wrapper, but that's very intrusive! (and requires a lot of rewriting, also it's not always possible.)

I really think the concept should be shared across different implementations of reactive patterns, so you get interop between libraries whenever they're running within the same JVM.
(E.g. If I use a bridge from Finagle's thrift to Hystrix/RxJava, I would love to have the context transferred, without having to write custom code.)
Should this even be defined as a "mini standard" a.la. Reactive Streams? What are your thoughts?

[jbripley]

Count me as and interested party in this. We're also switching a lot between Finagle, at service in/out end points, and Observable business logic in the middle.

We've been meaning to start using Zipkin for request tracking and we suspected that this might become an issue.

[eirslett]

@paulp @odersky @retronym It would be nice to get some input from a Scala language point of view as well! (Seen in the light of Scala's DynamicVariable)

[retronym]

Scala's DynamicVariable is basically a wrapper around InheritableThreadLocal. It also suffers from the threadpool problem:

scala> val dv = new scala.util.DynamicVariable[String](null)
dv: scala.util.DynamicVariable[String] = DynamicVariable(null)

scala> { dv.withValue("here") { new Thread { override def run = println(dv.value) } start() } }
here

scala> { dv.withValue("here") { import concurrent.ExecutionContext.Implicits.global; concurrent.Future( println(dv.value) ) } }
null
res14: scala.concurrent.Future[Unit] = scala.concurrent.impl.Promise$DefaultPromise@3b74ac8

Perhaps @viktorklang / @richdougherty might be able to chime in with their experience from Akka / Play.

[retronym]

A few related links:

http://stevenskelton.ca/threadlocal-variables-scala-futures/
https://www.playframework.com/documentation/2.3.x/ThreadPools
http://letitcrash.com/post/37482539796/propagating-threadlocals-amongst-actors

[eirslett]

With Akka, it could be hairy - since it would be natural to propagate such data between actors across services, and Akka has location transparency. Given that you store a trace ID in a continuation local (or whatever to call it), and you also store a Trace service (like in the Zipkin case), which is the implementation of the transport (how to store the trace ID to e.g. Cassandra later on). You probably wouldn't want to propagate the service implementation across actors that run on different machines, while it would be natural to propagate it within the same application - which is how Zipkin does it.

[beku8]

+1

[akarnokd]

it's a real pain when working with legacy code, and Trace debugging doesn't work because we lose the state inside e.g. a ForkJoinPool. One has to decorate every ExecutorService in the application with a custom wrapper, but that's very intrusive! (and requires a lot of rewriting, also it's not always possible.

Yes, this is what happens with many of the concurrency logic we hide behind schedulers and operators, but it is generally painful. In addition, I think such construct poses considerable overhead for every task scheduled and since it is opt-in, there is no way to optimize against it. The RxJava way is to use composite objects and pass along any relevant state in the stream. Besides, I'd say that local-passing works in Akka because it is a framework: the framework is in control when you are in or out of an actor and as such, can manage the lifecycle and pass-around of the local state for you. RxJava, however, is a library which can't impose such constraints.

[eirslett]

I think such construct poses considerable overhead for every task scheduled and since it is opt-in, there is no way to optimize against it

Well, as you said, it would be opt-in. Every time you make a new continuation-local variable, we record the fact that there exists at least one continuation-local. If there is at least one, then we transfer all continuation-local variables' state to the new thread. If there are no continuation-locals, we noop (so there is no overhead).

The RxJava way is to use composite objects and pass along any relevant state in the stream.

You're missing a key point here - the reference to the continuation-local must be implicit, outside of the traditional scope where you explicitly pass around references. If you were to pass around trace data explicitly, that would mean you'd have to rewrite your entire application's internal (and external?) API to support tracing. An important feature of tracing systems is that they should be transparent and not intrusive in userland code. (That's one of the things ThreadLocals are made for in the first place, what we're after is a kind of improved ThreadLocal)

[beku8]

Can't there be a way to attach an EventListener, which gets triggered every time a thread is "assigned"?Therefore we will configure it only once for each chain, ThreadLocals will be bound & removed when only appropriate. I'm pretty new to the library, so I'm aware I might be oversimplifying.

[eirslett]

You can use RxJavaSchedulersHook to achieve this already, but you'll have to provide your own continuation local (in this example it's Twitter's), and you'll have to run your application with a system property for it to work:

package com.example
final class Hook extends RxJavaSchedulersHook {
  import com.twitter.util.Local
  override def onSchedule(action: Action0) : Action0 = {
    val ctx = Local.save()
    new Action0 {
      override def call() {
        Local.let(ctx) {
          action.call()
        }
      }
    }
  }
}

This is a Scala snippet that transfers Twitter's Locals between threads in RxJava, but you'll have to run your app with the property

java -jar myapp.jar -Drxjava.plugin.RxJavaSchedulersHook.implementation=com.example.Hook

or you can also configure the plugin with Java, but then you'll have to add code snippets somewhere in your application on startup. The ideal situation would be that it could just work out of the box.

A compromise maybe, would be to let rxjava read classpath:rxjava-plugins.properties (if it exists) and load plugins specified there? It could also be implemented with a ServiceLoader. That way, we could add an optional jar to the classpath that added continuation-local support. (Though I would still say the best option is to provide continuation-locals as a core feature, independent of scheduler hooks)

[beku8]

@eirslett Thanks for the direction! I found what I wanted.

[stealthcode]

Joining the party late but I have something that might be of interest to you. So the continuation-local variable would be accessible to a sequence of observable onNexts. For instance if an observable emitted [1, 2, 3] then you want a context to follow each value through the observable chain that you could add objects to and fetch from. I believe that would be equivalent to your continuation-local variables.

I have a proof of concept project that introduces a BiObservable (TriObservable and etc) that passes more than one data element per emission. This would allow you to map over either data element. For instance, you could do something like this...

Observable<Integer> nums = Observable.range(0,10);
BiObservable<Integer, String> vs = BiObservable.generate(nums, Object::toString);
BiObservable<String[], String> arr = vs.map1((n) -> { return new String[n]; })
    .doOnNext((a, s) -> {
        for (int i = 0; i < a.length; i++)
            a[i] = s;
    }).subscribe(System.out::println);

The operators for BiObservable generally can take a Func1<T1> or Func2<T1, T2> (or actions as the case would be). Your user functions (funcs you pass into map, doOnNext, etc) would remain simple funcs from String to String for instance instead of a complex aggregate object like String to Context where context contains all the data you may need later. However the funcs that need to append new state would be explicit as in the case of the BiObservable.generate above.

Here is a link to the proof of concept project. I would greatly appreciate any feedback or ideas.

[akarnokd]

I did a schedulers blog post with an example that did automatic context copying on the scheduling boundary.

As for the BiObserver, I'm quite interested how you implemented observeOn and the other queueing operators.