<commit_before>#include "upoll.h"

#define uhash_slot(K,S) (((K) ^ (K >> 8)) & (S - 1))

static uhash_t* uhash_create(uint32_t size) {
  int i;
  size--;
  size |= size >> 1;
  size |= size >> 2;
  size |= size >> 4;
  size |= size >> 8;
  size |= size >> 16;
  size++;

  uhash_t* hash = (uhash_t*)calloc(1, sizeof(uhash_t) + size * sizeof(ulist_t));
  hash->count = 0;
  hash->size  = size;
  hash->items = (ulist_t*)(((char*)hash) + sizeof(uhash_t));

  for (i = 0; i < size; i++) {
    ulist_init(&hash->items[i]);
  }

  return hash;
}

static void* uhash_lookup(uhash_t* hash, intptr_t key) {
  uint32_t slot = uhash_slot(key, hash->size);
  ulist_t* q;
  ulist_scan(q, &hash->items[slot]) {
    uitem_t* i = ulist_data(q, uitem_t, list);
    if (i->key == key) return i->val;
  }
  return NULL;
}
static void uhash_insert(uhash_t* hash, intptr_t key, void* val) {
  uint32_t slot = uhash_slot(key, hash->size);

  uitem_t* item = (uitem_t*)calloc(1, sizeof(uitem_t));
  ulist_init(&item->list);
  item->key = key;
  item->val = val;

  ulist_append(&hash->items[slot], &item->list);
}
static int uhash_delete(uhash_t* hash, intptr_t key) {
  uint32_t slot = uhash_slot(key, hash->size);
  ulist_t* q;
  ulist_scan(q, &hash->items[slot]) {
    uitem_t* i = ulist_data(q, uitem_t, list);
    if (i->key == key) {
      ulist_remove(q);
      free(q);
      return 1;
    }
  }
  return 0;
}
static int uhash_destroy(uhash_t* hash) {
  int i;
  for (i = 0; i < hash->size; i++) {
    while (!ulist_empty(&hash->items[i])) {
      ulist_t* q = ulist_next(&hash->items[i]);
      uitem_t* n = ulist_data(q, uitem_t, list);
      ulist_remove(q);
      free(n);
    }
  }
  return 0;
}
static int upoll_note(unote_t* note, uint32_t hint) {
  note->events = hint;
  note->events &= (note->event.events | UPOLLERR);
  if (ulist_empty(&note->ready)) {
    ulist_remove(&note->queue);
    ulist_append(&note->upoll->ready, &note->ready);
  }
  return note->events;
}

upoll_t* upoll_create(uint32_t size) {
  assert(size > 0);
  upoll_t* upq = (upoll_t*)calloc(1, sizeof(upoll_t));

  ulist_init(&upq->alive);
  ulist_init(&upq->clean);
  ulist_init(&upq->ready);

  upq->table = uhash_create(size);
#if defined(HAVE_KQUEUE)
  upq->fd = kqueue();
  assert(upq->fd >= 0);
#elif defined(HAVE_EPOLL)
  upq->fd = epoll_create(1);
  assert(upq->fd >= 0);
#endif
  return upq;
}

void upoll_destroy(upoll_t* upq) {
  assert(upq != NULL);
  uhash_destroy(upq->table);
  ulist_t* q;
  unote_t* n;
  while (!ulist_empty(&upq->alive)) {
    q = ulist_next(&upq->alive);
    n = ulist_data(n, unote_t, queue);
    ulist_remove(q);
    free(n);
  }
  while (!ulist_empty(&upq->clean)) {
    q = ulist_next(&upq->clean);
    n = ulist_data(n, unote_t, queue);
    ulist_remove(q);
    free(n);
  }
#if defined(HAVE_KQUEUE) || defined(HAVE_EPOLL)
  close(upq->fd);
#endif
  free(upq);
}

#if defined(HAVE_KQUEUE)
int upoll_ctl_kqueue(upoll_t* upq, int op, unote_t* note, upoll_event_t* event) {

  struct timespec ts;
  ts.tv_sec  = 0;
  ts.tv_nsec = 0;

  struct kevent kch[4];
  int nch = 0, rc = 0, i = 0;
  int flags = EV_RECEIPT;
  switch (op) {
    case UPOLL_CTL_ADD: {
      flags |= (EV_ADD | EV_ENABLE);
      if (event->events & UPOLLET) flags |= EV_CLEAR;
      if (event->events & UPOLLIN) {
        EV_SET(&kch[nch++], note->fd, EVFILT_READ, flags, 0, 0, note);
      }
      if (event->events & UPOLLOUT) {
        EV_SET(&kch[nch++], note->fd, EVFILT_WRITE, flags, 0, 0, note);
      }
    }
    case UPOLL_CTL_DEL: {
      EV_SET(&kch[nch++], note->fd, EVFILT_READ, EV_DELETE, 0, 0, note);
      EV_SET(&kch[nch++], note->fd, EVFILT_WRITE, EV_DELETE, 0, 0, note);
    }
    case UPOLL_CTL_MOD: {
      if (event->events & UPOLLIN) {
        flags |= (EV_ADD | EV_ENABLE);
        if (event->events & UPOLLET) flags |= EV_CLEAR;
        EV_SET(&kch[nch++], note->fd, EVFILT_READ, flags, 0, 0, note);
      }
      else {
        flags |= (EV_ADD | EV_DISABLE);
        EV_SET(&kch[nch++], note->fd, EVFILT_READ, flags, 0, 0, note);
      }
      flags = EV_RECEIPT;
      if (event->events & UPOLLOUT) {
        flags |= (EV_ADD | EV_ENABLE);
        if (event->events & UPOLLET) flags |= EV_CLEAR;
        EV_SET(&kch[nch++], note->fd, EVFILT_WRITE, flags, 0, 0, note);
      }
      else {
        flags |= (EV_ADD | EV_DISABLE);
        EV_SET(&kch[nch++], note->fd, EVFILT_WRITE, flags, 0, 0, note);
      }
    }
  }
  rc = kevent(upq->fd, kch, nch, kch, nch, &ts);
  if (rc < 0) return -errno;
  for (i = 0; i < rc; i++) {
    if (kch[i].data) return -kch[i].data;
  }
  return 0;
}
#elif defined(HAVE_EPOLL)
int upoll_ctl_epoll(upoll_t* upq, int op, unote_t* note, upoll_event_t* event) {
  assert(upq->fd >= 0);
  struct epoll_event evt;

  evt.events = 0;
  evt.data.ptr = note;

  if (event->events & UPOLLIN) {
    evt.events |= EPOLLIN;
  }
  if (event->events & UPOLLOUT) {
    evt.events |= EPOLLOUT;
  }
  if (event->events & UPOLLET) {
    evt.events |= EPOLLET;
  }

  int rc = 0;
  switch (op) {
    case UPOLL_CTL_ADD: {
      rc = epoll_ctl(upq->fd, EPOLL_CTL_ADD, note->fd, &evt);
      break;
    }
    case UPOLL_CTL_DEL: {
      rc = epoll_ctl(upq->fd, EPOLL_CTL_DEL, note->fd, &evt);
      break;
    }
    case UPOLL_CTL_MOD: {
      rc = epoll_ctl(upq->fd, EPOLL_CTL_MOD, note->fd, &evt);
      break;
    }
  }
  if (rc < 0) return -errno;
  return rc;
}
#endif

int upoll_ctl(upoll_t* upq, int op, intptr_t fd, upoll_event_t* event) {
  if (fd < 0) return EBADF;

  unote_t* note = NULL;
  switch (op) {
    case UPOLL_CTL_ADD: {
      note = (unote_t*)uhash_lookup(upq->table, fd);
      if (!note) {
        if (ulist_empty(&upq->clean)) {
          note = (unote_t*)calloc(1, sizeof(unote_t));
          ulist_init(&note->queue);
          ulist_init(&note->ready);
        }
        else {
          ulist_t* item = ulist_next(&upq->clean);
          ulist_remove(item);
          note = ulist_data(item, unote_t, queue);
        }
        note->upoll = upq;
        note->event = *event;
        note->events = 0;

        note->fd = fd;

        ulist_append(&upq->alive, &note->queue);
        uhash_insert(upq->table, fd, (void*)note);
      }
      break;
    }
    case UPOLL_CTL_DEL: {
      note = (unote_t*)uhash_lookup(upq->table, fd);
      if (!note) return -ENOENT;
      event = &note->event;
      ulist_remove(&note->ready);
      ulist_remove(&note->queue);
      ulist_append(&upq->clean, &note->queue);
      uhash_delete(upq->table, fd);
      break;
    }
    case UPOLL_CTL_MOD: {
      note = (unote_t*)uhash_lookup(upq->table, fd);
      if (!note) return -ENOENT;
      note->event = *event;
      note->events = 0;
      ulist_remove(&note->ready);
      ulist_remove(&note->queue);
      ulist_append(&upq->alive, &note->queue);
      break;
    }
    default: {
      return -EINVAL;
    }
  }
#if defined(HAVE_KQUEUE)
  return upoll_ctl_kqueue(upq, op, note, event);
#elif defined(HAVE_EPOLL)
  return upoll_ctl_epoll(upq, op, note, event);
#else
  return 0;
#endif
}

#if defined(HAVE_KQUEUE)
int upoll_wait_kqueue(upoll_t* upq, int nev, int timeout) {
  int i, n;
  int nkev = nev * 2;
  struct timespec  ts;
  struct timespec* tsp = &ts;

  struct kevent kevs[nkev];

  ts.tv_nsec = 0;
  if (timeout < 0) {
    tsp = NULL;
  }
  else if (timeout == 0)  
    ts.tv_sec = 0;
  else {
    ts.tv_sec  = (timeout / 1000);
    ts.tv_nsec = (timeout % 1000) * 1000000;
  }

  n = kevent(upq->fd, NULL, 0, kevs, nkev, tsp);
  if (n < 0) return -errno;

  for (i = 0; i < n; i++) {
    uint32_t hint = 0;
    unote_t* n = (unote_t*)kevs[i].udata;
    if (kevs[i].filter == EVFILT_READ) {
      hint = UPOLLIN;
    }
    else if (kevs[i].filter == EVFILT_WRITE) {
      hint = UPOLLOUT;
    }
    if (kevs[i].flags & EV_ERROR) {
      hint &= ~UPOLLOUT;
      hint |= UPOLLERR;
    }
    upoll_note(n, hint);
  }

  return n;
}
#elif defined(HAVE_EPOLL)
int upoll_wait_epoll(upoll_t* upq, int nev, int timeout) {
  struct epoll_event evts[nev];
  int i, n;
  n = epoll_wait(upq->fd, evts, nev, timeout);
  for (i = 0; i < n; i++) {
    uint32_t hint;
    unote_t* n = (unote_t*)evts[i].data.ptr;
    if (evts[i].events & EPOLLIN) hint |= UPOLLIN;
    if (evts[i].events & EPOLLOUT) hint |= UPOLLOUT;
    if (evts[i].events & (EPOLLERR|EPOLLHUP)) hint |= UPOLLERR;
    upoll_note(n, hint);
  }
  return n;
}
#elif defined(HAVE_POLL)
int upoll_wait_poll(upoll_t* upq, int nev, int timeout) {
  /* FD_SETSIZE should be smaller than OPEN_MAX, but OPEN_MAX isn't portable */
  if (nev > FD_SETSIZE) nev = FD_SETSIZE;

  unote_t*  nvec[nev];
  int r, i, nfds = 0;
  struct pollfd pfds[nev];

  unote_t* n = NULL;
  ulist_t* s = ulist_mark(&upq->alive);
  ulist_t* q = ulist_next(&upq->alive);

  while (q != s && nfds < nev) {
    n = ulist_data(q, unote_t, queue);
    q = ulist_next(q);

    ulist_remove(&n->queue);
    ulist_insert(&upq->alive, &n->queue);

    assert(ulist_empty(&n->ready));

    nvec[nfds] = n;
    pfds[nfds].events = 0;
    pfds[nfds].fd = n->fd;
    if (n->event.events & UPOLLIN) {
      pfds[nfds].events |= POLLIN;
    }
    if (n->event.events & UPOLLOUT) {
      pfds[nfds].events |= POLLOUT;
    }
    nfds++;
  }

  r = poll(pfds, nfds, timeout);
  if (r > 0) {
    uint32_t hint = 0;
    for (i = 0; i < nfds; i++) {
      hint = 0;
      if (pfds[i].revents) {
        if (pfds[i].revents & POLLIN ) hint |= UPOLLIN;
        if (pfds[i].revents & POLLOUT) hint |= UPOLLOUT;
        if (pfds[i].revents & (POLLERR|POLLNVAL)) hint |= UPOLLERR;
        upoll_note(nvec[i], hint);
      }
    }
  }

  return r;
}
#else
int upoll_wait_select(upoll_t* upq, int nev, int timeout) {
  if (nev > FD_SETSIZE) nev = FD_SETSIZE;

  unote_t* nvec[nev];
  int i, maxfd = 0, r = 0, nfds = 0;

  fd_set pollin, pollout, pollerr;

  FD_ZERO(&pollin);
  FD_ZERO(&pollout);
  FD_ZERO(&pollerr);

  struct timeval  tv;
  struct timeval* tvp = &tv;

  tv.tv_usec = 0;
  if (timeout < 0) {
    tvp = NULL;
  }
  else if (timeout == 0)  
    tv.tv_sec = 0;
  else {
    tv.tv_sec  = (timeout / 1000);
    tv.tv_usec = (timeout % 1000) * 1000;
  }

  unote_t* n = NULL;
  ulist_t* s = ulist_mark(&upq->alive);
  ulist_t* q = ulist_next(&upq->alive);

  while (q != s && nfds < nev) {
    n = ulist_data(q, unote_t, queue);
    q = ulist_next(q);

    ulist_remove(&n->queue);
    ulist_insert(&upq->alive, &n->queue);

    assert(ulist_empty(&n->ready));

    nvec[nfds] = n;
    if (n->event.events & UPOLLIN) {
      FD_SET(n->fd, &pollin);
    }
    if (n->event.events & UPOLLOUT) {
      FD_SET(n->fd, &pollout);
    }
    FD_SET(n->fd, &pollerr);
    if (maxfd < n->fd) maxfd = n->fd;
    nfds++;
  }

# if defined(__WINDOWS__)
  int rc = select(0, &pollin, &pollout, &pollerr, tvp);
  if (rc == SOCKET_ERROR) {
    assert(WSAGetLastError() == WSAENOTSOCK);
    return -1;
  }
# else
  int rc = select(maxfd + 1, &pollin, &pollout, &pollerr, tvp);
  if (rc == -1) {
    assert(errno == EINTR || errno == EBADF);
    return -1;
  }
# endif
  r = 0;
  for (i = 0; i < nfds; i++) {
    uint32_t hint = 0;
    if (FD_ISSET(nvec[i]->fd, &pollin)) {
      hint |= UPOLLIN;
    }

    if (FD_ISSET(nvec[i]->fd, &pollerr)) {
      hint |= UPOLLERR;
    }
    else if (FD_ISSET(nvec[i]->fd, &pollout)) {
      hint |= UPOLLOUT;
    }

    if (hint) {
      upoll_note(nvec[i], hint);
      r++;
    }
  }
  return r;
}
#endif

int upoll_wait(upoll_t* upq, upoll_event_t *evs, int nev, int timeout) {
  int e = 0;

  unote_t* n = NULL;
  ulist_t* s;
  ulist_t* q;

  if (ulist_empty(&upq->ready)) {
#if defined(HAVE_KQUEUE)
    upoll_wait_kqueue(upq, nev, timeout);
#elif defined(HAVE_EPOLL)
    upoll_wait_epoll(upq, nev, timeout);
#elif defined(HAVE_POLL)
    upoll_wait_poll(upq, nev, timeout);
#else
    upoll_wait_select(upq, nev, timeout);
#endif
  }

  s = ulist_mark(&upq->ready);
  q = ulist_next(&upq->ready);

  while (q != s && e < nev) {
    n = ulist_data(q, unote_t, ready);
    q = ulist_next(q);
    ulist_remove(&n->ready);
    ulist_remove(&n->queue);
    evs[e].events = n->events;
    evs[e].data   = n->event.data;
    ulist_append(&upq->alive, &n->queue);
    n->events = 0;
    e++;
  }

  return e;
}

intptr_t usocket(int domain, int type, int proto) {
  intptr_t fd = (intptr_t)socket(domain, type, proto);

#if defined(__WINDOWS__)
  if (fd < 0) return -WSAGetLastError();
  unsigned long flags = 1;
  int rc = ioctlsocket((SOCKET)fd, FIONBIO, &flags);
  if (rc < 0) return -WSAGetLastError();
#else
  if (fd < 0) return -errno;
  int rc = fcntl(fd, F_SETFL, fcntl(fd, F_GETFL, 0) | O_NONBLOCK);
  if (rc < 0) return -errno;
#endif

  return fd;
}

int ubind(intptr_t fd, const char* host, const char* serv) {
  struct addrinfo* info;
  struct addrinfo  hint;
  memset(&hint, 0, sizeof(hint));

  int optval = 0;
  unsigned int optlen = sizeof(optval);
#if defined(__WINDOWS__)
  int rc = getsockopt((SOCKET)fd, SOL_SOCKET, SO_TYPE, (char*)&optval, &optlen);
#else
  int rc = getsockopt(fd, SOL_SOCKET, SO_TYPE, &optval, &optlen);
#endif

  hint.ai_family   = AF_INET;
  hint.ai_socktype = optval;

  rc = getaddrinfo(host, serv, &hint, &info);

  optval = 1;
  if (!rc) {
#if defined(__WINDOWS__)
    rc = setsockopt(
      (SOCKET)fd, SOL_SOCKET, SO_REUSEADDR, (char*)&optval, sizeof(optval)
    );
#else
    rc = setsockopt(fd, SOL_SOCKET, SO_REUSEADDR, &optval, sizeof(optval));
#endif
    if (!rc) rc = bind(fd, info->ai_addr, info->ai_addrlen);
  }

  freeaddrinfo(info);
  if (rc) {
#if defined(__WINDOWS__)
    return WSAGetLastError();
#else
    return errno;
#endif
  }
  return 0;
}

int uconnect(intptr_t fd, const char* host, const char* serv) {
  struct addrinfo* info;

  struct addrinfo  hint;
  memset(&hint, 0, sizeof(hint));

  int optval = 0;
  unsigned int optlen;

#if defined(__WINDOWS__)
  int rc = getsockopt(fd, SOL_SOCKET, SO_TYPE, (char*)&optval, &optlen);
#else
  int rc = getsockopt(fd, SOL_SOCKET, SO_TYPE, &optval, &optlen);
#endif

  hint.ai_family   = AF_INET;
  hint.ai_socktype = optval;

  rc = getaddrinfo(host, serv, &hint, &info);

  if (!rc) {
#if defined(__WINDOWS__)
    rc = connect((SOCKET)fd, info->ai_addr, info->ai_addrlen);
#else
    rc = connect(fd, info->ai_addr, info->ai_addrlen);
#endif
  }

  freeaddrinfo(info);

  if (rc) {
#if defined(__WINDOWS__)
    if (WSAGetLastError() != WSAEWOULDBLOCK) return WSAGetLastError();
#else
    if (errno != EINPROGRESS) return errno;
#endif
  }
  return 0;
}

int ulisten(intptr_t sock, int backlog) {
  return listen(sock, backlog);
}

intptr_t uaccept(intptr_t sock) {
  struct sockaddr addr;

  addr.sa_family = AF_INET;
  socklen_t addr_len;

#if defined(__WINDOWS__)
  intptr_t fd = (intptr_t)accept((SOCKET)sock, &addr, &addr_len);
  if (fd == INVALID_SOCKET) return WSAGetLastError();
#else
  intptr_t fd = accept(sock, &addr, &addr_len);
  if (fd < 0) return errno;
#endif

  return fd;
}

int uclose(intptr_t sock) {
#if defined(__WINDOWS__)
  return closesocket((SOCKET)sock);
#else
  return close(sock);
#endif
}

int uread(intptr_t fd, char* buf, size_t len) {
  return recv(fd, buf, len, 0);
}
int uwrite(intptr_t fd, const char* buf, size_t len) {
  return send(fd, buf, len, 0);
}

<commit_msg>Revert "unconditionally remove read/write filters from kqueue"<commit_after>#include "upoll.h"

#define uhash_slot(K,S) (((K) ^ (K >> 8)) & (S - 1))

static uhash_t* uhash_create(uint32_t size) {
  int i;
  size--;
  size |= size >> 1;
  size |= size >> 2;
  size |= size >> 4;
  size |= size >> 8;
  size |= size >> 16;
  size++;

  uhash_t* hash = (uhash_t*)calloc(1, sizeof(uhash_t) + size * sizeof(ulist_t));
  hash->count = 0;
  hash->size  = size;
  hash->items = (ulist_t*)(((char*)hash) + sizeof(uhash_t));

  for (i = 0; i < size; i++) {
    ulist_init(&hash->items[i]);
  }

  return hash;
}

static void* uhash_lookup(uhash_t* hash, intptr_t key) {
  uint32_t slot = uhash_slot(key, hash->size);
  ulist_t* q;
  ulist_scan(q, &hash->items[slot]) {
    uitem_t* i = ulist_data(q, uitem_t, list);
    if (i->key == key) return i->val;
  }
  return NULL;
}
static void uhash_insert(uhash_t* hash, intptr_t key, void* val) {
  uint32_t slot = uhash_slot(key, hash->size);

  uitem_t* item = (uitem_t*)calloc(1, sizeof(uitem_t));
  ulist_init(&item->list);
  item->key = key;
  item->val = val;

  ulist_append(&hash->items[slot], &item->list);
}
static int uhash_delete(uhash_t* hash, intptr_t key) {
  uint32_t slot = uhash_slot(key, hash->size);
  ulist_t* q;
  ulist_scan(q, &hash->items[slot]) {
    uitem_t* i = ulist_data(q, uitem_t, list);
    if (i->key == key) {
      ulist_remove(q);
      free(q);
      return 1;
    }
  }
  return 0;
}
static int uhash_destroy(uhash_t* hash) {
  int i;
  for (i = 0; i < hash->size; i++) {
    while (!ulist_empty(&hash->items[i])) {
      ulist_t* q = ulist_next(&hash->items[i]);
      uitem_t* n = ulist_data(q, uitem_t, list);
      ulist_remove(q);
      free(n);
    }
  }
  return 0;
}
static int upoll_note(unote_t* note, uint32_t hint) {
  note->events = hint;
  note->events &= (note->event.events | UPOLLERR);
  if (ulist_empty(&note->ready)) {
    ulist_remove(&note->queue);
    ulist_append(&note->upoll->ready, &note->ready);
  }
  return note->events;
}

upoll_t* upoll_create(uint32_t size) {
  assert(size > 0);
  upoll_t* upq = (upoll_t*)calloc(1, sizeof(upoll_t));

  ulist_init(&upq->alive);
  ulist_init(&upq->clean);
  ulist_init(&upq->ready);

  upq->table = uhash_create(size);
#if defined(HAVE_KQUEUE)
  upq->fd = kqueue();
  assert(upq->fd >= 0);
#elif defined(HAVE_EPOLL)
  upq->fd = epoll_create(1);
  assert(upq->fd >= 0);
#endif
  return upq;
}

void upoll_destroy(upoll_t* upq) {
  assert(upq != NULL);
  uhash_destroy(upq->table);
  ulist_t* q;
  unote_t* n;
  while (!ulist_empty(&upq->alive)) {
    q = ulist_next(&upq->alive);
    n = ulist_data(n, unote_t, queue);
    ulist_remove(q);
    free(n);
  }
  while (!ulist_empty(&upq->clean)) {
    q = ulist_next(&upq->clean);
    n = ulist_data(n, unote_t, queue);
    ulist_remove(q);
    free(n);
  }
#if defined(HAVE_KQUEUE) || defined(HAVE_EPOLL)
  close(upq->fd);
#endif
  free(upq);
}

#if defined(HAVE_KQUEUE)
int upoll_ctl_kqueue(upoll_t* upq, int op, unote_t* note, upoll_event_t* event) {

  struct timespec ts;
  ts.tv_sec  = 0;
  ts.tv_nsec = 0;

  struct kevent kch[4];
  int nch = 0, rc = 0, i = 0;
  int flags = EV_RECEIPT;
  switch (op) {
    case UPOLL_CTL_ADD: {
      flags |= (EV_ADD | EV_ENABLE);
      if (event->events & UPOLLET) flags |= EV_CLEAR;
      if (event->events & UPOLLIN) {
        EV_SET(&kch[nch++], note->fd, EVFILT_READ, flags, 0, 0, note);
      }
      if (event->events & UPOLLOUT) {
        EV_SET(&kch[nch++], note->fd, EVFILT_WRITE, flags, 0, 0, note);
      }
    }
    case UPOLL_CTL_DEL: {
      if (event->events & UPOLLIN) {
        EV_SET(&kch[nch++], note->fd, EVFILT_READ, EV_DELETE, 0, 0, note);
      }
      if (event->events & UPOLLOUT) {
        EV_SET(&kch[nch++], note->fd, EVFILT_WRITE, EV_DELETE, 0, 0, note);
      }
    }
    case UPOLL_CTL_MOD: {
      if (event->events & UPOLLIN) {
        flags |= (EV_ADD | EV_ENABLE);
        if (event->events & UPOLLET) flags |= EV_CLEAR;
        EV_SET(&kch[nch++], note->fd, EVFILT_READ, flags, 0, 0, note);
      }
      else {
        flags |= (EV_ADD | EV_DISABLE);
        EV_SET(&kch[nch++], note->fd, EVFILT_READ, flags, 0, 0, note);
      }
      flags = EV_RECEIPT;
      if (event->events & UPOLLOUT) {
        flags |= (EV_ADD | EV_ENABLE);
        if (event->events & UPOLLET) flags |= EV_CLEAR;
        EV_SET(&kch[nch++], note->fd, EVFILT_WRITE, flags, 0, 0, note);
      }
      else {
        flags |= (EV_ADD | EV_DISABLE);
        EV_SET(&kch[nch++], note->fd, EVFILT_WRITE, flags, 0, 0, note);
      }
    }
  }
  rc = kevent(upq->fd, kch, nch, kch, nch, &ts);
  if (rc < 0) return -errno;
  for (i = 0; i < rc; i++) {
    if (kch[i].data) return -kch[i].data;
  }
  return 0;
}
#elif defined(HAVE_EPOLL)
int upoll_ctl_epoll(upoll_t* upq, int op, unote_t* note, upoll_event_t* event) {
  assert(upq->fd >= 0);
  struct epoll_event evt;

  evt.events = 0;
  evt.data.ptr = note;

  if (event->events & UPOLLIN) {
    evt.events |= EPOLLIN;
  }
  if (event->events & UPOLLOUT) {
    evt.events |= EPOLLOUT;
  }
  if (event->events & UPOLLET) {
    evt.events |= EPOLLET;
  }

  int rc = 0;
  switch (op) {
    case UPOLL_CTL_ADD: {
      rc = epoll_ctl(upq->fd, EPOLL_CTL_ADD, note->fd, &evt);
      break;
    }
    case UPOLL_CTL_DEL: {
      rc = epoll_ctl(upq->fd, EPOLL_CTL_DEL, note->fd, &evt);
      break;
    }
    case UPOLL_CTL_MOD: {
      rc = epoll_ctl(upq->fd, EPOLL_CTL_MOD, note->fd, &evt);
      break;
    }
  }
  if (rc < 0) return -errno;
  return rc;
}
#endif

int upoll_ctl(upoll_t* upq, int op, intptr_t fd, upoll_event_t* event) {
  if (fd < 0) return EBADF;

  unote_t* note = NULL;
  switch (op) {
    case UPOLL_CTL_ADD: {
      note = (unote_t*)uhash_lookup(upq->table, fd);
      if (!note) {
        if (ulist_empty(&upq->clean)) {
          note = (unote_t*)calloc(1, sizeof(unote_t));
          ulist_init(&note->queue);
          ulist_init(&note->ready);
        }
        else {
          ulist_t* item = ulist_next(&upq->clean);
          ulist_remove(item);
          note = ulist_data(item, unote_t, queue);
        }
        note->upoll = upq;
        note->event = *event;
        note->events = 0;

        note->fd = fd;

        ulist_append(&upq->alive, &note->queue);
        uhash_insert(upq->table, fd, (void*)note);
      }
      break;
    }
    case UPOLL_CTL_DEL: {
      note = (unote_t*)uhash_lookup(upq->table, fd);
      if (!note) return -ENOENT;
      event = &note->event;
      ulist_remove(&note->ready);
      ulist_remove(&note->queue);
      ulist_append(&upq->clean, &note->queue);
      uhash_delete(upq->table, fd);
      break;
    }
    case UPOLL_CTL_MOD: {
      note = (unote_t*)uhash_lookup(upq->table, fd);
      if (!note) return -ENOENT;
      note->event = *event;
      note->events = 0;
      ulist_remove(&note->ready);
      ulist_remove(&note->queue);
      ulist_append(&upq->alive, &note->queue);
      break;
    }
    default: {
      return -EINVAL;
    }
  }
#if defined(HAVE_KQUEUE)
  return upoll_ctl_kqueue(upq, op, note, event);
#elif defined(HAVE_EPOLL)
  return upoll_ctl_epoll(upq, op, note, event);
#else
  return 0;
#endif
}

#if defined(HAVE_KQUEUE)
int upoll_wait_kqueue(upoll_t* upq, int nev, int timeout) {
  int i, n;
  int nkev = nev * 2;
  struct timespec  ts;
  struct timespec* tsp = &ts;

  struct kevent kevs[nkev];

  ts.tv_nsec = 0;
  if (timeout < 0) {
    tsp = NULL;
  }
  else if (timeout == 0)  
    ts.tv_sec = 0;
  else {
    ts.tv_sec  = (timeout / 1000);
    ts.tv_nsec = (timeout % 1000) * 1000000;
  }

  n = kevent(upq->fd, NULL, 0, kevs, nkev, tsp);
  if (n < 0) return -errno;

  for (i = 0; i < n; i++) {
    uint32_t hint = 0;
    unote_t* n = (unote_t*)kevs[i].udata;
    if (kevs[i].filter == EVFILT_READ) {
      hint = UPOLLIN;
    }
    else if (kevs[i].filter == EVFILT_WRITE) {
      hint = UPOLLOUT;
    }
    if (kevs[i].flags & EV_ERROR) {
      hint &= ~UPOLLOUT;
      hint |= UPOLLERR;
    }
    upoll_note(n, hint);
  }

  return n;
}
#elif defined(HAVE_EPOLL)
int upoll_wait_epoll(upoll_t* upq, int nev, int timeout) {
  struct epoll_event evts[nev];
  int i, n;
  n = epoll_wait(upq->fd, evts, nev, timeout);
  for (i = 0; i < n; i++) {
    uint32_t hint;
    unote_t* n = (unote_t*)evts[i].data.ptr;
    if (evts[i].events & EPOLLIN) hint |= UPOLLIN;
    if (evts[i].events & EPOLLOUT) hint |= UPOLLOUT;
    if (evts[i].events & (EPOLLERR|EPOLLHUP)) hint |= UPOLLERR;
    upoll_note(n, hint);
  }
  return n;
}
#elif defined(HAVE_POLL)
int upoll_wait_poll(upoll_t* upq, int nev, int timeout) {
  /* FD_SETSIZE should be smaller than OPEN_MAX, but OPEN_MAX isn't portable */
  if (nev > FD_SETSIZE) nev = FD_SETSIZE;

  unote_t*  nvec[nev];
  int r, i, nfds = 0;
  struct pollfd pfds[nev];

  unote_t* n = NULL;
  ulist_t* s = ulist_mark(&upq->alive);
  ulist_t* q = ulist_next(&upq->alive);

  while (q != s && nfds < nev) {
    n = ulist_data(q, unote_t, queue);
    q = ulist_next(q);

    ulist_remove(&n->queue);
    ulist_insert(&upq->alive, &n->queue);

    assert(ulist_empty(&n->ready));

    nvec[nfds] = n;
    pfds[nfds].events = 0;
    pfds[nfds].fd = n->fd;
    if (n->event.events & UPOLLIN) {
      pfds[nfds].events |= POLLIN;
    }
    if (n->event.events & UPOLLOUT) {
      pfds[nfds].events |= POLLOUT;
    }
    nfds++;
  }

  r = poll(pfds, nfds, timeout);
  if (r > 0) {
    uint32_t hint = 0;
    for (i = 0; i < nfds; i++) {
      hint = 0;
      if (pfds[i].revents) {
        if (pfds[i].revents & POLLIN ) hint |= UPOLLIN;
        if (pfds[i].revents & POLLOUT) hint |= UPOLLOUT;
        if (pfds[i].revents & (POLLERR|POLLNVAL)) hint |= UPOLLERR;
        upoll_note(nvec[i], hint);
      }
    }
  }

  return r;
}
#else
int upoll_wait_select(upoll_t* upq, int nev, int timeout) {
  if (nev > FD_SETSIZE) nev = FD_SETSIZE;

  unote_t* nvec[nev];
  int i, maxfd = 0, r = 0, nfds = 0;

  fd_set pollin, pollout, pollerr;

  FD_ZERO(&pollin);
  FD_ZERO(&pollout);
  FD_ZERO(&pollerr);

  struct timeval  tv;
  struct timeval* tvp = &tv;

  tv.tv_usec = 0;
  if (timeout < 0) {
    tvp = NULL;
  }
  else if (timeout == 0)  
    tv.tv_sec = 0;
  else {
    tv.tv_sec  = (timeout / 1000);
    tv.tv_usec = (timeout % 1000) * 1000;
  }

  unote_t* n = NULL;
  ulist_t* s = ulist_mark(&upq->alive);
  ulist_t* q = ulist_next(&upq->alive);

  while (q != s && nfds < nev) {
    n = ulist_data(q, unote_t, queue);
    q = ulist_next(q);

    ulist_remove(&n->queue);
    ulist_insert(&upq->alive, &n->queue);

    assert(ulist_empty(&n->ready));

    nvec[nfds] = n;
    if (n->event.events & UPOLLIN) {
      FD_SET(n->fd, &pollin);
    }
    if (n->event.events & UPOLLOUT) {
      FD_SET(n->fd, &pollout);
    }
    FD_SET(n->fd, &pollerr);
    if (maxfd < n->fd) maxfd = n->fd;
    nfds++;
  }

# if defined(__WINDOWS__)
  int rc = select(0, &pollin, &pollout, &pollerr, tvp);
  if (rc == SOCKET_ERROR) {
    assert(WSAGetLastError() == WSAENOTSOCK);
    return -1;
  }
# else
  int rc = select(maxfd + 1, &pollin, &pollout, &pollerr, tvp);
  if (rc == -1) {
    assert(errno == EINTR || errno == EBADF);
    return -1;
  }
# endif
  r = 0;
  for (i = 0; i < nfds; i++) {
    uint32_t hint = 0;
    if (FD_ISSET(nvec[i]->fd, &pollin)) {
      hint |= UPOLLIN;
    }

    if (FD_ISSET(nvec[i]->fd, &pollerr)) {
      hint |= UPOLLERR;
    }
    else if (FD_ISSET(nvec[i]->fd, &pollout)) {
      hint |= UPOLLOUT;
    }

    if (hint) {
      upoll_note(nvec[i], hint);
      r++;
    }
  }
  return r;
}
#endif

int upoll_wait(upoll_t* upq, upoll_event_t *evs, int nev, int timeout) {
  int e = 0;

  unote_t* n = NULL;
  ulist_t* s;
  ulist_t* q;

  if (ulist_empty(&upq->ready)) {
#if defined(HAVE_KQUEUE)
    upoll_wait_kqueue(upq, nev, timeout);
#elif defined(HAVE_EPOLL)
    upoll_wait_epoll(upq, nev, timeout);
#elif defined(HAVE_POLL)
    upoll_wait_poll(upq, nev, timeout);
#else
    upoll_wait_select(upq, nev, timeout);
#endif
  }

  s = ulist_mark(&upq->ready);
  q = ulist_next(&upq->ready);

  while (q != s && e < nev) {
    n = ulist_data(q, unote_t, ready);
    q = ulist_next(q);
    ulist_remove(&n->ready);
    ulist_remove(&n->queue);
    evs[e].events = n->events;
    evs[e].data   = n->event.data;
    ulist_append(&upq->alive, &n->queue);
    n->events = 0;
    e++;
  }

  return e;
}

intptr_t usocket(int domain, int type, int proto) {
  intptr_t fd = (intptr_t)socket(domain, type, proto);

#if defined(__WINDOWS__)
  if (fd < 0) return -WSAGetLastError();
  unsigned long flags = 1;
  int rc = ioctlsocket((SOCKET)fd, FIONBIO, &flags);
  if (rc < 0) return -WSAGetLastError();
#else
  if (fd < 0) return -errno;
  int rc = fcntl(fd, F_SETFL, fcntl(fd, F_GETFL, 0) | O_NONBLOCK);
  if (rc < 0) return -errno;
#endif

  return fd;
}

int ubind(intptr_t fd, const char* host, const char* serv) {
  struct addrinfo* info;
  struct addrinfo  hint;
  memset(&hint, 0, sizeof(hint));

  int optval = 0;
  unsigned int optlen = sizeof(optval);
#if defined(__WINDOWS__)
  int rc = getsockopt((SOCKET)fd, SOL_SOCKET, SO_TYPE, (char*)&optval, &optlen);
#else
  int rc = getsockopt(fd, SOL_SOCKET, SO_TYPE, &optval, &optlen);
#endif

  hint.ai_family   = AF_INET;
  hint.ai_socktype = optval;

  rc = getaddrinfo(host, serv, &hint, &info);

  optval = 1;
  if (!rc) {
#if defined(__WINDOWS__)
    rc = setsockopt(
      (SOCKET)fd, SOL_SOCKET, SO_REUSEADDR, (char*)&optval, sizeof(optval)
    );
#else
    rc = setsockopt(fd, SOL_SOCKET, SO_REUSEADDR, &optval, sizeof(optval));
#endif
    if (!rc) rc = bind(fd, info->ai_addr, info->ai_addrlen);
  }

  freeaddrinfo(info);
  if (rc) {
#if defined(__WINDOWS__)
    return WSAGetLastError();
#else
    return errno;
#endif
  }
  return 0;
}

int uconnect(intptr_t fd, const char* host, const char* serv) {
  struct addrinfo* info;

  struct addrinfo  hint;
  memset(&hint, 0, sizeof(hint));

  int optval = 0;
  unsigned int optlen;

#if defined(__WINDOWS__)
  int rc = getsockopt(fd, SOL_SOCKET, SO_TYPE, (char*)&optval, &optlen);
#else
  int rc = getsockopt(fd, SOL_SOCKET, SO_TYPE, &optval, &optlen);
#endif

  hint.ai_family   = AF_INET;
  hint.ai_socktype = optval;

  rc = getaddrinfo(host, serv, &hint, &info);

  if (!rc) {
#if defined(__WINDOWS__)
    rc = connect((SOCKET)fd, info->ai_addr, info->ai_addrlen);
#else
    rc = connect(fd, info->ai_addr, info->ai_addrlen);
#endif
  }

  freeaddrinfo(info);

  if (rc) {
#if defined(__WINDOWS__)
    if (WSAGetLastError() != WSAEWOULDBLOCK) return WSAGetLastError();
#else
    if (errno != EINPROGRESS) return errno;
#endif
  }
  return 0;
}

int ulisten(intptr_t sock, int backlog) {
  return listen(sock, backlog);
}

intptr_t uaccept(intptr_t sock) {
  struct sockaddr addr;

  addr.sa_family = AF_INET;
  socklen_t addr_len;

#if defined(__WINDOWS__)
  intptr_t fd = (intptr_t)accept((SOCKET)sock, &addr, &addr_len);
  if (fd == INVALID_SOCKET) return WSAGetLastError();
#else
  intptr_t fd = accept(sock, &addr, &addr_len);
  if (fd < 0) return errno;
#endif

  return fd;
}

int uclose(intptr_t sock) {
#if defined(__WINDOWS__)
  return closesocket((SOCKET)sock);
#else
  return close(sock);
#endif
}

int uread(intptr_t fd, char* buf, size_t len) {
  return recv(fd, buf, len, 0);
}
int uwrite(intptr_t fd, const char* buf, size_t len) {
  return send(fd, buf, len, 0);
}


{'fraction_non_alphanumeric': 0.16622976603971426, 'fraction_numerical': 0.009895799200471853, 'mean_word_length': 2.9717595002602812, 'pattern_counts': {'":': 0, '<': 47, '<?xml version=': 0, '>': 319, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '4113053', 'n_tokens_mistral': 13387, 'n_tokens_neox': 12936, 'n_words': 2810}