future_tracker.hpp

Go to the documentation of this file.

// Licensed to the Apache Software Foundation (ASF) under one
// or more contributor license agreements.  See the NOTICE file
// distributed with this work for additional information
// regarding copyright ownership.  The ASF licenses this file
// to you under the Apache License, Version 2.0 (the
// "License"); you may not use this file except in compliance
// with the License.  You may obtain a copy of the License at
//
//     http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.

#ifndef __FUTURE_TRACKER_HPP__
#define __FUTURE_TRACKER_HPP__

#include <list>
#include <map>
#include <string>
#include <vector>

#include <process/defer.hpp>
#include <process/dispatch.hpp>
#include <process/future.hpp>
#include <process/id.hpp>
#include <process/owned.hpp>
#include <process/process.hpp>

namespace mesos {
namespace internal {

struct FutureMetadata
{
  std::string operation;
  std::string component;
  std::map<std::string, std::string> args;

  inline bool operator==(const FutureMetadata& that) const
  {
    return operation == that.operation &&
           component == that.component &&
           args == that.args;
  }
};


class PendingFutureTrackerProcess
  : public process::Process<PendingFutureTrackerProcess>
{
public:
  PendingFutureTrackerProcess()
    : ProcessBase(process::ID::generate("pending-future-tracker")) {}

  template <typename T>
  void addFuture(const process::Future<T>& future, FutureMetadata&& metadata)
  {
    auto it = pending.emplace(pending.end(), std::move(metadata));

    future
      .onAny(process::defer(
          self(), &PendingFutureTrackerProcess::eraseFuture, it))
      .onAbandoned(process::defer(
          self(), &PendingFutureTrackerProcess::eraseFuture, it));
  }

  void eraseFuture(typename std::list<FutureMetadata>::iterator it)
  {
    pending.erase(it);
  }

  process::Future<std::vector<FutureMetadata>> pendingFutures()
  {
    return std::vector<FutureMetadata>(pending.begin(), pending.end());
  }

private:
  std::list<FutureMetadata> pending;
};


class PendingFutureTracker
{
public:
  static Try<PendingFutureTracker*> create()
  {
    return new PendingFutureTracker(process::Owned<PendingFutureTrackerProcess>(
        new PendingFutureTrackerProcess));
  }

  ~PendingFutureTracker()
  {
    terminate(process.get());
    process::wait(process.get());
  }

  template <typename T>
  process::Future<T> track(
      const process::Future<T>& future,
      const std::string& operation,
      const std::string& component,
      const std::map<std::string, std::string>& args = {})
  {
    process::dispatch(
        process.get(),
        &PendingFutureTrackerProcess::addFuture<T>,
        future,
        FutureMetadata{operation, component, args});

    return future;
  }

  process::Future<std::vector<FutureMetadata>> pendingFutures()
  {
    return process::dispatch(
        process.get(),
        &PendingFutureTrackerProcess::pendingFutures);
  }

private:
  explicit PendingFutureTracker(
      const process::Owned<PendingFutureTrackerProcess>& _process)
    : process(_process)
  {
    spawn(process.get());
  }

  process::Owned<PendingFutureTrackerProcess> process;
};

} // namespace internal {
} // namespace mesos {

#endif // __FUTURE_TRACKER_HPP__