| 1219 | |
| 1220 | |
| 1221 | void StorageLocalResourceProviderProcess::watchProfiles() |
| 1222 | { |
| 1223 | auto err = [](const string& message) { |
| 1224 | LOG(ERROR) << "Failed to watch for DiskProfileAdaptor: " << message; |
| 1225 | }; |
| 1226 | |
| 1227 | // TODO(chhsiao): Consider retrying with backoff. |
| 1228 | loop( |
| 1229 | self(), |
| 1230 | [=] { |
| 1231 | return diskProfileAdaptor->watch(profileInfos.keys(), info); |
| 1232 | }, |
| 1233 | [=](const hashset<string>& profiles) { |
| 1234 | CHECK(info.has_id()); |
| 1235 | |
| 1236 | LOG(INFO) |
| 1237 | << "Updating profiles " << stringify(profiles) |
| 1238 | << " for resource provider " << info.id(); |
| 1239 | |
| 1240 | std::function<Future<Nothing>()> update = defer(self(), [=] { |
| 1241 | return updateProfiles(profiles) |
| 1242 | .then(defer(self(), &Self::reconcileResources, false)); |
| 1243 | }); |
| 1244 | |
| 1245 | // Update the profile mapping and storage pools in `sequence` to wait |
| 1246 | // for any pending operation that disallow reconciliation or the last |
| 1247 | // reconciliation (if any) to finish, and set up `reconciled` to drop |
| 1248 | // incoming operations that disallow reconciliation until the storage |
| 1249 | // pools are reconciled. |
| 1250 | reconciled = sequence.add(update); |
| 1251 | |
| 1252 | return reconciled |
| 1253 | .then(defer(self(), [=]() -> ControlFlow<Nothing> { |
| 1254 | return Continue(); |
| 1255 | })); |
| 1256 | }) |
| 1257 | .onFailed(std::bind(err, lambda::_1)) |
| 1258 | .onDiscarded(std::bind(err, "future discarded")); |
| 1259 | } |
| 1260 | |
| 1261 | |
| 1262 | Future<Nothing> StorageLocalResourceProviderProcess::updateProfiles( |