| 2479 | |
| 2480 | |
| 2481 | long ProcessManager::init_threads() |
| 2482 | { |
| 2483 | // We create no fewer than 8 threads because some tests require |
| 2484 | // more worker threads than `sysconf(_SC_NPROCESSORS_ONLN)` on |
| 2485 | // computers with fewer cores. |
| 2486 | // e.g. https://issues.apache.org/jira/browse/MESOS-818 |
| 2487 | // |
| 2488 | // TODO(xujyan): Use a smarter algorithm to allocate threads. |
| 2489 | // Allocating a static number of threads can cause starvation if |
| 2490 | // there are more waiting Processes than the number of worker |
| 2491 | // threads. On error assumes one core. |
| 2492 | long num_worker_threads = |
| 2493 | std::max(8L, os::cpus().isSome() ? os::cpus().get() : 1); |
| 2494 | |
| 2495 | // We allow the operator to set the number of libprocess worker |
| 2496 | // threads, using an environment variable. The motivation is that |
| 2497 | // for machines with a large number of cores, setting the number of |
| 2498 | // worker threads to sysconf(_SC_NPROCESSORS_ONLN) can be very high |
| 2499 | // and affect performance negatively. Furthermore, libprocess is |
| 2500 | // widely used in Mesos and there may be a large number of Mesos |
| 2501 | // processes (e.g., executors) on the same machine. |
| 2502 | // |
| 2503 | // See https://issues.apache.org/jira/browse/MESOS-4353. |
| 2504 | constexpr char env_var[] = "LIBPROCESS_NUM_WORKER_THREADS"; |
| 2505 | Option<string> value = os::getenv(env_var); |
| 2506 | if (value.isSome()) { |
| 2507 | constexpr long maxval = 1024; |
| 2508 | Try<long> number = numify<long>(value->c_str()); |
| 2509 | if (number.isSome() && number.get() > 0L && number.get() <= maxval) { |
| 2510 | VLOG(1) << "Overriding default number of worker threads " |
| 2511 | << num_worker_threads << ", using the value " |
| 2512 | << env_var << "=" << number.get() << " instead"; |
| 2513 | num_worker_threads = number.get(); |
| 2514 | } else { |
| 2515 | LOG(WARNING) << "Ignoring invalid value " << value.get() |
| 2516 | << " for " << env_var |
| 2517 | << ", using default value " << num_worker_threads |
| 2518 | << ". Valid values are integers in the range 1 to " |
| 2519 | << maxval; |
| 2520 | } |
| 2521 | } |
| 2522 | |
| 2523 | if (runq.capacity() < (size_t) num_worker_threads) { |
| 2524 | EXIT(EXIT_FAILURE) << "Number of worker threads can not exceed " |
| 2525 | << runq.capacity() << " at this time"; |
| 2526 | } |
| 2527 | |
| 2528 | threads.reserve(num_worker_threads + 1); |
| 2529 | |
| 2530 | // Create processing threads. |
| 2531 | for (long i = 0; i < num_worker_threads; i++) { |
| 2532 | // Retain the thread handles so that we can join when shutting down. |
| 2533 | threads.emplace_back(new std::thread( |
| 2534 | [this]() { |
| 2535 | running.fetch_add(1); |
| 2536 | do { |
| 2537 | ProcessBase* process = dequeue(); |
| 2538 | if (process == nullptr) { |