+++ /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.
- *
- */
-
-#include "common/errno.h"
-#include "EventKqueue.h"
-
-#define dout_subsys ceph_subsys_ms
-
-#undef dout_prefix
-#define dout_prefix *_dout << "KqueueDriver."
-
-#define KEVENT_NOWAIT 0
-
-int KqueueDriver::test_kqfd() {
- struct kevent ke[1];
- if (kevent(kqfd, ke, 0, NULL, 0, KEVENT_NOWAIT) == -1) {
- ldout(cct,0) << __func__ << " invalid kqfd = " << kqfd
- << cpp_strerror(errno) << dendl;
- return -errno;
- }
- return kqfd;
-}
-
-int KqueueDriver::restore_events() {
- struct kevent ke[2];
- int i;
-
- ldout(cct,30) << __func__ << " on kqfd = " << kqfd << dendl;
- for(i=0;i<size;i++) {
- int num = 0;
- if (sav_events[i].mask == 0 )
- continue;
- ldout(cct,30) << __func__ << " restore kqfd = " << kqfd
- << " fd = " << i << " mask " << sav_events[i].mask << dendl;
- if (sav_events[i].mask & EVENT_READABLE)
- EV_SET(&ke[num++], i, EVFILT_READ, EV_ADD, 0, 0, NULL);
- if (sav_events[i].mask & EVENT_WRITABLE)
- EV_SET(&ke[num++], i, EVFILT_WRITE, EV_ADD, 0, 0, NULL);
- if (num) {
- if (kevent(kqfd, ke, num, NULL, 0, KEVENT_NOWAIT) == -1) {
- ldout(cct,0) << __func__ << " unable to add event: "
- << cpp_strerror(errno) << dendl;
- return -errno;
- }
- }
- }
- return 0;
-}
-
-int KqueueDriver::test_thread_change(const char* funcname) {
- // check to see if we changed thread, because that invalidates
- // the kqfd and we need to restore that
- int oldkqfd = kqfd;
-
- if (!pthread_equal(mythread, pthread_self())) {
- ldout(cct,20) << funcname << " We changed thread from " << mythread
- << " to " << pthread_self() << dendl;
- mythread = pthread_self();
- kqfd = -1;
- } else if ((kqfd != -1) && (test_kqfd() < 0)) {
- // should this ever happen?
- // It would be strange to change kqfd with thread change.
- // Might nee to change this into an assert() in the future.
- ldout(cct,0) << funcname << " Warning: Recreating old kqfd. "
- << "This should not happen!!!" << dendl;
- kqfd = -1;
- }
- if (kqfd == -1) {
- kqfd = kqueue();
- ldout(cct,30) << funcname << " kqueue: new kqfd = " << kqfd
- << " (was: " << oldkqfd << ")"
- << dendl;
- if (kqfd < 0) {
- lderr(cct) << funcname << " unable to do kqueue: "
- << cpp_strerror(errno) << dendl;
- return -errno;
- }
- if (restore_events()< 0) {
- lderr(cct) << funcname << " unable restore all events "
- << cpp_strerror(errno) << dendl;
- return -errno;
- }
- }
- return 0;
-}
-
-int KqueueDriver::init(EventCenter *c, int nevent)
-{
- // keep track of possible changes of our thread
- // because change of thread kills the kqfd
- mythread = pthread_self();
-
- // Reserve the space to accept the kevent return events.
- res_events = (struct kevent*)malloc(sizeof(struct kevent)*nevent);
- if (!res_events) {
- lderr(cct) << __func__ << " unable to malloc memory: "
- << cpp_strerror(errno) << dendl;
- return -ENOMEM;
- }
- memset(res_events, 0, sizeof(struct kevent)*nevent);
- size = nevent;
-
- // Reserve the space to keep all of the events set, so it can be redone
- // when we change trhread ID.
- sav_events = (struct SaveEvent*)malloc(sizeof(struct SaveEvent)*nevent);
- if (!sav_events) {
- lderr(cct) << __func__ << " unable to malloc memory: "
- << cpp_strerror(errno) << dendl;
- return -ENOMEM;
- }
- memset(sav_events, 0, sizeof(struct SaveEvent)*nevent);
- sav_max = nevent;
-
- // Delay assigning a descriptor until it is really needed.
- // kqfd = kqueue();
- kqfd = -1;
- return 0;
-}
-
-int KqueueDriver::add_event(int fd, int cur_mask, int add_mask)
-{
- struct kevent ke[2];
- int num = 0;
-
- ldout(cct,30) << __func__ << " add event kqfd = " << kqfd << " fd = " << fd
- << " cur_mask = " << cur_mask << " add_mask = " << add_mask
- << dendl;
-
- int r = test_thread_change(__func__);
- if ( r < 0 )
- return r;
-
- if (add_mask & EVENT_READABLE)
- EV_SET(&ke[num++], fd, EVFILT_READ, EV_ADD|EV_CLEAR, 0, 0, NULL);
- if (add_mask & EVENT_WRITABLE)
- EV_SET(&ke[num++], fd, EVFILT_WRITE, EV_ADD|EV_CLEAR, 0, 0, NULL);
-
- if (num) {
- if (kevent(kqfd, ke, num, NULL, 0, KEVENT_NOWAIT) == -1) {
- lderr(cct) << __func__ << " unable to add event: "
- << cpp_strerror(errno) << dendl;
- return -errno;
- }
- }
- // keep what we set
- if (fd >= sav_max)
- resize_events(sav_max+5000);
- sav_events[fd].mask = cur_mask | add_mask;
- return 0;
-}
-
-int KqueueDriver::del_event(int fd, int cur_mask, int del_mask)
-{
- struct kevent ke[2];
- int num = 0;
- int mask = cur_mask & del_mask;
-
- ldout(cct,30) << __func__ << " delete event kqfd = " << kqfd
- << " fd = " << fd << " cur_mask = " << cur_mask
- << " del_mask = " << del_mask << dendl;
-
- int r = test_thread_change(__func__);
- if ( r < 0 )
- return r;
-
- if (mask & EVENT_READABLE)
- EV_SET(&ke[num++], fd, EVFILT_READ, EV_DELETE, 0, 0, NULL);
- if (mask & EVENT_WRITABLE)
- EV_SET(&ke[num++], fd, EVFILT_WRITE, EV_DELETE, 0, 0, NULL);
-
- if (num) {
- int r = 0;
- if ((r = kevent(kqfd, ke, num, NULL, 0, KEVENT_NOWAIT)) < 0) {
- lderr(cct) << __func__ << " kevent: delete fd=" << fd << " mask=" << mask
- << " failed." << cpp_strerror(errno) << dendl;
- return -errno;
- }
- }
- // keep the administration
- sav_events[fd].mask = cur_mask & ~del_mask;
- return 0;
-}
-
-int KqueueDriver::resize_events(int newsize)
-{
- ldout(cct,30) << __func__ << " kqfd = " << kqfd << "newsize = " << newsize
- << dendl;
- if(newsize > sav_max) {
- sav_events = (struct SaveEvent*)realloc( sav_events,
- sizeof(struct SaveEvent)*newsize);
- if (!sav_events) {
- lderr(cct) << __func__ << " unable to realloc memory: "
- << cpp_strerror(errno) << dendl;
- return -ENOMEM;
- }
- memset(&sav_events[size], 0, sizeof(struct SaveEvent)*(newsize-sav_max));
- sav_max = newsize;
- }
- return 0;
-}
-
-int KqueueDriver::event_wait(vector<FiredFileEvent> &fired_events, struct timeval *tvp)
-{
- int retval, numevents = 0;
- struct timespec timeout;
-
- ldout(cct,10) << __func__ << " kqfd = " << kqfd << dendl;
-
- int r = test_thread_change(__func__);
- if ( r < 0 )
- return r;
-
- if (tvp != NULL) {
- timeout.tv_sec = tvp->tv_sec;
- timeout.tv_nsec = tvp->tv_usec * 1000;
- ldout(cct,20) << __func__ << " "
- << timeout.tv_sec << " sec "
- << timeout.tv_nsec << " nsec"
- << dendl;
- retval = kevent(kqfd, NULL, 0, res_events, size, &timeout);
- } else {
- ldout(cct,30) << __func__ << " event_wait: " << " NULL" << dendl;
- retval = kevent(kqfd, NULL, 0, res_events, size, KEVENT_NOWAIT);
- }
-
- ldout(cct,25) << __func__ << " kevent retval: " << retval << dendl;
- if (retval < 0) {
- lderr(cct) << __func__ << " kqueue error: "
- << cpp_strerror(errno) << dendl;
- return -errno;
- } else if (retval == 0) {
- ldout(cct,5) << __func__ << " Hit timeout("
- << timeout.tv_sec << " sec "
- << timeout.tv_nsec << " nsec"
- << ")." << dendl;
- } else {
- int j;
-
- numevents = retval;
- fired_events.resize(numevents);
- for (j = 0; j < numevents; j++) {
- int mask = 0;
- struct kevent *e = res_events + j;
-
- if (e->filter == EVFILT_READ) mask |= EVENT_READABLE;
- if (e->filter == EVFILT_WRITE) mask |= EVENT_WRITABLE;
- if (e->flags & EV_ERROR) mask |= EVENT_READABLE|EVENT_WRITABLE;
- fired_events[j].fd = (int)e->ident;
- fired_events[j].mask = mask;
-
- }
- }
- return numevents;
-}
Code Review / stor4nfv.git / blobdiff