/* -*- Mode: C; c-basic-offset:4 ; indent-tabs-mode:nil ; -*- */

/*  

 *  (C) 2008 by Argonne National Laboratory.

 *      See COPYRIGHT in top-level directory.

 */

/* Implements a fast, lockfree, multi-producer, single-consumer queue.  It's

 * important to note the *single-consumer* piece of this, since multithreaded

 * consumption will surely lead to data corruption and/or other problems. */

#ifndef OPA_QUEUE_H_INCLUDED

#define OPA_QUEUE_H_INCLUDED

#include "opa_primitives.h"

#ifdef OPA_HAVE_STDDEF_H

#include <stddef.h>

#endif /* OPA_HAVE_STDDEF_H */

/* This value is used to indicate NULL in the OPA_Shm_asymm_base_addr

   variable.  It is non-zero because one of the likely base addresses is zero

   (indicating memory is actually symmetrically mapped).  The value 64 was used

   because it is unlikely that mmap will choose an address this low for a

   mapping.  */

/* XXX DJG TODO put some conditionally compiled error checking in that uses this */

#define OPA_SHM_ASYMM_NULL_VAL    64

extern char *OPA_Shm_asymm_base_addr;

/* Used to initialize the base address for relative pointers.  This interface

   assumes that there is only one shared memory segment.  If this turns out to

   not be the case in the future, we should probably add support for multiple

   shm segments.

   This function will return an error if it has already been called. */

int OPA_Shm_asymm_init(char *base);

/* Relative addressing macros.  These are for manipulating addresses relative

   to the start of a shared memory region. */

#define OPA_SHM_REL_NULL (0x0)

#define OPA_SHM_IS_REL_NULL(rel_ptr) (OPA_load_ptr(&(rel_ptr).offset) == OPA_SHM_REL_NULL)

#define OPA_SHM_SET_REL_NULL(rel_ptr) (OPA_store_ptr(&(rel_ptr).offset, OPA_SHM_REL_NULL))

#define OPA_SHM_REL_ARE_EQUAL(rel_ptr1, rel_ptr2) \

    (OPA_load_ptr(&(rel_ptr1).offset) == OPA_load_ptr(&(rel_ptr2).offset))

/* This structure exists such that it is possible to expand the expressiveness

   of a relative address at some point in the future.  It also provides a

   modicum of type safety to help prevent certain flavors of errors.

   For example, instead of referencing an offset from a global base address, it

   might make sense for there to be multiple base addresses.  These base

   addresses could correspond to the start of a segment or region of shared

   memory.  This structure could store the segment number that is used to lookup

   a base address in a non-shared table.  Note that you would have to be very

   careful about all of this because if you add the segment number as a separate

   field you can no longer (compare and) swap a relative address atomically.  So

   you'll either have to shave bits from the pointer or make some sort of

   requirement that relative addresses can only be swapped within the same

   segment.  */

typedef struct OPA_Shm_rel_addr_t {

    OPA_ptr_t offset;

} OPA_Shm_rel_addr_t;

/* converts a relative pointer to an absolute pointer */

static _opa_inline

void *OPA_Shm_rel_to_abs(OPA_Shm_rel_addr_t r)

{

    void *offset = OPA_load_ptr(&r.offset);

    OPA_assert((size_t)OPA_Shm_asymm_base_addr != OPA_SHM_ASYMM_NULL_VAL);

    return (void*)(OPA_Shm_asymm_base_addr + (size_t)offset);

}

/* converts an absolute pointer to a relative pointer */

static _opa_inline

OPA_Shm_rel_addr_t OPA_Shm_abs_to_rel(void *a)

{

    OPA_Shm_rel_addr_t ret;

    OPA_assert((size_t)OPA_Shm_asymm_base_addr != OPA_SHM_ASYMM_NULL_VAL);

    OPA_store_ptr(&ret.offset, (void*)((size_t)a - (size_t)OPA_Shm_asymm_base_addr));

    return ret;

}

static _opa_inline

OPA_Shm_rel_addr_t OPA_Shm_swap_rel(OPA_Shm_rel_addr_t *addr, OPA_Shm_rel_addr_t newv) {

    OPA_Shm_rel_addr_t oldv;

    OPA_store_ptr(&oldv.offset, OPA_swap_ptr(&addr->offset, OPA_load_ptr(&newv.offset)));

    return oldv;

}

/* Compare the relative pointer to (relative) null and swap if equal.  Prevents

   the guts of the _rel_addr_t abstraction from bleeding up into the

   enqueue/dequeue operations. */

static _opa_inline

OPA_Shm_rel_addr_t OPA_Shm_cas_rel_null(OPA_Shm_rel_addr_t *addr, OPA_Shm_rel_addr_t oldv) {

    OPA_Shm_rel_addr_t prev;

    OPA_store_ptr(&prev.offset, OPA_cas_ptr(&(addr->offset), OPA_load_ptr(&oldv.offset), (void*)OPA_SHM_REL_NULL));

    return prev;

}

/* The shadow head pointer makes this queue implementation perform much better

   than a standard queue.  Unfortunately, it also makes it a bit non-intuitive

   when read the code.  The following is an excerpt from "Design and Evaluation

   of Nemesis,  a Scalable, Low-Latency, Message-Passing Communication

   Subsystem" by D. Buntinas, G.  Mercier, and W. Gropp that gives an

   explanation:

      A process must access both the head and tail of the queue when it is

      enqueuing an element on an empty queue or when it is dequeuing an element

      that is the last element in the queue. In these cases, if the head and

      tail were in the same cache line, only one L2 cache miss would be

      encountered. If the queue has more elements in it, however, then the

      enqueuer only needs to access the tail, and the dequeuer only needs to

      access the head. If the head and tail were in the same cache line, then

      there would be L2 misses encountered as a result of false sharing each

      time a process enqueues an element after another has been dequeued from

      the same queue, and vice versa. In this case it would be better if the

      head and tail were in separate cache lines.

      Our solution is to put the head and tail in the same cache line and have a

      shadow head pointer in a separate cache line. The shadow head is

      initialized to NULL. The dequeuer uses the shadow head in place of the

      real head except when the shadow head is NULL, meaning that the queue has

      become empty. If the shadow head is NULL when the dequeuer tries to

      dequeue, it checks the value of the real head. If the real head is not

      NULL, meaning that an element has been enqueued on the queue since the

      last time the queue became empty, the dequeuer initializes its shadow head

      to the value of the real head and sets the real head to NULL. In this way,

      only one L2 cache miss is encountered when enqueuing onto an empty queue

      or dequeuing from a queue with one element. And because the tail and

      shadow head are in separate cache lines, there are no L2 cache misses from

      false sharing. 

      We found that using a shadow head pointer reduced one-way latency by about

      200 ns on a dual 2 GHz Xeon node.

*/

/* Pick an arbitrary cacheline size for now, we can setup a mechanism to detect

   it at build time later on.  This should work well on most intel systems at

   the very least. */

#define OPA_QUEUE_CACHELINE_PADDING 128

/* All absolute and relative pointers point to the start of the enclosing element. */

typedef struct OPA_Queue_info_t {

    OPA_Shm_rel_addr_t head; /* holds the offset pointer, not the abs ptr */

    OPA_Shm_rel_addr_t tail; /* holds the offset pointer, not the abs ptr */

    char padding1[OPA_QUEUE_CACHELINE_PADDING-2*sizeof(OPA_Shm_rel_addr_t)];

    OPA_Shm_rel_addr_t  shadow_head; /* holds the offset pointer, not the abs ptr */

    char padding2[OPA_QUEUE_CACHELINE_PADDING-sizeof(OPA_Shm_rel_addr_t)];

} OPA_Queue_info_t;

/* Using this header struct even though it's just one element gives us the

   opportunity to vary the implementation more easily in the future without

   updating all callers. */

typedef struct OPA_Queue_element_hdr_t {

    OPA_Shm_rel_addr_t next; /* holds the offset pointer, not the abs ptr */

} OPA_Queue_element_hdr_t;

/* Used to initialize a queue structure. */

void OPA_Queue_init(OPA_Queue_info_t *qhead);

/* Used to initialize a queue element header. */

void OPA_Queue_header_init(OPA_Queue_element_hdr_t *hdr);

static _opa_inline

int OPA_Queue_is_empty(OPA_Queue_info_t *qhead)

{

    int __ret = 0;

    if (OPA_SHM_IS_REL_NULL (qhead->shadow_head)) {

        if (OPA_SHM_IS_REL_NULL(qhead->head)) {

            __ret = 1;

        }

        else {

            qhead->shadow_head = qhead->head;

            OPA_SHM_SET_REL_NULL(qhead->head); /* reset it for next time */

        }

    }

    return __ret;

}

/* Returns a pointer to the element at the head of the queue.  The current

   implementation of these queue algorithms imposes several notable

   restrictions on the use of this function:

    - The caller must currently hold the critical section, just as if you were

      calling OPA_Queue_dequeue.  Failure to do could easily produce incorrect

      results.

    - OPA_Queue_is_empty must be called on this queue prior to calling this

      function.  Furthermore, there must be no intervening calls to any

      OPA_Queue_* functions (for this queue) between _is_empty and _peek_head.

      Failure to do so will produce incorrect results.

   This operation is effectively the same as the dequeue operation (insofar as

   it shares the same calling restrictions) but it does not disturb the actual

   contents of the queue. */

static _opa_inline

void *OPA_Queue_peek_head(OPA_Queue_info_t *qhead_ptr)

{

    OPA_assert(qhead_ptr != NULL);

    if (OPA_SHM_IS_REL_NULL(qhead_ptr->shadow_head)) {

        return NULL;

    }

    return OPA_Shm_rel_to_abs(qhead_ptr->shadow_head);

}

/* This macro atomically enqueues an element (elt for short) into the queue

   indicated by qhead_ptr.  You need to pass several arguments:

     qhead_ptr - a pointer to a OPA_Queue_info_t structure to which the

                 element should be enqueued

     elt_ptr - a pointer to an element structure type that is to be enqueued

     elt_type - The base type of elt_ptr.  That is, if elt_ptr is a

                '(struct example_t *)' then elt_type is 'struct example_t'.

     elt_hdr_field - This is the member name of elt_type that is a

                     OPA_Queue_element_hdr_t.

    This queue implementation is loosely modeled after the linked lists used in

    the linux kernel.  You put the list header structure inside of the client

    structure, rather than the other way around.

   */

#define OPA_Queue_enqueue(qhead_ptr, elt_ptr, elt_type, elt_hdr_field)    \

do {                                                                      \

    OPA_Shm_rel_addr_t r_prev;                                            \

    OPA_Shm_rel_addr_t r_elt = OPA_Shm_abs_to_rel(elt_ptr);               \

                                                                          \

    OPA_SHM_SET_REL_NULL((elt_ptr)->elt_hdr_field.next);                  \

                                                                          \

    OPA_write_barrier();                                                  \

                                                                          \

    r_prev = OPA_Shm_swap_rel(&((qhead_ptr)->tail), r_elt);               \

                                                                          \

    if (OPA_SHM_IS_REL_NULL(r_prev)) {                                    \

        (qhead_ptr)->head = r_elt;                                        \

    }                                                                     \

    else {                                                                \

        elt_type *abs_prev = (elt_type *)OPA_Shm_rel_to_abs(r_prev);      \

        abs_prev->elt_hdr_field.next = r_elt;                             \

    }                                                                     \

} while (0)

/* This macro atomically dequeues an element (elt for short) from the queue

   indicated by qhead_ptr.  You need to pass several arguments:

     qhead_ptr - a pointer to a OPA_Queue_info_t structure to which the

                 element should be dequeued

     elt_ptr - A pointer to an element structure type that should be populated

               with the dequeued value.  Must be an lvalue.

     elt_type - The base type of elt_ptr.  That is, if elt_ptr is a

                '(struct example_t *)' then elt_type is 'struct example_t'.

     elt_hdr_field - This is the member name of elt_type that is a

                     OPA_Queue_element_hdr_t.

    This queue implementation is loosely modeled after the linked lists used in

    the linux kernel.  You put the list header structure inside of the client

    structure, rather than the other way around.

    NOTE: you must *always* call _is_empty() prior to this function */

#define OPA_Queue_dequeue(qhead_ptr, elt_ptr, elt_type, elt_hdr_field)        \

do {                                                                          \

    elt_type *_e;                                                             \

    OPA_Shm_rel_addr_t _r_e;                                                  \

                                                                              \

    _r_e = (qhead_ptr)->shadow_head;                                          \

    _e = OPA_Shm_rel_to_abs(_r_e);                                            \

                                                                              \

    if (!OPA_SHM_IS_REL_NULL(_e->elt_hdr_field.next)) {                       \

        (qhead_ptr)->shadow_head = _e->elt_hdr_field.next;                    \

    }                                                                         \

    else {                                                                    \

        OPA_Shm_rel_addr_t old_tail;                                          \

                                                                              \

        OPA_SHM_SET_REL_NULL((qhead_ptr)->shadow_head);                       \

                                                                              \

        old_tail = OPA_Shm_cas_rel_null(&((qhead_ptr)->tail), _r_e);          \

                                                                              \

        if (!OPA_SHM_REL_ARE_EQUAL(old_tail, _r_e)) {                         \

            while (OPA_SHM_IS_REL_NULL(_e->elt_hdr_field.next)) {             \

                OPA_busy_wait();                                              \

            }                                                                 \

            (qhead_ptr)->shadow_head = _e->elt_hdr_field.next;                \

        }                                                                     \

    }                                                                         \

    OPA_SHM_SET_REL_NULL(_e->elt_hdr_field.next);                             \

    OPA_read_barrier();                                                       \

    elt_ptr = _e;                                                             \

} while (0)

#endif /* OPA_QUEUE_H_INCLUDED */