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

   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) 2016 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 #include "UserspaceEvent.h"
  17
  18 #include "common/dout.h"
  19 #include "include/ceph_assert.h"
  20
  21 #define dout_subsys ceph_subsys_dpdk
  22 #undef dout_prefix
  23 #define dout_prefix *_dout << "dpdk "
  24
  25 int UserspaceEventManager::get_eventfd()
  26 {
  27   int fd;
  28   if (!unused_fds.empty()) {
  29     fd = unused_fds.front();
  30     unused_fds.pop_front();
  31   } else {
  32     fd = ++max_fd;
  33     fds.resize(fd + 1);
  34   }
  35
  36   std::optional<UserspaceFDImpl> &impl = fds[fd];
  37   ceph_assert(!impl);
  38   impl.emplace();
  39   ldout(cct, 20) << __func__ << " fd=" << fd << dendl;
  40   return fd;
  41 }
  42
  43 int UserspaceEventManager::notify(int fd, int mask)
  44 {
  45   ldout(cct, 20) << __func__ << " fd=" << fd << " mask=" << mask << dendl;
  46   if ((size_t)fd >= fds.size())
  47     return -ENOENT;
  48
  49   std::optional<UserspaceFDImpl> &impl = fds[fd];
  50   if (!impl)
  51     return -ENOENT;
  52
  53   ldout(cct, 20) << __func__ << " activing=" << int(impl->activating_mask)
  54                  << " listening=" << int(impl->listening_mask)
  55                  << " waiting_idx=" << int(impl->waiting_idx) << dendl;
  56
  57   impl->activating_mask |= mask;
  58   if (impl->waiting_idx)
  59     return 0;
  60
  61   if (impl->listening_mask & mask) {
  62     if (waiting_fds.size() <= max_wait_idx)
  63       waiting_fds.resize(waiting_fds.size()*2);
  64     impl->waiting_idx = ++max_wait_idx;
  65     waiting_fds[max_wait_idx] = fd;
  66   }
  67
  68   ldout(cct, 20) << __func__ << " activing=" << int(impl->activating_mask)
  69                  << " listening=" << int(impl->listening_mask)
  70                  << " waiting_idx=" << int(impl->waiting_idx) << " done " << dendl;
  71   return 0;
  72 }
  73
  74 void UserspaceEventManager::close(int fd)
  75 {
  76   ldout(cct, 20) << __func__ << " fd=" << fd << dendl;
  77   if ((size_t)fd >= fds.size())
  78     return ;
  79
  80   std::optional<UserspaceFDImpl> &impl = fds[fd];
  81   if (!impl)
  82     return ;
  83
  84   if (fd == max_fd)
  85     --max_fd;
  86   else
  87     unused_fds.push_back(fd);
  88
  89   if (impl->activating_mask) {
  90     if (waiting_fds[max_wait_idx] == fd) {
  91       ceph_assert(impl->waiting_idx == max_wait_idx);
  92       --max_wait_idx;
  93     }
  94     waiting_fds[impl->waiting_idx] = -1;
  95   }
  96   impl.reset();
  97 }
  98
  99 int UserspaceEventManager::poll(int *events, int *masks, int num_events, struct timeval *tp)
 100 {
 101   int fd;
 102   uint32_t i = 0;
 103   int count = 0;
 104   ceph_assert(num_events);
 105   // leave zero slot for waiting_fds
 106   while (i < max_wait_idx) {
 107     fd = waiting_fds[++i];
 108     if (fd == -1)
 109       continue;
 110
 111     events[count] = fd;
 112     std::optional<UserspaceFDImpl> &impl = fds[fd];
 113     ceph_assert(impl);
 114     masks[count] = impl->listening_mask & impl->activating_mask;
 115     ceph_assert(masks[count]);
 116     ldout(cct, 20) << __func__ << " fd=" << fd << " mask=" << masks[count] << dendl;
 117     impl->activating_mask &= (~masks[count]);
 118     impl->waiting_idx = 0;
 119     if (++count >= num_events)
 120       break;
 121   }
 122   if (i < max_wait_idx) {
 123     memmove(&waiting_fds[1], &waiting_fds[i+1], sizeof(int)*(max_wait_idx-i));
 124   }
 125   max_wait_idx -= i;
 126   return count;
 127 }