openmpi/ompi/mca/osc/pt2pt/osc_pt2pt.h

/* -*- Mode: C; c-basic-offset:4 ; indent-tabs-mode:nil -*- */
/*
 * Copyright (c) 2004-2005 The Trustees of Indiana University.
 *                         All rights reserved.
 * Copyright (c) 2004-2006 The Trustees of the University of Tennessee.
 *                         All rights reserved.
 * Copyright (c) 2004-2005 High Performance Computing Center Stuttgart,
 *                         University of Stuttgart.  All rights reserved.
 * Copyright (c) 2004-2005 The Regents of the University of California.
 *                         All rights reserved.
 * Copyright (c) 2007-2017 Los Alamos National Security, LLC.  All rights
 *                         reserved.
 * Copyright (c) 2010      Cisco Systems, Inc.  All rights reserved.
 * Copyright (c) 2012-2013 Sandia National Laboratories.  All rights reserved.
 * Copyright (c) 2015-2017 Research Organization for Information Science
 *                         and Technology (RIST). All rights reserved.
 * Copyright (c) 2016      FUJITSU LIMITED.  All rights reserved.
 * Copyright (c) 2016-2017 IBM Corporation. All rights reserved.
 * $COPYRIGHT$
 *
 * Additional copyrights may follow
 *
 * $HEADER$
 */

#ifndef OMPI_OSC_PT2PT_H
#define OMPI_OSC_PT2PT_H

#include "ompi_config.h"
#include "opal/class/opal_list.h"
#include "opal/class/opal_free_list.h"
#include "opal/class/opal_hash_table.h"
#include "opal/threads/threads.h"
#include "opal/util/output.h"

#include "ompi/win/win.h"
#include "ompi/info/info.h"
#include "ompi/communicator/communicator.h"
#include "ompi/datatype/ompi_datatype.h"
#include "ompi/request/request.h"
#include "ompi/mca/osc/osc.h"
#include "ompi/mca/osc/base/base.h"
#include "ompi/memchecker.h"

#include "osc_pt2pt_header.h"
#include "osc_pt2pt_sync.h"

BEGIN_C_DECLS

struct ompi_osc_pt2pt_frag_t;
struct ompi_osc_pt2pt_receive_t;

struct ompi_osc_pt2pt_component_t {
    /** Extend the basic osc component interface */
    ompi_osc_base_component_t super;

    /** lock access to modules */
    opal_mutex_t lock;

    /** cid -> module mapping */
    opal_hash_table_t modules;

    /** module count */
    int module_count;

    /** number of buffers per window */
    int receive_count;

    /** free list of ompi_osc_pt2pt_frag_t structures */
    opal_free_list_t frags;

    /** Free list of requests */
    opal_free_list_t requests;

    /** PT2PT component buffer size */
    unsigned int buffer_size;

    /** Lock for pending_operations */
    opal_mutex_t pending_operations_lock;

    /** List of operations that need to be processed */
    opal_list_t pending_operations;

    /** List of receives to be processed */
    opal_list_t pending_receives;

    /** Lock for pending_receives */
    opal_mutex_t pending_receives_lock;

    /** Is the progress function enabled? */
    bool progress_enable;
};
typedef struct ompi_osc_pt2pt_component_t ompi_osc_pt2pt_component_t;

enum {
    /** peer has sent an unexpected post message (no matching start) */
    OMPI_OSC_PT2PT_PEER_FLAG_UNEX = 1,
    /** eager sends are active on this peer */
    OMPI_OSC_PT2PT_PEER_FLAG_EAGER = 2,
    /** peer has been locked (on-demand locking for lock_all) */
    OMPI_OSC_PT2PT_PEER_FLAG_LOCK = 4,
};


struct ompi_osc_pt2pt_peer_t {
    /** make this an opal object */
    opal_object_t super;

    /** rank of this peer */
    int rank;

    /** pointer to the current send fragment for each outgoing target */
    struct ompi_osc_pt2pt_frag_t *active_frag;

    /** lock for this peer */
    opal_mutex_t lock;

    /** fragments queued to this target */
    opal_list_t queued_frags;

    /** number of fragments incomming (negative - expected, positive - unsynchronized) */
    volatile int32_t passive_incoming_frag_count;

    /** peer flags */
    volatile int32_t flags;
};
typedef struct ompi_osc_pt2pt_peer_t ompi_osc_pt2pt_peer_t;

OBJ_CLASS_DECLARATION(ompi_osc_pt2pt_peer_t);

static inline bool ompi_osc_pt2pt_peer_locked (ompi_osc_pt2pt_peer_t *peer)
{
    return !!(peer->flags & OMPI_OSC_PT2PT_PEER_FLAG_LOCK);
}

static inline bool ompi_osc_pt2pt_peer_unex (ompi_osc_pt2pt_peer_t *peer)
{
    return !!(peer->flags & OMPI_OSC_PT2PT_PEER_FLAG_UNEX);
}

static inline bool ompi_osc_pt2pt_peer_eager_active (ompi_osc_pt2pt_peer_t *peer)
{
    return !!(peer->flags & OMPI_OSC_PT2PT_PEER_FLAG_EAGER);
}

static inline void ompi_osc_pt2pt_peer_set_flag (ompi_osc_pt2pt_peer_t *peer, int32_t flag, bool value)
{
    if (value) {
        OPAL_ATOMIC_OR_FETCH32 (&peer->flags, flag);
    } else {
        OPAL_ATOMIC_AND_FETCH32 (&peer->flags, ~flag);
    }
}

static inline void ompi_osc_pt2pt_peer_set_locked (ompi_osc_pt2pt_peer_t *peer, bool value)
{
    ompi_osc_pt2pt_peer_set_flag (peer, OMPI_OSC_PT2PT_PEER_FLAG_LOCK, value);
}

static inline void ompi_osc_pt2pt_peer_set_unex (ompi_osc_pt2pt_peer_t *peer, bool value)
{
    ompi_osc_pt2pt_peer_set_flag (peer, OMPI_OSC_PT2PT_PEER_FLAG_UNEX, value);
}

static inline void ompi_osc_pt2pt_peer_set_eager_active (ompi_osc_pt2pt_peer_t *peer, bool value)
{
    ompi_osc_pt2pt_peer_set_flag (peer, OMPI_OSC_PT2PT_PEER_FLAG_EAGER, value);
}

OBJ_CLASS_DECLARATION(ompi_osc_pt2pt_peer_t);

/** Module structure.  Exactly one of these is associated with each
    PT2PT window */
struct ompi_osc_pt2pt_module_t {
    /** Extend the basic osc module interface */
    ompi_osc_base_module_t super;

    /** window should have accumulate ordering... */
    bool accumulate_ordering;

    /** no locks info key value */
    bool no_locks;

    /** pointer to free on cleanup (may be NULL) */
    void *free_after;

    /** Base pointer for local window */
    void *baseptr;

    /** communicator created with this window.  This is the cid used
        in the component's modules mapping. */
    ompi_communicator_t *comm;

    /** Local displacement unit. */
    int disp_unit;

    /** Mutex lock protecting module data */
    opal_recursive_mutex_t lock;

    /** condition variable associated with lock */
    opal_condition_t cond;

    /** hash table of peer objects */
    opal_hash_table_t peer_hash;

    /** lock protecting peer_hash */
    opal_mutex_t peer_lock;

    /** Nmber of communication fragments started for this epoch, by
        peer.  Not in peer data to make fence more manageable. */
    uint32_t *epoch_outgoing_frag_count;

    /** cyclic counter for a unique tage for long messages. */
    volatile uint32_t tag_counter;

    /** number of outgoing fragments still to be completed */
    volatile int32_t outgoing_frag_count;

    /** number of incoming fragments */
    volatile int32_t active_incoming_frag_count;

    /** Number of targets locked/being locked */
    unsigned int passive_target_access_epoch;

    /** Indicates the window is in a pcsw or all access (fence, lock_all) epoch */
    ompi_osc_pt2pt_sync_t all_sync;

    /* ********************* PWSC data ************************ */
    struct ompi_group_t *pw_group;

    /** Number of "count" messages from the remote complete group
        we've received */
    volatile int32_t num_complete_msgs;

    /* ********************* LOCK data ************************ */

    /** Status of the local window lock.  One of 0 (unlocked),
        MPI_LOCK_EXCLUSIVE, or MPI_LOCK_SHARED. */
    int32_t lock_status;

    /** lock for locks_pending list */
    opal_mutex_t locks_pending_lock;

    /** target side list of lock requests we couldn't satisfy yet */
    opal_list_t locks_pending;

    /** origin side list of locks currently outstanding */
    opal_hash_table_t outstanding_locks;

    /** receive fragments */
    struct ompi_osc_pt2pt_receive_t *recv_frags;

    /** number of receive fragments */
    unsigned int recv_frag_count;

    /* enforce accumulate semantics */
    opal_atomic_lock_t accumulate_lock;

    /** accumulate operations pending the accumulation lock */
    opal_list_t pending_acc;

    /** lock for pending_acc */
    opal_mutex_t pending_acc_lock;

    /** Lock for garbage collection lists */
    opal_mutex_t gc_lock;

    /** List of buffers that need to be freed */
    opal_list_t buffer_gc;
};
typedef struct ompi_osc_pt2pt_module_t ompi_osc_pt2pt_module_t;
OMPI_MODULE_DECLSPEC extern ompi_osc_pt2pt_component_t mca_osc_pt2pt_component;

static inline ompi_osc_pt2pt_peer_t *ompi_osc_pt2pt_peer_lookup (ompi_osc_pt2pt_module_t *module,
                                                                 int rank)
{
    ompi_osc_pt2pt_peer_t *peer = NULL;
    (void) opal_hash_table_get_value_uint32 (&module->peer_hash, rank, (void **) &peer);

    if (OPAL_UNLIKELY(NULL == peer)) {
        OPAL_THREAD_LOCK(&module->peer_lock);
        (void) opal_hash_table_get_value_uint32 (&module->peer_hash, rank, (void **) &peer);

        if (NULL == peer) {
            peer = OBJ_NEW(ompi_osc_pt2pt_peer_t);
            peer->rank = rank;

            (void) opal_hash_table_set_value_uint32 (&module->peer_hash, rank, (void *) peer);
        }
        OPAL_THREAD_UNLOCK(&module->peer_lock);
    }

    return peer;
}


struct ompi_osc_pt2pt_pending_t {
    opal_list_item_t super;
    ompi_osc_pt2pt_module_t *module;
    int source;
    ompi_osc_pt2pt_header_t header;
};
typedef struct ompi_osc_pt2pt_pending_t ompi_osc_pt2pt_pending_t;
OBJ_CLASS_DECLARATION(ompi_osc_pt2pt_pending_t);

struct ompi_osc_pt2pt_receive_t {
    opal_list_item_t super;
    ompi_osc_pt2pt_module_t *module;
    ompi_request_t *pml_request;
    void *buffer;
};
typedef struct ompi_osc_pt2pt_receive_t ompi_osc_pt2pt_receive_t;
OBJ_CLASS_DECLARATION(ompi_osc_pt2pt_receive_t);

#define GET_MODULE(win) ((ompi_osc_pt2pt_module_t*) win->w_osc_module)

extern bool ompi_osc_pt2pt_no_locks;

int ompi_osc_pt2pt_attach(struct ompi_win_t *win, void *base, size_t len);
int ompi_osc_pt2pt_detach(struct ompi_win_t *win, const void *base);

int ompi_osc_pt2pt_free(struct ompi_win_t *win);

int ompi_osc_pt2pt_put(const void *origin_addr,
                             int origin_count,
                             struct ompi_datatype_t *origin_dt,
                             int target,
                             ptrdiff_t target_disp,
                             int target_count,
                             struct ompi_datatype_t *target_dt,
                             struct ompi_win_t *win);

int ompi_osc_pt2pt_accumulate(const void *origin_addr,
			      int origin_count,
			      struct ompi_datatype_t *origin_dt,
			      int target,
			      ptrdiff_t target_disp,
			      int target_count,
			      struct ompi_datatype_t *target_dt,
			      struct ompi_op_t *op,
			      struct ompi_win_t *win);

int ompi_osc_pt2pt_get(void *origin_addr,
                       int origin_count,
                       struct ompi_datatype_t *origin_dt,
                       int target,
                       ptrdiff_t target_disp,
                       int target_count,
                       struct ompi_datatype_t *target_dt,
                       struct ompi_win_t *win);

int ompi_osc_pt2pt_compare_and_swap(const void *origin_addr,
                                   const void *compare_addr,
                                   void *result_addr,
                                   struct ompi_datatype_t *dt,
                                   int target,
                                   ptrdiff_t target_disp,
                                   struct ompi_win_t *win);

int ompi_osc_pt2pt_fetch_and_op(const void *origin_addr,
                               void *result_addr,
                               struct ompi_datatype_t *dt,
                               int target,
                               ptrdiff_t target_disp,
                               struct ompi_op_t *op,
                               struct ompi_win_t *win);

int ompi_osc_pt2pt_get_accumulate(const void *origin_addr,
                                 int origin_count,
                                 struct ompi_datatype_t *origin_datatype,
                                 void *result_addr,
                                 int result_count,
                                 struct ompi_datatype_t *result_datatype,
                                 int target_rank,
                                 MPI_Aint target_disp,
                                 int target_count,
                                 struct ompi_datatype_t *target_datatype,
                                 struct ompi_op_t *op,
                                 struct ompi_win_t *win);

int ompi_osc_pt2pt_rput(const void *origin_addr,
                       int origin_count,
                       struct ompi_datatype_t *origin_dt,
                       int target,
                       ptrdiff_t target_disp,
                       int target_count,
                       struct ompi_datatype_t *target_dt,
                       struct ompi_win_t *win,
                       struct ompi_request_t **request);

int ompi_osc_pt2pt_rget(void *origin_addr,
                       int origin_count,
                       struct ompi_datatype_t *origin_dt,
                       int target,
                       ptrdiff_t target_disp,
                       int target_count,
                       struct ompi_datatype_t *target_dt,
                       struct ompi_win_t *win,
                       struct ompi_request_t **request);

int ompi_osc_pt2pt_raccumulate(const void *origin_addr,
                              int origin_count,
                              struct ompi_datatype_t *origin_dt,
                              int target,
                              ptrdiff_t target_disp,
                              int target_count,
                              struct ompi_datatype_t *target_dt,
                              struct ompi_op_t *op,
                              struct ompi_win_t *win,
                              struct ompi_request_t **request);

int ompi_osc_pt2pt_rget_accumulate(const void *origin_addr,
                                  int origin_count,
                                  struct ompi_datatype_t *origin_datatype,
                                  void *result_addr,
                                  int result_count,
                                  struct ompi_datatype_t *result_datatype,
                                  int target_rank,
                                  MPI_Aint target_disp,
                                  int target_count,
                                  struct ompi_datatype_t *target_datatype,
                                  struct ompi_op_t *op,
                                  struct ompi_win_t *win,
                                  struct ompi_request_t **request);

int ompi_osc_pt2pt_fence(int assert, struct ompi_win_t *win);

/* received a post message */
void osc_pt2pt_incoming_post (ompi_osc_pt2pt_module_t *module, int source);

/* received a complete message */
void osc_pt2pt_incoming_complete (ompi_osc_pt2pt_module_t *module, int source, int frag_count);

int ompi_osc_pt2pt_start(struct ompi_group_t *group,
                        int assert,
                        struct ompi_win_t *win);
int ompi_osc_pt2pt_complete(struct ompi_win_t *win);

int ompi_osc_pt2pt_post(struct ompi_group_t *group,
                              int assert,
                              struct ompi_win_t *win);

int ompi_osc_pt2pt_wait(struct ompi_win_t *win);

int ompi_osc_pt2pt_test(struct ompi_win_t *win,
                              int *flag);

int ompi_osc_pt2pt_lock(int lock_type,
                              int target,
                              int assert,
                              struct ompi_win_t *win);

int ompi_osc_pt2pt_unlock(int target,
                                struct ompi_win_t *win);

int ompi_osc_pt2pt_lock_all(int assert,
                           struct ompi_win_t *win);

int ompi_osc_pt2pt_unlock_all(struct ompi_win_t *win);

int ompi_osc_pt2pt_sync(struct ompi_win_t *win);

int ompi_osc_pt2pt_flush(int target,
                        struct ompi_win_t *win);
int ompi_osc_pt2pt_flush_all(struct ompi_win_t *win);
int ompi_osc_pt2pt_flush_local(int target,
                              struct ompi_win_t *win);
int ompi_osc_pt2pt_flush_local_all(struct ompi_win_t *win);

int ompi_osc_pt2pt_set_info(struct ompi_win_t *win, struct opal_info_t *info);
int ompi_osc_pt2pt_get_info(struct ompi_win_t *win, struct opal_info_t **info_used);

int ompi_osc_pt2pt_component_irecv(ompi_osc_pt2pt_module_t *module,
                                  void *buf,
                                  size_t count,
                                  struct ompi_datatype_t *datatype,
                                  int src,
                                  int tag,
                                  struct ompi_communicator_t *comm);

int ompi_osc_pt2pt_lock_remote (ompi_osc_pt2pt_module_t *module, int target, ompi_osc_pt2pt_sync_t *lock);

/**
 * ompi_osc_pt2pt_progress_pending_acc:
 *
 * @short Progress one pending accumulation or compare and swap operation.
 *
 * @param[in] module   - OSC PT2PT module
 *
 * @long If the accumulation lock can be aquired progress one pending
 *       accumulate or compare and swap operation.
 */
int ompi_osc_pt2pt_progress_pending_acc (ompi_osc_pt2pt_module_t *module);


/**
 * mark_incoming_completion:
 *
 * @short Increment incoming completeion count.
 *
 * @param[in] module - OSC PT2PT module
 * @param[in] source - Passive target source or MPI_PROC_NULL (active target)
 *
 * @long This function incremements either the passive or active incoming counts.
 *       If the count reaches the signal count we signal the module's condition.
 *       This function uses atomics if necessary so it is not necessary to hold
 *       the module lock before calling this function.
 */
static inline void mark_incoming_completion (ompi_osc_pt2pt_module_t *module, int source)
{
    int32_t new_value;

    if (MPI_PROC_NULL == source) {
        OPAL_OUTPUT_VERBOSE((50, ompi_osc_base_framework.framework_output,
                             "mark_incoming_completion marking active incoming complete. module %p, count = %d",
                             (void *) module, (int) module->active_incoming_frag_count + 1));
        new_value = OPAL_THREAD_ADD_FETCH32(&module->active_incoming_frag_count, 1);
        if (new_value >= 0) {
            OPAL_THREAD_LOCK(&module->lock);
            opal_condition_broadcast(&module->cond);
            OPAL_THREAD_UNLOCK(&module->lock);
        }
    } else {
        ompi_osc_pt2pt_peer_t *peer = ompi_osc_pt2pt_peer_lookup (module, source);

        OPAL_OUTPUT_VERBOSE((50, ompi_osc_base_framework.framework_output,
                             "mark_incoming_completion marking passive incoming complete. module %p, source = %d, count = %d",
                             (void *) module, source, (int) peer->passive_incoming_frag_count + 1));
        new_value = OPAL_THREAD_ADD_FETCH32((int32_t *) &peer->passive_incoming_frag_count, 1);
        if (0 == new_value) {
            OPAL_THREAD_LOCK(&module->lock);
            opal_condition_broadcast(&module->cond);
            OPAL_THREAD_UNLOCK(&module->lock);
        }
    }
}

/**
 * mark_outgoing_completion:
 *
 * @short Increment outgoing count.
 *
 * @param[in] module - OSC PT2PT module
 *
 * @long This function is used to signal that an outgoing send is complete. It
 *       incrememnts only the outgoing fragment count and signals the module
 *       condition the fragment count is >= the signal count. This function
 *       uses atomics if necessary so it is not necessary to hold the module
 *       lock before calling this function.
 */
static inline void mark_outgoing_completion (ompi_osc_pt2pt_module_t *module)
{
    int32_t new_value = OPAL_THREAD_ADD_FETCH32((int32_t *) &module->outgoing_frag_count, 1);
    OPAL_OUTPUT_VERBOSE((50, ompi_osc_base_framework.framework_output,
                         "mark_outgoing_completion: outgoing_frag_count = %d", new_value));
    if (new_value >= 0) {
        OPAL_THREAD_LOCK(&module->lock);
        opal_condition_broadcast(&module->cond);
        OPAL_THREAD_UNLOCK(&module->lock);
    }
}

/**
 * ompi_osc_signal_outgoing:
 *
 * @short Increment outgoing signal counters.
 *
 * @param[in] module - OSC PT2PT module
 * @param[in] target - Passive target rank or MPI_PROC_NULL (active target)
 * @param[in] count  - Number of outgoing messages to signal.
 *
 * @long This function uses atomics if necessary so it is not necessary to hold
 *       the module lock before calling this function.
 */
static inline void ompi_osc_signal_outgoing (ompi_osc_pt2pt_module_t *module, int target, int count)
{
    OPAL_THREAD_ADD_FETCH32((int32_t *) &module->outgoing_frag_count, -count);
    if (MPI_PROC_NULL != target) {
        OPAL_OUTPUT_VERBOSE((50, ompi_osc_base_framework.framework_output,
                             "ompi_osc_signal_outgoing_passive: target = %d, count = %d, total = %d", target,
                             count, module->epoch_outgoing_frag_count[target] + count));
        OPAL_THREAD_ADD_FETCH32((int32_t *) (module->epoch_outgoing_frag_count + target), count);
    }
}

/**
 * osc_pt2pt_copy_on_recv:
 *
 * @short Helper function. Copies data from source to target through the
 * convertor.
 *
 * @param[in] target     - destination for the data
 * @param[in] source     - packed data
 * @param[in] source_len - length of source buffer
 * @param[in] proc       - proc that packed the source data
 * @param[in] count      - count of datatype items
 * @param[in] datatype   - datatype used for unpacking
 *
 * @long This functions unpacks data from the source buffer into the target
 *       buffer. The copy is done with a convertor generated from proc,
 *       datatype, and count.
 */
static inline void osc_pt2pt_copy_on_recv (void *target, void *source, size_t source_len, ompi_proc_t *proc,
                                          int count, ompi_datatype_t *datatype)
{
    opal_convertor_t convertor;
    uint32_t iov_count = 1;
    struct iovec iov;
    size_t max_data;

    /* create convertor */
    OBJ_CONSTRUCT(&convertor, opal_convertor_t);

    /* initialize convertor */
    opal_convertor_copy_and_prepare_for_recv(proc->super.proc_convertor, &datatype->super, count, target,
                                             0, &convertor);

    iov.iov_len  = source_len;
    iov.iov_base = (IOVBASE_TYPE *) source;
    max_data     = iov.iov_len;
    MEMCHECKER(memchecker_convertor_call(&opal_memchecker_base_mem_defined, &convertor));

    opal_convertor_unpack (&convertor, &iov, &iov_count, &max_data);

    MEMCHECKER(memchecker_convertor_call(&opal_memchecker_base_mem_noaccess, &convertor));

    OBJ_DESTRUCT(&convertor);
}

/**
 * osc_pt2pt_copy_for_send:
 *
 * @short: Helper function. Copies data from source to target through the
 * convertor.
 *
 * @param[in] target     - destination for the packed data
 * @param[in] target_len - length of the target buffer
 * @param[in] source     - original data
 * @param[in] proc       - proc this data will be sent to
 * @param[in] count      - count of datatype items
 * @param[in] datatype   - datatype used for packing
 *
 * @long This functions packs data from the source buffer into the target
 *       buffer. The copy is done with a convertor generated from proc,
 *       datatype, and count.
 */
static inline void osc_pt2pt_copy_for_send (void *target, size_t target_len, const void *source, ompi_proc_t *proc,
                                           int count, ompi_datatype_t *datatype)
{
    opal_convertor_t convertor;
    uint32_t iov_count = 1;
    struct iovec iov;
    size_t max_data;

    OBJ_CONSTRUCT(&convertor, opal_convertor_t);

    opal_convertor_copy_and_prepare_for_send(proc->super.proc_convertor, &datatype->super,
                                             count, source, 0, &convertor);

    iov.iov_len = target_len;
    iov.iov_base = (IOVBASE_TYPE *) target;
    opal_convertor_pack(&convertor, &iov, &iov_count, &max_data);

    OBJ_DESTRUCT(&convertor);
}

/**
 * osc_pt2pt_gc_clean:
 *
 * @short Release finished PML requests and accumulate buffers.
 *
 * @long This function exists because it is not possible to free a buffer from
 *     a request completion callback. We instead put requests and buffers on the
 *     module's garbage collection lists and release then at a later time.
 */
static inline void osc_pt2pt_gc_clean (ompi_osc_pt2pt_module_t *module)
{
    opal_list_item_t *item;

    OPAL_THREAD_LOCK(&module->gc_lock);
    while (NULL != (item = opal_list_remove_first (&module->buffer_gc))) {
        OBJ_RELEASE(item);
    }
    OPAL_THREAD_UNLOCK(&module->gc_lock);
}

static inline void osc_pt2pt_gc_add_buffer (ompi_osc_pt2pt_module_t *module, opal_list_item_t *buffer)
{
    OPAL_THREAD_SCOPED_LOCK(&module->gc_lock,
                            opal_list_append (&module->buffer_gc, buffer));
}

static inline void osc_pt2pt_add_pending (ompi_osc_pt2pt_pending_t *pending)
{
    OPAL_THREAD_SCOPED_LOCK(&mca_osc_pt2pt_component.pending_operations_lock,
                            opal_list_append (&mca_osc_pt2pt_component.pending_operations, &pending->super));
}

#define OSC_PT2PT_FRAG_TAG   0x10000
#define OSC_PT2PT_FRAG_MASK  0x0ffff

/**
 * get_tag:
 *
 * @short Get a send/recv base tag for large memory operations.
 *
 * @param[in] module - OSC PT2PT module
 *
 * @long This function acquires a 16-bit tag for use with large memory operations. The
 *       tag will be odd or even depending on if this is in a passive target access
 *       or not. An actual tag that will be passed to PML send/recv function is given
 *       by tag_to_target or tag_to_origin function depending on the communication
 *       direction.
 */
static inline int get_tag(ompi_osc_pt2pt_module_t *module)
{
    /* the LSB of the tag is used be the receiver to determine if the
       message is a passive or active target (ie, where to mark
       completion). */
    int32_t tmp = OPAL_THREAD_ADD_FETCH32((volatile int32_t *) &module->tag_counter, 4);
    return (tmp & OSC_PT2PT_FRAG_MASK) | !!(module->passive_target_access_epoch);
}

/**
 * tag_to_target:
 *
 * @short Get a tag used for PML send/recv communication from an origin to a target.
 *
 * @param[in] tag - base tag given by get_tag function.
 */
static inline int tag_to_target(int tag)
{
    /* (returned_tag >> 1) & 0x1 == 0 */
    return tag + 0;
}

/**
 * tag_to_origin:
 *
 * @short Get a tag used for PML send/recv communication from a target to an origin.
 *
 * @param[in] tag - base tag given by get_tag function.
 */
static inline int tag_to_origin(int tag)
{
    /* (returned_tag >> 1) & 0x1 == 1 */
    return tag + 2;
}

/**
 * ompi_osc_pt2pt_accumulate_lock:
 *
 * @short Internal function that spins until the accumulation lock has
 *        been aquired.
 *
 * @param[in] module - OSC PT2PT module
 *
 * @returns 0
 *
 * @long This functions blocks until the accumulation lock has been aquired. This
 *       behavior is only acceptable from a user-level call as blocking in a
 *       callback may cause deadlock. If a callback needs the accumulate lock and
 *       it is not available it should be placed on the pending_acc list of the
 *       module. It will be released by ompi_osc_pt2pt_accumulate_unlock().
 */
static inline int ompi_osc_pt2pt_accumulate_lock (ompi_osc_pt2pt_module_t *module)
{
    while (opal_atomic_trylock (&module->accumulate_lock)) {
        opal_progress ();
    }

    return 0;
}

/**
 * ompi_osc_pt2pt_accumulate_trylock:
 *
 * @short Try to aquire the accumulation lock.
 *
 * @param[in] module - OSC PT2PT module
 *
 * @returns 0 if the accumulation lock was aquired
 * @returns 1 if the lock was not available
 *
 * @long This function will try to aquire the accumulation lock. This function
 *       is safe to call from a callback.
 */
static inline int ompi_osc_pt2pt_accumulate_trylock (ompi_osc_pt2pt_module_t *module)
{
    return opal_atomic_trylock (&module->accumulate_lock);
}

/**
 * @brief check if this process has this process is in a passive target access epoch
 *
 * @param[in] module          osc pt2pt module
 */
static inline bool ompi_osc_pt2pt_in_passive_epoch (ompi_osc_pt2pt_module_t *module)
{
    return 0 != module->passive_target_access_epoch;
}

/**
 * ompi_osc_pt2pt_accumulate_unlock:
 *
 * @short Unlock the accumulation lock and release a pending accumulation operation.
 *
 * @param[in] module - OSC PT2PT module
 *
 * @long This function unlocks the accumulation lock and release a single pending
 *       accumulation operation if one exists. This function may be called recursively.
 */
static inline void ompi_osc_pt2pt_accumulate_unlock (ompi_osc_pt2pt_module_t *module)
{
    opal_atomic_unlock (&module->accumulate_lock);
    if (0 != opal_list_get_size (&module->pending_acc)) {
        ompi_osc_pt2pt_progress_pending_acc (module);
    }
}

/**
 * Find the first outstanding lock of the target.
 *
 * @param[in]  module   osc pt2pt module
 * @param[in]  target   target rank
 * @param[out] peer     peer object associated with the target
 *
 * @returns an outstanding lock on success
 *
 * This function looks for an outstanding lock to the target. If a lock exists it is returned.
 */
static inline ompi_osc_pt2pt_sync_t *ompi_osc_pt2pt_module_lock_find (ompi_osc_pt2pt_module_t *module, int target,
                                                                      ompi_osc_pt2pt_peer_t **peer)
{
    ompi_osc_pt2pt_sync_t *outstanding_lock = NULL;

    (void) opal_hash_table_get_value_uint32 (&module->outstanding_locks, (uint32_t) target, (void **) &outstanding_lock);
    if (NULL != outstanding_lock && peer) {
        *peer = outstanding_lock->peer_list.peer;
    }

    return outstanding_lock;
}

/**
 * Add an outstanding lock
 *
 * @param[in] module   osc pt2pt module
 * @param[in] lock     lock object
 *
 * This function inserts a lock object to the list of outstanding locks. The caller must be holding the module
 * lock.
 */
static inline void ompi_osc_pt2pt_module_lock_insert (struct ompi_osc_pt2pt_module_t *module, ompi_osc_pt2pt_sync_t *lock)
{
    (void) opal_hash_table_set_value_uint32 (&module->outstanding_locks, (uint32_t) lock->sync.lock.target, (void *) lock);
}


/**
 * Remove an outstanding lock
 *
 * @param[in] module   osc pt2pt module
 * @param[in] lock     lock object
 *
 * This function removes a lock object to the list of outstanding locks. The caller must be holding the module
 * lock.
 */
static inline void ompi_osc_pt2pt_module_lock_remove (struct ompi_osc_pt2pt_module_t *module, ompi_osc_pt2pt_sync_t *lock)
{

    (void) opal_hash_table_remove_value_uint32 (&module->outstanding_locks, (uint32_t) lock->sync.lock.target);
}

/**
 * Lookup a synchronization object associated with the target
 *
 * @param[in] module   osc pt2pt module
 * @param[in] target   target rank
 * @param[out] peer    peer object
 *
 * @returns NULL if the target is not locked, fenced, or part of a pscw sync
 * @returns synchronization object on success
 *
 * This function returns the synchronization object associated with an access epoch for
 * the target. If the target is not part of any current access epoch then NULL is returned.
 */
static inline ompi_osc_pt2pt_sync_t *ompi_osc_pt2pt_module_sync_lookup (ompi_osc_pt2pt_module_t *module, int target,
                                                                        struct ompi_osc_pt2pt_peer_t **peer)
{
    ompi_osc_pt2pt_peer_t *tmp;

    if (NULL == peer) {
        peer = &tmp;
    }

    OPAL_OUTPUT_VERBOSE((50, ompi_osc_base_framework.framework_output,
                         "osc/pt2pt: looking for synchronization object for target %d", target));

    switch (module->all_sync.type) {
    case OMPI_OSC_PT2PT_SYNC_TYPE_NONE:
        if (!module->no_locks) {
            return ompi_osc_pt2pt_module_lock_find (module, target, peer);
        }

        return NULL;
    case OMPI_OSC_PT2PT_SYNC_TYPE_FENCE:
    case OMPI_OSC_PT2PT_SYNC_TYPE_LOCK:
        OPAL_OUTPUT_VERBOSE((50, ompi_osc_base_framework.framework_output,
                             "osc/pt2pt: found fence/lock_all access epoch for target %d", target));

        /* fence epoch is now active */
        module->all_sync.epoch_active = true;
        *peer = ompi_osc_pt2pt_peer_lookup (module, target);
        if (OMPI_OSC_PT2PT_SYNC_TYPE_LOCK == module->all_sync.type && !ompi_osc_pt2pt_peer_locked (*peer)) {
            (void) ompi_osc_pt2pt_lock_remote (module, target, &module->all_sync);
        }

        return &module->all_sync;
    case OMPI_OSC_PT2PT_SYNC_TYPE_PSCW:
        if (ompi_osc_pt2pt_sync_pscw_peer (module, target, peer)) {
            OPAL_OUTPUT_VERBOSE((50, ompi_osc_base_framework.framework_output,
                                 "osc/pt2pt: found PSCW access epoch target for %d", target));
            return &module->all_sync;
        }
    }

    return NULL;
}

/**
 * @brief check if an access epoch is active
 *
 * @param[in] module        osc pt2pt module
 *
 * @returns true if any type of access epoch is active
 * @returns false otherwise
 *
 * This function is used to check for conflicting access epochs.
 */
static inline bool ompi_osc_pt2pt_access_epoch_active (ompi_osc_pt2pt_module_t *module)
{
    return (module->all_sync.epoch_active || ompi_osc_pt2pt_in_passive_epoch (module));
}

static inline bool ompi_osc_pt2pt_peer_sends_active (ompi_osc_pt2pt_module_t *module, int rank)
{
    ompi_osc_pt2pt_sync_t *sync;
    ompi_osc_pt2pt_peer_t *peer;

    sync = ompi_osc_pt2pt_module_sync_lookup (module, rank, &peer);
    if (!sync) {
        return false;
    }

    return sync->eager_send_active || ompi_osc_pt2pt_peer_eager_active (peer);
}

END_C_DECLS

#endif /* OMPI_OSC_PT2PT_H */