#include <errno.h>
#include <fcntl.h>
#include <pthread.h>
#include <stdint.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <time.h>
#include <unistd.h>
#include <sys/types.h>
#include <sys/epoll.h>
#include <sys/socket.h>
#include <sys/un.h>

static char temp[11];

static void die(const char* msg, int err) {
  fprintf(stderr, "%s: %s\n", msg, strerror(err));
  exit(EXIT_FAILURE);
}

struct condlock {
  pthread_mutex_t mutex;
  pthread_cond_t cond;
  int ready;
};

void condlock_init(struct condlock* cl) {
  int err;

  err = pthread_mutex_init(&cl->mutex, NULL);
  if (err != 0) {
    die("pthread_mutex_init", err);
  }
  err = pthread_cond_init(&cl->cond, NULL);
  if (err != 0) {
    die("pthread_cond_init", err);
  }
  cl->ready = 0;
}

void condlock_signal(struct condlock* cl) {
  int err;

  err = pthread_mutex_lock(&cl->mutex);
  if (err != 0) {
    die("pthread_mutex_lock", err);
  }
  cl->ready = 1;
  err = pthread_cond_signal(&cl->cond);
  if (err != 0) {
    die("pthread_cond_signal", err);
  }
  err = pthread_mutex_unlock(&cl->mutex);
  if (err != 0) {
    die("pthread_mutex_unlock", err);
  }
}

void condlock_wait(struct condlock* cl) {
  int err;
  struct timespec ts;

  /* If everything worked as expected, the poller would always report
     that a descriptor is ready and call condlock_signal.  We give it
     20 seconds to do so.  If it fails, we are seeing a bug.  */

  memset(&ts, 0, sizeof ts);

  err = pthread_mutex_lock(&cl->mutex);
  if (err != 0) {
    die("pthread_mutex_lock", err);
  }

  if (!cl->ready) {
    ts.tv_sec = time(NULL) + 20;
    err = pthread_cond_timedwait(&cl->cond, &cl->mutex, &ts);
    if (err != 0) {
      if (err == ETIMEDOUT) {
	fprintf(stderr, "waited more than 20 seconds for descriptor to be ready\n");
	exit(EXIT_FAILURE);
      } else {
	die("pthread_cond_timedwait", err);
      }
    }
  }

  cl->ready = 0;

  err = pthread_mutex_unlock(&cl->mutex);
  if (err != 0) {
    die("pthread_mutex_unlock", err);
  }
}

static int pollfd;

static void poll_init() {
  int d;

  d = epoll_create(20);
  if (d < 0) {
    die("epoll_create", errno);
  }
  pollfd = d;
}

static void poll_add(int d, struct condlock *cond) {
  struct epoll_event ev;

  memset(&ev, 0, sizeof ev);
  ev.events = EPOLLIN | EPOLLOUT | EPOLLRDHUP | EPOLLET;
  ev.data.u64 = (uint64_t)(uintptr_t)(void*)(cond);
  if (epoll_ctl(pollfd, EPOLL_CTL_ADD, d, &ev) < 0) {
    die("epoll_ctl add", errno);
  }
}

static void poll_del(int d) {
  if (epoll_ctl(pollfd, EPOLL_CTL_DEL, d, NULL) < 0) {
    die("epoll_ctl del", errno);
  }
}

static void *poller(void *arg __attribute__((unused))) {
  int c;
  struct epoll_event events[128];
  int i;
  struct condlock *cl;

  while (1) {
    c = epoll_wait(pollfd, events, 128, -1);
    for (i = 0; i < c; i++) {
      cl = (struct condlock*)(void*)(uintptr_t)(events[i].data.u64);
      condlock_signal(cl);
    }
  }

  abort(); /* unreachable */
}

static void *server(void *arg __attribute__((unused))) {
  struct sockaddr_un sun;
  struct condlock cl;
  int s;
  int i;
  char buf[4096];

  memset(&sun, 0, sizeof sun);
  sun.sun_family = AF_UNIX;
  memcpy(sun.sun_path, temp, sizeof temp);

  condlock_init(&cl);

  while (1) {
    s = socket(AF_UNIX, SOCK_DGRAM | SOCK_NONBLOCK, 0);
    if (s < 0) {
      die("socket", errno);
    }

    if (bind(s, (struct sockaddr*)(&sun), sizeof sun) < 0) {
      die("bind", errno);
    }

    poll_add(s, &cl);

    for (i = 0; i < 20; i++) {
      if (recv(s, buf, sizeof buf, 0) < 0) {
	if (errno != EAGAIN && errno != EWOULDBLOCK) {
	  die("recv", errno);
	}
	condlock_wait(&cl);
      }
    }

    poll_del(s);
    if (close(s) < 0) {
      die("close", errno);
    }
    if (remove(temp) < 0) {
      die("remove", errno);
    }

    usleep(1000); /* one millisecond */
  }

  abort(); /* unreachable */
}

static void *client(void *arg __attribute__((unused))) {
  const char * const msg = "test line";
  struct sockaddr_un sun;
  struct condlock cl;
  int s;
  int i;
  int flags;

  memset(&sun, 0, sizeof sun);
  sun.sun_family = AF_UNIX;
  memcpy(sun.sun_path, temp, sizeof temp);

  condlock_init(&cl);

  s = -1;
  for (i = 0; i < 500000; i++) {
    while (s < 0) {
      s = socket(AF_UNIX, SOCK_DGRAM, 0);
      if (s < 0) {
	die("socket", errno);
      }
      if (connect(s, (struct sockaddr *)(&sun), sizeof sun) < 0) {
	if (errno != ECONNREFUSED && errno != ENOENT) {
	  die("connect", errno);
	}
	if (close(s) < 0) {
	  die("close", errno);
	}
	s = -1;
	usleep(1); /* one microsecond */
      } else {
	flags = fcntl(s, F_GETFL, 0);
	if (flags < 0) {
	  die("fcntl", errno);
	}
	if (fcntl(s, F_SETFL, flags | O_NONBLOCK) < 0) {
	  die("fcntl", errno);
	}

	poll_add(s, &cl);
      }
    }

    if (write(s, msg, strlen(msg)) < 0) {
      if (errno == EAGAIN || errno == EWOULDBLOCK) {
	condlock_wait(&cl);
      } else if (errno == ECONNREFUSED) {
	poll_del(s);
	if (close(s) < 0) {
	  die("close", errno);
	}
	s = -1;
      } else {
	die("write", errno);
      }
    }

    if (i % 10000 == 0) {
      usleep(1000); /* one millisecond */
    }
  }

  if (s > 0) {
    poll_del(s);
    if (close(s) < 0) {
      die("close", errno);
    }
  }

  return NULL;
}

#define CLIENT_COUNT (20)

int main() {
  int d;
  pthread_attr_t attr;
  int err;
  pthread_t tid;
  int i;
  pthread_t clients[CLIENT_COUNT];

  memcpy(temp, "testXXXXXX", sizeof temp);
  d = mkstemp(temp);
  if (d < 0) {
    die("mkstemp", errno);
  }
  close(d);
  remove(temp);

  err = pthread_attr_init(&attr);
  if (err != 0) {
    die("pthread_attr_init", err);
  }
  err = pthread_attr_setdetachstate(&attr, PTHREAD_CREATE_DETACHED);
  if (err != 0) {
    die("pthread_attr_setdetachstate", err);
  }

  poll_init();
  err = pthread_create(&tid, &attr, poller, NULL);
  if (err != 0) {
    die("pthread_create", err);
  }

  err = pthread_create(&tid, &attr, server, NULL);
  if (err != 0) {
    die("pthread_create", err);
  }

  for (i = 0; i < CLIENT_COUNT; i++) {
    err = pthread_create(&clients[i], NULL, client, NULL);
    if (err != 0) {
      die("pthread_create", err);
    }
  }

  for (i = 0; i < CLIENT_COUNT; i++) {
    err = pthread_join(clients[i], NULL);
    if (err != 0) {
      die("pthread_join", err);
    }
  }

  return 0;
}