bsp_mesgqueue.h

Go to the documentation of this file.

/*
    BSPonMPI. This is an implementation of the BSPlib standard on top of MPI
    Copyright (C) 2006  Wijnand J. Suijlen
                                                                                
    This library is free software; you can redistribute it and/or
    modify it under the terms of the GNU Lesser General Public
    License as published by the Free Software Foundation; either
    version 2.1 of the License, or (at your option) any later version.
                                                                                
    This library is distributed in the hope that it will be useful,
    but WITHOUT ANY WARRANTY; without even the implied warranty of
    MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
    Lesser General Public License for more details.
                                                                                
    You should have received a copy of the GNU Lesser General Public
    License along with this library; if not, write to the Free Software
    Foundation, Inc., 51 Franklin St, Fifth Floor, Boston, MA  02110-1301  USA
                                                                                
    You may contact me via electronic mail:
      wjsuijle@users.sourceforge.net
    or snail mail:
      W.J. Suijlen
      Kraaiheidelaan 10
      2803 VP Gouda
      The Netherlands
*/

#ifndef BSP_MESGQUEUE_H
#define BSP_MESGQUEUE_H
#include "bsp_exptable.h"
#include <config.h>

static inline void
messageQueue_initialize (ExpandableTable * restrict table, const int rows)
{
  union SpecInfo info;
  info.mesgq.tag_size = 0;
  info.mesgq.slot_offset = 0;
  info.mesgq.inslot_offset = 0;
  info.mesgq.n_mesg_deq = 0;
  info.mesgq.n_mesg = 0;
  info.mesgq.accum_size = 0;
  varElSizeTable_initialize (table, 1, rows, info);
}

static inline void
messageQueue_destruct (ExpandableTable * restrict table)
{
  expandableTable_destruct (table);
}

static inline void
messageQueue_reset( ExpandableTable * restrict table)
{
  table->info.mesgq.slot_offset = 0;
  table->info.mesgq.inslot_offset = 0;
  table->info.mesgq.n_mesg_deq=0;
  table->info.mesgq.n_mesg = 0;
  table->info.mesgq.accum_size = 0;
  expandableTable_reset(table);
}  

static inline void
messageQueue_expand (ExpandableTable * restrict table, const int rows)
{
  expandableTable_expand (table, rows, table->info);
}

static inline void
messageQueue_push (ExpandableTable * restrict table, char * restrict data, int size)
{
  int free_space = table->rows - table->used_slot_count[0];
  VarSizeElement * restrict pointer ;
  size = no_slots(size, sizeof(VarSizeElement));

  if (size > free_space) 
    messageQueue_expand(table, MAX(table->rows * 2, size));

  pointer =  (VarSizeElement *) table->data + table->used_slot_count[0];
  memcpy(pointer, data, size * sizeof(VarSizeElement));
  
  table->info.mesgq.accum_size+=pointer->info.send.payload_size * pointer->info.send.item_count;
  table->info.mesgq.n_mesg += pointer->info.send.item_count;
  table->used_slot_count[0]+=size;
  table->count[0]++;
}

static inline int
messageQueue_setTagSize (ExpandableTable * restrict table, const int tag_size)
{
  const int a = table->info.mesgq.tag_size;
  table->info.mesgq.tag_size = tag_size;
  return a;
}

static inline int
messageQueue_getTagSize (const ExpandableTable * restrict table)
{
  return table->info.mesgq.tag_size;
}

/* Query number of messages in queue
  @param table Reference to MessageQueue
  @return Number of messages in queue
  */
static inline int
messageQueue_getNumber (const ExpandableTable * restrict table)
{
   return table->info.mesgq.n_mesg - table->info.mesgq.n_mesg_deq;
}

static inline int
messageQueue_getAccumSize (const ExpandableTable * restrict table)
{
  return table->info.mesgq.accum_size;
}

static inline int
messageQueue_getTag (ExpandableTable * restrict table, void * restrict tag)
{
  VarSizeElement * restrict element = (VarSizeElement *) table->data + table->info.mesgq.slot_offset;
  int subslot = 
    table->info.mesgq.inslot_offset * (element->info.send.tag_size + element->info.send.payload_size);
  
  memcpy (tag, (char *) (element + 1) + subslot, element->info.send.tag_size);
  return element->info.send.payload_size;
}

static inline int
messageQueue_dequeue (ExpandableTable * restrict table, void * restrict payload, int size)
{
  VarSizeElement * restrict element = (VarSizeElement *) table->data + table->info.mesgq.slot_offset;
  int subslot = 
    table->info.mesgq.inslot_offset * 
      (element->info.send.tag_size + element->info.send.payload_size) +
       element->info.send.tag_size;

  size = MIN (size, element->info.send.payload_size);
  memcpy (payload, (char *) (element + 1) + subslot, size);
  
  table->info.mesgq.inslot_offset++;
  table->info.mesgq.slot_offset += 
    (table->info.mesgq.inslot_offset / element->info.send.item_count) *
     no_slots(sizeof (VarSizeElement) + element->size, sizeof(VarSizeElement));
  table->info.mesgq.inslot_offset %= element->info.send.item_count;

  table->info.mesgq.n_mesg_deq++;
  table->info.mesgq.accum_size -= element->info.send.payload_size;
  return size;
}

static inline int
messageQueue_hpdequeue (ExpandableTable * restrict table, const void ** const tag,
                        const void ** const payload)
{
  VarSizeElement * restrict element = (VarSizeElement *) table->data + table->info.mesgq.slot_offset;
  int subslot = 
    table->info.mesgq.inslot_offset * 
      (element->info.send.tag_size + element->info.send.payload_size) ;
  char * restrict data = (char*) (element + 1) + subslot ;

  *tag = data;
  *payload = (char *) data + element->info.send.tag_size;
  
  table->info.mesgq.inslot_offset++;
  table->info.mesgq.slot_offset += 
    (table->info.mesgq.inslot_offset / element->info.send.item_count) *
     no_slots(sizeof (VarSizeElement) + element->size, sizeof(VarSizeElement));
  table->info.mesgq.inslot_offset %= element->info.send.item_count;

  table->info.mesgq.n_mesg_deq++;
  table->info.mesgq.accum_size -= element->info.send.payload_size;
  return element->info.send.payload_size;
}

#endif

---