source: libcfa/src/concurrency/ready_subqueue.hfa @ aaf92de

#pragma once

#define __CFA_NO_SCHED_STATS__

#include "containers/queueLockFree.hfa"

// Intrusives lanes which are used by the relaxed ready queue
struct __attribute__((aligned(128))) __intrusive_lane_t {

        #if defined(USE_MPSC)
                mpsc_queue($thread) queue;
                __attribute__((aligned(128)))
        #else
                // anchor for the head and the tail of the queue
                __attribute__((aligned(128))) struct __sentinel_t {
                        // Link lists fields
                        // instrusive link field for threads
                        // must be exactly as in $thread
                        __thread_desc_link link;
                } before, after;
        #endif

        // spin lock protecting the queue
        volatile bool lock;

        // Optional statistic counters
        #if !defined(__CFA_NO_SCHED_STATS__)
                struct __attribute__((aligned(64))) {
                        // difference between number of push and pops
                        ssize_t diff;

                        // total number of pushes and pops
                        size_t  push;
                        size_t  pop ;
                } stat;
        #endif
};

void  ?{}(__intrusive_lane_t & this);
void ^?{}(__intrusive_lane_t & this);

// Get the head pointer (one before the first element) from the anchor
static inline $thread * head(const __intrusive_lane_t & this) {
        #if defined(USE_MPSC)
                return this.queue.head;
        #else
                $thread * rhead = ($thread *)(
                        (uintptr_t)( &this.before ) - offsetof( $thread, link )
                );
                /* paranoid */ verify(rhead);
                return rhead;
        #endif
}

// Get the tail pointer (one after the last element) from the anchor
static inline $thread * tail(const __intrusive_lane_t & this) {
        #if defined(USE_MPSC)
                return this.queue.tail;
        #else
                $thread * rtail = ($thread *)(
                        (uintptr_t)( &this.after ) - offsetof( $thread, link )
                );
                /* paranoid */ verify(rtail);
                return rtail;
        #endif
}

// Ctor
void ?{}( __intrusive_lane_t & this ) {
        this.lock = false;

        #if !defined(USE_MPSC)
                this.before.link.prev = 0p;
                this.before.link.next = tail(this);
                this.before.link.ts   = 0;

                this.after .link.prev = head(this);
                this.after .link.next = 0p;
                this.after .link.ts   = 0;

                #if !defined(__CFA_NO_SCHED_STATS__)
                        this.stat.diff = 0;
                        this.stat.push = 0;
                        this.stat.pop  = 0;
                #endif

                // We add a boat-load of assertions here because the anchor code is very fragile
                /* paranoid */ verify(((uintptr_t)( head(this) ) + offsetof( $thread, link )) == (uintptr_t)(&this.before));
                /* paranoid */ verify(((uintptr_t)( tail(this) ) + offsetof( $thread, link )) == (uintptr_t)(&this.after ));
                /* paranoid */ verify(head(this)->link.prev == 0p );
                /* paranoid */ verify(head(this)->link.next == tail(this) );
                /* paranoid */ verify(tail(this)->link.next == 0p );
                /* paranoid */ verify(tail(this)->link.prev == head(this) );
                /* paranoid */ verify(&head(this)->link.prev == &this.before.link.prev );
                /* paranoid */ verify(&head(this)->link.next == &this.before.link.next );
                /* paranoid */ verify(&tail(this)->link.prev == &this.after .link.prev );
                /* paranoid */ verify(&tail(this)->link.next == &this.after .link.next );
                /* paranoid */ verify(__alignof__(__intrusive_lane_t) == 128);
                /* paranoid */ verify(__alignof__(this) == 128);
                /* paranoid */ verifyf(((intptr_t)(&this) % 128) == 0, "Expected address to be aligned %p %% 128 == %zd", &this, ((intptr_t)(&this) % 128));
        #endif
}

// Dtor is trivial
void ^?{}( __intrusive_lane_t & this ) {
        #if !defined(USE_MPSC)
                // Make sure the list is empty
                /* paranoid */ verify(head(this)->link.prev == 0p );
                /* paranoid */ verify(head(this)->link.next == tail(this) );
                /* paranoid */ verify(tail(this)->link.next == 0p );
                /* paranoid */ verify(tail(this)->link.prev == head(this) );
        #endif
}

// Push a thread onto this lane
// returns true of lane was empty before push, false otherwise
bool push(__intrusive_lane_t & this, $thread * node) {
        #if defined(USE_MPSC)
                inline $thread * volatile & ?`next ( $thread * this )  __attribute__((const)) {
                        return this->link.next;
                }
                push(this.queue, node);
        #else
                #if defined(__CFA_WITH_VERIFY__)
                        /* paranoid */ verify(this.lock);
                        /* paranoid */ verify(node->link.ts != 0);
                        /* paranoid */ verify(node->link.next == 0p);
                        /* paranoid */ verify(node->link.prev == 0p);
                        /* paranoid */ verify(tail(this)->link.next == 0p);
                        /* paranoid */ verify(head(this)->link.prev == 0p);

                        if(this.before.link.ts == 0l) {
                                /* paranoid */ verify(tail(this)->link.prev == head(this));
                                /* paranoid */ verify(head(this)->link.next == tail(this));
                        } else {
                                /* paranoid */ verify(tail(this)->link.prev != head(this));
                                /* paranoid */ verify(head(this)->link.next != tail(this));
                        }
                #endif

                // Get the relevant nodes locally
                $thread * tail = tail(this);
                $thread * prev = tail->link.prev;

                // Do the push
                node->link.next = tail;
                node->link.prev = prev;
                prev->link.next = node;
                tail->link.prev = node;

                // Update stats
                #if !defined(__CFA_NO_SCHED_STATS__)
                        this.stat.diff++;
                        this.stat.push++;
                #endif

                verify(node->link.next == tail(this));

                // Check if the queue used to be empty
                if(this.before.link.ts == 0l) {
                        this.before.link.ts = node->link.ts;
                        /* paranoid */ verify(node->link.prev == head(this));
                        return true;
                }
                return false;
        #endif
}

// Pop a thread from this lane (must be non-empty)
// returns popped
// returns true of lane was empty before push, false otherwise
$thread * pop(__intrusive_lane_t & this) {
        /* paranoid */ verify(this.lock);
        #if defined(USE_MPSC)
                inline $thread * volatile & ?`next ( $thread * this )  __attribute__((const)) {
                        return this->link.next;
                }
                return pop(this.queue);
        #else
                /* paranoid */ verify(this.before.link.ts != 0ul);

                // Get anchors locally
                $thread * head = head(this);
                $thread * tail = tail(this);

                // Get the relevant nodes locally
                $thread * node = head->link.next;
                $thread * next = node->link.next;

                /* paranoid */ verify(node != tail);
                /* paranoid */ verify(node);

                // Do the pop
                head->link.next = next;
                next->link.prev = head;
                node->link.next = 0p;
                node->link.prev = 0p;

                // Update head time stamp
                this.before.link.ts = next->link.ts;

                // Update stats
                #ifndef __CFA_NO_SCHED_STATS__
                        this.stat.diff--;
                        this.stat.pop ++;
                #endif

                // Check if we emptied list and return accordingly
                /* paranoid */ verify(tail(this)->link.next == 0p);
                /* paranoid */ verify(head(this)->link.prev == 0p);
                if(next == tail) {
                        /* paranoid */ verify(this.before.link.ts == 0);
                        /* paranoid */ verify(tail(this)->link.prev == head(this));
                        /* paranoid */ verify(head(this)->link.next == tail(this));
                        return node;
                }
                else {
                        /* paranoid */ verify(next->link.ts != 0);
                        /* paranoid */ verify(tail(this)->link.prev != head(this));
                        /* paranoid */ verify(head(this)->link.next != tail(this));
                        /* paranoid */ verify(this.before.link.ts != 0);
                        return node;
                }
        #endif
}

// Check whether or not list is empty
static inline bool is_empty(__intrusive_lane_t & this) {
        #if defined(USE_MPSC)
                return this.queue.head == 0p;
        #else
                // Cannot verify here since it may not be locked
                return this.before.link.ts == 0;
        #endif
}

// Return the timestamp
static inline unsigned long long ts(__intrusive_lane_t & this) {
        #if defined(USE_MPSC)
                $thread * tl = this.queue.head;
                if(!tl) return -1ull;
                return tl->link.ts;
        #else
                // Cannot verify here since it may not be locked
                return this.before.link.ts;
        #endif
}

// Aligned timestamps which are used by the relaxed ready queue
struct __attribute__((aligned(128))) __timestamp_t {
        volatile unsigned long long tv;
};

void  ?{}(__timestamp_t & this) { this.tv = 0; }
void ^?{}(__timestamp_t & this) {}