Function run

src/examples/balloon_executor.cpp:87–153

This function will increase the memory footprint gradually. `TaskInfo.data` specifies the upper limit (in MB) of the memory footprint. The upper limit can be larger than the amount of memory allocated to this executor. In that case, the isolator (e.g. cgroups) may detect that and destroy the container before the executor can sent a TASK_FINISHED update.

// may detect that and destroy the container before the executor can
// sent a TASK_FINISHED update.
void run(ExecutorDriver* driver, const TaskInfo& task)
{
  TaskStatus status;
  status.mutable_task_id()->MergeFrom(task.task_id());
  status.set_state(TASK_RUNNING);

  driver->sendStatusUpdate(status);

  // Get the balloon limit (in MB).
  Try<Bytes> _limit = Bytes::parse(task.data());
  CHECK(_limit.isSome());
  const size_t limit = _limit->bytes() / Bytes::MEGABYTES;

  // On systems with swap partitions we explicitly prevent memory used by the
  // executor from being swapped out with `mlock`. Since the amount of memory a
  // process can lock is controlled by an rlimit, we only `mlock` when
  // strictly necessary to prevent complicating the test setup.
  Try<bool> hasSwap_ = hasSwap();
  const bool lockMemory = hasSwap_.isError() || hasSwap_.get();

  if (lockMemory) {
    LOG(INFO)
      << "System might use swap partitions, will explicitly"
      << " lock memory to prevent swapping";
  }

  const size_t chunk = BALLOON_STEP_MB * 1024 * 1024;
  for (size_t i = 0; i < limit / BALLOON_STEP_MB; i++) {
    LOG(INFO)
      << "Increasing memory footprint by " << BALLOON_STEP_MB << " MB";

    // Allocate page-aligned virtual memory.
    void* buffer = nullptr;
    if (posix_memalign(&buffer, os::pagesize(), chunk) != 0) {
      LOG(FATAL) << ErrnoError(
          "Failed to allocate page-aligned memory, posix_memalign").message;
    }

    // We use `memset` and possibly `mlock` here to make sure that the
    // memory actually gets paged in and thus accounted for.
    if (memset(buffer, 1, chunk) != buffer) {
      LOG(FATAL) << ErrnoError("Failed to fill memory, memset").message;
    }

    if (lockMemory) {
      if (mlock(buffer, chunk) != 0) {
        LOG(FATAL) << ErrnoError("Failed to lock memory, mlock").message;
      }
    }

    // Try not to increase the memory footprint too fast.
    os::sleep(Seconds(1));
  }

  LOG(INFO) << "Finishing task " << task.task_id().value();

  status.set_state(TASK_FINISHED);