#include <stddef.h>
#include <stdint.h>
#include <sys/ioctl.h>
+#include <sys/poll.h>
#include <sys/types.h>
#include <unistd.h>
#define atomic_add(ptr, n) (void)atomic_fetch_and_add(ptr, n)
#define atomic_fetch_and_sub(ptr, n) __sync_fetch_and_sub(ptr, n)
#define atomic_sub(ptr, n) (void)atomic_fetch_and_sub(ptr, n)
+#define atomic_xchg(ptr, v) ((typeof(*(ptr)))__sync_lock_test_and_set(ptr, v))
#define ALWAYS_INLINE __attribute__((always_inline)) inline
/* atomic */ int nthreads;
/* atomic */ int waiters;
/* atomic */ int parked;
+ /* atomic */ int overcommit_count;
union {
struct {
pwqr_signal_relaxed_n(1);
}
+static int pwqr_fd_read_overcommit(int fd)
+{
+ int buf;
+
+ if (read(fd, &buf, sizeof(buf)) == sizeof(buf))
+ return buf;
+ return 0;
+}
+
+__attribute__((cold))
+static int pwqr_fd_overcommit_check(void)
+{
+ if (atomic_xchg(&pwqr_g.overcommit_count, 0)) {
+ int oc = pwqr_fd_read_overcommit(pwqr_g.fd);
+
+ if (oc) {
+ access_once(pwqr_g.overcommit_count) = (oc - 1);
+ return 1;
+ }
+ }
+ return 0;
+}
+
+__attribute__((unused))
+static void pwqr_overcommit_poll_loop(void)
+{
+ struct pollfd pfd = {
+ .fd = pwqr_g.fd,
+ .events = POLLIN,
+ };
+
+ for (;;) {
+ pfd.revents = 0;
+ if (poll(&pfd, 1, -1) >= 0) {
+ if (pfd.revents & POLLIN) {
+ access_once(pwqr_g.overcommit_count) =
+ pwqr_fd_read_overcommit(pwqr_g.fd);
+ }
+ if (pfd.revents & (POLLHUP | POLLERR))
+ return;
+ } else if (errno != EINTR && errno != EAGAIN) {
+ return;
+ }
+ }
+}
+
/* returns:
- INPOOL if we were woken to run jobs.
#endif
if ((rc = pwqr_ctl(pwqr_g.fd, PWQR_PARK, 0, NULL)) == 0) {
+ access_once(pwqr_g.overcommit_count) = 0;
if (atomic_fetch_and_sub(&pwqr_g.parked, 1) == 0) {
pwqr_spawn_thread(PARKED, 1);
}
do {
uint64_t ticket;
+ park:
if (state == PARKED && (state = pwqr_do_park()) < 0)
break;
+ for (;;) {
+ if (unlikely(pwqr_g.overcommit_count)) {
+ if (pwqr_fd_overcommit_check()) {
+ state = PARKED;
+ goto park;
+ }
+ }
#ifdef PSEUDO_CODE
- while ((job = find_some_job(ANY_JOB)))
+ job = find_some_job(ANY_JOB);
+ if (!job)
+ break;
run_job(job);
#endif
+ }
ticket = pwqr_get_ticket();
#ifdef PSEUDO_CODE
if ((job = find_some_job(ANY_JOB))) {