<commit_before>/*
 *  Copyright (c) 2016, Facebook, Inc.
 *  All rights reserved.
 *
 *  This source code is licensed under the BSD-style license found in the
 *  LICENSE file in the root directory of this source tree. An additional grant
 *  of patent rights can be found in the PATENTS file in the same directory.
 *
 */
#include "bistro/bistro/remote/RemoteWorkers.h"

#include <thrift/lib/cpp2/protocol/DebugProtocol.h>

#include "bistro/bistro/if/gen-cpp2/common_types_custom_protocol.h"
#include "bistro/bistro/remote/RemoteWorker.h"
#include "bistro/bistro/remote/RemoteWorkerUpdate.h"
#include "bistro/bistro/remote/WorkerSetID.h"
#include "bistro/bistro/utils/Exception.h"

DEFINE_int32(
  CAUTION_startup_wait_for_workers, -1,
  "At startup, the scheduler has to wait for workers to connect, so that "
  "we do not accidentally re-start tasks that are already running elsewhere. "
  "The default of -1 computes a 'minimum safe wait' from your healthcheck "
  "and worker loss timeouts. DANGER 1: If you reduce these timeouts from one "
  "scheduler run to the next, the new default ('safe') wait may not be long "
  "enough, so either increase this when reducing timeouts, or take down "
  "all your workers. DANGER 2: Do not use an 'unsafe' value. Imagine a "
  "network partition (e.g. your discovery service is down), which makes some "
  "workers unable to connect longer than your unsafe initial wait.  Then, "
  "it's a virual certainty that the scheduler will at some point be (a) out "
  "of initial wait, (b) will not have all of the workers that have running "
  "tasks -- and therefore it will double-start some already-running tasks. "
  "NOTE 1: As a precaution, the scheduler will not exit initial wait while "
  "it has 'new' workers that have not yet replied with their running tasks. "
  "NOTE 2: Once the initial wait time passes, the scheduler becomes "
  "eligible to exit initial wait, even if the healthy workers's worker set "
  "consensus does not perfectly match our non-MUST_DIE worker set -- at the "
  "time of writing, I see no benefit to waiting longer. For more detail, "
  "see README.worker_set_consensus."
);
DEFINE_int32(
  CAUTION_worker_suicide_backoff_safety_margin_sec, 60,
  "When a worker commits suicide, it will TERM-wait-KILLs its tasks. Since "
  "task termination is not instant, the scheduler needs a safety margin to "
  "add into its 'safe initial wait' calculation, and also into the "
  "'minimum backoff for tasks of lost workers'. This margin must also "
  "indirectly account for the fact that Bistro's workers poll running tasks "
  "tasks only once every 'poll_ms' -- so keep this margin large."
);
DEFINE_int32(
  CAUTION_worker_suicide_task_kill_wait_ms, 5000,
  "The scheduler (NOT the worker -- the worker ignores this flag) includes "
  "this value with every task it starts. When, in the future, the worker "
  "commits suicide (i.e. due to a too-long network partition, or a scheduler "
  "request), it will SIGTERM the task, wait this long, and then SIGKILL. "
  "The scheduler needs this value to add into its 'safe initial wait' "
  "calculation, and into the 'minimum backoff for tasks of lost workers'. "
  "CAUTION: Therefore, if you lower this delay on a scheduler with running "
  "tasks, you will significantly increase the risk that the scheduler "
  "will start a second copy of a task before the previous one has exited."
);

namespace facebook { namespace bistro {

using namespace std;
using apache::thrift::debugString;

// A section of remote/README.worker_set_consensus documents this call.
bool RemoteWorkers::consensusPermitsBecomingHealthy(const RemoteWorker& w)
    const {
  CHECK_EQ(
    nonMustDieWorkerSetID_.hash.numWorkers, initialWorkerSetIDs_.size()
  );
  CHECK_GE(
    initialWorkerSetIDs_.size(), indirectVersionsOfNonMustDieWorkers_.size()
  );
  bool can_become_healthy =
    // Is this worker aware of every non-MUST_DIE worker?
    w.workerSetID().hasValue()
    // Future: think if this can be relaxed to test indirectWorkerSetID().
    && *w.workerSetID() == nonMustDieWorkerSetID_
    // Does every non-MUST_DIE worker have an indirect version? -- we must
    // test this, since the version is set only after a worker first echoes
    // this scheduler's WorkerSetID.  If any worker lacks it, we don't have
    // a safe consensus -- that worker might declare consensus with itself.
    && indirectVersionsOfNonMustDieWorkers_.size()
         == nonMustDieWorkerSetID_.hash.numWorkers
    // Does each non-MUST_DIE worker's WorkerSetID indirectly require this
    // worker?  "first associated" == "first WorkerSetID containing this
    // worker", since RemoteWorkers::processHeartbeat guarantees it.
    && w.firstAssociatedWorkerSetID().hasValue()
    && !WorkerSetIDEarlierThan()(
         indirectVersionsOfNonMustDieWorkers_.begin()->first,
         w.firstAssociatedWorkerSetID()->version
       );
  if (!w.hasBeenHealthy() && can_become_healthy) {
    LOG(INFO) << "Worker " << w.getBistroWorker().shard << " has not been "
      << "healthy, but WorkerSetID consensus allows it.";
  }
  return can_become_healthy;
}

namespace {
void addToVersionShardSet(
    RemoteWorkers::VersionShardSet* vss,
    const cpp2::WorkerSetID& id,
    const std::string& shard) {
  CHECK(vss->emplace(id.version, shard).second)
    << "Duplicate: v" << id.version << " in shard " << shard;
}

void removeFromVersionShardSet(
    RemoteWorkers::VersionShardSet* vss,
    const cpp2::WorkerSetID& id,
    const std::string& shard) {
  CHECK_NE(0, vss->erase({id.version, shard}))
    << "Not found: v" << id.version << " in shard " << shard;
}
}  // anonymous namespace

void RemoteWorkers::updateIndirectWorkerSetVersion(
    RemoteWorker* w,
    const cpp2::WorkerSetID& new_id) {

  auto& maybe_id = w->indirectWorkerSetID();
  CHECK(!maybe_id.hasValue() || maybe_id->schedulerID == schedulerID_);<commit_msg>Logs better explain why workers fail to achieve consensus<commit_after>/*
 *  Copyright (c) 2016, Facebook, Inc.
 *  All rights reserved.
 *
 *  This source code is licensed under the BSD-style license found in the
 *  LICENSE file in the root directory of this source tree. An additional grant
 *  of patent rights can be found in the PATENTS file in the same directory.
 *
 */
#include "bistro/bistro/remote/RemoteWorkers.h"

#include <thrift/lib/cpp2/protocol/DebugProtocol.h>

#include "bistro/bistro/if/gen-cpp2/common_types_custom_protocol.h"
#include "bistro/bistro/remote/RemoteWorker.h"
#include "bistro/bistro/remote/RemoteWorkerUpdate.h"
#include "bistro/bistro/remote/WorkerSetID.h"
#include "bistro/bistro/utils/Exception.h"

DEFINE_int32(
  CAUTION_startup_wait_for_workers, -1,
  "At startup, the scheduler has to wait for workers to connect, so that "
  "we do not accidentally re-start tasks that are already running elsewhere. "
  "The default of -1 computes a 'minimum safe wait' from your healthcheck "
  "and worker loss timeouts. DANGER 1: If you reduce these timeouts from one "
  "scheduler run to the next, the new default ('safe') wait may not be long "
  "enough, so either increase this when reducing timeouts, or take down "
  "all your workers. DANGER 2: Do not use an 'unsafe' value. Imagine a "
  "network partition (e.g. your discovery service is down), which makes some "
  "workers unable to connect longer than your unsafe initial wait.  Then, "
  "it's a virual certainty that the scheduler will at some point be (a) out "
  "of initial wait, (b) will not have all of the workers that have running "
  "tasks -- and therefore it will double-start some already-running tasks. "
  "NOTE 1: As a precaution, the scheduler will not exit initial wait while "
  "it has 'new' workers that have not yet replied with their running tasks. "
  "NOTE 2: Once the initial wait time passes, the scheduler becomes "
  "eligible to exit initial wait, even if the healthy workers's worker set "
  "consensus does not perfectly match our non-MUST_DIE worker set -- at the "
  "time of writing, I see no benefit to waiting longer. For more detail, "
  "see README.worker_set_consensus."
);
DEFINE_int32(
  CAUTION_worker_suicide_backoff_safety_margin_sec, 60,
  "When a worker commits suicide, it will TERM-wait-KILLs its tasks. Since "
  "task termination is not instant, the scheduler needs a safety margin to "
  "add into its 'safe initial wait' calculation, and also into the "
  "'minimum backoff for tasks of lost workers'. This margin must also "
  "indirectly account for the fact that Bistro's workers poll running tasks "
  "tasks only once every 'poll_ms' -- so keep this margin large."
);
DEFINE_int32(
  CAUTION_worker_suicide_task_kill_wait_ms, 5000,
  "The scheduler (NOT the worker -- the worker ignores this flag) includes "
  "this value with every task it starts. When, in the future, the worker "
  "commits suicide (i.e. due to a too-long network partition, or a scheduler "
  "request), it will SIGTERM the task, wait this long, and then SIGKILL. "
  "The scheduler needs this value to add into its 'safe initial wait' "
  "calculation, and into the 'minimum backoff for tasks of lost workers'. "
  "CAUTION: Therefore, if you lower this delay on a scheduler with running "
  "tasks, you will significantly increase the risk that the scheduler "
  "will start a second copy of a task before the previous one has exited."
);

namespace facebook { namespace bistro {

using namespace std;
using apache::thrift::debugString;

namespace {
template <typename... Args>
bool consensusFail(const RemoteWorker& w, Args&&... args) {
  if (!w.hasBeenHealthy()) {
    LOG(WARNING) << "Worker " << w.getBistroWorker().shard
      << " is not yet able to achieve consensus: "
      << folly::to<std::string>(std::forward<Args>(args)...);
  }
  return false;
}
}  // anonymous namespace

// A section of remote/README.worker_set_consensus documents this call.
bool RemoteWorkers::consensusPermitsBecomingHealthy(const RemoteWorker& w)
    const {
  CHECK_EQ(
    nonMustDieWorkerSetID_.hash.numWorkers, initialWorkerSetIDs_.size()
  );
  CHECK_GE(
    initialWorkerSetIDs_.size(), indirectVersionsOfNonMustDieWorkers_.size()
  );

  // Is this worker aware of every non-MUST_DIE worker?
  if (!w.workerSetID().hasValue()) {
    return consensusFail(w, "It has no WorkerSetID");
  }
  // Future: think if this can be relaxed to test indirectWorkerSetID().
  if (*w.workerSetID() != nonMustDieWorkerSetID_) {
    return consensusFail(
      w, "It has WorkerSetID ", debugString(*w.workerSetID()), " while "
      "the scheduler has ", debugString(nonMustDieWorkerSetID_)
    );
  }

  // Does every non-MUST_DIE worker have an indirect version? -- we must
  // test this, since the version is set only after a worker first echoes
  // this scheduler's WorkerSetID.  If any worker lacks it, we don't have
  // a safe consensus -- that worker might declare consensus with itself.
  if (indirectVersionsOfNonMustDieWorkers_.size()
      != nonMustDieWorkerSetID_.hash.numWorkers) {
    return consensusFail(
      w, "Scheduler has ", indirectVersionsOfNonMustDieWorkers_.size(),
      " non-MUST_DIE workers with indirect versions versus ",
      nonMustDieWorkerSetID_.hash.numWorkers, " non-MUST_DIE workers"
    );
  }

  // Does each non-MUST_DIE worker's WorkerSetID indirectly require this
  // worker?  "first associated" == "first WorkerSetID containing this
  // worker", since RemoteWorkers::processHeartbeat guarantees it.
  if (!w.firstAssociatedWorkerSetID().hasValue()) {
    return consensusFail(
      w, "It has not yet echoed any WorkerSetID from the scheduler"
    );
  }
  if (WorkerSetIDEarlierThan()(
    indirectVersionsOfNonMustDieWorkers_.begin()->first,
    w.firstAssociatedWorkerSetID()->version
  )) {
    return consensusFail(
      w, "It first appeared in the following WorkerSetID: ",
      w.firstAssociatedWorkerSetID()->version, " but an earlier one "
      "is indirectly required by all non-MUST_DIE workers: ",
      indirectVersionsOfNonMustDieWorkers_.begin()->first
    );
  }

  if (!w.hasBeenHealthy()) {
    LOG(INFO) << "Worker " << w.getBistroWorker().shard << " has not been "
      << "healthy, but WorkerSetID consensus allows it.";
  }
  return true;
}

namespace {
void addToVersionShardSet(
    RemoteWorkers::VersionShardSet* vss,
    const cpp2::WorkerSetID& id,
    const std::string& shard) {
  CHECK(vss->emplace(id.version, shard).second)
    << "Duplicate: v" << id.version << " in shard " << shard;
}

void removeFromVersionShardSet(
    RemoteWorkers::VersionShardSet* vss,
    const cpp2::WorkerSetID& id,
    const std::string& shard) {
  CHECK_NE(0, vss->erase({id.version, shard}))
    << "Not found: v" << id.version << " in shard " << shard;
}
}  // anonymous namespace

void RemoteWorkers::updateIndirectWorkerSetVersion(
    RemoteWorker* w,
    const cpp2::WorkerSetID& new_id) {

  auto& maybe_id = w->indirectWorkerSetID();
  CHECK(!maybe_id.hasValue() || maybe_id->schedulerID == schedulerID_);
{'fraction_non_alphanumeric': 0.0958714538325517, 'fraction_numerical': 0.0043053740293688015, 'mean_word_length': 3.97628156082632, 'pattern_counts': {'":': 0, '<': 64, '<?xml version=': 0, '>': 22, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 2, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '2990226', 'n_tokens_mistral': 4005, 'n_tokens_neox': 3916, 'n_words': 1417}