Futures, Promises, and shared state in local vars

[jrochkind]

I wrote up this issue more fully in my blog, at https://bibwild.wordpress.com/2015/01/15/ruby-threads-gotcha-with-local-vars-and-shared-state/

But consider this little demo showing the issue:

value = 100

future = Concurrent::Future.execute do
  sleep 1
  # DANGER will robinson!
  value + 1
end

value = 200

puts future.value # you get 201, not 101!

That 'value' local var is the same one as the outer scope, changes to it in the scope where the thread/future was created effect the thread too.

With stdlib Thread.new, you can get around this with the args to Thread.new:

value = "original"
t = Thread.new(value) do |t_value|
   sleep 1
   puts value
end
value = "new"
t.join

You still get "original", not "new", because of the use of the arg to Thread.new to pass the value to the block argument.

A raw Concurrent::ThreadPoolExecutor lets you do this too, eg pool.post(work) do |t_work|

But Future (and Promise?) don't seem to have an API to do this, to pass an arg whose value gets passed to the block arg, to break the 'connection' with the local var in lexical scope, so you can actually fix it's value in the task you are creating and not worry about the outer scope changing it from under you.

Am I missing something? Or do the expected ways one might use Future/Promise mean that this probably won't come up? Or should Future/Promise have an API similar to Thread/ThreadPoolExecutor, to allow parameters to be passed in and fixed?

[jdantonio]

@jrochkind I read your blog post and to be perfectly honest, the behavior you demonstrate for Concurrent::FixedThreadPool.new surprises me. That's not intentional behavior. I'm going to do some more research to try and understand it better.

All of our high-level abstractions (such as Concurrent::Future) post work to an internal thread pool--a pool just like the one you manually create in your other example. In neither case do we use arguments to Thread.new. Threads are created when the thread pool is created. Each of the internal threads loops forever, pulling jobs off an internal Queue. When we post to a thread pool we simply create a 2-element array in which the first element is the argument array and the second element is the proc object. We literally do nothing more than pass the arguments along through the Queue then pass them to the Proc when it is executed. We don't dup or clone the args. Since all Ruby variables are references I would not expect them to be safe.

I'm going to test this on several versions of Ruby and several interpreters (MRI, JRuby, and Rbx) and see if this is consistent across all. Once I have a better understanding of why you get safe variables when posting directly to a thread pool we'll chat more. If there is a consistent and reliable way that we can make variables safe with our high-level abstractions we'll definitely look into extending the APIs of those classes.

PS: The short answer to your question about closures is that closures in Ruby are inherently not thread-safe. There's nothing we can do about that. Functional languages with advanced concurrency capabilities (Erlang, Clojure, Haskell, F#) have immutable variables. That's why they can have closures and advanced concurrency. In Ruby we just have to be more careful and use thread-safe objects (like Concurrent::Atomic). :-(

[chrisseaton]

I wouldn't say that anything here is unsafe. The Ruby semantics may not be what you'd expect, but when you understand that Ruby closures take a reference to the frame, not a copy, they do make sense and I think they're consistent.

If you've ever written code like this:

n = 0
[1, 2, 3].each do |i|
    n += i
end
puts n

Then you already know that closures can mutate the original frame in Ruby. If threads did anything special to isolate frames, then that would be surprising.

[jrochkind]

I believe it the behavior is a result of how ruby works, and you'll have trouble making it any different. It's the reason stdlib Thread.new takes arguments.

I understand that closures are inherently unthreadsafe. So what you have to do is avoid shared state. The trick is passing input parameters into the destined-for-a-thread closure in a way that avoids shared state. Thread.new gives you a way to do that (as far as local-variable-as-shared-state goes; you are still responsible for passing in an immutable object, or not mutating it) -- Thread.new(variable) {|new_variable| .... Concurrent::ThreadPoolExecutor gives you the very same way to do that.

But Concurrent::Future and Concurrent::Promise do not. Perhaps they need to?

Does this make sense?

[jrochkind]

@chrisseaton Right, I agree. The thing is that Thread.new(variable) {|new_variable| ... gives you a way to capture your value separately from the original frame. Concurrent::Future and Concurrent::Promise do not, but may need such?

[chrisseaton]

Yeah I agree with that. It's a lot like the situation in JS where people think they've captured a loop variable in a closure when actually subsequent iterations are modifying what was captured.

[jdantonio]

@jrochkind @chrisseaton It's not the closure's behavior that was surprising me, it's the variable change within the block. Ruby's variables are references and not thread-safe. When we pass an argument to a thread pool we're not doing anything special to it so it shouldn't be safe within the block. Upon further examination I see the issue isn't with what's happening within the block, it's that the original example performs an assignment outside of the block.

Here's an alternative to the Thread.new example:

require 'thread'

value = ['original']

t = Thread.new(value) do |t_value|
  sleep 1
  puts t_value.first #=> "changed" - not safe
end

value[0] = 'changed'

t.join

And the same example using a Concurrent::FixedThreadPool:

require 'concurrent'

work = ['original']
pool = Concurrent::FixedThreadPool.new(5)

pool.post(work) do |t_work|
  sleep 1
  puts t_work.first #=> "new" -- is NOT safe
end

work[0] = 'new'

pool.shutdown
pool.wait_for_termination

In this example, the puts statement displays "changed", thus demonstrating the unsafe nature of Ruby's variables. Passing a variable to a Thread.new isn't really a safe operation, it just presents the illusion of safety in certain circumstances.

[chrisseaton]

Ah so what you're saying is that the key problem is that passing the variable into the block isn't useful, because it provides the illusion of isolation (I think 'isolation' is much better term that 'safety' here), but that doesn't hold for mutable objects. And so you don't think passing variables into blocks to be run potentially on other threads is not a good idea.

The example with a local assignment outside the block and then again inside the block doesn't surprise me. That behaviour seems consistent to me with Ruby's handling of frames: Identifiers defined in an enclosing lexical scope are references to the frame that corresponds to that scope. If a block contains an identifier not present in an enclosing lexical scope, you get a new entry in the frame for it.

[jrochkind]

Right, that's exactly what's going on @jdantonio. The thing is that you often need a way out of it -- you need a way to pass input, in a variable (rather than a literal string) to a `unit of work' (whether using raw Thread, ThreadPoolExecutor, Future, or anything else), while making the variable fixed, no longer tied to the variable in the stack frame.

I tried to explain this above and in my blog post, but it is confusing, I'll try one more time. Imagine:

## UNSAFE:
input.each do |unit_of_input|
   Thread.new { do_something ; do_something_with unit_of_input }
end

That is NOT safe, becuase the unit_of_input reference in do_something_with unit_of_input will change each time the loop comes around in the main thread. You have a race condition.

This is a very common use pattern, and one I've run into multiple times. The solution is using the Thread.new argument:

input.each do |unit_of_input|
  # SAFE
   Thread.new(unit_of_input) { |arg_input| do_something;  do_something_with arg_input }

We have the same issue with Futures, but with no way to get around it:

input.each do |unit_of_input|
  # UNSAFE
   Concurrent::Future.execute { do_something; do_something_with unit_of_input }
end

But we do not have access to what we did with Thread.new to make it safe.

Yes, on top of that, the actual object of unit_of_work needs to be immutable (or you need not to mutate it). But this is a separate, higher-level concern. Even if the object in unit_of_work is immutable -- those unsafe examples are still unsafe. Not because the object in unit_of_work is mutating, but because the value referenced by the variable unit_of_work is changing, and the variable is shared accross all threads.

The stdlib Thread solution is the arguments on Thread.new. But perhaps there's another solution? I think looping over a collection and giving each item in the collection to a thread-based executor of some sort (whether a raw Thread, or a Future, or a Promise, or a ThreadPoolExecutor) is a fairly pedestrian use case, isn't it? How can I do it in a safe way with Future, protected from race conditions? I don't believe I can.

[jdantonio]

@chrisseaton Yes. When I first looked at the examples I only thought about the mutability of the objects within the block, I didn't fully think through the assignment statements and the scope of the different variables. Hence my initial confusion.

[jrochkind]

And I think ThreadPoolExecutor supports the same pattern as stdlib Thread just because it allows you to pass arguments to post that get copied to block arguments in the task block. Just having this API:

 pool.post(value) {|some_value| .... }

Is what does it. It's possible the people implementing that API were just copying it from somewhere else without thinking of what it did -- or were just thinking of it in terms of being able to reuse a closure with different parameters or something. But, accidentally or not, what it does is let you do this safely:

some_input.each do |unit_of_input|
   # SAFE (but yes, you can't mutate the object in unit_of_input)
    pool.post(unit_of_input) {|arg_input| do_something; do_something_with arg_input }
end

It works just becuase it's taking the arg and supplying it as a param to the block.call, it doesn't actually need to use the underlying Thread's actual similar API, it just needs to give you a way to pass the value into a variable that is not from the parent frame, which it does.

[jdantonio]

@jrochkind "But we do not have access to what we did with Thread.new to make it safe." But that's the thing, passing a variable through Thread.new isn't actually "safe." The reference gets copied but it's still a reference to the same object, which is mutable. So, as @chrisseaton suggests, we should probably not use the word "safe" to describe this phenomenon.

One possible implementation could be this:

arg1 = 'foo'
arg2 = 'bar'
pool = Concurrent::FixedThreadPool.new(5)

future1 = Concurrent::Future.execute{ do_something }
future2 = Concurrent::Future.execute(args: arg1){|a| do_something }
future3 = Concurrent::Future.execute(args: [arg1, arg2]){|a, b| do_something }
future4 = Concurrent::Future.execute(executor: pool, args: [arg1, arg2]){|a, b| do_something }

(Note: The execute method signature currently accepts an options hash so we can't have it accept *args, which is why my example uses an :args option.)

My question is this: Would this be more confusing for the average Rubyist since it gives the illusion of thread-safety. Whatever is passed in is still a mutable reference and misuse can lead to many problems. Do we do the average user a disservice by giving them a potentially dangerous and unsafe option?

[jrochkind]

Assume the value is immutable. THEN it's safe. If "safe" is the wrong word, that's fine, this stuff is super confusing to talk about, maybe I don't have the right unsafe terminology.

But the problem here is entirely separate from the mutability of the object. Even if the object is immutable (say, you freeze it with ruby #freeze), is is STILL unsafe. It's not about mutability, it's about the variable reference itself being shared, with no way to make it unshared.

Whether it's confusing or not -- it's the only way I know of to perform a very common sort of task. Creating a threaded executor (raw thread, future, whatever) in the body of a loop.

Okay, actually, I thought of another way, you can do this:

def create_task(arg)
     Concurrent::Future.execute { do_something_with arg }
end
input.each do |i|
    create_task(i)
end

That is safe, by moving the thread creation into a method, so the local variable arg no longer gets shared over loop iterations.

But is it less confusing that that is safe (I know you don't like me using the word 'safe', I don't know what other adjective to use, sorry!), where this is not?

#UNSAFE
input.each do |i|
    i.freeze # does NOT make it safe
    Concurrent::Future.execute { do_something_with i }
end

The current situation is super confusing! So confusing that I'm only coming to terms with it finally now after a couple years of ruby multi-threaded concurrency dev! Yes, I am not super happy with any of the solutions, I think it is an inherently confusing situation. But copying the pattern Thread.new uses seems appropriate to me -- if it's the best the stdlib can do, makes sense to support it for consistentcy. Of course, developers are still free to use the "Make sure you only create futures in isolated methods instead" technique, either way. And maybe there are more techniques I'm not thinking of.

[chrisseaton]

Well this is kind of what the data flow pattern does. There if you have any data dependencies you're supposed to pass them through arguments. Of course there's no way to enforce that, and we have the same problem here.

If only Ruby could tell us if the proc we have been passed contains a reference to an enclosing lexical frame. Then we could issue a warning. Maybe Ruby should do that anyway for blocks passed to threads.

[jrochkind]

(And yeah, Actor pattern avoids this problem entirely, where Future and Promise do not, it is true! I think anyway!)

[jrochkind]

@chrisseaton Okay, as a thought experiment, let's just pretend that magically, we could warn if your block included a reference to an enclosing lexical frame -- or even refuse to run the block if it did.

The problem then is, you have NO way to pass a parameter into a Future -- the only way to construct a future in terms of something you have in a variable (rather than a literal hard-coded into source code) is to do it as a reference to an enclosing frame! No?

The Thread.new(value) {|arg_value| ... trick is for exactly that -- being able to pass in a parameter without it being a reference to an enclosing frame. Without that... the best you can do, that I can think of, is making sure the enclosing frame gets abandoned immediately after creating the Future, as in my example where I use a separate method for no reason other than to get an enclosing frame that gets abandoned after Future creation.