/*
 * Dynomite - A thin, distributed replication layer for multi non-distributed
 * storages. Copyright (C) 2014 Netflix, Inc.
 */

/*
 * twemproxy - A fast and lightweight proxy for memcached protocol.
 * Copyright (C) 2011 Twitter, Inc.
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 * http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

#include <dyn_core.h>

#ifdef DN_HAVE_EPOLL

#include <dyn_event.h>
#include <sys/epoll.h>

struct event_base *event_base_create(int nevent, event_cb_t cb) {
  struct event_base *evb;
  int status, ep;
  struct epoll_event *event;

  ASSERT(nevent > 0);

  ep = epoll_create(nevent);
  if (ep < 0) {
    log_error("epoll create of size %d failed: %s", nevent, strerror(errno));
    return NULL;
  }

  event = dn_calloc(nevent, sizeof(*event));
  if (event == NULL) {
    status = close(ep);
    if (status < 0) {
      log_error("close e %d failed, ignored: %s", ep, strerror(errno));
    }
    return NULL;
  }

  evb = dn_alloc(sizeof(*evb));
  if (evb == NULL) {
    dn_free(event);
    status = close(ep);
    if (status < 0) {
      log_error("close e %d failed, ignored: %s", ep, strerror(errno));
    }
    return NULL;
  }

  evb->ep = ep;
  evb->event = event;
  evb->nevent = nevent;
  evb->cb = cb;

  log_debug(LOG_INFO, "e %d with nevent %d", evb->ep, evb->nevent);

  return evb;
}

void event_base_destroy(struct event_base *evb) {
  int status;

  if (evb == NULL) {
    return;
  }

  ASSERT(evb->ep >= 0);

  dn_free(evb->event);

  status = close(evb->ep);
  if (status < 0) {
    log_error("close e %d failed, ignored: %s", evb->ep, strerror(errno));
  }
  evb->ep = -1;

  dn_free(evb);
}

int event_add_in(struct event_base *evb, struct conn *c) {
  int status;
  struct epoll_event event;
  int ep = evb->ep;

  ASSERT(ep >= 0);
  ASSERT(c != NULL);
  ASSERT(c->sd > 0);

  if (c->recv_active) {
    return 0;
  }

  event.events = (uint32_t)(EPOLLIN);  // | EPOLLET);
  event.data.ptr = c;

  status = epoll_ctl(ep, EPOLL_CTL_MOD, c->sd, &event);
  if (status < 0) {
    log_error("epoll ctl on e %d sd %d failed: %s", ep, c->sd, strerror(errno));
  } else {
    c->recv_active = 1;
  }

  return status;
}

int event_del_in(struct event_base *evb, struct conn *c) { return 0; }

int event_add_out(struct event_base *evb, struct conn *c) {
  int status;
  struct epoll_event event;
  int ep = evb->ep;

  ASSERT(ep >= 0);
  ASSERT(c != NULL);
  ASSERT(c->sd > 0);
  ASSERT(c->recv_active);

  if (c->send_active) {
    return 0;
  }

  event.events = (uint32_t)(EPOLLIN | EPOLLOUT);  // | EPOLLET);
  event.data.ptr = c;

  log_debug(LOG_DEBUG, "adding conn %s to active", print_obj(c));
  status = epoll_ctl(ep, EPOLL_CTL_MOD, c->sd, &event);
  if (status < 0) {
    log_error("epoll ctl on e %d sd %d failed: %s", ep, c->sd, strerror(errno));
  } else {
    c->send_active = 1;
  }

  return status;
}

int event_del_out(struct event_base *evb, struct conn *c) {
  int status;
  struct epoll_event event;
  int ep = evb->ep;

  ASSERT(ep >= 0);
  ASSERT(c != NULL);
  ASSERT(c->sd > 0);
  ASSERT(c->recv_active);

  if (!c->send_active) {
    return 0;
  }

  event.events = (uint32_t)(EPOLLIN | EPOLLET);
  event.data.ptr = c;

  log_debug(LOG_DEBUG, "removing conn %s from active", print_obj(c));
  status = epoll_ctl(ep, EPOLL_CTL_MOD, c->sd, &event);
  if (status < 0) {
    log_error("epoll ctl on e %d sd %d failed: %s", ep, c->sd, strerror(errno));
  } else {
    c->send_active = 0;
  }

  return status;
}

int event_add_conn(struct event_base *evb, struct conn *c) {
  int status;
  struct epoll_event event;
  int ep = evb->ep;

  ASSERT(ep >= 0);
  ASSERT(c != NULL);
  ASSERT(c->sd > 0);

  event.events = (uint32_t)(EPOLLIN | EPOLLOUT | EPOLLET);
  event.data.ptr = c;

  log_debug(LOG_DEBUG, "adding conn %s to active", print_obj(c));
  status = epoll_ctl(ep, EPOLL_CTL_ADD, c->sd, &event);
  if (status < 0) {
    log_error("epoll ctl on e %d sd %d failed: %s", ep, c->sd, strerror(errno));
  } else {
    c->send_active = 1;
    c->recv_active = 1;
  }

  return status;
}

int event_del_conn(struct event_base *evb, struct conn *c) {
  int status;
  int ep = evb->ep;

  ASSERT(ep >= 0);
  ASSERT(c != NULL);
  ASSERT(c->sd > 0);

  log_debug(LOG_DEBUG, "removing conn %s from active", print_obj(c));
  status = epoll_ctl(ep, EPOLL_CTL_DEL, c->sd, NULL);
  if (status < 0) {
    log_error("epoll ctl on e %d sd %d failed: %s", ep, c->sd, strerror(errno));
  } else {
    c->recv_active = 0;
    c->send_active = 0;
  }

  return status;
}

int event_wait(struct event_base *evb, int timeout) {
  int ep = evb->ep;
  struct epoll_event *event = evb->event;
  int nevent = evb->nevent;

  ASSERT(ep >= 0);
  ASSERT(event != NULL);
  ASSERT(nevent > 0);

  for (;;) {
    int i, nsd;

    nsd = epoll_wait(ep, event, nevent, timeout);
    if (nsd > 0) {
      for (i = 0; i < nsd; i++) {
        struct epoll_event *ev = &evb->event[i];
        uint32_t events = 0;

        log_debug(LOG_VVVERB, "epoll %04" PRIX32 " triggered on conn %p",
                  ev->events, ev->data.ptr);

        if (ev->events & (EPOLLERR | EPOLLRDHUP)) {
          events |= EVENT_ERR;
        }

        if (ev->events & (EPOLLIN | EPOLLHUP)) {
          events |= EVENT_READ;
        }

        if (ev->events & EPOLLOUT) {
          events |= EVENT_WRITE;
        }

        if (evb->cb != NULL) {
          evb->cb(ev->data.ptr, events);
        }
      }
      return nsd;
    }

    if (nsd == 0) {
      if (timeout == -1) {
        log_error(
            "epoll wait on e %d with %d events and %d timeout "
            "returned no events",
            ep, nevent, timeout);
        return -1;
      }

      return 0;
    }

    if (errno == EINTR) {
      continue;
    }

    log_error("epoll wait on e %d with %d events failed: %s", ep, nevent,
              strerror(errno));
    return -1;
  }

  NOT_REACHED();
}

void event_loop_stats(event_stats_cb_t cb, void *arg) {
  struct stats *st = arg;
  int status, ep;
  struct epoll_event ev;

  ep = epoll_create(1);
  if (ep < 0) {
    log_error("epoll create failed: %s", strerror(errno));
    return;
  }

  ev.data.fd = st->sd;
  ev.events = EPOLLIN;

  status = epoll_ctl(ep, EPOLL_CTL_ADD, st->sd, &ev);
  if (status < 0) {
    log_error("epoll ctl on e %d sd %d failed: %s", ep, st->sd,
              strerror(errno));
    goto error;
  }

  for (;;) {
    int n;

    n = epoll_wait(ep, &ev, 1, st->interval);
    if (n < 0) {
      if (errno == EINTR) {
        continue;
      }
      log_error("epoll wait on e %d with m %d failed: %s", ep, st->sd,
                strerror(errno));
      break;
    }

    cb(st, &n);
  }

error:
  status = close(ep);
  if (status < 0) {
    log_error("close e %d failed, ignored: %s", ep, strerror(errno));
  }
  ep = -1;
}

void event_loop_entropy(event_entropy_cb_t cb, void *arg) {
  struct entropy *ent = arg;
  int status, ep;
  struct epoll_event ev;
  ent->interval = 30;

  ep = epoll_create(1);
  if (ep < 0) {
    log_error("entropy epoll create failed: %s", strerror(errno));
    return;
  }

  ev.data.fd = ent->sd;
  ev.events = EPOLLIN;

  status = epoll_ctl(ep, EPOLL_CTL_ADD, ent->sd, &ev);
  if (status < 0) {
    log_error("entropy epoll ctl on e %d sd %d failed: %s", ep, ent->sd,
              strerror(errno));
    goto error;
  }

  for (;;) {
    int n;

    n = epoll_wait(ep, &ev, 1, ent->interval);
    if (n < 0) {
      if (errno == EINTR) {
        continue;
      }
      log_error("entropy epoll wait on e %d with m %d failed: %s", ep, ent->sd,
                strerror(errno));
      break;
    }

    cb(ent, &n);
  }

error:
  status = close(ep);
  if (status < 0) {
    log_error("close e %d failed, ignored: %s", ep, strerror(errno));
  }
  ep = -1;
}

#endif /* DN_HAVE_EPOLL */
