src/ceph/src/msg/async/dpdk/UserspaceEvent.h

   1 // -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*-
   2 /*
   3  * Ceph - scalable distributed file system
   4  *
   5  * Copyright (C) 2015 XSky <haomai@xsky.com>
   6  *
   7  * Author: Haomai Wang <haomaiwang@gmail.com>
   8  *
   9  * This is free software; you can redistribute it and/or
  10  * modify it under the terms of the GNU Lesser General Public
  11  * License version 2.1, as published by the Free Software
  12  * Foundation.  See file COPYING.
  13  *
  14  */
  15
  16 #ifndef CEPH_USERSPACEEVENT_H
  17 #define CEPH_USERSPACEEVENT_H
  18
  19 #include <cstddef>
  20 #include <errno.h>
  21 #include <string.h>
  22
  23 #include <vector>
  24 #include <list>
  25
  26 #include "include/assert.h"
  27 #include "include/int_types.h"
  28 #include "common/Tub.h"
  29
  30 class CephContext;
  31
  32 class UserspaceEventManager {
  33   struct UserspaceFDImpl {
  34     uint32_t waiting_idx = 0;
  35     int16_t read_errno = 0;
  36     int16_t write_errno = 0;
  37     int8_t listening_mask = 0;
  38     int8_t activating_mask = 0;
  39     uint32_t magic = 4921;
  40   };
  41   CephContext *cct;
  42   int max_fd = 0;
  43   uint32_t max_wait_idx = 0;
  44   std::vector<Tub<UserspaceFDImpl> > fds;
  45   std::vector<int> waiting_fds;
  46   std::list<uint32_t> unused_fds;
  47
  48  public:
  49   UserspaceEventManager(CephContext *c): cct(c) {
  50     waiting_fds.resize(1024);
  51   }
  52
  53   int get_eventfd();
  54
  55   int listen(int fd, int mask) {
  56     if ((size_t)fd >= fds.size())
  57       return -ENOENT;
  58
  59     Tub<UserspaceFDImpl> &impl = fds[fd];
  60     if (!impl)
  61       return -ENOENT;
  62
  63     impl->listening_mask |= mask;
  64     if (impl->activating_mask & impl->listening_mask && !impl->waiting_idx) {
  65       if (waiting_fds.size() <= max_wait_idx)
  66         waiting_fds.resize(waiting_fds.size()*2);
  67       impl->waiting_idx = ++max_wait_idx;
  68       waiting_fds[max_wait_idx] = fd;
  69     }
  70     return 0;
  71   }
  72
  73   int unlisten(int fd, int mask) {
  74     if ((size_t)fd >= fds.size())
  75       return -ENOENT;
  76
  77     Tub<UserspaceFDImpl> &impl = fds[fd];
  78     if (!impl)
  79       return -ENOENT;
  80
  81     impl->listening_mask &= (~mask);
  82     if (!(impl->activating_mask & impl->listening_mask) && impl->waiting_idx) {
  83       if (waiting_fds[max_wait_idx] == fd) {
  84         assert(impl->waiting_idx == max_wait_idx);
  85         --max_wait_idx;
  86       }
  87       waiting_fds[impl->waiting_idx] = -1;
  88       impl->waiting_idx = 0;
  89     }
  90     return 0;
  91   }
  92
  93   int notify(int fd, int mask);
  94   void close(int fd);
  95   int poll(int *events, int *masks, int num_events, struct timeval *tp);
  96
  97   bool check() {
  98     for (auto &&m : fds) {
  99       if (m && m->magic != 4921)
 100         return false;
 101     }
 102     return true;
 103   }
 104 };
 105
 106 #endif //CEPH_USERSPACEEVENT_H