class ProcessBase : public EventVisitor
{
public:
  ...
  UPID self() const { return pid; }
  ...
  virtual void initialize() {}
  virtual void finalize() {}
  virtual void exited(const UPID& pid) {}
  virtual void lost(const UPID& pid) {}
  ...

  // Puts a message at front of queue.
  void inject(
      const UPID& from,
      const std::string& name,
      const char* data = NULL,
      size_t length = 0);

  // Sends a message with data to PID.
  void send(
      const UPID& to,
      const std::string& name,
      const char* data = NULL,
      size_t length = 0);
  ...
  UPID link(const UPID& pid);

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

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

private:
  ...

  std::deque<Event*> events;

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

  // Handlers for messages and HTTP requests.
  struct {
    std::map<std::string, MessageHandler> message;
    std::map<std::string, HttpRequestHandler> http;
  } handlers;

  // Active references.
  int refs;

  // Process PID.
  UPID pid;
};

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

  // Returns pid of process; valid even before calling spawn.
  PID<T> self() const { return PID<T>(dynamic_cast<const T*>(this)); }

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

...
struct Message
{
  std::string name;
  UPID from;
  UPID to;
  std::string body;
};
...

struct UPID
{
...
  UPID(const std::string& id_, uint32_t ip_, uint16_t port_)
    : id(id_), ip(ip_), port(port_) {}

  UPID(const char* s);

  UPID(const std::string& s);
...
  std::string id;
  uint32_t ip;
  uint16_t port;
};

template <typename T = ProcessBase>
struct PID : UPID
{
...
};

ostream& operator << (ostream& stream, const UPID& pid)
{
  // Call inet_ntop since inet_ntoa is not thread-safe!
  char ip[INET_ADDRSTRLEN];
  if (inet_ntop(AF_INET, (in_addr *) &pid.ip, ip, INET_ADDRSTRLEN) == NULL)
    memset(ip, 0, INET_ADDRSTRLEN);

  stream << pid.id << "@" << ip << ":" << pid.port;
  return stream;
}

class ProcessBase : public EventVisitor
{
...
  typedef std::tr1::function<Future<http::Response>(const http::Request&)>
  HttpRequestHandler;

  // Setup a handler for an HTTP request.
  bool route(
      const std::string& name,
      const HttpRequestHandler& handler)
  {
    if (name.find('/') != 0) {
      return false;
    }
    handlers.http[name.substr(1)] = handler;
    return true;
  }
...
};

template <typename T>
class ProtobufProcess : public process::Process<T>
{
...
protected:
  void send(const process::UPID& to,
            const google::protobuf::Message& message)
  {
    std::string data;
    message.SerializeToString(&data);
    process::Process<T>::send(to, message.GetTypeName(),
                              data.data(), data.size());
  }
...
  template <typename M>
  void install(void (T::*method)(const M&))
  {
    google::protobuf::Message* m = new M();
    T* t = static_cast<T*>(this);
    protobufHandlers[m->GetTypeName()] =
      std::tr1::bind(&handlerM<M>,
                     t, method,
                     std::tr1::placeholders::_1);
    delete m;
  }
...
  template <typename M>
  static void handlerM(T* t, void (T::*method)(const M&),
                       const std::string& data)
  {
    M m;
    m.ParseFromString(data);
    if (m.IsInitialized()) {
      (t->*method)(m);
    } else {
      LOG(WARNING) << "Initialization errors: "
                   << m.InitializationErrorString();
    }
  }
...
};

template <typename T>
class ProtobufProcess : public process::Process<T>
{
...
  virtual void visit(const process::MessageEvent& event)
  {
    if (protobufHandlers.count(event.message->name) > 0) {
      from = event.message->from; // For 'reply'.
      protobufHandlers[event.message->name](event.message->body);
      from = process::UPID();
    } else {
      process::Process<T>::visit(event);
    }
  }
...
  process::UPID from; // Sender of "current" message, accessible by subclasses.
...
  void reply(const google::protobuf::Message& message)
  {
    CHECK(from) << "Attempting to reply without a sender";
    std::string data;
    message.SerializeToString(&data);
    send(from, message);
  }
};

class MyProcess : public Process<MyProcess>
{
...
};

...

int main(int argc, char** argv)
{
  MyProcess process;
  PID<MyProcess> pid = spawn(&process);
...
  dispatch(pid, &MyProcess::func1);
  dispatch(pid, &MyProcess::func2, 42);
  wait(pid);
  return 0;
}

...
// Active ProcessManager (eventually will probably be thread-local).
static ProcessManager* process_manager = NULL;
...
UPID spawn(ProcessBase* process, bool manage)
{
  process::initialize();

  if (process != NULL) {
...
    return process_manager->spawn(process, manage);
  } else {
    return UPID();
  }
}

class ProcessManager
{
public:
  ProcessManager(const string& delegate);
  ~ProcessManager();

  ProcessReference use(const UPID& pid);

  bool handle(
      const Socket& socket,
      Request* request);

  bool deliver(
      ProcessBase* receiver,
      Event* event,
      ProcessBase* sender = NULL);

  bool deliver(
      const UPID& to,
      Event* event,
      ProcessBase* sender = NULL);

  UPID spawn(ProcessBase* process, bool manage);
  void resume(ProcessBase* process);
  void cleanup(ProcessBase* process);
  void link(ProcessBase* process, const UPID& to);
  void terminate(const UPID& pid, bool inject, ProcessBase* sender = NULL);
  bool wait(const UPID& pid);

  void enqueue(ProcessBase* process);
  ProcessBase* dequeue();

  void settle();

private:
  // Map of all local spawned and running processes.
  map<string, ProcessBase*> processes;
  synchronizable(processes);

  ...

  // Queue of runnable processes (implemented using list).
  list<ProcessBase*> runq;
  synchronizable(runq);

  // Number of running processes, to support Clock::settle operation.
  int running;
};

void post(const UPID& to, const string& name, const char* data, size_t length)
{
  process::initialize();

  if (!to) {
    return;
  }

  // Encode and transport outgoing message.
  transport(encode(UPID(), to, name, string(data, length)));
}

...

static void transport(Message* message, ProcessBase* sender = NULL)
{
  if (message->to.ip == __ip__ && message->to.port == __port__) {
    // Local message.
    process_manager->deliver(message->to, new MessageEvent(message), sender);
  } else {
    // Remote message.
    socket_manager->send(message);
  }
}

bool ProcessManager::deliver(
    ProcessBase* receiver,
    Event* event,
    ProcessBase* sender)
{
...
  receiver->enqueue(event);

  return true;
}

void ProcessBase::enqueue(Event* event, bool inject)
{
  CHECK(event != NULL);

  // 事前に登録された条件に基づきメッセージをフィルタする仕組み.
  synchronized (filterer) {
    if (filterer != NULL) {
      bool filter = false;
      struct FilterVisitor : EventVisitor
      {
         ...
      } visitor(&filter);

      event->visit(&visitor);

      if (filter) {
        delete event;
        return;
      }
    }
  }

  lock();
  {
    if (state != FINISHED) {
      if (!inject) {
        events.push_back(event);
      } else {
        events.push_front(event);
      }

      if (state == BLOCKED) {
        state = READY;
        // ProcessManager に通知してワーカスレッドを起こす
        process_manager->enqueue(this);
      }

      CHECK(state == BOTTOM ||
            state == READY ||
            state == RUNNING);
    } else {
      delete event;
    }
  }
  unlock();
}

void ProcessManager::enqueue(ProcessBase* process)
{
  ...
  synchronized (runq) {
    CHECK(find(runq.begin(), runq.end(), process) == runq.end());
    runq.push_back(process);
  }

  // Wake up the processing thread if necessary.
  gate->open();
}

ProcessBase* ProcessManager::dequeue()
{
  ...

  ProcessBase* process = NULL;

  synchronized (runq) {
    if (!runq.empty()) {
      process = runq.front();
      runq.pop_front();
      // Increment the running count of processes in order to support
      // the Clock::settle() operation (this must be done atomically
      // with removing the process from the runq).
      __sync_fetch_and_add(&running, 1);
    }
  }

  return process;
}

void* schedule(void* arg)
{
  do {
    ProcessBase* process = process_manager->dequeue();
    if (process == NULL) {
      Gate::state_t old = gate->approach();
      process = process_manager->dequeue();
      if (process == NULL) {
        gate->arrive(old); // Wait at gate if idle.
        continue;
      } else {
        gate->leave();
      }
    }
    process_manager->resume(process);
  } while (true);
}

void initialize(const string& delegate)
{
  ...
  for (int i = 0; i < cpus; i++) {
    pthread_t thread; // For now, not saving handles on our threads.
    if (pthread_create(&thread, NULL, schedule, NULL) != 0) {
      LOG(FATAL) << "Failed to initialize, pthread_create";
    }
  }
  ...
}

void ProcessManager::resume(ProcessBase* process)
{
  __process__ = process; // __process__ はスレッドローカル変数

  ...
  bool blocked = false;
  ...

  while (!terminate && !blocked) {
    Event* event = NULL;

    process->lock();
    {
      if (process->events.size() > 0) {
        event = process->events.front();
        process->events.pop_front();
        process->state = ProcessBase::RUNNING;
      } else {
        process->state = ProcessBase::BLOCKED;
        blocked = true;
      }
    }
    process->unlock();

    if (!blocked) {
      ...

      // Now service the event.
      try {
        process->serve(*event);
      } catch (const std::exception& e) {
        std::cerr << "libprocess: " << process->pid
                  << " terminating due to "
                  << e.what() << std::endl;
        terminate = true;
      } catch (...) {
        std::cerr << "libprocess: " << process->pid
                  << " terminating due to unknown exception" << std::endl;
        terminate = true;
      }

      delete event;

      if (terminate) {
        cleanup(process);
      }
    }
  }

  __process__ = NULL;

  CHECK_GE(running, 1);
  __sync_fetch_and_sub(&running, 1);
}

class SocketManager
{
public:
...
  void send(Encoder* encoder, int s, bool persist);
  void send(const Response& response, int s, bool persist);
  void send(Message* message);
...
  Socket accepted(int s);
  void close(int s);
...

  // Map from UPID (local/remote) to process.
  map<UPID, set<ProcessBase*> > links;

  // Collection of all actice sockets.
  map<int, Socket> sockets;

  // Collection of sockets that should be disposed when they are
  // finished being used (e.g., when there is no more data to send on
  // them).
  set<int> dispose;

  // Map from socket to node (ip, port).
  map<int, Node> nodes;

  // Maps from node (ip, port) to temporary sockets (i.e., they will
  // get closed once there is no more data to send on them).
  map<Node, int> temps;

  // Maps from node (ip, port) to persistent sockets (i.e., they will
  // remain open even if there is no more data to send on them).  We
  // distinguish these from the 'temps' collection so we can tell when
  // a persistant socket has been lost (and thus generate
  // ExitedEvents).
  map<Node, int> persists;

  // Map from socket to outgoing queue.
  map<int, queue<Encoder*> > outgoing;
...
};

class Node
{
public:
  Node(uint32_t _ip = 0, uint16_t _port = 0)
    : ip(_ip), port(_port) {}
...
  uint32_t ip;
  uint16_t port;
};

void* serve(void* arg)
{
  ev_loop(((struct ev_loop*) arg), 0);

  return NULL;
}
...
void initialize(const string& delegate)
{
  ...
#ifdef __sun__
  loop = ev_default_loop(EVBACKEND_POLL | EVBACKEND_SELECT);
#else
  loop = ev_default_loop(EVFLAG_AUTO);
#endif // __sun__

  ev_async_init(&async_watcher, handle_async);
  ev_async_start(loop, &async_watcher);

  ev_timer_init(&timeouts_watcher, handle_timeouts, 0., 2100000.0);
  ev_timer_again(loop, &timeouts_watcher);

  ev_io_init(&server_watcher, accept, __s__, EV_READ);
  ev_io_start(loop, &server_watcher);
  ...
  pthread_t thread; // For now, not saving handles on our threads.
  if (pthread_create(&thread, NULL, serve, loop) != 0) {
    LOG(FATAL) << "Failed to initialize, pthread_create";
  }
  ...
}

void recv_data(struct ev_loop* loop, ev_io* watcher, int revents)
{
  DataDecoder* decoder = (DataDecoder*) watcher->data;
  ...
  while (true) {
    ...
    length = recv(s, data, size, 0);
    ...
      // Decode as much of the data as possible into HTTP requests.
      const deque<Request*>& requests = decoder->decode(data, length);

      if (!requests.empty()) {
        foreach (Request* request, requests) {
          process_manager->handle(decoder->socket(), request);
        }
      }
    ...
  }
}

bool ProcessManager::handle(
    const Socket& socket,
    Request* request)
{
  CHECK(request != NULL);

  // Check if this is a libprocess request (i.e., 'User-Agent:
  // libprocess/id@ip:port') and if so, parse as a message.
  if (libprocess(request)) {
    Message* message = parse(request);
    if (message != NULL) {
      delete request;
      // TODO(benh): Use the sender PID in order to capture
      // happens-before timing relationships for testing.
      return deliver(message->to, new MessageEvent(message));
    }

    VLOG(1) << "Failed to handle libprocess request: "
            << request->method << " " << request->path
            << " (User-Agent: " << request->headers["User-Agent"] << ")";

    delete request;
    return false;
  }
  ...
}

class Statistics
{
public:
  Statistics(const Duration& window);
  ~Statistics();

  // Returns the time series of a statistic.
  process::Future<std::map<Seconds, double> > timeseries(
      const std::string& context,
      const std::string& name,
      const Option<Seconds>& start = None(),
      const Option<Seconds>& stop = None());

  // Returns the latest value of a statistic.
  process::Future<Option<double> > get(
      const std::string& context,
      const std::string& name);
  ...
};

// Definition of a "shared" future. A future can hold any
// copy-constructible value. A future is considered "shared" because
// by default a future can be accessed concurrently.
template <typename T>
class Future
{
public:
  // Constructs a failed future.
  static Future<T> failed(const std::string& message);


  // Helpers to get the current state of this future.
  bool isPending() const;
  bool isReady() const;
  bool isDiscarded() const;
  bool isFailed() const;
  ...
  bool discard();
  ...
  // Waits for this future to become ready, discarded, or failed.
  bool await(const Duration& duration = Seconds(-1.0)) const;

  // Return the value associated with this future, waits indefinitely
  // until a value gets associated or until the future is discarded.
  T get() const;

  // Returns the failure message associated with this future.
  std::string failure() const;

  // Installs callbacks for the specified events and returns a const
  // reference to 'this' in order to easily support chaining.
  const Future<T>& onReady(const ReadyCallback& callback) const;
  const Future<T>& onFailed(const FailedCallback& callback) const;
  const Future<T>& onDiscarded(const DiscardedCallback& callback) const;
  const Future<T>& onAny(const AnyCallback& callback) const;

  // Installs callbacks that get executed when this future is ready
  // and associates the result of the callback with the future that is
  // returned to the caller (which may be of a different type).
  template <typename X>
  Future<X> then(const std::tr1::function<Future<X>(const T&)>& f) const;

  template <typename X>
  Future<X> then(const std::tr1::function<X(const T&)>& f) const;

  // Helpers for the compiler to be able to forward std::tr1::bind results.
  template <typename X>
  Future<X> then(const std::tr1::_Bind<X(*(void))(void)>& b) const
  {
    return then(std::tr1::function<X(const T&)>(b));
  }
...
private:
  friend class Promise<T>;
...
  enum State {
    PENDING,
    READY,
    FAILED,
    DISCARDED,
  };

  int* refs;
  int* lock;
  State* state;
  T** t;
  std::string** message; // Message associated with failure.
  std::queue<ReadyCallback>* onReadyCallbacks;
  std::queue<FailedCallback>* onFailedCallbacks;
  std::queue<DiscardedCallback>* onDiscardedCallbacks;
  std::queue<AnyCallback>* onAnyCallbacks;
  Latch* latch;
};

... // Promise<T> の定義がつづく...

...
class Synchronized
{
public:
  Synchronized(Synchronizable *_synchronizable)
    : synchronizable(_synchronizable)
  {
    synchronizable->acquire();
  }

  ~Synchronized()
  {
    synchronizable->release();
  }

  operator bool () { return true; }

private:
  Synchronizable *synchronizable;
};


#define synchronized(s)                                                 \
  if (Synchronized __synchronized ## s = Synchronized(&__synchronizable_ ## s))
...

// Represents a uniquely owned pointer.
//
// TODO(bmahler): For now, Owned only provides shared_ptr semantics.
// When we make the switch to C++11, we will change to provide
// unique_ptr semantics. Consequently, each usage of Owned that
// invoked a copy will have to be adjusted to use move semantics.
template <typename T>
class Owned : public boost::shared_ptr<T>
{
public:
  Owned(T* t) : boost::shared_ptr<T>(t) {}
};