process.hpp

Go to the documentation of this file.

// Licensed 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 __PROCESS_PROCESS_HPP__
#define __PROCESS_PROCESS_HPP__

#include <stdint.h>

#include <memory>
#include <map>
#include <queue>
#include <vector>

#include <process/address.hpp>
#include <process/authenticator.hpp>
#include <process/clock.hpp>
#include <process/event.hpp>
#include <process/filter.hpp>
#include <process/firewall.hpp>
#include <process/http.hpp>
#include <process/message.hpp>
#include <process/mime.hpp>
#include <process/owned.hpp>
#include <process/pid.hpp>

#include <stout/duration.hpp>
#include <stout/hashmap.hpp>
#include <stout/lambda.hpp>
#include <stout/option.hpp>
#include <stout/synchronized.hpp>

namespace process {

// Forward declaration.
class EventQueue;
class Gate;
class Logging;
class Sequence;

namespace firewall {

void install(std::vector<Owned<FirewallRule>>&& rules);

} // namespace firewall {

class ProcessBase : public EventConsumer
{
public:
  explicit ProcessBase(const std::string& id = "");

  virtual ~ProcessBase();

  const UPID& self() const { return pid; }

protected:
  virtual void serve(Event&& event)
  {
    std::move(event).consume(this);
  }

  // Callbacks used to consume (i.e., handle) a specific event.
  void consume(MessageEvent&& event) override;
  void consume(DispatchEvent&& event) override;
  void consume(HttpEvent&& event) override;
  void consume(ExitedEvent&& event) override;
  void consume(TerminateEvent&& event) override;

  virtual void initialize() {}

  virtual void finalize() {}

  virtual void exited(const UPID&) {}

  virtual void lost(const UPID&) {}

  void send(
      const UPID& to,
      const std::string& name,
      const char* data = nullptr,
      size_t length = 0);

  void send(
      const UPID& to,
      std::string&& name);

  void send(
      const UPID& to,
      std::string&& name,
      std::string&& data);

  enum class RemoteConnection
  {
    REUSE,

    RECONNECT,
  };

  UPID link(
      const UPID& pid,
      const RemoteConnection remote = RemoteConnection::REUSE);

  typedef lambda::function<void(const UPID&, const std::string&)>
  MessageHandler;

  void install(
      const std::string& name,
      const MessageHandler& handler)
  {
    handlers.message[name] = handler;
  }

  template <typename T>
  void install(
      const std::string& name,
      void (T::*method)(const UPID&, const std::string&))
  {
    // Note that we use dynamic_cast here so a process can use
    // multiple inheritance if it sees so fit (e.g., to implement
    // multiple callback interfaces).
    MessageHandler handler =
      lambda::bind(method, dynamic_cast<T*>(this), lambda::_1, lambda::_2);
    install(name, handler);
  }

  void delegate(const std::string& name, const UPID& pid)
  {
    delegates[name] = pid;
  }

  typedef lambda::function<Future<http::Response>(const http::Request&)>
  HttpRequestHandler;

  // Options to control the behavior of a route.
  struct RouteOptions
  {
    RouteOptions()
      : requestStreaming(false) {}

    // Set to true if the endpoint supports request streaming.
    // Default: false.
    bool requestStreaming;
  };

  void route(
      const std::string& name,
      const Option<std::string>& help,
      const HttpRequestHandler& handler,
      const RouteOptions& options = RouteOptions());

  template <typename T>
  void route(
      const std::string& name,
      const Option<std::string>& help,
      Future<http::Response> (T::*method)(const http::Request&),
      const RouteOptions& options = RouteOptions())
  {
    // Note that we use dynamic_cast here so a process can use
    // multiple inheritance if it sees so fit (e.g., to implement
    // multiple callback interfaces).
    HttpRequestHandler handler =
      lambda::bind(method, dynamic_cast<T*>(this), lambda::_1);
    route(name, help, handler, options);
  }

  typedef lambda::function<Future<http::Response>(
      const http::Request&,
      const Option<http::authentication::Principal>&)>
          AuthenticatedHttpRequestHandler;

  // TODO(arojas): Consider introducing an `authentication::Realm` type.
  void route(
      const std::string& name,
      const std::string& realm,
      const Option<std::string>& help,
      const AuthenticatedHttpRequestHandler& handler,
      const RouteOptions& options = RouteOptions());

  template <typename T>
  void route(
      const std::string& name,
      const std::string& realm,
      const Option<std::string>& help,
      Future<http::Response> (T::*method)(
          const http::Request&,
          const Option<http::authentication::Principal>&),
      const RouteOptions& options = RouteOptions())
  {
    // Note that we use dynamic_cast here so a process can use
    // multiple inheritance if it sees so fit (e.g., to implement
    // multiple callback interfaces).
    AuthenticatedHttpRequestHandler handler =
      lambda::bind(method, dynamic_cast<T*>(this), lambda::_1, lambda::_2);
    route(name, realm, help, handler, options);
  }

  void provide(
      const std::string& name,
      const std::string& path,
      const std::map<std::string, std::string>& types = mime::types)
  {
    // TODO(benh): Check that name is only alphanumeric (i.e., has no
    // '/') and that path is absolute.
    Asset asset;
    asset.path = path;
    asset.types = types;
    assets[name] = asset;
  }

  template <typename T>
  size_t eventCount();

private:
  friend class SocketManager;
  friend class ProcessManager;
  friend void* schedule(void*);

  // Process states.
  //
  // Transitioning from BLOCKED to READY also requires enqueueing the
  // process in the run queue otherwise the events will never be
  // processed!
  enum class State
  {
    BOTTOM, // Uninitialized but events may be enqueued.
    BLOCKED, // Initialized, no events enqueued.
    READY, // Initialized, events enqueued.
    TERMINATING // Initialized, no more events will be enqueued.
  };

  std::atomic<State> state = ATOMIC_VAR_INIT(State::BOTTOM);

  // Flag for indicating that a terminate event has been injected.
  std::atomic<bool> termination = ATOMIC_VAR_INIT(false);

  // Enqueue the specified message, request, or function call.
  void enqueue(Event* event);

  // Delegates for messages.
  std::map<std::string, UPID> delegates;

  // Definition of an HTTP endpoint. The endpoint can be
  // associated with an authentication realm, in which case:
  //
  //  (1) `realm` and `authenticatedHandler` will be set.
  //      Libprocess will perform HTTP authentication for
  //      all requests to this endpoint (by default during
  //      HttpEvent consumption). The authentication principal
  //      will be passed to the `authenticatedHandler`.
  //
  //  Otherwise, if the endpoint is not associated with an
  //  authentication realm:
  //
  //  (2) Only `handler` will be set, and no authentication
  //      takes place.
  struct HttpEndpoint
  {
    Option<HttpRequestHandler> handler;

    Option<std::string> realm;
    Option<AuthenticatedHttpRequestHandler> authenticatedHandler;
    RouteOptions options;
  };

  // Handlers for messages and HTTP requests.
  struct {
    hashmap<std::string, MessageHandler> message;
    hashmap<std::string, HttpEndpoint> http;

    // Used for delivering HTTP requests in the correct order.
    // Initialized lazily to avoid ProcessBase requiring
    // another Process!
    Owned<Sequence> httpSequence;
  } handlers;

  // Definition of a static asset.
  struct Asset
  {
    std::string path;
    std::map<std::string, std::string> types;
  };

  // Continuation for `consume(HttpEvent&&)`.
  Future<http::Response> _consume(
      const HttpEndpoint& endpoint,
      const std::string& name,
      const Owned<http::Request>& request);

  // JSON representation of process. MUST be invoked from within the
  // process itself in order to safely examine events.
  operator JSON::Object();

  // Static assets(s) to provide.
  std::map<std::string, Asset> assets;

  // Queue of received events. We employ the PIMPL idiom here and use
  // a pointer so we can hide the implementation of `EventQueue`.
  std::unique_ptr<EventQueue> events;

  // NOTE: this is a shared pointer to a _pointer_, hence this is not
  // responsible for the ProcessBase itself.
  std::shared_ptr<ProcessBase*> reference;

  std::shared_ptr<Gate> gate;

  // Whether or not the runtime should delete this process after it
  // has terminated. Note that failure to spawn the process will leave
  // the process unmanaged and thus it may leak!
  bool manage = false;

  // Process PID.
  UPID pid;
};


template <typename T>
class Process : public virtual ProcessBase {
public:
  virtual ~Process() {}

  PID<T> self() const { return PID<T>(static_cast<const T*>(this)); }

protected:
  // Useful typedefs for dispatch/delay/defer to self()/this.
  typedef T Self;
  typedef T This;
};


bool initialize(
    const Option<std::string>& delegate = None(),
    const Option<std::string>& readwriteAuthenticationRealm = None(),
    const Option<std::string>& readonlyAuthenticationRealm = None());


void finalize(bool finalize_wsa = false);


std::string absolutePath(const std::string& path);


network::inet::Address address();


PID<Logging> logging();


long workers();


UPID spawn(ProcessBase* process, bool manage = false);

inline UPID spawn(ProcessBase& process, bool manage = false)
{
  return spawn(&process, manage);
}

template <typename T>
PID<T> spawn(T* t, bool manage = false)
{
  // We save the pid before spawn is called because it's possible that
  // the process has already been deleted after spawn returns (e.g.,
  // if 'manage' is true).
  PID<T> pid(t);

  if (!spawn(static_cast<ProcessBase*>(t), manage)) {
    return PID<T>();
  }

  return pid;
}

template <typename T>
PID<T> spawn(T& t, bool manage = false)
{
  return spawn(&t, manage);
}


void terminate(const UPID& pid, bool inject = true);
void terminate(const ProcessBase& process, bool inject = true);
void terminate(const ProcessBase* process, bool inject = true);


bool wait(const UPID& pid, const Duration& duration = Seconds(-1));
bool wait(const ProcessBase& process, const Duration& duration = Seconds(-1));
bool wait(const ProcessBase* process, const Duration& duration = Seconds(-1));


void post(const UPID& to,
          const std::string& name,
          const char* data = nullptr,
          size_t length = 0);


void post(const UPID& from,
          const UPID& to,
          const std::string& name,
          const char* data = nullptr,
          size_t length = 0);


inline void terminate(const ProcessBase& process, bool inject)
{
  terminate(process.self(), inject);
}


inline void terminate(const ProcessBase* process, bool inject)
{
  terminate(process->self(), inject);
}


inline bool wait(const ProcessBase& process, const Duration& duration)
{
  return process::wait(process.self(), duration); // Explicit to disambiguate.
}


inline bool wait(const ProcessBase* process, const Duration& duration)
{
  return process::wait(process->self(), duration); // Explicit to disambiguate.
}


// Per thread process pointer.
extern thread_local ProcessBase* __process__;

// NOTE: Methods in this namespace should only be used in tests to
// inject arbitrary events.
namespace inject {

bool exited(const UPID& from, const UPID& to);

} // namespace inject {

} // namespace process {

#endif // __PROCESS_PROCESS_HPP__