You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@mesos.apache.org by ya...@apache.org on 2017/07/18 23:07:43 UTC
mesos git commit: Perform agent GC asynchronously.
Repository: mesos
Updated Branches:
refs/heads/master d5b85d2b9 -> 91b3abcfa
Perform agent GC asynchronously.
Previously, rmdir operations were serialized
through the garbage collector. Expensive removal
events had the potential to delay task launches.
Review: https://reviews.apache.org/r/53479/
Project: http://git-wip-us.apache.org/repos/asf/mesos/repo
Commit: http://git-wip-us.apache.org/repos/asf/mesos/commit/91b3abcf
Tree: http://git-wip-us.apache.org/repos/asf/mesos/tree/91b3abcf
Diff: http://git-wip-us.apache.org/repos/asf/mesos/diff/91b3abcf
Branch: refs/heads/master
Commit: 91b3abcfa98de3c2b6231eec7ab0f3df3c53dcd8
Parents: d5b85d2
Author: Jacob Janco <jj...@gmail.com>
Authored: Tue Jul 18 11:23:48 2017 -0700
Committer: Jiang Yan Xu <xu...@apple.com>
Committed: Tue Jul 18 15:50:25 2017 -0700
----------------------------------------------------------------------
src/slave/gc.cpp | 116 +++++++++++++++++++++++++++++++++++---------------
src/slave/gc.hpp | 28 ++++++++----
2 files changed, 101 insertions(+), 43 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/mesos/blob/91b3abcf/src/slave/gc.cpp
----------------------------------------------------------------------
diff --git a/src/slave/gc.cpp b/src/slave/gc.cpp
index 961f547..f270fa4 100644
--- a/src/slave/gc.cpp
+++ b/src/slave/gc.cpp
@@ -16,10 +16,14 @@
#include <list>
+#include <process/async.hpp>
+#include <process/check.hpp>
+#include <process/defer.hpp>
#include <process/delay.hpp>
#include <process/dispatch.hpp>
#include <stout/foreach.hpp>
+#include <stout/lambda.hpp>
#include <stout/os/rmdir.hpp>
@@ -42,8 +46,8 @@ namespace slave {
GarbageCollectorProcess::~GarbageCollectorProcess()
{
- foreachvalue (const PathInfo& info, paths) {
- info.promise->discard();
+ foreachvalue (const Owned<PathInfo>& info, paths) {
+ info->promise.discard();
}
}
@@ -57,15 +61,21 @@ Future<Nothing> GarbageCollectorProcess::schedule(
// If there's an existing schedule for this path, we must remove
// it here in order to reschedule.
if (timeouts.contains(path)) {
- CHECK(unschedule(path));
+ return unschedule(path)
+ .then(defer(
+ self(),
+ &Self::schedule,
+ d,
+ path));
}
- Owned<Promise<Nothing>> promise(new Promise<Nothing>());
-
Timeout removalTime = Timeout::in(d);
timeouts[path] = removalTime;
- paths.put(removalTime, PathInfo(path, promise));
+
+ Owned<PathInfo> info(new PathInfo(path));
+
+ paths.put(removalTime, info);
// If the timer is not yet initialized or the timeout is sooner than
// the currently active timer, update it.
@@ -74,11 +84,11 @@ Future<Nothing> GarbageCollectorProcess::schedule(
reset(); // Schedule the timer for next event.
}
- return promise->future();
+ return info->promise.future();
}
-bool GarbageCollectorProcess::unschedule(const string& path)
+Future<bool> GarbageCollectorProcess::unschedule(const string& path)
{
LOG(INFO) << "Unscheduling '" << path << "' from gc";
@@ -90,14 +100,23 @@ bool GarbageCollectorProcess::unschedule(const string& path)
CHECK(paths.contains(timeout));
// Locate the path.
- foreach (const PathInfo& info, paths.get(timeout)) {
- if (info.path == path) {
+ foreach (const Owned<PathInfo>& info, paths.get(timeout)) {
+ if (info->path == path) {
+ // If the path is currently undergoing removal, we cannot
+ // prevent path removal and wait for removal completion.
+ if (info->removing) {
+ // Return false to be consistent with the behavior when
+ // `unschedule` is called after the path is removed.
+ return info->promise.future()
+ .then([]() { return false; });
+ }
+
// Discard the promise.
- info.promise->discard();
+ info->promise.discard();
// Clean up the maps.
CHECK(paths.remove(timeout, info));
- CHECK(timeouts.erase(path) > 0);
+ CHECK_EQ(timeouts.erase(info->path), 1u);
return true;
}
@@ -125,41 +144,70 @@ void GarbageCollectorProcess::reset()
void GarbageCollectorProcess::remove(const Timeout& removalTime)
{
- // TODO(bmahler): Other dispatches can block waiting for a removal
- // operation. To fix this, the removal operation can be done
- // asynchronously in another thread.
if (paths.count(removalTime) > 0) {
- foreach (const PathInfo& info, paths.get(removalTime)) {
- LOG(INFO) << "Deleting " << info.path;
-
- // Run rmdir with 'continueOnError = true'. It's possible for
- // tasks and isolators to lay down files that are not deletable by
- // GC. In the face of such errors GC needs to free up disk space
- // wherever it can because it's already re-offered to frameworks.
- Try<Nothing> rmdir = os::rmdir(info.path, true, true, true);
-
- if (rmdir.isError()) {
- LOG(WARNING) << "Failed to delete '" << info.path << "': "
- << rmdir.error();
- info.promise->fail(rmdir.error());
- } else {
- LOG(INFO) << "Deleted '" << info.path << "'";
- info.promise->set(rmdir.get());
+ list<Owned<PathInfo>> infos;
+
+ foreach (const Owned<PathInfo> info, paths.get(removalTime)) {
+ if (info->removing) {
+ VLOG(1) << "Skipping deletion of '" << info-> path
+ << "' as it is already in progress";
+ continue;
}
- timeouts.erase(info.path);
+ infos.push_back(info);
+
+ // Set `removing` to signify that the path is being cleaned up.
+ info->removing = true;
}
- paths.remove(removalTime);
+ auto rmdirs = [infos]() {
+ foreach (const Owned<PathInfo>& info, infos) {
+ // Run the removal operation with 'continueOnError = true'.
+ // It's possible for tasks and isolators to lay down files
+ // that are not deletable by GC. In the face of such errors
+ // GC needs to free up disk space wherever it can because the
+ // disk space has already been re-offered to frameworks.
+ LOG(INFO) << "Deleting " << info->path;
+ Try<Nothing> rmdir = os::rmdir(info->path, true, true, true);
+
+ if (rmdir.isError()) {
+ LOG(WARNING) << "Failed to delete '" << info->path << "': "
+ << rmdir.error();
+ info->promise.fail(rmdir.error());
+ } else {
+ LOG(INFO) << "Deleted '" << info->path << "'";
+ info->promise.set(rmdir.get());
+ }
+ }
+
+ return Nothing();
+ };
+
+ async(rmdirs)
+ .onAny(defer(self(), &Self::_remove, lambda::_1, infos));
} else {
// This occurs when either:
// 1. The path(s) has already been removed (e.g. by prune()).
// 2. All paths under the removal time were unscheduled.
LOG(INFO) << "Ignoring gc event at " << removalTime.remaining()
<< " as the paths were already removed, or were unscheduled";
+ reset();
+ }
+}
+
+
+void GarbageCollectorProcess::_remove(const Future<Nothing>& result,
+ const list<Owned<PathInfo>> infos)
+{
+ CHECK_READY(result);
+
+ // Remove path records from `paths` and `timeouts` data structures.
+ foreach (const Owned<PathInfo>& info, infos) {
+ CHECK(paths.remove(timeouts[info->path], info));
+ CHECK_EQ(timeouts.erase(info->path), 1u);
}
- reset(); // Schedule the timer for next event.
+ reset();
}
http://git-wip-us.apache.org/repos/asf/mesos/blob/91b3abcf/src/slave/gc.hpp
----------------------------------------------------------------------
diff --git a/src/slave/gc.hpp b/src/slave/gc.hpp
index 5ea8245..b8d5301 100644
--- a/src/slave/gc.hpp
+++ b/src/slave/gc.hpp
@@ -17,6 +17,7 @@
#ifndef __SLAVE_GC_HPP__
#define __SLAVE_GC_HPP__
+#include <list>
#include <string>
#include <vector>
@@ -70,8 +71,9 @@ public:
// Unschedules the specified path for removal.
// The future will be true if the path has been unscheduled.
// The future will be false if the path is not scheduled for
- // removal, or the path has already being removed.
- // Note that you currently cannot discard a returned future.
+ // removal, or the path has already being removed. If path
+ // removal is in progress, the future will be false when the
+ // removal finishes or will fail on unsuccessful removal.
virtual process::Future<bool> unschedule(const std::string& path);
// Deletes all the directories, whose scheduled garbage collection time
@@ -96,7 +98,7 @@ public:
const Duration& d,
const std::string& path);
- bool unschedule(const std::string& path);
+ process::Future<bool> unschedule(const std::string& path);
void prune(const Duration& d);
@@ -107,23 +109,31 @@ private:
struct PathInfo
{
- PathInfo(const std::string& _path,
- process::Owned<process::Promise<Nothing>> _promise)
- : path(_path), promise(_promise) {}
+ PathInfo(const std::string& _path)
+ : path(_path) {}
bool operator==(const PathInfo& that) const
{
- return path == that.path && promise == that.promise;
+ return path == that.path;
}
const std::string path;
- const process::Owned<process::Promise<Nothing>> promise;
+
+ // This promise tracks the scheduled gc for the path.
+ process::Promise<Nothing> promise;
+
+ bool removing = false;
};
+ // Callback for `remove` for bookkeeping after path removal.
+ void _remove(
+ const process::Future<Nothing>& result,
+ const std::list<process::Owned<PathInfo>> infos);
+
// Store all the timeouts and corresponding paths to delete.
// NOTE: We are using Multimap here instead of Multihashmap, because
// we need the keys of the map (deletion time) to be sorted.
- Multimap<process::Timeout, PathInfo> paths;
+ Multimap<process::Timeout, process::Owned<PathInfo>> paths;
// We also need efficient lookup for a path, to determine whether
// it exists in our paths mapping.