Implementation of the join operation #63

dns2utf8 · 2017-07-07T11:40:55Z

it restricts the operation to observer only.

Because the std::sync::mpsc::channel does not offer a way to query the queue length I added another counter field: stored_jobs_counter

frewsxcv · 2017-07-07T17:06:09Z

What about if we use Condvar for this?

https://doc.rust-lang.org/std/sync/struct.Condvar.html

Seems like we could store a Condvar in the ThreadPool, and if the user calls join, we call wait on the Condvar, and every time the ThreadPool finishes, we call notify_all.

dns2utf8 · 2017-07-08T10:39:20Z

Good point, we trade polling for a little memory. I am working on it.

dns2utf8 · 2017-07-08T14:27:56Z

The implementation could be optimized further by moving more/all of the data to ThreadPoolSharedData.

While thinking about the problem I was thinking what if the pool were to be used to sync other threads like in the example below?
It currently does not work and it would require the Sync flag:

unsafe impl Sync for ThreadPool { }

        let pool = Arc::new(
                    ThreadPool::new_with_name("multi join test", 8));
        let test_count = Arc::new(AtomicUsize::new(0));

        let pool_r0 = pool.clone();
        let pool_r1 = pool.clone();
        let test_count_r0 = test_count.clone();
        let test_count_r1 = test_count.clone();

        let t0 = thread::spawn(move||{
            for _ in 0..21 {
                let test_count = test_count_r0.clone();
                pool_r0.execute(move || {
                    sleep(Duration::from_secs(2));
                    test_count.fetch_add(1, Ordering::Release);
                });
            }
            pool_r0.join();
            assert_eq!(42, test_count_r0.load(Ordering::Acquire));
        });

        let t1 = thread::spawn(move||{
            for _ in 0..21 {
                let test_count = test_count_r1.clone();
                pool_r1.execute(move || {
                    sleep(Duration::from_secs(2));
                    test_count.fetch_add(1, Ordering::Release);
                });
            }
            pool_r1.join();
            assert_eq!(42, test_count_r1.load(Ordering::Acquire));
        });

        
        pool.join();
        assert_eq!(42, test_count.load(Ordering::Acquire));

        t0.join().unwrap();
        t1.join().unwrap();

frewsxcv · 2017-07-08T18:11:50Z

The implementation could be optimized further by moving more/all of the data to ThreadPoolSharedData.

Is it possible you could pull this change out into a separate pull request? We can merge that first without this join method. Then we can rebase this join pull request off those changes once those merge.

Open interface new_with_name to accept anything convertable into String

Regarding this commit, see my response in #58

dns2utf8 · 2017-07-08T19:18:58Z

Well it is going to take some time, since I decided to create the shared data struct when I realized the many times I would have to pass around 6 counters for every operation and handle their inconsistent names after I implemented the join. Let me see what I can do.

…st-threadpool#63

dns2utf8 · 2017-07-08T19:46:05Z

I created two new branches for the PRs with one commit each. It was simpler than doing some git magic 😉

frewsxcv · 2017-07-09T01:32:12Z

This usage of sleep can be replaced with join, right?. If so, there are a few other examples and tests that can be replaced:

https://github.com/dns2utf8/rust-threadpool/blob/cc5fc936e71954c1e938dd8d6bc872c22bf5d6ac/lib.rs#L457
https://github.com/dns2utf8/rust-threadpool/blob/cc5fc936e71954c1e938dd8d6bc872c22bf5d6ac/lib.rs#L478
https://github.com/dns2utf8/rust-threadpool/blob/cc5fc936e71954c1e938dd8d6bc872c22bf5d6ac/lib.rs#L492
https://github.com/dns2utf8/rust-threadpool/blob/cc5fc936e71954c1e938dd8d6bc872c22bf5d6ac/lib.rs#L691

dns2utf8 · 2017-07-09T18:01:52Z

Yes, let me replace them quickly. As mentioned by @bblancha this should solve #66 as well.

dns2utf8 · 2017-07-09T18:39:52Z

I added support for a pool where all threads panic. This would have lead to a stall until at least one job would have succeeded.

Some of the sleeps you mentioned are measuring the active_count. Joining the pool would cause the number to be zero so I left them as they were.

Do you see any unhandled cases that we should test for?

frewsxcv

thanks for this! added a few comments/questions

frewsxcv · 2017-07-09T18:53:20Z

lib.rs

@@ -207,6 +226,9 @@ impl ThreadPool {
        let rx = Arc::new(Mutex::new(rx));

        let shared_data = Arc::new(ThreadPoolSharedData {
+            empty_condvar: Condvar::new(),
+            empty_trigger: Mutex::new(false),
+            stored_jobs_counter: AtomicUsize::new(0),


can we call this queued_count? i think that better parallels the name active_count

frewsxcv · 2017-07-09T18:54:24Z

lib.rs

@@ -250,6 +273,8 @@ impl ThreadPool {
    ///         sleep(Duration::from_secs(5));
    ///     });
    /// }
+    ///
+    /// // wait for the pool to start working


frewsxcv · 2017-07-09T18:57:56Z

lib.rs

@@ -310,6 +333,43 @@ impl ThreadPool {
            }
        }
    }
+
+    /// Block the current thread until all jobs in the pool are completed.
+    /// Once waiting for the pool to complete you can no longer add new jobs, &mut self ensures that.


Does this need to be &mut self? If the user wants to add more jobs to the threadpool, shouldn't they be allowed to?

The reason I did it in the first place was because without the mut many threads can join at the same time.

Since the join is blocking, the current thread can not send new jobs anyways.
Accessing the pool from many threads requires the Sync flag. But I have to reason about that some more because there is an edge case, see this test.

frewsxcv · 2017-07-09T19:05:13Z

lib.rs

+
+    /// Notify all observers joining this pool if there is no more work to do.
+    fn no_work_notify_all(&self) {
+        if self.has_work() == false {


if !self.has_work()

frewsxcv · 2017-07-09T19:15:34Z

lib.rs

+    /// Once waiting for the pool to complete you can no longer add new jobs, &mut self ensures that.
+    ///
+    /// ```
+    /// # use threadpool::ThreadPool;


I think we should uncomment this so we have a complete example a user can just copy/paste

frewsxcv · 2017-07-09T19:22:30Z

lib.rs

@@ -128,6 +128,7 @@ impl<'a> Drop for Sentinel<'a> {
            self.shared_data.active_count.fetch_sub(1, Ordering::SeqCst);
            if panicking() {
                self.shared_data.panic_count.fetch_add(1, Ordering::SeqCst);
+                self.shared_data.no_work_notify_all();


If this line was moved outside this if panicking() conditional, could we get rid of this line? Would that be the same thing?

No, because a worker thread keeps on working on jobs for as long as possible and reducing the expensive create/cleanup cycle of threads.

dns2utf8 · 2017-07-09T21:53:08Z

I added test for the new cases. The PR is currently mergable up to 254fa9d.

…tead.

dns2utf8 · 2017-07-10T21:57:07Z

I found a very nasty deadlock with the fn test_multi_join(). The reason was some threads could end up in an infinite loop because they did not exit the false-positive loop fast enough. Since testing for the Condvar requires the thread to acquire the Mutex shared_data.empty_trigger they would block until the thread was joinable.

While debugging I added the .field("queued_count", &self.queued_count()) to the debug output and exposed the queued_count like the other counters with a function.

What do you think?

frewsxcv · 2017-07-12T04:20:14Z

LGTM, thanks for doing this!

dns2utf8 · 2017-07-12T06:26:03Z

It is a pleasure solving the hard problems 😄

dns2utf8 added a commit to dns2utf8/rust-threadpool that referenced this pull request Jul 8, 2017

Create the internal data structure required for rust-threadpool#63

6b298c5

dns2utf8 added a commit to dns2utf8/rust-threadpool that referenced this pull request Jul 8, 2017

Create the internal data structure required for the join operation ru…

dd7b17b

…st-threadpool#63

dns2utf8 added a commit to dns2utf8/rust-threadpool that referenced this pull request Jul 8, 2017

Create the internal data structure required for the join operation ru…

795ff20

…st-threadpool#63

dns2utf8 changed the title ~~This is a naive implementation of the join operation~~ Implementation of the join operation Jul 8, 2017

dns2utf8 force-pushed the join branch from c06f4ca to f7f0871 Compare July 8, 2017 19:36

Implement join operation with Condvar

cc5fc93

dns2utf8 force-pushed the join branch from f7f0871 to cc5fc93 Compare July 8, 2017 22:32

bblancha mentioned this pull request Jul 9, 2017

Spurious failure with 'test_massive_task_creation' test #66

Closed

Handle join with panic! threads. Use join in examples and tests.

09dbeaa

frewsxcv requested changes Jul 9, 2017

View reviewed changes

dns2utf8 added 3 commits July 9, 2017 22:32

Include feedback

a0a277e

Cleanup use statements

254fa9d

First impl of fn join(&self) with Sync flag.

78e54dd

dns2utf8 added 3 commits July 10, 2017 08:03

fixup use cleanup

f7c3ed5

Moving all Sync fields from ThreadPool to ThreadPoolSharedData

e237d0f

Iter::sum is available with rust 1.11 and later. Using Iter::fold ins…

900bf98

…tead.

dns2utf8 force-pushed the join branch from 121d761 to 53a67e4 Compare July 10, 2017 21:52

Fix nasty deadlock

2694e24

dns2utf8 force-pushed the join branch from 53a67e4 to 2694e24 Compare July 11, 2017 09:15

dns2utf8 mentioned this pull request Jul 11, 2017

On drop join guard #69

Closed

frewsxcv merged commit 0824dce into rust-threadpool:master Jul 12, 2017

frewsxcv mentioned this pull request Jul 12, 2017

Correct way to wait until all threads are finished #19

Closed

dns2utf8 deleted the join branch July 12, 2017 06:20

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Implementation of the join operation #63

Implementation of the join operation #63

dns2utf8 commented Jul 7, 2017

frewsxcv commented Jul 7, 2017

dns2utf8 commented Jul 8, 2017

dns2utf8 commented Jul 8, 2017

frewsxcv commented Jul 8, 2017

dns2utf8 commented Jul 8, 2017

dns2utf8 commented Jul 8, 2017 •

edited

frewsxcv commented Jul 9, 2017

dns2utf8 commented Jul 9, 2017

dns2utf8 commented Jul 9, 2017

frewsxcv left a comment

frewsxcv Jul 9, 2017

dns2utf8 Jul 9, 2017

frewsxcv Jul 9, 2017

frewsxcv Jul 9, 2017

dns2utf8 Jul 9, 2017

frewsxcv Jul 9, 2017

frewsxcv Jul 9, 2017

dns2utf8 Jul 9, 2017

frewsxcv Jul 9, 2017

dns2utf8 Jul 9, 2017

dns2utf8 commented Jul 9, 2017

dns2utf8 commented Jul 10, 2017

frewsxcv commented Jul 12, 2017

dns2utf8 commented Jul 12, 2017

Implementation of the join operation #63

Implementation of the join operation #63

Conversation

dns2utf8 commented Jul 7, 2017

frewsxcv commented Jul 7, 2017

dns2utf8 commented Jul 8, 2017

dns2utf8 commented Jul 8, 2017

frewsxcv commented Jul 8, 2017

dns2utf8 commented Jul 8, 2017

dns2utf8 commented Jul 8, 2017 • edited

frewsxcv commented Jul 9, 2017

dns2utf8 commented Jul 9, 2017

dns2utf8 commented Jul 9, 2017

frewsxcv left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

dns2utf8 commented Jul 9, 2017

dns2utf8 commented Jul 10, 2017

frewsxcv commented Jul 12, 2017

dns2utf8 commented Jul 12, 2017

dns2utf8 commented Jul 8, 2017 •

edited