bool wait_on_child_and_print_progress (process &child_proc) {
  const size_t BUF_SIZE = 4096;
  char buf[BUF_SIZE];
  ssize_t bytes_read;
  bool success = true;
  while( (bytes_read = child_proc.read_from_child(buf, BUF_SIZE)) > 0 ) {
    logprogress_stream << std::string(buf, buf + bytes_read);
    if (cppipc::must_cancel()) {
      logprogress_stream << "Cancel by user" << std::endl;
      child_proc.kill(false);
      success = false;
      break;
    }
  }
  logprogress_stream << std::endl;

  return success;
}

void Mesos::stop()
{
  if (process != nullptr) {
    terminate(process);
    wait(process);

    delete process;
    process = nullptr;
  }
}

bool processSearch(process &currProcess)
{
    bool result = false;;
    if((currProcess.get_pname() != "") && (currProcess.get_pid() == 0)) //pname set, pid not set, search with pname
    {
#ifdef DEBUG
	print_string("Using Name Search");
#endif
	int pid = 0; //will be set in currprocess, can't be done by this function
	result = processSearch(currProcess.get_pname(), &pid);
	currProcess.set_pid(pid);
    } else {
#ifdef DEBUG
	print_string("Using PID Search");
#endif
	result = processSearch(currProcess.get_pid());
    }
    return result;
}

Future<Nothing> HealthCheckerProcess::__httpHealthCheck(
    const tuple<
        Future<Option<int>>,
        Future<string>,
        Future<string>>& t)
{
  Future<Option<int>> status = std::get<0>(t);
  if (!status.isReady()) {
    return Failure(
        "Failed to get the exit status of the " + string(HTTP_CHECK_COMMAND) +
        " process: " + (status.isFailed() ? status.failure() : "discarded"));
  }

  if (status->isNone()) {
    return Failure(
        "Failed to reap the " + string(HTTP_CHECK_COMMAND) + " process");
  }

  int statusCode = status->get();
  if (statusCode != 0) {
    Future<string> error = std::get<2>(t);
    if (!error.isReady()) {
      return Failure(
          string(HTTP_CHECK_COMMAND) + " returned " +
          WSTRINGIFY(statusCode) + "; reading stderr failed: " +
          (error.isFailed() ? error.failure() : "discarded"));
    }

    return Failure(
        string(HTTP_CHECK_COMMAND) + " returned " +
        WSTRINGIFY(statusCode) + ": " + error.get());
  }

  Future<string> output = std::get<1>(t);
  if (!output.isReady()) {
    return Failure(
        "Failed to read stdout from " + string(HTTP_CHECK_COMMAND) + ": " +
        (output.isFailed() ? output.failure() : "discarded"));
  }

  // Parse the output and get the HTTP response code.
  Try<int> code = numify<int>(output.get());
  if (code.isError()) {
    return Failure(
        "Unexpected output from " + string(HTTP_CHECK_COMMAND) + ": " +
        output.get());
  }

  if (code.get() < process::http::Status::OK ||
      code.get() >= process::http::Status::BAD_REQUEST) {
    return Failure(
        "Unexpected HTTP response code: " +
        process::http::Status::string(code.get()));
  }

  return Nothing();
}

TEST_F(SubprocessTest, Default)
{
  Try<Subprocess> s = subprocess("echo hello world");

  ASSERT_SOME(s);

  // Advance time until the internal reaper reaps the subprocess.
  Clock::pause();
  while (s.get().status().isPending()) {
    Clock::advance(MAX_REAP_INTERVAL());
    Clock::settle();
  }
  Clock::resume();

  AWAIT_ASSERT_READY(s.get().status());
  ASSERT_SOME(s.get().status().get());

  int status = s.get().status().get().get();
  EXPECT_TRUE(WIFEXITED(status));
  EXPECT_EQ(0, WEXITSTATUS(status));
}

// Print individual statistics for processes:
void individual_stats(process &curr_process)
{
  cout << "Process " << curr_process.get_pid() << endl;
  cout << "Burst Time: " << curr_process.get_burst() << " ms" << endl;
  cout << "Priority: " << curr_process.get_priority() << endl;
  cout << "Arrival Time: " << curr_process.get_arrival() << " ms" << endl;
  cout << "Initial Wait Time: " << curr_process.get_initial_wait() << " ms" << endl;
  cout << "Total Wait Time: " << curr_process.get_total_wait_time() << " ms" << endl;
  cout << "Status: " << curr_process.get_status() << endl;
  cout << "=============================================" << endl;
}

        void putProcess(process p){
            if(!readyQueue.empty()){
                for(i=0;i<readyQueue.size();i++){
                    if(readyQueue[i].getRem()>p.getRem())
                        break;
                }
                readyQueue.insert(readyQueue.begin()+i,p);
            }
            else{
                readyQueue.push_back(p);
            }

        }

// Wait for a subprocess and test the status code for the following
// conditions of 'expected_status':
//   1. 'None' = Anything but '0'.
//   2. 'Some' = the value of 'expected_status'.
// Returns Nothing if the resulting status code matches the
// expectation otherwise a Failure with the output of the subprocess.
// TODO(jmlvanre): Turn this into a generally useful abstraction for
// gtest where we can have a more straigtforward 'expected_status'.
Future<Nothing> await_subprocess(
    const Subprocess& subprocess,
    const Option<int>& expected_status = None())
{
  // Dup the pipe fd of the subprocess so we can read the output if
  // needed.
  Try<int_fd> dup = os::dup(subprocess.out().get());
  if (dup.isError()) {
    return Failure(dup.error());
  }

  int_fd out = dup.get();

  // Once we get the status of the process.
  return subprocess.status()
    .then([=](const Option<int>& status) -> Future<Nothing> {
      // If the status is not set, fail out.
      if (status.isNone()) {
        return Failure("Subprocess status is none");
      }

      // If the status is not what we expect then fail out with the
      // output of the subprocess. The failure message will include
      // the assertion failures of the subprocess.
      if ((expected_status.isSome() && status.get() != expected_status.get()) ||
          (expected_status.isNone() && status.get() == 0)) {
        return io::read(out)
          .then([](const string& output) -> Future<Nothing> {
            return Failure("\n[++++++++++] Subprocess output.\n" + output +
                           "[++++++++++]\n");
          });
      }

      // If the subprocess ran successfully then return nothing.
      return Nothing();
    }).onAny([=]() {
      os::close(out);
    });
}

void stdin_filler::fill(process& p) {
    if (fd == -1) return; //don't need to fill anything...
    char buf[512];
    int numbytes;
    do {
        numbytes = read(fd, buf, 512);
        if (numbytes == -1) {
            perror("read");
            std::exit(1);
        }
        if (!writeall(p.in(), buf, numbytes)) {
            //it is easy to imagine this write failing - so if that happens
            //just kill the process, as it's not useful to us...
            if (settings.verbose) {
                std::cerr << "write failed to subprocess, killing...\n";
            }
            p.term();
            break;
        }
    } while (numbytes != 0);
    lseek(fd, 0, SEEK_SET); //return to beginning of file
}

// Like the 'remote' test but uses http::connect.
TEST(ProcessTest, Http1)
{
  ASSERT_TRUE(GTEST_IS_THREADSAFE);

  RemoteProcess process;
  spawn(process);

  http::URL url = http::URL(
      "http",
      process.self().address.ip,
      process.self().address.port,
      process.self().id + "/handler");

  Future<http::Connection> connect = http::connect(url);
  AWAIT_READY(connect);

  http::Connection connection = connect.get();

  Future<UPID> pid;
  Future<string> body;
  EXPECT_CALL(process, handler(_, _))
    .WillOnce(DoAll(FutureArg<0>(&pid),
                    FutureArg<1>(&body)));

  http::Request request;
  request.method = "POST";
  request.url = url;
  request.headers["User-Agent"] = "libprocess/";
  request.body = "hello world";

  // Send the libprocess request. Note that we will not
  // receive a 202 due to the use of the `User-Agent`
  // header, therefore we need to explicitly disconnect!
  Future<http::Response> response = connection.send(request);

  AWAIT_READY(body);
  ASSERT_EQ("hello world", body.get());

  AWAIT_READY(pid);
  ASSERT_EQ(UPID(), pid.get());

  EXPECT_TRUE(response.isPending());

  AWAIT_READY(connection.disconnect());

  terminate(process);
  wait(process);
}

// The net_cls handles aren't treated as resources. Further, they have fixed
// values and hence don't have a notion of usage. We are therefore returning an
// empty 'ResourceStatistics' object.
Future<ResourceStatistics> CgroupsNetClsIsolatorProcess::usage(
    const ContainerID& containerId)
{
  if (!infos.contains(containerId)) {
    return Failure("Unknown container");
  }

  return ResourceStatistics();
}

Future<bool> RegistrarProcess::apply(Owned<Operation> operation)
{
  if (recovered.isNone()) {
    return Failure("Attempted to apply the operation before recovering");
  }

  return recovered.get()->future()
    .then(defer(self(), &Self::_apply, operation));
}

void HealthCheckerProcess::reschedule()
{
  VLOG(1) << "Rescheduling health check in "
          << Seconds(static_cast<int64_t>(check.interval_seconds()));

  delay(Seconds(static_cast<int64_t>(check.interval_seconds())),
        self(),
        &Self::_healthCheck);
}

Future<Nothing> NetworkCniIsolatorProcess::_cleanup(
    const ContainerID& containerId,
    const list<Future<Nothing>>& detaches)
{
  CHECK(infos.contains(containerId));

  vector<string> messages;
  foreach (const Future<Nothing>& detach, detaches) {
    if (!detach.isReady()) {
      messages.push_back(
          detach.isFailed() ? detach.failure() : "discarded");
    }
  }

  if (!messages.empty()) {
    return Failure(strings::join("\n", messages));
  }

  const string containerDir =
      paths::getContainerDir(rootDir.get(), containerId.value());

  const string target =
      paths::getNamespacePath(rootDir.get(), containerId.value());

  if (os::exists(target)) {
    Try<Nothing> unmount = fs::unmount(target);
    if (unmount.isError()) {
      return Failure(
          "Failed to unmount the network namespace handle '" +
          target + "': " + unmount.error());
    }
  }

  Try<Nothing> rmdir = os::rmdir(containerDir);
  if (rmdir.isError()) {
    return Failure(
        "Failed to remove the container directory '" +
        containerDir + "': " + rmdir.error());
  }

  infos.erase(containerId);

  return Nothing();
}

Future<Nothing> Fetcher::fetch(
    const URI& uri,
    const string& directory) const
{
  if (!plugins.contains(uri.scheme())) {
    return Failure("Scheme '" + uri.scheme() + "' is not supported");
  }

  return plugins.at(uri.scheme())->fetch(uri, directory);
}

TEST(ProcessTest, Defer3)
{
  ASSERT_TRUE(GTEST_IS_THREADSAFE);

  std::atomic_bool bool1(false);
  std::atomic_bool bool2(false);

  Deferred<void(bool)> set1 =
    defer([&bool1](bool b) { bool1.store(b); });

  set1(true);

  Deferred<void(bool)> set2 =
    defer([&bool2](bool b) { bool2.store(b); });

  set2(true);

  while (bool1.load() == false);
  while (bool2.load() == false);
}

Future<HttpResponse> SlavesManager::deactivate(const HttpRequest& request)
{
  // Parse the query to get out the slave hostname and port.
  string hostname = "";
  uint16_t port = 0;

  map<string, vector<string> > pairs =
    strings::pairs(request.query, ",", "=");

  // Make sure there is at least a 'hostname=' and 'port='.
  if (pairs.count("hostname") == 0) {
    LOG(WARNING) << "Slaves manager expecting 'hostname' in query string"
                 << " when trying to deactivate a slave";
    return HttpNotFoundResponse();
  } else if (pairs.count("port") == 0) {
    LOG(WARNING) << "Slaves manager expecting 'port' in query string"
                 << " when trying to deactivate a slave";
    return HttpNotFoundResponse();
  }

  hostname = pairs["hostname"].front();

  // Check that 'port' is valid.
  try {
    port = lexical_cast<uint16_t>(pairs["port"].front());
  } catch (const bad_lexical_cast&) {
    LOG(WARNING) << "Slaves manager failed to parse 'port = "
		 << pairs["port"].front()
                 << "'  when trying to deactivate a slave";
    return HttpNotFoundResponse();
  }

  LOG(INFO) << "Slaves manager received HTTP request to deactivate slave at "
	    << hostname << ":" << port;

  if (deactivate(hostname, port)) {
    return HttpOKResponse();
  } else {
    return HttpInternalServerErrorResponse();
  }
}

TEST(LoopTest, DiscardIterate)
{
  Promise<int> promise;

  promise.future().onDiscard([&]() { promise.discard(); });

  Future<Nothing> future = loop(
      [&]() {
        return promise.future();
      },
      [&](int i) -> ControlFlow<Nothing> {
        return Break();
      });

  EXPECT_TRUE(future.isPending());

  future.discard();

  AWAIT_DISCARDED(future);
  EXPECT_TRUE(promise.future().hasDiscard());
}

TEST(LoopTest, Sync)
{
  std::atomic_int value = ATOMIC_VAR_INIT(1);

  Future<Nothing> future = loop(
      [&]() {
        return value.load();
      },
      [](int i) -> ControlFlow<Nothing> {
        if (i != 0) {
          return Continue();
        }
        return Break();
      });

  EXPECT_TRUE(future.isPending());

  value.store(0);

  AWAIT_READY(future);
}

v1::scheduler::Event evolve(const StatusUpdateMessage& message)
{
  v1::scheduler::Event event;
  event.set_type(v1::scheduler::Event::UPDATE);

  v1::scheduler::Event::Update* update = event.mutable_update();

  update->mutable_status()->CopyFrom(evolve(message.update().status()));

  if (message.update().has_slave_id()) {
    update->mutable_status()->mutable_agent_id()->CopyFrom(
        evolve(message.update().slave_id()));
  }

  if (message.update().has_executor_id()) {
    update->mutable_status()->mutable_executor_id()->CopyFrom(
        evolve(message.update().executor_id()));
  }

  update->mutable_status()->set_timestamp(message.update().timestamp());

  // If the update does not have a 'uuid', it does not need
  // acknowledging. However, prior to 0.23.0, the update uuid
  // was required and always set. In 0.24.0, we can rely on the
  // update uuid check here, until then we must still check for
  // this being sent from the driver (from == UPID()) or from
  // the master (pid == UPID()).
  // TODO(vinod): Get rid of this logic in 0.25.0 because master
  // and slave correctly set task status in 0.24.0.
  if (!message.update().has_uuid() || message.update().uuid() == "") {
    update->mutable_status()->clear_uuid();
  } else if (UPID(message.pid()) == UPID()) {
    update->mutable_status()->clear_uuid();
  } else {
    update->mutable_status()->set_uuid(message.update().uuid());
  }

  return event;
}