rt_benchs/communication_techniques/src/communication/batch_queue.c

91 lines
2.3 KiB
C

#include <stdio.h>
#include <stdlib.h>
#include <pthread.h>
/* Non standard include */
#include <commtech.h>
#include <specific_comm.h>
void *create_comm_channel(void)
{
struct channel *channel;
if (!posix_memalign((void **) &channel, CACHE_LINE_SIZE, sizeof(struct channel)))
{
channel->receiver_idx = 0;
channel->state = 0;
channel->sender_idx = 0;
return channel;
}
return NULL;
}
int destroy_comm_channel(void *channel)
{
free(channel);
return 0;
}
/*
* Copy at max count received data into buf
* @param buf The buffer in which received data must be copied into
* @return Number of data received and copied into buf
*
* @warning recv_one_data should not be used in conjonction of
* recv_some_data
*/
void *recv_one_data(struct channel *channel)
{
void *result;
if (unlikely(!(channel->receiver_idx % (BUF_SIZE / sizeof(void *)))))
while (!channel->state);
result = channel->buf[channel->receiver_idx++];
channel->receiver_idx %= (2 * BUF_SIZE) / sizeof(void *);
if (unlikely(!(channel->receiver_idx % (BUF_SIZE / sizeof(void *)))))
channel->state = 0;
return result;
}
/*
* Copy at max count received data into buf
* @param buf The buffer in which received data must be copied into
* @return Number of data received and copied into buf
*
* @warning recv_some_data should not be used in conjonction of
* recv_one_data
* @warning count must be a multiple of BUF_SIZE
*/
ssize_t recv_some_data(struct channel *channel, void **buf, size_t count)
{
int nb_read;
nb_read = 0;
while (channel->state && nb_read < count)
{
int i, n;
/*
* cur->receiver_idx point to the last buffer we have read.
* We go to the next cache line "+ (BUF_SIZE / sizeof(void *))"
* (because the line is full of void * and then if we are after
* the second cache line we correct the pointer to point to the
* first one (this is done by the modulo).
*/
i = channel->receiver_idx;
n = channel->receiver_idx + (BUF_SIZE / sizeof(void *));
channel->receiver_idx = n % ((2 * BUF_SIZE) / sizeof(void *));
for(; i < n; i++)
{
/*
* The behaviour of this is not documented but we know
* the values inside buf won't change during this affectation
*/
*buf++ = channel->buf[i];
}
nb_read += BUF_SIZE / sizeof(void *);
channel->state = 0;
}
return nb_read;
}