The term, I used, binary thread scheduler means a scheduler that orders the execution of two threads for two tasks that has a specific dependency. Rather than using some mathematical expressions to explain this, think about scheduling two threads for double buffering. Thread 1 ends writing in buffer 1 and signals thread 2 that it can read buffer 1. Now thread 1 and 2 can run in parallel. If thread 2 ends reading buffer 1 before thread 1 has not finished writing in buffer 2, it has to wait for thread 1 to finish, and so on. I tried implemented such a scheduler with minimum overhead possible.
Before reading the code, I recommend running it first and see how each state is encoded. Rather than me trying to explain the pattern, I think you'll be quicker to catch the pattern by actually looking at the output of each state before a task has been done.
Sorry for the short variable names, but that's what I prefer when writing my own code. Somehow it's less confusing with short names when following the code to verify the algorithm.
#include <stdatomic.h>
#include <stdio.h>
#include <pthread.h>
typedef _Atomic int binSched_t;
#define astr_(a, b) atomic_store_explicit(a, b, memory_order_relaxed)
#define aldr_(a) atomic_load_explicit(a, memory_order_relaxed)
#define acmpxchgr_(a, b, c) atomic_compare_exchange_strong_explicit(a, b, c,\
memory_order_relaxed, memory_order_relaxed)
#define bsch_state_(i) (1 << (i))
static void bsch_init(binSched_t *_) {
astr_(_, bsch_state_(2));
}
static int bsch_recv0(binSched_t *_) {
int s;
while (!(s = aldr_(_) & 055));
return s > bsch_state_(2);
}
static void bsch_send0(binSched_t *_, int t) {
if (t) {
int s = bsch_state_(3);
if (!acmpxchgr_(_, &s, bsch_state_(4))) {
astr_(_, bsch_state_(0));
}
} else {
int s = bsch_state_(0);
if (!acmpxchgr_(_, &s, bsch_state_(1))) {
astr_(_, bsch_state_(3));
}
}
}
static int bsch_recv1(binSched_t *_) {
int s;
while (!(s = aldr_(_) & 033));
return s < bsch_state_(2);
}
static void bsch_send1(binSched_t *_, int t) {
if (t) {
int s = bsch_state_(0);
if (!acmpxchgr_(_, &s, bsch_state_(2))) {
astr_(_, bsch_state_(3));
}
} else {
int s = bsch_state_(3);
if (!acmpxchgr_(_, &s, bsch_state_(5))) {
astr_(_, bsch_state_(0));
}
}
}
static binSched_t bs;
static void printState() {
static const char *const s[] = {
"0 2\n2 0\n",
"2 1\n1 0\n",
"0 1\n1 2\n",
"2 0\n0 2\n",
"1 2\n0 1\n",
"1 0\n2 1\n"
};
switch (aldr_(&bs)) {
case bsch_state_(0): puts(s[0]); break;
case bsch_state_(1): puts(s[1]); break;
case bsch_state_(2): puts(s[2]); break;
case bsch_state_(3): puts(s[3]); break;
case bsch_state_(4): puts(s[4]); break;
case bsch_state_(5): puts(s[5]); break;
default: __builtin_unreachable();
}
}
static void *main2(void *_) {
for (;;) {
int t = bsch_recv1(&bs);
printf("thread 2 task %d\n", t + 1);
printState();
bsch_send1(&bs, t);
}
return _;
}
int main() {
bsch_init(&bs);
pthread_t th;
pthread_create(&th, NULL, main2, NULL);
for (;;) {
int t = bsch_recv0(&bs);
printf("thread 1 task %d\n", t + 1);
printState();
bsch_send0(&bs, t);
}
return 0;
}
The reason I wrote this was to minimize the overhead when there is no need to wait, that is when the thread can acquire the lock immediately. In my tests, spinlocks do have the minimal overhead in such cases. However minimal waiting time does not necessarily mean fastest overall throughput. Actually the original implementation has a slightly worse throughput than an alternative implementation using semaphores. Semaphores have a larger base overhead having to deal with the kernel, but they don't mess up with the OS scheduler like spinlocks. Unless the important thread which is doing something is a realtime thread, the OS thinks the busy-waiting thread is also important, so the job-doing thread occasionally gets blocked because of the busy-waiting thread.
I also tested a version with Posix condition variables. They have a surprisingly low overhead to acquire the lock. Even faster than a spinlock! But the overall throughput is worse than both the semaphore and the spinlock version.
- Throughput
- Semaphore > Spinlock > Condition Variable
- Lock Overhead
- Semaphore > Spinlock > Condition Variable
- Stability
- Condition Variable > Semaphore > Spinlock
This is the semaphore version.
static void chk(int ok) {
if (!__builtin_expect(ok, true)) abort();
}
typedef struct {
sem_t *p[2];
sem_t s[2][2];
} binSched_t;
static void bsch_init(binSched_t *_) {
*_->p = *_->s;
_->p[1] = _->s[1];
for (int i = 0; i < 2; ++i) {
for (int j = 0; j < 2; ++j) {
chk(!sem_init(_->s[i] + j, false, !i));
}
}
}
static int bsch_recv0(binSched_t *_) {
chk(!sem_wait(*_->p));
return *_->p == *_->s + 1;
}
static void bsch_send0(binSched_t *_, int t) {
chk(!sem_post(*_->p + 2));
*_->p = *_->s + !t;
}
static int bsch_recv1(binSched_t *_) {
chk(!sem_wait(_->p[1]));
return _->p[1] == _->s[1] + 1;
}
static void bsch_send1(binSched_t *_, int t) {
chk(!sem_post(_->p[1] - 2));
_->p[1] = _->s[1] + !t;
}
And this is the condition variable version.
typedef struct {
pthread_mutex_t m;
int f;
pthread_cond_t _;
} cond_t;
typedef struct {
cond_t *p[2];
cond_t c[2][2];
} binSched_t;
static void bsch_init(binSched_t *_) {
*_->p = *_->c;
_->p[1] = _->c[1];
for (int i = 0; i < 2; ++i) {
for (int j = 0; j < 2; ++j) {
cond_t *c = _->c[i] + j;
chk(!pthread_mutex_init(&c->m, NULL));
c->f = !i;
chk(!pthread_cond_init(&c->_, NULL));
}
}
}
static int bsch_recv0(binSched_t *_) {
cond_t *c = *_->p;
chk(!pthread_mutex_lock(&c->m));
if (!c->f) chk(!pthread_cond_wait(&c->_, &c->m));
c->f = false;
chk(!pthread_mutex_unlock(&c->m));
return c == *_->c + 1;
}
static void bsch_send0(binSched_t *_, int t) {
cond_t *c = *_->p + 2;
chk(!pthread_mutex_lock(&c->m));
c->f = true;
chk(!pthread_mutex_unlock(&c->m));
chk(!pthread_cond_signal(&c->_));
*_->p = *_->c + !t;
}
static int bsch_recv1(binSched_t *_) {
cond_t *c = _->p[1];
chk(!pthread_mutex_lock(&c->m));
if (!c->f) chk(!pthread_cond_wait(&c->_, &c->m));
c->f = false;
chk(!pthread_mutex_unlock(&c->m));
return c == _->c[1] + 1;
}
static void bsch_send1(binSched_t *_, int t) {
cond_t *c = _->p[1] - 2;
chk(!pthread_mutex_lock(&c->m));
c->f = true;
chk(!pthread_mutex_unlock(&c->m));
chk(!pthread_cond_signal(&c->_));
_->p[1] = _->c[1] + !t;
}