An unhandled race condition exists in `ThreadPool`

[framlog]

As the title mentioned, I suspect that an unhandled race condition exists here:

arrow/cpp/src/arrow/util/thread_pool.cc

Lines 238 to 257 in f48cabe

	if (pid_ != current_pid) {
	// Reinitialize internal state in child process after fork()
	// Ideally we would use pthread_at_fork(), but that doesn't allow
	// storing an argument, hence we'd need to maintain a list of all
	// existing ThreadPools.
	int capacity = state_->desired_capacity_;
	auto new_state = std::make_shared<ThreadPool::State>();
	new_state->please_shutdown_ = state_->please_shutdown_;
	new_state->quick_shutdown_ = state_->quick_shutdown_;
	pid_ = current_pid;
	sp_state_ = new_state;
	state_ = sp_state_.get();
	// Launch worker threads anew
	if (!state_->please_shutdown_) {
	ARROW_UNUSED(SetCapacity(capacity));
	}
	}

Assume that we have two threads, one executes SpawnReal to push the task into the pending_tasks, the other is running at the ProtectAgainstFork, which may reset the whole state which the former one is using, which makes the loss of tasks.

In my scenario, when I fork multiple processes to read data after reading something in the parent process, it has chances to stall at Future::Wait. (However, I'm not very confident that the race condition is the root cause of my scenario since I only reproduce it in python code, but fail to reproduce it in cpp.)

A quick fix could be change the type of pid_ to std::atomic<pid_t> and use compare_exchange_strong to protect the critical section.

[westonpace]

It's been a little bit since I've looked at this but if I recall correctly the condition pid_ != current_pid will only be true after a fork. When a fork happens the child (the one that gets the new pid) will not have any running threads except the thread that called fork.

We would then go into this branch and reinitialize the state before any worker thread could possibly be launched. I suppose however, if there were two user threads (e.g. created by the arrow library user and not worker threads in the thread pool) that simultaneously trigger calls to SpawnReal then we could get into a race condition.

Our options may be somewhat limited here due to the nature of fork. I'm not entirely sure how compare_exchange_strong can be used here. What would the values of expected and desired be? Since it is a negative check we would need "expected" to be "anything but current_pid" and desired to be "current_pid" but I don't think we can use compare_exchange_strong in that negative way.

[framlog]

You are right.

The fix that I give is to use 0 as an intermediate state(we set pid_ to 0 when we try to reset everything and then set pid_ to the current_pid_). Other threads entering this block checks old_pid != 0 && pid_.compare_exchange_strong(old_pid, 0) to figure out whether they need to spin or not. (old_pid is the value read before we do the check). At least it passes my local tests.

[westonpace]

I see now. You have a lock that is only taken if the pid doesn't match up. That should mean it's highly likely to be clean after a fork.

It's not guaranteed though. You open yourself up to a potential deadlock if there is a fork while you are in the middle of resetting the state. In that case the child would come online with pid_ == 0 and it would be impossible to reset the thread pool.

However, that seems like too unlikely a scenario to worry about (e.g. you'd have to be forking on a forked child almost immediately after the initial fork call).

Do you want to open a JIRA and create a PR?

[framlog]

OK. A JIRA is created. The pr will be submitted in a few days.

[pitrou]

When a fork happens the child (the one that gets the new pid) will not have any running threads except the thread that called fork.

This is right. fork() only preserves the thread which called fork().

We would then go into this branch and reinitialize the state before any worker thread could possibly be launched. I suppose however, if there were two user threads (e.g. created by the arrow library user and not worker threads in the thread pool) that simultaneously trigger calls to SpawnReal then we could get into a race condition.

You mean user threads created in the forked child? Yes, that's basically possible. Probably not very common, but possible :-)