ports/openmpi
ports/openmpi
1
1
Форкнуть 0
Код Релизы Активность

Improve shared-memory allocation:

Просмотр исходного кода 
* compute mmap-file size more wisely and pass requested size to allocator
* change MCA parameters:
  - get rid of mpool_sm_per_peer_size
  - get rid of mpool_sm_max_size
  - set default mpool_sm_min_size to 0
* no longer pad sm allocations to page boundaries
* have sm_btl_first_time_init check return codes on free-list creations

Have mca_btl_sm_prepare_src() check to see if it can allocate an EAGER fragment
rather than a MAX fragment if the smaller size works.

Remove ompi/class/ompi_[circular_buffer_]fifo.h and references thereto.

Remove opal/util/pow2.[c|h] and references thereto.

This commit was SVN r20614.
Этот коммит содержится в:
Eugene Loh 2009-02-20 19:51:57 +00:00
родитель 47f875bca1
Коммит 463f11f993
16 изменённых файлов: 63 добавлений и 1622 удалений
Показать статистику Скачать Patch файл Скачать Diff файл Показать все файлы Свернуть все файлы

4
ompi/class/Makefile.am
Просмотреть файл

@ -20,10 +20,8 @@
# This makefile.am does not stand on its own - it is included from ompi/Makefile.am
headers += \
class/ompi_circular_buffer_fifo.h \
class/ompi_fifo.h \
class/ompi_free_list.h \
class/ompi_bitmap.h \
class/ompi_free_list.h \
class/ompi_rb_tree.h \
class/ompi_seq_tracker.h

373
ompi/class/ompi_circular_buffer_fifo.h
Просмотреть файл

@ -1,373 +0,0 @@
/*
* Copyright (c) 2004-2005 The Trustees of Indiana University and Indiana
* University Research and Technology
* Corporation. All rights reserved.
* Copyright (c) 2004-2006 The University of Tennessee and The University
* of Tennessee Research Foundation. 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$
*
* Additional copyrights may follow
*
* $HEADER$
*/
#ifndef _OMPI_CIRCULAR_BUFFER_FIFO
#define _OMPI_CIRCULAR_BUFFER_FIFO
#ifdef HAVE_UNISTD_H
#include <unistd.h> /* for getpagesize() */
#endif
#include "ompi/constants.h"
#include "opal/sys/cache.h"
#include "opal/sys/atomic.h"
#include "ompi/mca/mpool/mpool.h"
#include "opal/util/pow2.h"
/** @file
*
* This defines a set of functions to create, and manipulate a FIFO
* set up in a circular buffer. FIFO elements are assumed to be
* pointers. Pointers are written to the head, and read from the
* tail. For thread safety, a spin lock is provided in the
* ompi_cb_fifo_ctl_t structure, but it's use must be managed by
* the calling routines - this is not by these set of routines.
* Queues are addressed relative to an offset from the base of
* a memory pool, in this way, different processes with different
* base addresses can access these queue at the same time.
*/
/* error code */
#define OMPI_CB_ERROR -1
#define OMPI_CB_FREE (void *)-2
#define OMPI_CB_RESERVED (void *)-3
#define OMPI_CB_NULL (void *)-4
/*
* Structure used to keep track of the fifo status
*/
struct ompi_cb_fifo_ctl_t {
/* spin-lock for access control */
opal_atomic_lock_t lock;
/* current queue index */
volatile int fifo_index;
/* number of entries that have been used, but not invalidated. used
* for lazy resource reclamation */
int num_to_clear;
};
typedef struct ompi_cb_fifo_ctl_t ompi_cb_fifo_ctl_t;
/* data structure used to describe the fifo */
struct ompi_cb_fifo_t {
/* head of queue - where next entry will be written (sender address)*/
ompi_cb_fifo_ctl_t *head;
/* tail of queue - next element to read (receiver address) */
ompi_cb_fifo_ctl_t *tail;
/* head of queue - where next entry will be written (receiver address) */
ompi_cb_fifo_ctl_t *recv_head;
/* circular buffer array (sender address) */
volatile void **queue;
/* circular buffer array (receiver address) */
volatile void **recv_queue;
/* frequency of lazy free */
int lazy_free_frequency;
/* mask - to handle wrap around */
unsigned int mask;
};
typedef struct ompi_cb_fifo_t ompi_cb_fifo_t;
/**
* Return the fifo size
*
* @param fifo Pointer to data structure defining this fifo (IN)
*
* @returncode fifo size
*
*/
static inline int ompi_cb_fifo_size(ompi_cb_fifo_t *fifo) {
return fifo->mask + 1;
}
/**
* Initialize a fifo
*
* @param size_of_fifo Length of fifo array (IN)
*
* @param fifo_memory_locality_index Locality index to apply to
* the fifo array. Not currently
* in use (IN)
*
* @param tail_memory_locality_index Locality index to apply to the
* head control structure. Not
* currently in use (IN)
*
* @param tail_memory_locality_index Locality index to apply to the
* tail control structure. Not
* currently in use (IN)
*
* @param fifo Pointer to data structure defining this fifo (IN)
*
* @param memory_allocator Pointer to the memory allocator to use
* to allocate memory for this fifo. (IN)
*
* @returncode Error code
*
*/
static inline int ompi_cb_fifo_init(int size_of_fifo,
int lazy_free_freq,
mca_mpool_base_module_t *head_mpool,
mca_mpool_base_module_t *tail_mpool,
ompi_cb_fifo_t *fifo, ptrdiff_t offset)
{
int i, size;
char *buf;
/* verify that size is power of 2, and greater that 0 - if not,
* round up */
if(size_of_fifo <= 0) {
return OMPI_ERROR;
}
/* set fifo size */
size = opal_round_up_to_nearest_pow2(size_of_fifo);
/* set lazy free frequence */
if((lazy_free_freq <= 0) || (lazy_free_freq > size)) {
return OMPI_ERROR;
}
fifo->lazy_free_frequency = lazy_free_freq;
/* this will be used to mask off the higher order bits,
* and use the & operator for the wrap-around */
fifo->mask = (size - 1);
/* allocate fifo array */
buf = (char *) tail_mpool->mpool_alloc(tail_mpool,
sizeof(void *) * size + CACHE_LINE_SIZE, getpagesize(), 0, NULL);
if (NULL == buf) {
return OMPI_ERR_OUT_OF_RESOURCE;
}
fifo->queue = (volatile void**)(buf + CACHE_LINE_SIZE);
/* buffer address in a receiver address space */
fifo->recv_queue = (volatile void**)((char*)fifo->queue - offset);
/* initialize the queue entries */
for (i = 0; i < size; i++) {
fifo->queue[i] = OMPI_CB_FREE;
}
fifo->tail = (ompi_cb_fifo_ctl_t*)buf;
/* initialize the tail structure */
opal_atomic_unlock(&(fifo->tail->lock));
fifo->tail->fifo_index=0;
fifo->tail->num_to_clear=0;
/* recalculate tail address in a receiver address space */
fifo->tail = (ompi_cb_fifo_ctl_t*)((char*)fifo->tail - offset);
fifo->head = (ompi_cb_fifo_ctl_t*)head_mpool->mpool_alloc(head_mpool,
sizeof(ompi_cb_fifo_ctl_t), getpagesize(), 0, NULL);
if ( NULL == fifo->head ){
return OMPI_ERR_OUT_OF_RESOURCE;
}
/* head address in a receiver address space */
fifo->recv_head = (ompi_cb_fifo_ctl_t*)((char*)fifo->head - offset);
/* initialize the head structure */
opal_atomic_unlock(&(fifo->head->lock));
fifo->head->fifo_index=0;
fifo->head->num_to_clear=0;
/* return */
return OMPI_SUCCESS;
}
/**
* function to cleanup the fifo
*
* @param fifo Pointer to data structure defining this fifo (IN)
*
* @param memory_allocator Pointer to the memory allocator to use
* to allocate memory for this fifo. (IN)
*
*/
static inline int ompi_cb_fifo_free(ompi_cb_fifo_t *fifo,
mca_mpool_base_module_t *memory_allocator)
{
char *ptr;
/* make sure null fifo is not passed in */
if(NULL == fifo) {
return OMPI_ERROR;
}
/* free fifo array */
if(OMPI_CB_NULL != fifo->head){
ptr=(char *)(fifo->head);
memory_allocator->mpool_free(memory_allocator, ptr, NULL);
fifo->queue = (volatile void**)OMPI_CB_NULL;
}
return OMPI_SUCCESS;
}
/**
* Write pointer to the specified slot
*
* @param slot Slot index (IN)
*
* @param data Pointer value to write in specified slot (IN)
*
* @param fifo Pointer to data structure defining this fifo (IN)
*
* @returncode Slot index to which data is written
*
*/
static inline int ompi_cb_fifo_write_to_slot(int slot, void* data,
ompi_cb_fifo_t *fifo)
{
volatile void **ptr;
/* make sure that this slot is already reserved */
ptr=fifo->queue;
if (ptr[slot] == OMPI_CB_RESERVED ) {
opal_atomic_rmb();
ptr[slot] = data;
opal_atomic_wmb();
return slot;
}
return OMPI_CB_ERROR;
}
/**
* Try to write pointer to the head of the queue
*
* @param data Pointer value to write in specified slot (IN)
*
* @param fifo Pointer to data structure defining this fifo (IN)
*
* @returncode Slot index to which data is written
*
*/
static inline int ompi_cb_fifo_write_to_head(void *data, ompi_cb_fifo_t *fifo)
{
volatile void **ptr;
ompi_cb_fifo_ctl_t *h_ptr;
int index;
h_ptr=fifo->head;
ptr=fifo->queue;
index = h_ptr->fifo_index;
/* make sure the head is pointing at a free element */
if (ptr[index] == OMPI_CB_FREE) {
opal_atomic_rmb();
ptr[index] = data;
opal_atomic_wmb();
h_ptr->fifo_index = (index + 1) & fifo->mask;
return index;
}
/* return */
return OMPI_CB_ERROR;
}
/**
* Reserve slot in the fifo array
*
* @param fifo Pointer to data structure defining this fifo (IN)
*
* @returncode Slot index to which data is written
*
* @returncode OMPI_CB_ERROR failed to allocate index
*
*/
static inline int ompi_cb_fifo_get_slot(ompi_cb_fifo_t *fifo)
{
return ompi_cb_fifo_write_to_head(OMPI_CB_RESERVED, fifo);
}
/**
* Try to read pointer from the tail of the queue
*
* @param data Pointer to where data was be written (OUT)
*
* @param fifo Pointer to data structure defining this fifo (IN)
*
* @param flush_entries_read force the lazy free to happen (IN)
*
* @param queue_empty checks to see if the fifo is empty, but only if
* flush_entries_read is set (OUT)
*
* @returncode Slot index to which data is written
*
*/
static inline void *ompi_cb_fifo_read_from_tail(
ompi_cb_fifo_t *fifo,
bool flush_entries_read, bool *queue_empty)
{
int index, i;
volatile void **q_ptr;
ompi_cb_fifo_ctl_t *t_ptr;
void *read_from_tail;
*queue_empty=false;
t_ptr=fifo->tail;
q_ptr=fifo->recv_queue;
index = t_ptr->fifo_index;
read_from_tail = (void *)q_ptr[index];
opal_atomic_rmb();
/* check to see that the data is valid */
if ((read_from_tail == OMPI_CB_FREE) ||
(read_from_tail == OMPI_CB_RESERVED)) {
return (void*)OMPI_CB_FREE;
}
/* increment counter for later lazy free */
t_ptr->num_to_clear++;
t_ptr->fifo_index = (index + 1) & fifo->mask;
/* check to see if time to do a lazy free of queue slots */
if ( (t_ptr->num_to_clear == fifo->lazy_free_frequency) ||
flush_entries_read ) {
ompi_cb_fifo_ctl_t *h_ptr = fifo->recv_head;
index = (index - t_ptr->num_to_clear + 1) & fifo->mask;
for (i = 0; i < t_ptr->num_to_clear; i++) {
q_ptr[index] = OMPI_CB_FREE;
index = (index + 1) & fifo->mask;
}
opal_atomic_wmb();
t_ptr->num_to_clear = 0;
/* check to see if queue is empty */
if( flush_entries_read &&
(t_ptr->fifo_index == h_ptr->fifo_index) ) {
*queue_empty=true;
}
}
return read_from_tail;
}
#endif /* !_OMPI_CIRCULAR_BUFFER_FIFO */

445
ompi/class/ompi_fifo.h
Просмотреть файл

@ -1,445 +0,0 @@
/*
* Copyright (c) 2004-2005 The Trustees of Indiana University and Indiana
* University Research and Technology
* Corporation. All rights reserved.
* Copyright (c) 2004-2006 The University of Tennessee and The University
* of Tennessee Research Foundation. 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$
*
* Additional copyrights may follow
*
* $HEADER$
*/
#ifndef _OMPI_FIFO
#define _OMPI_FIFO
#include "ompi/constants.h"
#include "opal/sys/cache.h"
#include "opal/sys/atomic.h"
#include "ompi/mca/mpool/mpool.h"
#include "ompi/class/ompi_circular_buffer_fifo.h"
/** @file
*
* This defines a set of functions to create, and manipulate a FIFO
* implemented as a link list of circular buffer FIFO's. FIFO
* elements are assumed to be pointers. Pointers are written to the
* head, and read from the tail. For thread safety, a spin lock is
* provided in the !!!!!ompi_cb_fifo_ctl_t!!!! structure, but it's use
* must be managed by the calling routines - this is not by these set
* of routines. When a write to a circular buffer queue will overflow
* that queue, the next circular buffer queue if the link list is
* used, if it is empty, or a new one is inserted into the list.
*
* This set of routines is currently exclusively used by the sm btl,
* and has been tailored to meet its needs (i.e., it is probably not
* suitable as a general purpose fifo).
*
* Before describing any further, a note about mmap() is in order.
* mmap() is used to create/attach shared memory segments to a
* process. It is used by OMPI to manage shared memory.
* Specifically, each process ends up calling mmap() to create or
* attach shared memory; the end result is that multiple processes
* have the same shared memory segment attached to their process.
* This shared memory is therefore used here in the fifo code.
*
* However, it is important to note that when attaching the same
* shared memory segment to multiple processes, mmap() does *not* need
* to return the same virtual address to the beginning of the shared
* memory segment to each process. That is, the virtual address
* returned in each process will point to the same shared memory
* segment as all others, but its virtual address value may be
* different. Specifically, process A may get the value X back from
* mmap(), while process B, who attached the same shared memory
* segment as process A, may get back the value Y from mmap().
* Process C may attach the same shared memory segment and get back
* value X from mmap(). This is perfectly legal mmap() behavior.
*
* As such, our code -- including this fifo code -- needs to be able
* to handle the cases where the base address is the same and the
* cases where it is different.
*
* There are four main interface functions:
*
* ompi_fifo_init_same_base_addr(): create a fifo for the case where
* the creating process shares a common shared memory segment base
* address.
*
* ompi_fifo_write_to_head_same_base_addr(): write a value to the head
* of the fifo for the case where the shared memory segment virtual
* address is the same as the process who created the fifo.
*
* ompi_fifo_read_from_tail_same_base_addr(): read a value from the
* tail of the fifo for the case where the shared memory segment
* virtual address is the same as the process who created the fifo.
*
* ompi_fifo_read_from_tail(): read a value from the tail of the fifo
* for the case where the shared memory segment virtual address is
* *not* the same as the process who created the fifo.
*
* The data structures used in these fifos are carefully structured to
* be lockless, even when used in shared memory. However, this is
* predicated upon there being only exactly *ONE* concurrent writer
* and *ONE* concurrent reader (in terms of the sm btl, two fifos are
* established between each process pair; one for data flowing A->B
* and one for data flowing B->A). Hence, the writer always looks at
* the "head" and the reader always looks at the "tail."
*
* The general scheme of the fifo is that this class is an upper-level
* manager for the ompi_circular_buffer_fifo_t class. When an
* ompi_fifo_t instance is created, it creates an
* ompi_circular_buffer_fifo_t. Items can then be put into the fifo
* until the circular buffer fills up (i.e., items have not been
* removed from the circular buffer, so it gets full). The
* ompi_fifo_t class will manage this case and create another
* circular_buffer and start putting items in there. This can
* continue indefinitely; the ompi_fifo_t class will create a linked
* list of circular buffers in order to create storage for any items
* that need to be put in the fifo.
*
* The tail will then read from these circular buffers in order,
* draining them as it goes.
*
* The linked list of circular buffers is created in a circle, so if
* you have N circular buffers, the fill pattern will essentially go
* in a circle (assuming that the reader is dutifully reading/draining
* behind the writer). Yes, this means that we have a ring of
* circular buffers. A single circular buffer is treated as a
* standalone entitle, a reader/writer pair can utilize it
* indefinitely; they will never move on to the next circular buffer
* unless the writer gets so far ahead of the reader that the current
* circular buffer fills up and the writer moves on to the next
* circular buffer. In this case, the reader will eventually drain
* the current circular buffer and then move on to the next circular
* buffer (and assumedly eventually catch up to the writer).
*
* The natural question of "why bother doing this instead of just
* having an array of pointers that you realloc?" arises. The intent
* with this class is to have a lockless structure -- using realloc,
* by definition, means that you would have to lock every single
* access to the array to ensure that it doesn't get realloc'ed from
* underneath you. This is definitely something we want to avoid for
* performance reasons.
*
* Hence, once you get your head wrapped around this scheme, it
* actually does make sense (and give good performance).
*
********************************* NOTE *******************************
*
* Although the scheme is designed to be lockless, there is currently
* one lock used in this scheme. There is a nasty race condition
* between multiple processes that if the writer fills up a circular
* buffer before anything this read, it can make the decision to
* create a new circular buffer (because that one is full). However,
* if, at the same time, the reader takes over -- after the decision
* has been made to make a new circular buffer, and after some [but
* not all] of the data fields are updated to reflect this -- the
* reader can drain the entire current circular buffer, obviating the
* need to make a new circular buffer (because there's now space
* available in the current one). The reader will then update some
* data fields in the fifo.
*
* This can lead to a fifo management consistency error -- the reader
* thinks it is advancing to the next circular bufer but it really
* ends up back on the same circular buffer (because the writer had
* not updated the "next cb" field yet). The reader is then stuck in
* a cb where nothing will arrive until the writer loops all the way
* around (i.e., through all other existing circular buffers) and
* starts writing to the circular buffer where the reader is waiting.
* This effectively means that the reader will miss a lot of messages.
*
* So we had to add a lock to protect this -- when the writer decides
* to make a new circular buffer and when the reader decides to move
* to the new circular buffer. It is a rather coarse-grained lock; it
* convers a relatively large chunk of code in the writing_to_head
* function, but, interestingly enough, this seems to create *better*
* performance for sending large messages via shared memory (i.e.,
* netpipe graphs with and without this lock show that using the lock
* gives better overall bandwidth for large messages). We do lose a
* bit of overall bandwidth for mid-range message sizes, though.
*
* We feel that this lock can probably be eventually removed from the
* implementation; we recognized this race condition and ran out of
* time to fix is properly (i.e., in a lockless way). As such, we
* employed a lock to serialize the access and protect it that way.
* This issue should be revisited someday to remove the lock.
*
* See the notes in the writer function for more details on the lock.
*/
/*
* Structure by the the ompi_fifo routines to keep track of some
* extra queue information not needed by the ompi_cb_fifo routines.
*/
struct ompi_cb_fifo_wrapper_t {
/* pointer to ompi_cb_fifo_ctl_t structure in use */
ompi_cb_fifo_t cb_fifo;
/* pointer to next ompi_cb_fifo_ctl_t structure. This is always
stored as an absolute address. */
struct ompi_cb_fifo_wrapper_t *next_fifo_wrapper;
/* flag indicating if cb_fifo has over flown - need this to force
* release of entries already read */
volatile bool cb_overflow;
};
typedef struct ompi_cb_fifo_wrapper_t ompi_cb_fifo_wrapper_t;
/* data structure used to describe the fifo */
struct ompi_fifo_t {
/* locks for multi-process synchronization */
opal_atomic_lock_t fifo_lock;
/* locks for thread synchronization */
opal_atomic_lock_t *head_lock;
/* locks for thread synchronization */
opal_atomic_lock_t *tail_lock;
/* size of fifo */
int size;
/* number of allocated circular buffers */
int cb_count;
/* fifo memory locality index */
mca_mpool_base_module_t *fifo_mpool;
/* head memory locality index */
mca_mpool_base_module_t *head_mpool;
/* tail memory locality index */
mca_mpool_base_module_t *tail_mpool;
/* offset between sender and receiver shared mapping */
ptrdiff_t offset;
/* pointer to head (write) ompi_cb_fifo_t structure. This is
always stored as an sender size address. */
ompi_cb_fifo_wrapper_t *head;
/* pointer to tail (read) ompi_cb_fifo_t structure. This is
always stored as an receiver size address. */
ompi_cb_fifo_wrapper_t *tail;
};
typedef struct ompi_fifo_t ompi_fifo_t;
/**
* Initialize a fifo
*
* @param size_of_cb_fifo Length of fifo array (IN)
*
* @param fifo_memory_locality_index Locality index to apply to
* the fifo array. Not currently
* in use (IN)
*
* @param head_memory_locality_index Locality index to apply to the
* head control structure. Not
* currently in use (IN)
*
* @param tail_memory_locality_index Locality index to apply to the
* tail control structure. Not
* currently in use (IN)
*
* @param fifo Pointer to data structure defining this fifo (IN)
*
* @param memory_allocator Pointer to the memory allocator to use
* to allocate memory for this fifo. (IN)
*
* @returncode Error code
*
*/
static inline int ompi_fifo_init(int size_of_cb_fifo,
int lazy_free_freq, int cb_num_limit,
mca_mpool_base_module_t *fifo_mpool,
mca_mpool_base_module_t *head_mpool,
mca_mpool_base_module_t *tail_mpool,
ompi_fifo_t *fifo, ptrdiff_t offset)
{
int error_code;
fifo->offset = offset;
fifo->size = size_of_cb_fifo;
/*we allocate one cb below so subtract one here */
fifo->cb_count = cb_num_limit - 1;
fifo->fifo_mpool = fifo_mpool;
fifo->head_mpool = head_mpool;
fifo->tail_mpool = tail_mpool;
/* allocate head ompi_cb_fifo_t structure and place for head and tail locks
* on different cache lines */
fifo->head = (ompi_cb_fifo_wrapper_t*)fifo_mpool->mpool_alloc(
fifo_mpool, sizeof(ompi_cb_fifo_wrapper_t), getpagesize(),
0, NULL);
if(NULL == fifo->head) {
return OMPI_ERR_OUT_OF_RESOURCE;
}
/* initialize the circular buffer fifo head structure */
error_code = ompi_cb_fifo_init(size_of_cb_fifo,
lazy_free_freq, head_mpool, tail_mpool, &(fifo->head->cb_fifo),
offset);
if ( OMPI_SUCCESS != error_code ) {
return error_code;
}
/* finish head initialization */
opal_atomic_init(&(fifo->fifo_lock), OPAL_ATOMIC_UNLOCKED);
fifo->head->next_fifo_wrapper = fifo->head;
fifo->head->cb_overflow=false; /* no attempt to overflow the queue */
/* set the tail */
fifo->tail = (ompi_cb_fifo_wrapper_t*)((char*)fifo->head - offset);
/* return */
return error_code;
}
/**
* Try to write pointer to the head of the queue
*
* @param data Pointer value to write in specified slot (IN)
*
* @param fifo Pointer to data structure defining this fifo (IN)
*
* @returncode Slot index to which data is written
*
*/
static inline int ompi_fifo_write_to_head(void *data, ompi_fifo_t *fifo)
{
int error_code;
ompi_cb_fifo_wrapper_t *next_ff;
/* attempt to write data to head ompi_fifo_cb_fifo_t */
error_code = ompi_cb_fifo_write_to_head(data, &fifo->head->cb_fifo);
/* If the queue is full, create a new circular buffer and put the
data in it. */
if(OMPI_CB_ERROR == error_code) {
/* NOTE: This is the lock described in the top-level comment
in this file. There are corresponding uses of this lock in
both of the read routines. We need to protect this whole
section -- setting cb_overflow to true through setting the
next_fifo_wrapper to the next circular buffer. It is
likely possible to do this in a finer grain; indeed, it is
likely that we can get rid of this lock altogther, but it
will take some refactoring to make the data updates
safe. */
opal_atomic_lock(&fifo->fifo_lock);
/* mark queue as overflown */
fifo->head->cb_overflow = true;
/* We retry to write to the old head before creating new one just in
* case consumer read all entries after first attempt failed, but
* before we set cb_overflow to true */
error_code = ompi_cb_fifo_write_to_head(data, &fifo->head->cb_fifo);
if(error_code != OMPI_CB_ERROR) {
fifo->head->cb_overflow = false;
opal_atomic_unlock(&(fifo->fifo_lock));
return OMPI_SUCCESS;
}
/* see if next queue is available - while the next queue
* has not been emptied, it will be marked as overflowen*/
next_ff = fifo->head->next_fifo_wrapper;
/* if next queue not available, allocate new queue */
if (next_ff->cb_overflow) {
/* allocate head ompi_cb_fifo_t structure */
if(0 == fifo->cb_count)
next_ff = NULL;
else
next_ff = (ompi_cb_fifo_wrapper_t*)
fifo->fifo_mpool->mpool_alloc(fifo->fifo_mpool,
sizeof(ompi_cb_fifo_wrapper_t), getpagesize(), 0,
NULL);
if (NULL == next_ff) {
opal_atomic_unlock(&fifo->fifo_lock);
return OMPI_ERR_OUT_OF_RESOURCE;
}
/* initialize the circular buffer fifo head structure */
error_code = ompi_cb_fifo_init(fifo->size,
fifo->head->cb_fifo.lazy_free_frequency,
fifo->head_mpool, fifo->tail_mpool,
&(next_ff->cb_fifo), fifo->offset);
if (OMPI_SUCCESS != error_code) {
fifo->fifo_mpool->mpool_free(fifo->fifo_mpool, next_ff, NULL);
opal_atomic_unlock(&fifo->fifo_lock);
return error_code;
}
fifo->cb_count--;
/* finish new element initialization */
/* only one element in the link list */
next_ff->next_fifo_wrapper = fifo->head->next_fifo_wrapper;
next_ff->cb_overflow = false; /* no attempt to overflow the queue */
fifo->head->next_fifo_wrapper = next_ff;
}
/* reset head pointer */
fifo->head = next_ff;
opal_atomic_unlock(&fifo->fifo_lock);
/* write data to new head structure */
error_code=ompi_cb_fifo_write_to_head(data, &fifo->head->cb_fifo);
if( OMPI_CB_ERROR == error_code ) {
return OMPI_ERROR;
}
}
return OMPI_SUCCESS;
}
/**
* Try to read pointer from the tail of the queue
*
* @param fifo Pointer to data structure defining this fifo (IN)
*
* @returncode Pointer - OMPI_CB_FREE indicates no data to read
*
*/
static inline
void *ompi_fifo_read_from_tail(ompi_fifo_t *fifo)
{
/* local parameters */
void *return_value;
bool queue_empty;
/* get next element */
return_value = ompi_cb_fifo_read_from_tail(&fifo->tail->cb_fifo,
fifo->tail->cb_overflow, &queue_empty);
/* check to see if need to move on to next cb_fifo in the link list */
if(queue_empty) {
/* queue_emptied - move on to next element in fifo */
/* See the big comment at the top of this file about this
lock. */
opal_atomic_lock(&(fifo->fifo_lock));
if(fifo->tail->cb_overflow == true) {
fifo->tail->cb_overflow = false;
if(fifo->head == (ompi_cb_fifo_wrapper_t*) (((char*)fifo->tail) + fifo->offset)) {
opal_atomic_unlock(&(fifo->fifo_lock));
return return_value;
}
fifo->tail = (ompi_cb_fifo_wrapper_t*)
((char*)fifo->tail->next_fifo_wrapper - fifo->offset);
}
opal_atomic_unlock(&(fifo->fifo_lock));
}
return return_value;
}
#endif /* !_OMPI_FIFO */

1
ompi/mca/btl/self/btl_self.c
Просмотреть файл

@ -32,7 +32,6 @@
#include "opal/util/if.h"
#include "orte/util/proc_info.h"
#include "opal/util/printf.h"
#include "ompi/class/ompi_fifo.h"
#include "ompi/class/ompi_free_list.h"
#include "ompi/mca/pml/pml.h"
#include "ompi/mca/btl/btl.h"

1
ompi/mca/btl/self/btl_self.h
Просмотреть файл

@ -29,7 +29,6 @@
#endif /* HAVE_SYS_TYPES_H */
#include "ompi/class/ompi_free_list.h"
#include "ompi/class/ompi_bitmap.h"
#include "ompi/class/ompi_fifo.h"
#include "opal/event/event.h"
#include "ompi/mca/pml/pml.h"
#include "ompi/mca/btl/btl.h"

61
ompi/mca/btl/sm/btl_sm.c
Просмотреть файл

@ -187,8 +187,21 @@ static int sm_btl_first_time_init(mca_btl_sm_t *sm_btl, int n)
/* create mpool for each memory node */
for(i = 0; i < num_mem_nodes; i++) {
mca_mpool_base_resources_t res;
mca_btl_sm_component_t* m = &mca_btl_sm_component;
/* disable memory binding if there is only one memory node */
res.mem_node = (num_mem_nodes == 1) ? -1 : i;
/* determine how much memory to create */
res.size = m->nfifos * ( sizeof(sm_fifo_t) + sizeof(void *) * m->fifo_size )
+ ( 2 * n + m->sm_free_list_inc ) * ( m->eager_limit + CACHE_LINE_SIZE )
+ m->sm_free_list_num * ( m->max_frag_size + CACHE_LINE_SIZE );
if ( ((double) res.size) * n > LONG_MAX )
res.size = LONG_MAX;
else
res.size *= n;
/* now, create it */
mca_btl_sm_component.sm_mpools[i] =
mca_mpool_base_module_create(mca_btl_sm_component.sm_mpool_name,
sm_btl, &res);
@ -305,29 +318,35 @@ static int sm_btl_first_time_init(mca_btl_sm_t *sm_btl, int n)
length = sizeof(mca_btl_sm_frag1_t);
length_payload =
sizeof(mca_btl_sm_hdr_t) + mca_btl_sm_component.eager_limit;
ompi_free_list_init_new(&mca_btl_sm_component.sm_frags_eager, length,
CACHE_LINE_SIZE, OBJ_CLASS(mca_btl_sm_frag1_t),
length_payload, CACHE_LINE_SIZE,
mca_btl_sm_component.sm_free_list_num,
mca_btl_sm_component.sm_free_list_max,
mca_btl_sm_component.sm_free_list_inc,
mca_btl_sm_component.sm_mpool);
i = ompi_free_list_init_new(&mca_btl_sm_component.sm_frags_eager, length,
CACHE_LINE_SIZE, OBJ_CLASS(mca_btl_sm_frag1_t),
length_payload, CACHE_LINE_SIZE,
mca_btl_sm_component.sm_free_list_num,
mca_btl_sm_component.sm_free_list_max,
mca_btl_sm_component.sm_free_list_inc,
mca_btl_sm_component.sm_mpool);
if ( OMPI_SUCCESS != i )
return i;
length = sizeof(mca_btl_sm_frag2_t);
length_payload =
sizeof(mca_btl_sm_hdr_t) + mca_btl_sm_component.max_frag_size;
ompi_free_list_init_new(&mca_btl_sm_component.sm_frags_max, length,
CACHE_LINE_SIZE, OBJ_CLASS(mca_btl_sm_frag2_t),
length_payload, CACHE_LINE_SIZE,
mca_btl_sm_component.sm_free_list_num,
mca_btl_sm_component.sm_free_list_max,
mca_btl_sm_component.sm_free_list_inc,
mca_btl_sm_component.sm_mpool);
i = ompi_free_list_init_new(&mca_btl_sm_component.sm_frags_max, length,
CACHE_LINE_SIZE, OBJ_CLASS(mca_btl_sm_frag2_t),
length_payload, CACHE_LINE_SIZE,
mca_btl_sm_component.sm_free_list_num,
mca_btl_sm_component.sm_free_list_max,
mca_btl_sm_component.sm_free_list_inc,
mca_btl_sm_component.sm_mpool);
if ( OMPI_SUCCESS != i )
return i;
opal_free_list_init(&mca_btl_sm_component.pending_send_fl,
sizeof(btl_sm_pending_send_item_t),
OBJ_CLASS(opal_free_list_item_t),
16, -1, 32);
i = opal_free_list_init(&mca_btl_sm_component.pending_send_fl,
sizeof(btl_sm_pending_send_item_t),
OBJ_CLASS(opal_free_list_item_t),
16, -1, 32);
if ( OMPI_SUCCESS != i )
return i;
/* set flag indicating btl has been inited */
sm_btl->btl_inited = true;
@ -628,7 +647,11 @@ struct mca_btl_base_descriptor_t* mca_btl_sm_prepare_src(
size_t max_data = *size;
int rc;
MCA_BTL_SM_FRAG_ALLOC_MAX(frag, rc);
if ( reserve + max_data <= mca_btl_sm_component.eager_limit ) {
MCA_BTL_SM_FRAG_ALLOC_EAGER(frag,rc);
} else {
MCA_BTL_SM_FRAG_ALLOC_MAX(frag, rc);
}
if(OPAL_UNLIKELY(NULL == frag)) {
return NULL;
}

10
ompi/mca/btl/sm/btl_sm_component.c
Просмотреть файл

@ -131,8 +131,6 @@ int mca_btl_sm_component_open(void)
mca_btl_sm_param_register_int("free_list_inc", 64);
mca_btl_sm_component.sm_max_procs =
mca_btl_sm_param_register_int("max_procs", -1);
mca_btl_sm_component.sm_extra_procs =
mca_btl_sm_param_register_int("sm_extra_procs", -1);
mca_btl_sm_component.sm_mpool_name =
mca_btl_sm_param_register_string("mpool", "sm");
mca_btl_sm_component.fifo_size =
@ -415,10 +413,13 @@ int mca_btl_sm_component_progress(void)
* memory address, to a true virtual address */
hdr = (mca_btl_sm_hdr_t *) RELATIVE2VIRTUAL(hdr);
peer_smp_rank = hdr->my_smp_rank;
if ( FIFO_MAP(peer_smp_rank) != j )
#if OMPI_ENABLE_DEBUG
if ( FIFO_MAP(peer_smp_rank) != j ) {
opal_output(0, "mca_btl_sm_component_progress: "
"rank %d got %d on FIFO %d, but this sender should send to FIFO %d\n",
my_smp_rank, peer_smp_rank, j, FIFO_MAP(peer_smp_rank));
}
#endif
/* recv upcall */
reg = mca_btl_base_active_message_trigger + hdr->tag;
Frag.segment.seg_addr.pval = ((char*)hdr) +
@ -428,6 +429,7 @@ int mca_btl_sm_component_progress(void)
Frag.base.des_dst = &(Frag.segment);
reg->cbfunc(&mca_btl_sm.super, hdr->tag, &(Frag.base),
reg->cbdata);
/* return the fragment */
MCA_BTL_SM_FIFO_WRITE(
mca_btl_sm_component.sm_peers[peer_smp_rank],
my_smp_rank, peer_smp_rank, hdr->frag, false, rc);
@ -464,7 +466,7 @@ int mca_btl_sm_component_progress(void)
/* unknown */
/*
* This code path should presumably never be called.
* It's unclear how it should be written.
* It's unclear if it should exist or, if so, how it should be written.
* If we want to return it to the sending process,
* we have to figure out who the sender is.
* It seems we need to subtract the mask bits.

1
ompi/mca/mpool/sm/mpool_sm.h
Просмотреть файл

@ -43,6 +43,7 @@ struct mca_mpool_sm_component_t {
typedef struct mca_mpool_sm_component_t mca_mpool_sm_component_t;
typedef struct mca_mpool_base_resources_t {
size_t size;
int32_t mem_node;
} mca_mpool_base_resources_t;

101
ompi/mca/mpool/sm/mpool_sm_component.c
Просмотреть файл

@ -72,21 +72,8 @@ mca_mpool_sm_component_t mca_mpool_sm_component = {
}
};
static char *max_size_param = NULL, *min_size_param = NULL, *peer_size_param = NULL;
static long default_max, default_min, default_peer;
#if SIZEOF_LONG == 4
/* For 32-bit libraries, set the default maximum to 2Gbytes - 1. This
* is the absolute maximum possible as this is the largest value that
* can be given to the ftruncate() call. This means we scale at 32
* Mbytes per process until np=64 where we hit the maximum. */
#define OMPI_MAX_SM_MPOOL_FILE_SIZE 2147483647
#else
/* For 64-bit libraries, set the default maximum to 8Gbytes - 1. This
* allows the scaling of 32 Mbytes per process to be in effect up until
* about np=256 and then get capped. */
#define OMPI_MAX_SM_MPOOL_FILE_SIZE 8589934591
#endif
static char *min_size_param = NULL;
static long default_min;
/**
* component open/close/init function
@ -96,9 +83,7 @@ static int mca_mpool_sm_open(void)
int value = 0;
char *size_str = NULL;
default_max = OMPI_MAX_SM_MPOOL_FILE_SIZE;
default_min = 128*1024*1024;
default_peer = 32*1024*1024;
default_min = 0;
/* register SM component parameters */
mca_base_param_reg_string(&mca_mpool_sm_component.super.mpool_version,
@ -109,28 +94,14 @@ static int mca_mpool_sm_open(void)
&mca_mpool_sm_component.sm_allocator_name);
/* register values as string instead of int. A string-converted
* signed long int allows the max_size or the sm_size
* signed long int allows the min_size or the sm_size
* to be set up to 2GB-1 for 32 bit and much greater for 64 bit. */
asprintf(&size_str, "%ld", default_max);
mca_base_param_reg_string(&mca_mpool_sm_component.super.mpool_version,
"max_size",
"Maximum size of the sm mpool shared memory file",
false, false, size_str, &max_size_param);
free(size_str);
asprintf(&size_str, "%ld", default_min);
mca_base_param_reg_string(&mca_mpool_sm_component.super.mpool_version,
"min_size",
"Minimum size of the sm mpool shared memory file",
false, false, size_str, &min_size_param);
free(size_str);
asprintf(&size_str, "%ld", default_peer);
mca_base_param_reg_string(&mca_mpool_sm_component.super.mpool_version,
"per_peer_size",
"Size (in bytes) to allocate per local peer in "
"the sm mpool shared memory file, bounded by "
"min_size and max_size",
false, false, size_str, &peer_size_param);
free(size_str);
mca_base_param_reg_int(&mca_mpool_sm_component.super.mpool_version,
"verbose",
"Enable verbose output for mpool sm component",
@ -167,15 +138,9 @@ static int mca_mpool_sm_close( void )
if (NULL != mca_mpool_sm_component.sm_allocator_name) {
free(mca_mpool_sm_component.sm_allocator_name);
}
if (NULL != max_size_param) {
free(max_size_param);
}
if (NULL != min_size_param) {
free(min_size_param);
}
if (NULL != peer_size_param) {
free(peer_size_param);
}
return OMPI_SUCCESS;
}
@ -186,7 +151,7 @@ static mca_mpool_base_module_t* mca_mpool_sm_init(
int len;
mca_mpool_sm_module_t* mpool_module;
mca_allocator_base_component_t* allocator_component;
long max_size, min_size, peer_size;
long min_size;
ompi_proc_t **procs;
size_t num_all_procs, i, num_local_procs = 0;
@ -200,18 +165,8 @@ static mca_mpool_base_module_t* mca_mpool_sm_init(
}
free(procs);
/* parse the max, min and peer sizes, and validate them */
/* absolutely necessary to reset errno each time */
errno = 0;
max_size = strtol(max_size_param, (char **)NULL, 10);
if (errno == ERANGE) {
opal_output(0, "mca_mpool_sm_init: max_size overflows! set to default (%ld)", default_max);
max_size = default_max;
} else if (errno == EINVAL) {
opal_output(0, "mca_mpool_sm_init: invalid max_size entered. set it to (%ld)", default_max);
max_size = default_max;
}
/* parse the min size and validate it */
/* if other parameters are added, absolutely necessary to reset errno each time */
errno = 0;
min_size = strtol(min_size_param, (char **)NULL, 10);
if (errno == ERANGE) {
@ -222,44 +177,12 @@ static mca_mpool_base_module_t* mca_mpool_sm_init(
min_size = default_min;
}
errno = 0;
peer_size = strtol(peer_size_param, (char **)NULL, 10);
if (errno == ERANGE) {
opal_output(0, "mca_mpool_sm_init: peer_size overflows! set to default (%ld)", default_peer);
peer_size = default_peer;
} else if (errno == EINVAL) {
opal_output(0, "mca_mpool_sm_init: invalid peer_size entered. set it to (%ld)", default_peer);
peer_size = default_peer;
}
/* set sm_size */
mca_mpool_sm_component.sm_size = resources->size;
/* more checks... */
if (min_size > max_size) {
opal_output(0, "mca_mpool_sm_init: adjusting max_size to be min_size (%ld)",
min_size);
max_size = min_size;
}
/* sm_size is a product of peer_size * num_local_procs. To prevent the
* sm_size from overflowing SIZE_MAX, we first calculate the quotient.
* If quotient is less than the peer_size, it means the product
* (peer_size * num_local_procs) is going to overflow SIZE_MAX, then we'll
* set sm_size to max_size. */
if ((double)LONG_MAX / num_local_procs < peer_size) {
/* enable verbose would show if sm_size overflows */
opal_output(mca_mpool_sm_component.verbose,
"mca_mpool_sm_init: sm_size overflows, set sm_size to max_size (%ld)",
LONG_MAX);
mca_mpool_sm_component.sm_size = max_size;
} else {
mca_mpool_sm_component.sm_size = peer_size * num_local_procs;
}
if (min_size > mca_mpool_sm_component.sm_size) {
mca_mpool_sm_component.sm_size = min_size;
}
if (max_size < mca_mpool_sm_component.sm_size) {
mca_mpool_sm_component.sm_size = max_size;
}
/* clip at the min size */
if ( mca_mpool_sm_component.sm_size < min_size )
mca_mpool_sm_component.sm_size = min_size;
allocator_component = mca_allocator_component_lookup(
mca_mpool_sm_component.sm_allocator_name);

3
ompi/mca/mpool/sm/mpool_sm_module.c
Просмотреть файл

@ -76,8 +76,7 @@ void* mca_mpool_sm_alloc(
opal_maffinity_base_segment_t mseg;
mseg.mbs_start_addr =
mpool_sm->sm_allocator->alc_alloc(mpool_sm->sm_allocator, size,
OPAL_ALIGN(align, getpagesize(), size_t), registration);
mpool_sm->sm_allocator->alc_alloc(mpool_sm->sm_allocator, size, align, registration);
if(mpool_sm->mem_node >= 0) {
mseg.mbs_len = size;

2
opal/util/Makefile.am
Просмотреть файл

@ -49,7 +49,6 @@ headers = \
os_path.h \
output.h \
path.h \
pow2.h \
printf.h \
qsort.h \
show_help.h \
@ -82,7 +81,6 @@ libopalutil_la_SOURCES = \
os_path.c \
output.c \
path.c \
pow2.c \
printf.c \
qsort.c \
show_help.c \

59
opal/util/pow2.c
Просмотреть файл

@ -1,59 +0,0 @@
/*
* Copyright (c) 2004-2005 The Trustees of Indiana University and Indiana
* University Research and Technology
* Corporation. All rights reserved.
* Copyright (c) 2004-2005 The University of Tennessee and The University
* of Tennessee Research Foundation. 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$
*
* Additional copyrights may follow
*
* $HEADER$
*/
#include "opal_config.h"
#include "opal/util/pow2.h"
/**
* This routine takes in an interger, and returns the nearest
* power of 2 integer, same or greater than the input.
*
* @param input_integer input value
*
* @returnvalue power of 2 integer same or greater than the input
* parameter.
*/
int opal_round_up_to_nearest_pow2(int input_integer)
{
int pop_count, highest_used_bit, tmp_input_integer, return_value;
/* init counters */
pop_count=0;
highest_used_bit=-1;
/* get population count, and highest non-zero bit */
tmp_input_integer=input_integer;
while ( tmp_input_integer > 0 ){
pop_count+=(tmp_input_integer&1);
highest_used_bit++;
tmp_input_integer>>=1;
};
if( 1 < pop_count ) {
/* "round" up */
highest_used_bit++;
}
/* generate return value */
return_value=1<<highest_used_bit;
return return_value;
}

41
opal/util/pow2.h
Просмотреть файл

@ -1,41 +0,0 @@
/*
* Copyright (c) 2004-2005 The Trustees of Indiana University and Indiana
* University Research and Technology
* Corporation. All rights reserved.
* Copyright (c) 2004-2006 The University of Tennessee and The University
* of Tennessee Research Foundation. 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$
*
* Additional copyrights may follow
*
* $HEADER$
*/
#ifndef OPAL_POW2_H
#define OPAL_POW2_H
#if defined(c_plusplus) || defined(__cplusplus)
extern "C" {
#endif
/**
* This routine takes in an interger, and returns the nearest
* power of 2 integer, same or greater than the input.
*
* @param input_integer input value
*
* @returnvalue power of 2 integer same or greater than the input
* parameter.
*/
OPAL_DECLSPEC int opal_round_up_to_nearest_pow2(int input_integer);
#if defined(c_plusplus) || defined(__cplusplus)
}
#endif
#endif /* OPAL_POW2_H */

16
test/class/Makefile.am
Просмотреть файл

@ -39,22 +39,6 @@ ompi_bitmap_LDADD = \
$(top_builddir)/test/support/libsupport.a
ompi_bitmap_DEPENDENCIES = $(ompi_bitmap_LDADD)
#ompi_circular_buffer_fifo_SOURCES = ompi_circular_buffer_fifo.c
#ompi_circular_buffer_fifo_LDADD = \
# $(top_builddir)/ompi/libmpi.la \
# $(top_builddir)/orte/libopen-rte.la \
# $(top_builddir)/opal/libopen-pal.la \
# $(top_builddir)/test/support/libsupport.a
#ompi_circular_buffer_fifo_DEPENDENCIES = $(ompi_circular_buffer_fifo_LDADD)
#ompi_fifo_SOURCES = ompi_fifo.c
#ompi_fifo_LDADD = \
# $(top_builddir)/ompi/libmpi.la \
# $(top_builddir)/orte/libopen-rte.la \
# $(top_builddir)/opal/libopen-pal.la \
# $(top_builddir)/test/support/libsupport.a
#ompi_fifo_DEPENDENCIES = $(ompi_fifo_LDADD)
opal_list_SOURCES = opal_list.c
opal_list_LDADD = \
$(top_builddir)/opal/libopen-pal.la \

294
test/class/ompi_circular_buffer_fifo.c
Просмотреть файл

@ -1,294 +0,0 @@
/*
* Copyright (c) 2004-2005 The Trustees of Indiana University and Indiana
* University Research and Technology
* Corporation. All rights reserved.
* Copyright (c) 2004-2005 The University of Tennessee and The University
* of Tennessee Research Foundation. 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$
*
* Additional copyrights may follow
*
* $HEADER$
*/
#include "ompi_config.h"
#include <stdlib.h>
#include "support.h"
#include "ompi/mca/mpool/mpool.h"
static void *malloc_noalign(mca_mpool_base_module_t* mpool, size_t size, size_t dummy, void* user_out) {
return malloc(size);
}
size_t offset;
static void *malloc_base_addr(mca_mpool_base_module_t* mpool){
return (void *)offset;
}
static void my_free(mca_mpool_base_module_t* mpool, void* addr)
{
free(addr);
}
static void* my_realloc(mca_mpool_base_module_t* mpool, void* addr, size_t size, void* user_out){
return realloc(addr, size);
}
#include "ompi/class/ompi_circular_buffer_fifo.h"
/* simple allocator for some simple tests */
mca_mpool_base_module_t pool = {
NULL, /* component structure */
malloc_base_addr, /* mca_mpool_base_module_address_fn_t */
malloc_noalign, /* mca_mpool_base_module_alloc_fn_t */
my_realloc, /* ca_mpool_base_module_realloc_fn_t */
my_free, /*mca_mpool_base_module_free_fn_t */
NULL, /* mca_mpool_base_module_register_fn_t */
NULL, /* mca_mpool_base_module_deregister_fn_t */
NULL /* mca_mpool_base_module_finalize_fn_t */
};
int main(int argc, char **argv) {
/* local variables */
ompi_cb_fifo_t fifo;
int i,size_of_fifo,lazy_free,return_status,error_cnt,loop_cnt;
bool queue_empty;
union {
int ivalue;
void *vvalue;
} value;
#if 0
/* get queue size */
size_of_fifo=atoi(argv[1]);
lazy_free=atoi(argv[2]);
loop_cnt=atoi(argv[3]);
offset=atol(argv[4]);
#else
size_of_fifo = 5;
lazy_free = 1;
loop_cnt = 2;
offset = 2;
#endif
/* init result tracking */
test_init("ompi_circular_buffer_fifo");
/* init fifo */
return_status=ompi_cb_fifo_init(size_of_fifo,lazy_free,0,0,0,&fifo,
&pool);
/* check to see that retrun status is success */
if( OMPI_SUCCESS == return_status ) {
test_success();
} else {
test_failure(" ompi_cv_fifo_init");
}
/* populate fifo */
error_cnt=0;
for( i=0 ; i < ompi_cb_fifo_size(&fifo); i++ ) {
value.ivalue = i + 5;
return_status=ompi_cb_fifo_write_to_head(value.vvalue, &fifo,
(size_t)pool.mpool_base(&pool));
if( OMPI_CB_ERROR == return_status ) {
test_failure(" ompi_cb_fifo_write_to_head");
error_cnt++;
}
}
if( 0 == error_cnt ) {
test_success();
}
/* try an over-fill the queue */
error_cnt=0;
for( i=0 ; i < 3 ; i++ ) {
value.ivalue = i;
return_status=ompi_cb_fifo_write_to_head(value.vvalue, &fifo,
(size_t)pool.mpool_base(&pool));
if( OMPI_CB_ERROR != return_status ) {
test_failure(" ompi_cb_fifo_write_to_head :: over-fill queue");
error_cnt++;
}
}
if( 0 == error_cnt ) {
test_success();
}
/* pop items off the queue */
error_cnt=0;
for( i=0 ; i < ompi_cb_fifo_size(&fifo); i++ ) {
value.vvalue =ompi_cb_fifo_read_from_tail(&fifo,0,&queue_empty,
(size_t)pool.mpool_base(&pool));
if( (i+5) != value.ivalue ) {
test_failure(" ompi_cb_fifo_read_from_tail (1)");
error_cnt++;
}
}
if( 0 == error_cnt ) {
test_success();
}
/* free fifo */
return_status=ompi_cb_fifo_free(&fifo,&pool);
if( OMPI_SUCCESS == return_status ) {
test_success();
} else {
test_failure(" ompi_cv_fifo_init");
}
/* init fifo - lazy_free greater than size ==> should return error*/
return_status=ompi_cb_fifo_init(size_of_fifo,size_of_fifo*2,0,0,0,&fifo,
&pool);
/* check to see that retrun status is success */
if( OMPI_SUCCESS != return_status ) {
test_success();
} else {
test_failure(" ompi_cv_fifo_init with lazy_free too large");
}
/* split the writting of data to the slot to a reserve, and then a
* write */
/* init fifo */
return_status=ompi_cb_fifo_init(size_of_fifo,lazy_free,0,0,0,&fifo,
&pool);
/* check to see that retrun status is success */
if( OMPI_SUCCESS == return_status ) {
test_success();
} else {
test_failure(" ompi_cv_fifo_init");
}
/* populate fifo */
error_cnt=0;
for( i=0 ; i < ompi_cb_fifo_size(&fifo); i++ ) {
return_status=ompi_cb_fifo_get_slot(&fifo,(size_t)pool.mpool_base(&pool));
if( OMPI_CB_ERROR == return_status ) {
test_failure(" ompi_cb_fifo_get_slot");
error_cnt++;
}
}
if( 0 == error_cnt ) {
test_success();
}
/* try an over-fill the queue */
error_cnt=0;
for( i=0 ; i < 3 ; i++ ) {
return_status=ompi_cb_fifo_get_slot(&fifo,(size_t)pool.mpool_base(&pool));
if( OMPI_CB_ERROR != return_status ) {
test_failure(" ompi_cb_fifo_get_slot :: over-fill queue");
error_cnt++;
}
}
if( 0 == error_cnt ) {
test_success();
}
/* write to slot - all slots previously reserved, so just use
* them now */
error_cnt=0;
for( i=0 ; i < ompi_cb_fifo_size(&fifo); i++ ) {
value.ivalue = i + 5;
return_status=ompi_cb_fifo_write_to_slot(i, value.vvalue, &fifo,
(size_t)pool.mpool_base(&pool));
if( OMPI_CB_ERROR == return_status ) {
test_failure(" ompi_cb_fifo_write_to_slot");
error_cnt++;
}
}
if( 0 == error_cnt ) {
test_success();
}
/* pop items off the queue */
error_cnt=0;
for( i=0 ; i < ompi_cb_fifo_size(&fifo); i++ ) {
value.vvalue = ompi_cb_fifo_read_from_tail(&fifo,0,&queue_empty,
(size_t)pool.mpool_base(&pool));
if( (i+5) != value.ivalue ) {
test_failure(" ompi_cb_fifo_read_from_tail (2)");
error_cnt++;
}
}
if( 0 == error_cnt ) {
test_success();
}
/* free fifo */
return_status=ompi_cb_fifo_free(&fifo,&pool);
if( OMPI_SUCCESS == return_status ) {
test_success();
} else {
test_failure(" ompi_cv_fifo_init");
}
/* go through the fifo multiple times */
/* init fifo */
return_status=ompi_cb_fifo_init(size_of_fifo,lazy_free,0,0,0,&fifo,&pool);
/* check to see that retrun status is success */
if( OMPI_SUCCESS == return_status ) {
test_success();
} else {
test_failure(" ompi_cv_fifo_init");
}
error_cnt=0;
for( i=0 ; i < ompi_cb_fifo_size(&fifo)*loop_cnt ; i++ ) {
/* populate fifo */
return_status=ompi_cb_fifo_get_slot(&fifo,
(size_t)pool.mpool_base(&pool));
if( OMPI_CB_ERROR == return_status ) {
test_failure(" ompi_cb_fifo_get_slot");
error_cnt++;
}
/* write to the slot */
value.ivalue = i + 5;
return_status=ompi_cb_fifo_write_to_slot(i%(ompi_cb_fifo_size(&fifo)),
value.vvalue, &fifo,
(size_t)pool.mpool_base(&pool));
if( OMPI_CB_ERROR == return_status ) {
test_failure(" ompi_cb_fifo_write_to_slot");
error_cnt++;
}
/* pop items off the queue */
if( (i % ompi_cb_fifo_size(&fifo) ) ==
ompi_cb_fifo_size(&fifo)/2 ) {
/* force a flush */
value.vvalue = ompi_cb_fifo_read_from_tail(&fifo,1,&queue_empty,
(size_t)pool.mpool_base(&pool));
} else {
value.vvalue = ompi_cb_fifo_read_from_tail(&fifo,0,&queue_empty,
(size_t)pool.mpool_base(&pool));
}
if( (i+5) != value.ivalue ) {
test_failure(" ompi_cb_fifo_read_from_tail (3)");
error_cnt++;
}
}
if( 0 == error_cnt ) {
test_success();
}
/* free fifo */
return_status=ompi_cb_fifo_free(&fifo,&pool);
if( OMPI_SUCCESS == return_status ) {
test_success();
} else {
test_failure(" ompi_cv_fifo_init");
}
/* finalize result tracking */
return test_finalize();
}

273
test/class/ompi_fifo.c
Просмотреть файл

@ -1,273 +0,0 @@
/*
* Copyright (c) 2004-2005 The Trustees of Indiana University and Indiana
* University Research and Technology
* Corporation. All rights reserved.
* Copyright (c) 2004-2005 The University of Tennessee and The University
* of Tennessee Research Foundation. 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$
*
* Additional copyrights may follow
*
* $HEADER$
*/
#include "ompi_config.h"
#include <stdlib.h>
#include "support.h"
#include "ompi/mca/mpool/mpool.h"
static void *malloc_noalign(mca_mpool_base_module_t* mpool,
size_t size,
size_t align,
uint32_t flags,
mca_mpool_base_registration_t** reg) {
return malloc(size);
}
size_t offset;
static void *malloc_base_addr(mca_mpool_base_module_t* mpool){
return (void *)offset;
}
static void my_free(mca_mpool_base_module_t* mpool,
void* addr,
mca_mpool_base_registration_t* reg)
{
free(addr);
}
static void* my_realloc(mca_mpool_base_module_t* mpool,
void* addr,
size_t size,
mca_mpool_base_registration_t** reg){
return realloc(addr, size);
}
#include "ompi/class/ompi_fifo.h"
/* simple allocator for some simple tests */
mca_mpool_base_module_t pool = {
NULL, /* component structure */
malloc_base_addr, /* mca_mpool_base_module_address_fn_t */
malloc_noalign, /* mca_mpool_base_module_alloc_fn_t */
my_realloc, /* ca_mpool_base_module_realloc_fn_t */
my_free, /*mca_mpool_base_module_free_fn_t */
NULL, /* mca_mpool_base_module_register_fn_t */
NULL, /* mca_mpool_base_module_deregister_fn_t */
NULL /* mca_mpool_base_module_finalize_fn_t */
};
int main(int argc, char **argv) {
/* local variables */
ompi_fifo_t fifo;
int i,j,size_of_fifo,lazy_free,return_status,error_cnt,loop_cnt;
cb_slot_t *slot_data;
size_t cnt;
union {
int ivalue;
void *vvalue;
} value;
#if 0
/* get queue size */
size_of_fifo=atoi(argv[1]);
lazy_free=atoi(argv[2]);
loop_cnt=atoi(argv[3]);
offset=atol(argv[4]);
#else
size_of_fifo = 5;
lazy_free = 1;
loop_cnt = 2;
offset = 2;
#endif
/* init result tracking */
test_init("ompi_circular_buffer_fifo");
/* init fifo */
return_status=ompi_fifo_init(size_of_fifo,lazy_free,0,0,0,&fifo,&pool);
/* check to see that retrun status is success */
if( OMPI_SUCCESS == return_status ) {
test_success();
} else {
test_failure(" ompi_fifo_init \n");
}
/* populate fifo */
error_cnt=0;
for( i=0 ; i < loop_cnt*ompi_cb_fifo_size( (ompi_cb_fifo_t *)
&(fifo.head->cb_fifo)); i++ ) {
value.ivalue = i + 5;
return_status=ompi_fifo_write_to_head(value.vvalue,
(ompi_fifo_t *)&(fifo), &pool, (size_t)pool.mpool_base(&pool));
if( OMPI_CB_ERROR == return_status ) {
test_failure(" ompi_cb_fifo_write_to_head\n");
error_cnt++;
}
}
if( 0 == error_cnt ) {
test_success();
}
/* pop items off the queue */
error_cnt=0;
for( i=0 ; i < loop_cnt*ompi_cb_fifo_size( (ompi_cb_fifo_t *)
&(fifo.head->cb_fifo)); i++ ) {
value.vvalue = ompi_fifo_read_from_tail(&fifo,
(size_t)pool.mpool_base(&pool));
if( (i+5) != value.ivalue ) {
test_failure(" ompi_cb_fifo_read_from_tail\n");
error_cnt++;
}
}
if( 0 == error_cnt ) {
test_success();
}
/* free fifo */
return_status=ompi_fifo_free(&fifo,&pool);
if( OMPI_SUCCESS == return_status ) {
test_success();
} else {
test_failure(" ompi_fifo_free \n");
}
/* init fifo */
return_status=ompi_fifo_init(size_of_fifo,lazy_free,0,0,0,&fifo,&pool);
/* check to see that retrun status is success */
if( OMPI_SUCCESS == return_status ) {
test_success();
} else {
test_failure(" ompi_fifo_init \n");
}
/*
* test slot reservation
*/
cnt = sizeof(cb_slot_t) * loop_cnt * fifo.head->cb_fifo.size;
slot_data=malloc(cnt);
if( !slot_data ) {
test_failure(" can't allocate memory for slot_data");
goto ERRORS;
}
/* reserve slot fifo */
error_cnt=0;
for( i=0 ; i < loop_cnt*ompi_cb_fifo_size( (ompi_cb_fifo_t *)
&(fifo.head->cb_fifo)); i++ ) {
slot_data[i]=ompi_fifo_get_slot(&fifo,
&pool, (size_t)pool.mpool_base(&pool));
if( slot_data[i].index < 0 ) {
test_failure(" ompi_fifo_get_slot \n");
error_cnt++;
}
}
if( 0 == error_cnt ) {
test_success();
}
/* populate the reserved slots */
error_cnt=0;
for( i=0 ; i < loop_cnt*ompi_cb_fifo_size( (ompi_cb_fifo_t *)
&(fifo.head->cb_fifo)); i++ ) {
value.ivalue = i + 5;
return_status=ompi_fifo_write_to_slot(&(slot_data[i]),
value.vvalue,
(size_t)pool.mpool_base(&pool));
if( OMPI_CB_ERROR == return_status ) {
test_failure(" ompi_fifo_write_to_slot \n");
error_cnt++;
}
}
if( 0 == error_cnt ) {
test_success();
}
/* pop items off the queue */
error_cnt=0;
for( i=0 ; i < loop_cnt*ompi_cb_fifo_size( (ompi_cb_fifo_t *)
&(fifo.head->cb_fifo)); i++ ) {
value.vvalue = ompi_fifo_read_from_tail(&fifo,
(size_t)pool.mpool_base(&pool));
if( (i+5) != value.ivalue ) {
test_failure(" ompi_cb_fifo_read_from_tail II\n");
error_cnt++;
}
}
if( 0 == error_cnt ) {
test_success();
}
/* free fifo */
return_status=ompi_fifo_free(&fifo,&pool);
if( OMPI_SUCCESS == return_status ) {
test_success();
} else {
test_failure(" ompi_fifo_free II\n");
}
/*
* re-use slots
*/
/* init fifo */
return_status=ompi_fifo_init(size_of_fifo,lazy_free,0,0,0,&fifo,&pool);
/* check to see that retrun status is success */
if( OMPI_SUCCESS == return_status ) {
test_success();
} else {
test_failure(" ompi_fifo_init \n");
}
/* populate fifo */
for( j=0 ; j < loop_cnt ; j++ ) {
error_cnt=0;
for( i=0 ; i < ompi_cb_fifo_size( (ompi_cb_fifo_t *)
&(fifo.head->cb_fifo)); i++ ) {
value.ivalue = (i + 5) * (j + 1);
return_status=ompi_fifo_write_to_head(value.vvalue,
(ompi_fifo_t *)&(fifo), &pool,
(size_t)pool.mpool_base(&pool));
if( OMPI_CB_ERROR == return_status ) {
test_failure(" ompi_cb_fifo_write_to_head\n");
error_cnt++;
}
}
if( 0 == error_cnt ) {
test_success();
}
/* pop items off the queue */
error_cnt=0;
for( i=0 ; i < ompi_cb_fifo_size( (ompi_cb_fifo_t *)
&(fifo.head->cb_fifo)); i++ ) {
value.vvalue = ompi_fifo_read_from_tail(&fifo,
(size_t)pool.mpool_base(&pool));
if( ((i+5)*(j+1)) != value.ivalue ) {
test_failure(" ompi_cb_fifo_read_from_tail\n");
error_cnt++;
}
}
if( 0 == error_cnt ) {
test_success();
}
}
/* free fifo */
return_status=ompi_fifo_free(&fifo,&pool);
if( OMPI_SUCCESS == return_status ) {
test_success();
} else {
test_failure(" ompi_fifo_free \n");
}
ERRORS:
/* finalize result tracking */
return test_finalize();
}