--- /dev/null
+// -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*-
+// vim: ts=8 sw=2 smarttab
+/*
+ * Ceph - scalable distributed file system
+ *
+ * Copyright (C) 2014 UnitedStack <haomai@unitedstack.com>
+ *
+ * Author: Haomai Wang <haomaiwang@gmail.com>
+ *
+ * This is free software; you can redistribute it and/or
+ * modify it under the terms of the GNU Lesser General Public
+ * License version 2.1, as published by the Free Software
+ * Foundation. See file COPYING.
+ *
+ */
+
+#ifndef CEPH_MSG_EVENT_H
+#define CEPH_MSG_EVENT_H
+
+#ifdef __APPLE__
+#include <AvailabilityMacros.h>
+#endif
+
+// We use epoll, kqueue, evport, select in descending order by performance.
+#if defined(__linux__)
+#define HAVE_EPOLL 1
+#endif
+
+#if (defined(__APPLE__) && defined(MAC_OS_X_VERSION_10_6)) || defined(__FreeBSD__) || defined(__OpenBSD__) || defined (__NetBSD__)
+#define HAVE_KQUEUE 1
+#endif
+
+#ifdef __sun
+#include <sys/feature_tests.h>
+#ifdef _DTRACE_VERSION
+#define HAVE_EVPORT 1
+#endif
+#endif
+
+#include <atomic>
+#include <mutex>
+#include <condition_variable>
+
+#include "common/ceph_time.h"
+#include "common/dout.h"
+#include "net_handler.h"
+
+#define EVENT_NONE 0
+#define EVENT_READABLE 1
+#define EVENT_WRITABLE 2
+
+class EventCenter;
+
+class EventCallback {
+
+ public:
+ virtual void do_request(int fd_or_id) = 0;
+ virtual ~EventCallback() {} // we want a virtual destructor!!!
+};
+
+typedef EventCallback* EventCallbackRef;
+
+struct FiredFileEvent {
+ int fd;
+ int mask;
+};
+
+/*
+ * EventDriver is a wrap of event mechanisms depends on different OS.
+ * For example, Linux will use epoll(2), BSD will use kqueue(2) and select will
+ * be used for worst condition.
+ */
+class EventDriver {
+ public:
+ virtual ~EventDriver() {} // we want a virtual destructor!!!
+ virtual int init(EventCenter *center, int nevent) = 0;
+ virtual int add_event(int fd, int cur_mask, int mask) = 0;
+ virtual int del_event(int fd, int cur_mask, int del_mask) = 0;
+ virtual int event_wait(vector<FiredFileEvent> &fired_events, struct timeval *tp) = 0;
+ virtual int resize_events(int newsize) = 0;
+ virtual bool need_wakeup() { return true; }
+};
+
+/*
+ * EventCenter maintain a set of file descriptor and handle registered events.
+ */
+class EventCenter {
+ public:
+ // should be enough;
+ static const int MAX_EVENTCENTER = 24;
+
+ private:
+ using clock_type = ceph::coarse_mono_clock;
+
+ struct AssociatedCenters {
+ EventCenter *centers[MAX_EVENTCENTER];
+ AssociatedCenters(CephContext *c) {
+ memset(centers, 0, MAX_EVENTCENTER * sizeof(EventCenter*));
+ }
+ };
+
+ struct FileEvent {
+ int mask;
+ EventCallbackRef read_cb;
+ EventCallbackRef write_cb;
+ FileEvent(): mask(0), read_cb(NULL), write_cb(NULL) {}
+ };
+
+ struct TimeEvent {
+ uint64_t id;
+ EventCallbackRef time_cb;
+
+ TimeEvent(): id(0), time_cb(NULL) {}
+ };
+
+ public:
+ /**
+ * A Poller object is invoked once each time through the dispatcher's
+ * inner polling loop.
+ */
+ class Poller {
+ public:
+ explicit Poller(EventCenter* center, const string& pollerName);
+ virtual ~Poller();
+
+ /**
+ * This method is defined by a subclass and invoked once by the
+ * center during each pass through its inner polling loop.
+ *
+ * \return
+ * 1 means that this poller did useful work during this call.
+ * 0 means that the poller found no work to do.
+ */
+ virtual int poll() = 0;
+
+ private:
+ /// The EventCenter object that owns this Poller. NULL means the
+ /// EventCenter has been deleted.
+ EventCenter* owner;
+
+ /// Human-readable string name given to the poller to make it
+ /// easy to identify for debugging. For most pollers just passing
+ /// in the subclass name probably makes sense.
+ string poller_name;
+
+ /// Index of this Poller in EventCenter::pollers. Allows deletion
+ /// without having to scan all the entries in pollers. -1 means
+ /// this poller isn't currently in EventCenter::pollers (happens
+ /// after EventCenter::reset).
+ int slot;
+ };
+
+ private:
+ CephContext *cct;
+ std::string type;
+ int nevent;
+ // Used only to external event
+ pthread_t owner;
+ std::mutex external_lock;
+ std::atomic_ulong external_num_events;
+ deque<EventCallbackRef> external_events;
+ vector<FileEvent> file_events;
+ EventDriver *driver;
+ std::multimap<clock_type::time_point, TimeEvent> time_events;
+ // Keeps track of all of the pollers currently defined. We don't
+ // use an intrusive list here because it isn't reentrant: we need
+ // to add/remove elements while the center is traversing the list.
+ std::vector<Poller*> pollers;
+ std::map<uint64_t, std::multimap<clock_type::time_point, TimeEvent>::iterator> event_map;
+ uint64_t time_event_next_id;
+ int notify_receive_fd;
+ int notify_send_fd;
+ NetHandler net;
+ EventCallbackRef notify_handler;
+ unsigned idx;
+ AssociatedCenters *global_centers = nullptr;
+
+ int process_time_events();
+ FileEvent *_get_file_event(int fd) {
+ assert(fd < nevent);
+ return &file_events[fd];
+ }
+
+ public:
+ explicit EventCenter(CephContext *c):
+ cct(c), nevent(0),
+ external_num_events(0),
+ driver(NULL), time_event_next_id(1),
+ notify_receive_fd(-1), notify_send_fd(-1), net(c),
+ notify_handler(NULL), idx(0) { }
+ ~EventCenter();
+ ostream& _event_prefix(std::ostream *_dout);
+
+ int init(int nevent, unsigned idx, const std::string &t);
+ void set_owner();
+ pthread_t get_owner() const { return owner; }
+ unsigned get_id() const { return idx; }
+
+ EventDriver *get_driver() { return driver; }
+
+ // Used by internal thread
+ int create_file_event(int fd, int mask, EventCallbackRef ctxt);
+ uint64_t create_time_event(uint64_t milliseconds, EventCallbackRef ctxt);
+ void delete_file_event(int fd, int mask);
+ void delete_time_event(uint64_t id);
+ int process_events(int timeout_microseconds, ceph::timespan *working_dur = nullptr);
+ void wakeup();
+
+ // Used by external thread
+ void dispatch_event_external(EventCallbackRef e);
+ inline bool in_thread() const {
+ return pthread_equal(pthread_self(), owner);
+ }
+
+ private:
+ template <typename func>
+ class C_submit_event : public EventCallback {
+ std::mutex lock;
+ std::condition_variable cond;
+ bool done = false;
+ func f;
+ bool nonwait;
+ public:
+ C_submit_event(func &&_f, bool nw)
+ : f(std::move(_f)), nonwait(nw) {}
+ void do_request(int id) override {
+ f();
+ lock.lock();
+ cond.notify_all();
+ done = true;
+ bool del = nonwait;
+ lock.unlock();
+ if (del)
+ delete this;
+ }
+ void wait() {
+ assert(!nonwait);
+ std::unique_lock<std::mutex> l(lock);
+ while (!done)
+ cond.wait(l);
+ }
+ };
+
+ public:
+ template <typename func>
+ void submit_to(int i, func &&f, bool nowait = false) {
+ assert(i < MAX_EVENTCENTER && global_centers);
+ EventCenter *c = global_centers->centers[i];
+ assert(c);
+ if (!nowait && c->in_thread()) {
+ f();
+ return ;
+ }
+ if (nowait) {
+ C_submit_event<func> *event = new C_submit_event<func>(std::move(f), true);
+ c->dispatch_event_external(event);
+ } else {
+ C_submit_event<func> event(std::move(f), false);
+ c->dispatch_event_external(&event);
+ event.wait();
+ }
+ };
+};
+
+#endif
Code Review / stor4nfv.git / blobdiff