Shared memory API targeting SystemTap

shared_memory.h

/*
Shared memory (SHM_FIFO) is a byte ring buffer in the virtual memory. Driver pushes the data to
the tail of the FIFO (producer) and user space application reads the data from the head of the
buffer. Driver always places data structures in consecutive memory. If there is not enough
room to place the whole structure into the ring buffer without wrap around driver places special
"skip" code until the last byte of the allocated virtual memory and writes the structure
starting with offset zero. Application knows to handle the wrap around correctly by checking
if there is enough space for a strcuture of minimum size and skipping the data to the end of the
virtual memory.
*/

typedef struct shm_fifo_descr_s
{
	u32 tail;   // volatile is missing here for the user space?
	u32 head;
	u32 size;
} shm_fifo_descr_t;


typedef struct shm_fifo_s
{
	shm_fifo_descr_t descr;
	u8 data[1];
} shm_fifo_t;

typedef rwlock_t SHM_MUTEX_TYPE;
static void __sendIncidentSkip(u8* data, u64 size);

static inline void shm_fifo_commit(shm_fifo_t *fifo, u8* addr, int bytes);



/**
 * Allocate a region after the tail, move the tail forward
 * Make sure to call shm_fifo_commit() after the data is set
 * I need mutex/spinlock because of the call to sendIncidentSkip().
 * I have an indication how often the race happens in the 'commit' part of the API
 */
static u8* shm_fifo_allocate(shm_fifo_t *fifo, SHM_MUTEX_TYPE* mutex, int bytes)
{
	u8* res = NULL;
	u32 tail_orig, tail_new;
	shm_fifo_descr_t *descr = &fifo->descr;
	u32 fifo_size = descr->size;
	u32 head_orig = descr->head;

	/* TODO probably I can run lock free
	 * I am giving up immediately. If the message is critical sendIncident can retry
	 * Failure to get a lock shall hot happen often. There is a FIFO and a lock per core, the
	 * function is short, probes start with spinlocks */
	if (!write_trylock(mutex))
	{
 		HIT_MAP_INC(HIT_MAP_FIFO_SPINLOCK_FAIL);
		return NULL;
	}
	// SystemTap starts a probe with a mutex if I use a map (or I will overwrite events from time to time)
	tail_new = tail_orig = SHM_FIFO_VOLATILE(descr->tail);
	HIT_MAP_INC(HIT_MAP_FIFO_ALLOC);
	/* There is a contiguous block (no wraparound) between tail and end of the fifo->data I can use */
	if ((fifo_size - tail_orig) >= bytes)
	{
		HIT_MAP_INC(HIT_MAP_FIFO_ALLOC_NORMAL);
		if ( (tail_orig >= head_orig) || ((tail_orig + bytes) < head_orig) )
		{
			HIT_MAP_INC(HIT_MAP_FIFO_ALLOC_NORMAL_OK);
			res = &fifo->data[tail_orig];
			tail_new = tail_orig + bytes;
			goto Exit;
		}

		/* fifo->head is too close to the tail - return NULL */
		HIT_MAP_INC(HIT_MAP_FIFO_ALLOC_NORMAL_FAIL);
		goto Error;
	}

	/* I want to skip the area until the end of the FIFO */

	/* I can not skip the area if there is still data */
	if (tail_orig < head_orig)
	{
		HIT_MAP_INC(HIT_MAP_FIFO_ALLOC_FAIL_1);
		goto Error;
	}

	/* I can not allocate data from the offset 0 if the head is there too close */
	if (head_orig <= bytes)
	{
		HIT_MAP_INC(HIT_MAP_FIFO_ALLOC_FAIL_2);
		goto Error;
	}

	/* Skip to the end of the fifo->data */
	HIT_MAP_INC(HIT_MAP_FIFO_ALLOC_SKIP);
	sendIncidentSkip(&fifo->data[tail_orig], fifo_size - tail_orig);

	/* Allocate the data from the beginning of the fifo->data */
	res =  &fifo->data[0];
	tail_new = bytes;

Exit:
	/* this is not a mutex, just an indication if a race condition hits me and how often
	 * If I am running with a lock it shall never happen
	 */
	if (SHM_FIFO_VOLATILE(descr->tail) != tail_orig)
	{
		HIT_MAP_INC(HIT_MAP_FIFO_ALLOC_RACE);
	}
	/* spoil the magic word if there is one - avoid race when there is a valid magic word
	 * and the tail is new, but 'commit' is not called yet and the data is not set
	 * completely
	 */
	res[0] = 0;
	/* Insure that all writes are completed before setting the new tail */
	barrier();
	SHM_FIFO_VOLATILE(descr->tail) = tail_new;
Error:
	write_unlock(mutex);
	return res;
}