Must everything used in thread be `Send`able in Rust?

user_1103561 0
Viewed 297

I'm trying to implement concurrent processing in Rust. Here is the (simplified) code (playground):

struct Checker {
    unsendable: std::rc::Rc<String> // `Rc` does not implement Send + Sync (`Rc` just for simplicity, a graph actually)
}

impl Checker {
    pub fn check(&self, input: String) -> bool {
        // some logics (no matter)
        true
    }
}

struct ParallelChecker {
    allow_checker: Checker,
    block_checker: Checker
}

impl ParallelChecker {
    // `String` for simplicity here
    pub fn check(&self, input: String) -> bool {
        let thread_pool = rayon::ThreadPoolBuilder::new()
            .num_threads(2)
            .build()
            .unwrap();

        // scoped thread pool for simplicity (to avoid `self` is not `'static`)
        thread_pool.scope(move |s| {
            s.spawn(|_| {
                let result = self.allow_checker.check(input.clone());
                // + send result via channel
            });
        });

        thread_pool.scope(move |s| {
            s.spawn(|_| {
                let result = self.block_checker.check(input.clone());
                // + send result via channel
            });
        });

        true // for simplicity
        // + receive result form the channels
    }
}

The problem is that self.allow_checker and self.block_checker don't implement Send:

55 |         thread_pool.scope(move |s| {
   |                     ^^^^^ `Rc<String>` cannot be sent between threads safely
   |
   = help: within `Checker`, the trait `Send` is not implemented for `Rc<String>`
   = note: required because it appears within the type `Checker`
   = note: required because of the requirements on the impl of `Send` for `Mutex<Checker>`
   = note: 1 redundant requirements hidden
   = note: required because of the requirements on the impl of `Send` for `Arc<Mutex<Checker>>`
   = note: required because it appears within the type `[closure@src/parallel_checker.rs:55:27: 60:10]`

I was under the impression that only something that is sent via the channels must implement Send + Sync and i'm probably wrong here.

As you can see the threads code does not have any shared variables (except self). How can i make it working without implementing Send and paying no costs on sync?

I've tried to synchronize the access (though it looks like useless as there is no shared variable in threads), but no luck:

    let allow_checker = Arc::new(Mutex::new(self.allow_checker));
        thread_pool.scope(move |s| {
            s.spawn(|_| {
                let result = allow_checker.lock().unwrap().check(input.clone());
                // + send result via channel
            });
        });

PS. Migrating to Arc is highly undesired due to performance issue and lot's of related code to be migrated to Arc too.

rust multithreading
2 Answers

I was under the impression that only something that is sent via the channels must implement Send + Sync and I'm probably wrong here.

You are slightly wrong:

  • A type is Send if its values can be sent across threads. Many types are Send, String is, for example.
  • A type is Sync if references to its values can be accessed from multiple threads without incurring any data-race. Perhaps surprisingly this means that String is Sync -- by virtue of being immutable when shared -- and in general T is Sync for any &T that is Send.

Note that those rules do not care how values are sent or shared across threads, only that they are.

This is important here because the closure you use to start a thread is itself sent across threads: it is created on the "launcher" thread, and executed on the "launched" thread.

As a result, this closure must be Send, and this means that anything it captures must in turn be Send.

Why doesn't Rc implement Send?

Because its reference count is non-atomic.

That is, if you have:

  • Thread A: one Rc.
  • Thread B: one Rc (same pointee).

And then Thread A drops its handle at the same time Thread B creates a clone, you'd expect the count to be 2 (still) but due to non-atomic accesses it could be only 1 despite 2 handles still existing:

  • Thread A reads count (2).
  • Thread B reads count (2).
  • Thread B writes incremented count (3).
  • Thread A writes decremented count (1).

And then, the next time B drops a handle, the item is destructed and the memory released and any further access via the last handle will blow up in your face.

I've tried to synchronize the access (though it looks like useless as there is no shared variable in threads), but no luck:

You can't wrap a type to make it Send, it doesn't help because it doesn't change the fundamental properties.

The above race condition on Rc could happen even with a Rc wrapped in Arc<Mutex<...>>.

And therefore !Send is contagious and "infects" any containing type.

Migrating to Arc is highly undesired due to performance issue and lot's of related code to be migrated to Arc too.

Arc itself has relatively little performance overhead, so it seems unlikely it would matter unless you keep cloning those Checker, which you could probably improve on -- passing references, instead of clones.

The higher overhead here will come from Mutex (or RwLock) if Checker is not Sync. As I mentioned, any immutable value is trivially Sync, so if you can refactor the internal state of Checker to be Sync, then you can avoid the Mutex entirely and just have Checker contain Arc<State>.

If you have mutable state at the moment, consider extracting it, going towards:

struct Checker {
    unsendable: Arc<Immutable>,
}

impl Checker {
    pub fn check(&self, input: String) -> Mutable {
        unimplemented!()
    }
}

Or drop the idea of parallel checks.

user_1471920

The requirement for Send is not limited to channels. If you look at the documentation for thread::spawn, you will see that the closure it takes must be Send, and therefore can only capture Send values. The same applies to the closure taken by rayon::ThreadPool::scope.

Since you're cloning the strings in the closures, you might as well clone them before spawning the threads:

let input_clone = input.clone();
thread_pool.scope(move |s| {
    s.spawn(|_| {
        let result = self.allow_checker.check(input_clone);
        // + send result via channel
    });
});
user_53970090
Related