RPC Calls 90 Percent done

I need to work out a few synchronization bugs
Signed-off-by: Scotty Bauer <sbauer@eng.utah.edu>

kernel_rpc/2/Makefile

+obj-m := test.o 
+#CFLAGS_test.o = -DDEBUG -DDEBUG_MWAIT_RETRY  -DTIMING -g -Wall
+#CFLAGS_betaModule2.o = -DDEBUG -DDEBUG_MWAIT_RETRY  -DTIMING -g -Wall
+#CFLAGS_ring-channel.o = -DDEBUG -DDEBUG_MWAIT_RETRY -DTIMING -g -Wall
+
+CFLAGS_test.o =  -O2 -DPOLL  -DUSE_FLOOD #-DDEBUG_MWAIT_RETRY #-DDEBUG
+CFLAGS_betaModule_RPC2.o =  -O2 -DPOLL  -DUSE_FLOOD #-DDEBUG_MWAIT_RETRY #-DDEBUG
+CFLAGS_ring-channel.o = -O2
+CFLAGS_functions.o = -O0
+
+test-objs := ./ring-chan/ring-channel.o ./betaModule_RPC2.o ./functions.o
+
+KDIR := /lib/modules/`uname -r`/build
+
+default:
+	make -C $(KDIR) M=$(PWD) modules
+clean:
+	make -C $(KDIR) M=$(PWD) clean

kernel_rpc/2/functions.c

+#include "../betaModule.h"
+
+static unsigned long foo(unsigned long a, unsigned long b, unsigned long c)
+{
+	/* just some random math, doesnt mean anything just meant to take cycles */
+	return (((a << 20) + a) - ((b >> 20) + b) + c)<<4;
+
+}
+
+unsigned long foo_dispatch(struct ipc_message *msg)
+{
+	return foo(msg->reg1, msg->reg6, msg->reg4);
+}
+
+unsigned long bar_dispatch(void)
+{
+	return 10;
+}
+
+unsigned long baz_dispatch(void)
+{
+	return 11;
+}

kernel_rpc/2/ring-chan/ring-channel.c

+/*
+ * Ring channels provide a general producer-consumer interface for relaying
+ * data from to the LCD components. Specifically ring channels are used for
+ * establishing a high-performance communication channels between LCD components
+ *
+ * Authors: Anton Burtsev, Scotty Bauer
+ * Date:    October 2011,  Feburary 2015
+ *
+ */
+
+
+#include <asm-generic/getorder.h>
+#include <linux/mm.h>
+#include <linux/gfp.h>
+#include "ring-channel.h"
+
+
+static inline unsigned long bsrl(unsigned long x)
+{
+	unsigned long ret;
+	asm("bsr %1,%0" : "=r"(ret) : "r"(x));
+	return ret;
+
+}
+
+static inline unsigned long lower_power_of_two(unsigned long x)
+{
+	return 0x80000000000000UL >>  (__builtin_clzl(x)-1);
+}
+
+
+/* Stolen from  xen/mm.h */
+static inline int get_order_from_pages(unsigned long nr_pages)
+{
+	int order;
+	nr_pages--;
+	for ( order = 0; nr_pages; order++ )
+		nr_pages >>= 1;
+	return order;
+}
+
+int ttd_ring_channel_alloc(struct ttd_ring_channel *ring_channel,
+			   unsigned long size_in_pages,
+			   unsigned long size_of_a_rec) {
+	return ttd_ring_channel_alloc_with_metadata(ring_channel,
+						    size_in_pages, size_of_a_rec, 0);
+}
+
+void ttd_ring_channel_free(struct ttd_ring_channel *ring_channel) {
+
+	if (ring_channel->recs) {
+		free_pages((unsigned long) ring_channel->recs,
+			   ring_channel->buf_order);
+		ring_channel->recs = NULL;
+	}
+
+	if (ring_channel->buf) {
+		free_pages((unsigned long) ring_channel->buf,
+			   ring_channel->header_order);
+		ring_channel->buf = NULL;
+	}
+}
+
+
+int ttd_ring_channel_alloc_with_metadata(struct ttd_ring_channel *ring_channel,
+                                         unsigned long size_in_pages,
+                                         unsigned long size_of_a_rec,
+                                         unsigned long priv_metadata_size)
+{
+	int           ret;
+	unsigned long order, header_order;
+	unsigned long size_of_header_in_pages;
+	pr_debug("Allocating ring channel\n");
+	ttd_ring_channel_init(ring_channel);
+
+
+	/* number of pages required for this */
+	header_order = get_order(priv_metadata_size +
+					    sizeof(struct ttd_buf));
+
+
+	if ( (ring_channel->buf = (void *)  __get_free_pages(GFP_KERNEL, header_order)) == NULL ) {
+		pr_err("Xen deterministic time-travel buffers: memory allocation failed\n");
+		return -EINVAL;
+	}
+
+
+	size_of_header_in_pages = 1 << header_order;
+	pr_debug("Allocating ring channel: header area size:%lu, in pages:%lu\n",
+		 priv_metadata_size + sizeof(struct ttd_buf), size_of_header_in_pages);
+
+	order = get_order_from_pages(size_in_pages);
+	if ( (ring_channel->recs = (char *) __get_free_pages(GFP_KERNEL, order)) == NULL ) {
+		pr_err("Xen deterministic time-travel buffers: memory allocationcd failed, "
+		       "size in pages:%lu, order:%lu\n", size_in_pages, order);
+		ret = -EINVAL; goto cleanup;
+	}
+
+
+	ring_channel->priv_metadata = (char *) (ring_channel->buf + 1);
+	ring_channel->buf->cons = ring_channel->buf->prod = 0;
+
+	ttd_ring_channel_reinit_stats(ring_channel->buf);
+
+	ring_channel->size_of_a_rec = size_of_a_rec;
+	pr_debug("Size of a rec is %lu\n", size_of_a_rec);
+
+
+	/* ring_channel->size_in_recs  = (lower_power_of_two(size_in_pages * PAGE_SIZE))
+		/ ring_channel->size_of_a_rec;
+	*/
+	ring_channel->size_in_recs = (size_in_pages * PAGE_SIZE) /
+		ring_channel->size_of_a_rec;
+
+
+	pr_debug("size in recs is %lu lower_power_of_two returned %lu and in hex %lxwith input %lu and hex %lx\n",
+		 ring_channel->size_in_recs,
+		 lower_power_of_two(size_in_pages * PAGE_SIZE),
+		 lower_power_of_two(size_in_pages * PAGE_SIZE),
+		 (size_in_pages * PAGE_SIZE), (size_in_pages * PAGE_SIZE));
+	if (ring_channel->size_in_recs == 0) {
+		pr_err(" Size_in_recs was incorrectly 0\n");
+		ret = -EINVAL; goto cleanup;
+	}
+
+	/* Init shared buffer structures */
+
+	ring_channel->buf->payload_buffer_mfn =
+		(unsigned long) ring_channel->recs; /*NOTE*/
+
+
+	ring_channel->buf->payload_buffer_size = size_in_pages * PAGE_SIZE;
+	ring_channel->buf->size_of_a_rec = ring_channel->size_of_a_rec;
+
+	ring_channel->buf->size_in_recs = ring_channel->size_in_recs;
+
+	//ring_channel->highwater = ring_channel->size_in_recs >> 2; /* 25% high water */
+	ring_channel->highwater = ring_channel->size_in_recs >> 1; /* 50% high water */
+	ring_channel->emergency_margin = ring_channel->size_in_recs >> 4; /* 5% we are very close */
+
+	pr_debug("New ring channel: payload area {requested:%lu, allocated:%lu, wasted on allocation:%lu (order:%lu), \n"
+		 "metadata area {size:%lu, buffer size: %lu, full header: %lu, "
+		 "allocated:%lu, wasted on allocation:%lu (order:%lu)} \n"
+		 "rec: {requested size:%lu, size ^2:%lu, rownded down size in recs:%lu, "
+		 "possible size in recs:%lu, wasted:%lu} "
+		 "highwater: %lu, emergency margin: %lu\n",
+		 size_in_pages * PAGE_SIZE,
+		 ((unsigned long)1 << order) * PAGE_SIZE,
+		 ((1 << order) * PAGE_SIZE) - size_in_pages * PAGE_SIZE, order,
+		 priv_metadata_size, (unsigned long) sizeof(struct ttd_buf),
+		 priv_metadata_size + sizeof(struct ttd_buf),
+		 ((unsigned long)1 << header_order) * PAGE_SIZE,
+		 ((1 << header_order)* PAGE_SIZE) - priv_metadata_size - sizeof(struct ttd_buf),
+		 header_order, size_of_a_rec,
+		 ttd_ring_channel_size_of_a_rec(ring_channel),
+		 ttd_ring_channel_size_in_recs(ring_channel),
+		 (size_in_pages * PAGE_SIZE)/ring_channel->size_of_a_rec,
+		 (size_in_pages * PAGE_SIZE)/ring_channel->size_of_a_rec - ttd_ring_channel_size_in_recs(ring_channel),
+		 ttd_ring_channel_highwater(ring_channel),
+		 ttd_ring_channel_emergency_margin(ring_channel));
+
+	ring_channel->header_order = header_order;
+	ring_channel->buf_order = order;
+
+	return 0;
+
+ cleanup:
+
+	ttd_ring_channel_free(ring_channel);
+	return ret;
+
+}
+

kernel_rpc/2/ring-chan/ring-channel.h

+/*
+ * common/ring-channel.c
+ *
+ * This file is part of the Flux deterministic time-travel infrastructure.
+ * Ring channels provide a general producer-consumer interface for relaying
+ * data from to the guest components. Specifically ring channels are used for
+ * establishing a high-performance communication channels between Xen microkernel
+ * and loggin, replay, and devd daemons.
+ *
+ * Authors: Anton Burtsev Scotty Bauer
+ * Date:    October 2011 Feburary 2015
+ *
+ */
+
+#ifndef __XEN_RING_CHANNEL_H__
+#define __XEN_RING_CHANNEL_H__
+
+#include <linux/string.h>
+
+
+/*
+ * This structure contains the metadata for a single trace buffer.  The head
+ * field, indexes into an array of struct t_rec's.
+ */
+struct ttd_buf {
+	unsigned long   cons;      /* Next item to be consumed by control tools. */
+	unsigned long   prod;      /* Next item to be produced by Xen.           */
+
+	/* Shape of the buffer */
+	unsigned long      payload_buffer_mfn;
+	unsigned long      payload_buffer_size;
+
+	unsigned long      size_of_a_rec;        /* size of a single record */
+
+	unsigned long      size_in_recs;         /* size of the buffer in recs */
+
+	/* Stats for the buffer */
+
+	unsigned long long avg_queue_size;      /* average size of the request queue in the buffer */
+	unsigned long long msgs_sent;            /* total number of messages sent */
+	unsigned long msgs_by_type[128];          /* count how many messages of individual type we record */
+	unsigned long long tx_notifications;     /* total number of TX notifications */
+	unsigned long long tx_hit_emergency_margin; /* total number of times the channel gets emergently full */
+	unsigned long long rx_notifications;     /* total number of RX notifications */
+	unsigned long long msgs_with_extra_reg_space; /* total number of messages which asked for extra space in registers */
+	unsigned long msgs_with_extra_reg_space_by_type[128]; /* count how many times an individual event type asks for more space */
+	unsigned long long msgs_with_extra_data_space; /* total number of messages which asked for extra space for data */
+	unsigned long msgs_with_extra_data_space_by_type[128]; /* count how many times an individual event type asks for more space */
+
+	/* 'ttd_nr_recs' records follow immediately after the meta-data header.    */
+};
+
+
+struct ttd_ring_channel {
+	struct ttd_buf   *buf;                  /* pointer to the buffer metadata   */
+	unsigned long     buf_mfn;
+	unsigned long     buf_order;
+
+	void             *priv_metadata;        /* pointer to the private buffer metadata  */
+	unsigned long     priv_metadata_size;   /* size of the private buffer metadata  */
+	unsigned long     header_order;   /* size of the private buffer metadata  */
+
+	char             *recs;                 /* pointer to buffer data areas      */
+
+	unsigned long     size_of_a_rec;        /* size of a single record */
+
+
+	unsigned long     size_in_recs;         /* size of the buffer in recs */
+
+	unsigned long     highwater;            /* buffer is quite full, time to notify other end */ 
+	unsigned long     emergency_margin;     /* buffer is nearly full, time to freeze everything */
+
+};
+
+static inline void ttd_ring_channel_reinit_stats(struct ttd_buf *buf)
+{
+	buf->avg_queue_size = 0;
+	buf->msgs_sent = 0;
+	buf->tx_notifications = 0;
+	buf->rx_notifications = 0;
+	buf->tx_hit_emergency_margin = 0;
+	buf->msgs_with_extra_reg_space = 0;
+	buf->msgs_with_extra_data_space = 0;
+
+	memset(buf->msgs_by_type, 0,
+	       sizeof(buf->msgs_by_type));
+
+	memset(buf->msgs_with_extra_reg_space_by_type, 0,
+	       sizeof(buf->msgs_with_extra_reg_space_by_type));
+
+	memset(buf->msgs_with_extra_data_space_by_type, 0,
+	       sizeof(buf->msgs_with_extra_data_space_by_type));
+}
+
+static inline void ttd_ring_channel_init(struct ttd_ring_channel *ring_channel)
+{
+	memset(ring_channel, 0, sizeof(*ring_channel));
+	return;
+}
+
+static inline void ttd_ring_channel_buf_init(struct ttd_buf *buf)
+{
+	memset(buf, 0, sizeof(*buf));
+	return;
+}
+
+int ttd_ring_channel_alloc(struct ttd_ring_channel *ring_channel,
+			   unsigned long size_in_pages,
+			   unsigned long size_of_a_rec);
+
+int ttd_ring_channel_alloc_with_metadata(struct ttd_ring_channel *ring_channel,
+					 unsigned long size_in_pages,
+					 unsigned long size_of_a_rec,
+					 unsigned long priv_metadata_size);
+
+void ttd_ring_channel_free(struct ttd_ring_channel *ring_channel);
+
+static inline void
+*ttd_ring_channel_get_priv_metadata(struct ttd_ring_channel *ring_channel)
+{
+	return ring_channel->priv_metadata;
+}
+
+static inline unsigned long
+ttd_ring_channel_get_prod(struct ttd_ring_channel *ring_channel) {
+	return ring_channel->buf->prod;
+};
+
+static inline unsigned long
+ttd_ring_channel_inc_prod(struct ttd_ring_channel *ring_channel) {
+	return (ring_channel->buf->prod++);
+};
+
+static inline void
+ttd_ring_channel_set_prod(struct ttd_ring_channel *ring_channel, unsigned long prod) {
+	ring_channel->buf->prod = prod;
+	return;
+};
+
+static inline unsigned long
+ttd_ring_channel_get_cons(struct ttd_ring_channel *ring_channel) {
+	return ring_channel->buf->cons;
+};
+
+static inline unsigned long
+ttd_ring_channel_inc_cons(struct ttd_ring_channel *ring_channel) {
+	return (ring_channel->buf->cons++);
+};
+
+
+static inline void ttd_ring_channel_set_cons(struct ttd_ring_channel *ring_channel,
+					     unsigned long cons) {
+	ring_channel->buf->cons = cons;
+	return;
+};
+
+static inline char *ttd_ring_channel_get_rec_slow(struct ttd_ring_channel *ring_channel,
+						  unsigned long cons) {
+
+	return (ring_channel->recs + (cons % 1024) * 64);
+	//return (ring_channel->recs
+	//	+ (cons % ring_channel->size_in_recs)
+	//	* ring_channel->size_of_a_rec);*/
+};
+
+
+static inline unsigned long ttd_ring_channel_get_index_mod_slow(struct ttd_ring_channel *ring_channel, unsigned long index) {
+	return (index % ring_channel->size_in_recs);
+}
+
+
+static inline unsigned long ttd_ring_channel_size_in_recs(struct ttd_ring_channel *ring_channel) {
+	return ring_channel->size_in_recs;
+}
+
+static inline unsigned long ttd_ring_channel_size_of_a_rec(struct ttd_ring_channel *ring_channel) {
+	return ring_channel->size_of_a_rec;
+}
+
+static inline unsigned long ttd_ring_channel_size(struct ttd_ring_channel *ring_channel) {
+	return ring_channel->size_in_recs * ring_channel->size_of_a_rec;
+}
+
+
+static inline unsigned long ttd_ring_channel_highwater(struct ttd_ring_channel *ring_channel) {
+	return ring_channel->highwater;
+}
+
+static inline unsigned long ttd_ring_channel_emergency_margin(struct ttd_ring_channel *ring_channel) {
+	return ring_channel->emergency_margin;
+}
+
+#endif
+

kernel_rpc/Makefile

+obj-m :=  betaModule_RPC1.o 
+#CFLAGS_betaModule1.o = -DDEBUG -DDEBUG_MWAIT_RETRY  -DTIMING -g -Wall
+#CFLAGS_betaModule.o = -DDEBUG -DDEBUG_MWAIT_RETRY   -DTIMING -g -Wall
+#CFLAGS_ring-channel.o = -DDEBUG -DDEBUG_MWAIT_RETRY -DTIMING -g -Wall
+
+
+CFLAGS_betaModule_RPC1.o =  -g -O2  -DPOLL -DTIMING -DUSE_FLOOD #-DDEBUG #-DDEBUG_MWAIT_RETRY 
+CFLAGS_betaModule_RPC.o =   -g -O2 -DPOLL -DTIMING -DUSE_FLOOD #-DDEBUG #-DDEBUG_MWAIT_RETRY #-DDEBUG
+CFLAGS_ring-channel.o = -g -O2
+#CFLAGS_functions.o = -O2
+
+betaModule_RPC1-objs := ./ring-chan/ring-channel.o ./betaModule_RPC.o ./functions.o
+
+KDIR := /lib/modules/`uname -r`/build
+
+default:
+	make -C $(KDIR) M=$(PWD) modules
+clean:
+	make -C $(KDIR) M=$(PWD) clean

kernel_rpc/betaModule.h

+#ifndef _INCL_GUARD
+#define _INCL_GUARD
+
+#include <linux/types.h>
+
+
+
+
+typedef enum { FOO, BAR, BAZ } ftype;
+
+
+#if defined(USE_MWAIT)
+	unsigned long ecx = 1; /*break of interrupt flag */
+	unsigned long cstate_wait = 0x0; /* 4 states, 0x0, 0x1 0x10 0x20 */
+#endif
+
+
+
+struct ipc_container{
+	struct task_struct *thread;
+	struct ttd_ring_channel *channel_tx;
+	struct ttd_ring_channel *channel_rx;
+};
+
+/*Don't let gcc do anything cute, we need this to be 128 bytes */
+
+/* TODO CONFIRM ALIGNMENT REQUIREMENTS FOR ULONG! */
+
+struct ipc_message{
+	ftype type; /* 4 */
+	unsigned long reg1;
+	unsigned long reg2;
+	unsigned long reg3;
+	unsigned long reg4;
+	unsigned long reg5;
+	unsigned long reg6;
+	unsigned long reg7;
+	volatile uint32_t msg_status;
+}__attribute__((packed));
+
+
+
+#define BETA_GET_CPU_AFFINITY 1<<1
+#define BETA_CONNECT_SHARED_MEM 1<<2
+#define BETA_UNPARK_THREAD 1<<3
+#define BETA_ALLOC_MEM 1<<4
+#define BETA_GET_MEM 1<<5
+#define BETA_DUMP_TIME 1<<6
+
+#endif

kernel_rpc/functions.c

+
+#include "betaModule.h"
+
+
+static struct ttd_ring_channel *prod_channel;
+static struct ttd_ring_channel *cons_channel;
+
+extern unsigned int recv_ack;
+extern unsigned int send_slot_avail;
+extern unsigned int send_message;
+extern unsigned int recv_message;
+extern unsigned long send_slot;
+extern unsigned long recv_slot;
+
+
+extern struct ipc_message  *get_next_available_slot(struct ttd_ring_channel*, unsigned long);
+extern int wait_for_producer_slot(struct ipc_message*, unsigned int);
+extern void send(struct ipc_message *);
+extern struct ipc_message *recv(struct ttd_ring_channel *);
+
+void register_chans(struct ttd_ring_channel *prod, struct ttd_ring_channel *cons)
+{
+	prod_channel = prod;
+	cons_channel = cons;
+}
+
+unsigned long foo(unsigned long a, unsigned long b, unsigned long c)
+{
+	struct ipc_message *send_msg, *recv_msg;
+	unsigned long ret;
+	send_msg = get_next_available_slot(prod_channel, send_slot);
+	wait_for_producer_slot(prod_channel, send_slot_avail);
+	send_msg->type = FOO;
+	send_msg->reg1 = a;
+	send_msg->reg6 = b;
+	send_msg->reg4 = c;
+	send(send_msg);
+	recv_msg = recv(cons_channel);
+	ret = recv_msg->reg1;
+	recv_msg->msg_status = recv_ack;
+	return ret;
+}

kernel_rpc/num_configs.h

+int CHAN_NUM_PAGES = 16;
+int BUF_SIZE = 60;
+int NUM_LOOPS = 10000;
+int FLOOD_SIZE = 1;

kernel_rpc/ring-chan/ring-channel.c

+/*
+ * Ring channels provide a general producer-consumer interface for relaying
+ * data from to the LCD components. Specifically ring channels are used for
+ * establishing a high-performance communication channels between LCD components
+ *
+ * Authors: Anton Burtsev, Scotty Bauer
+ * Date:    October 2011,  Feburary 2015
+ *
+ */
+
+
+#include <asm-generic/getorder.h>
+#include <linux/mm.h>
+#include <linux/gfp.h>
+#include "ring-channel.h"
+
+
+static inline unsigned long bsrl(unsigned long x)
+{
+	unsigned long ret;
+	asm("bsr %1,%0" : "=r"(ret) : "r"(x));
+	return ret;
+
+}
+
+static inline unsigned long lower_power_of_two(unsigned long x)
+{
+	return 0x80000000000000UL >>  (__builtin_clzl(x)-1);
+}
+
+
+/* Stolen from  xen/mm.h */
+static inline int get_order_from_pages(unsigned long nr_pages)
+{
+	int order;
+	nr_pages--;
+	for ( order = 0; nr_pages; order++ )
+		nr_pages >>= 1;
+	return order;
+}
+
+int ttd_ring_channel_alloc(struct ttd_ring_channel *ring_channel,
+			   unsigned long size_in_pages,
+			   unsigned long size_of_a_rec) {
+	return ttd_ring_channel_alloc_with_metadata(ring_channel,
+						    size_in_pages, size_of_a_rec, 0);
+}
+
+void ttd_ring_channel_free(struct ttd_ring_channel *ring_channel) {
+
+	if (ring_channel->recs) {
+		free_pages((unsigned long) ring_channel->recs,
+			   ring_channel->buf_order);
+		ring_channel->recs = NULL;
+	}
+
+	if (ring_channel->buf) {
+		free_pages((unsigned long) ring_channel->buf,
+			   ring_channel->header_order);
+		ring_channel->buf = NULL;
+	}
+}
+
+
+int ttd_ring_channel_alloc_with_metadata(struct ttd_ring_channel *ring_channel,
+                                         unsigned long size_in_pages,
+                                         unsigned long size_of_a_rec,
+                                         unsigned long priv_metadata_size)
+{
+	int           ret;
+	unsigned long order, header_order;
+	unsigned long size_of_header_in_pages;
+	pr_debug("Allocating ring channel\n");
+	ttd_ring_channel_init(ring_channel);
+
+
+	/* number of pages required for this */
+	header_order = get_order(priv_metadata_size +
+					    sizeof(struct ttd_buf));
+
+
+	if ( (ring_channel->buf = (void *)  __get_free_pages(GFP_KERNEL, header_order)) == NULL ) {
+		pr_err("Xen deterministic time-travel buffers: memory allocation failed\n");
+		return -EINVAL;
+	}
+
+
+	size_of_header_in_pages = 1 << header_order;
+	pr_debug("Allocating ring channel: header area size:%lu, in pages:%lu\n",
+		 priv_metadata_size + sizeof(struct ttd_buf), size_of_header_in_pages);
+
+	order = get_order_from_pages(size_in_pages);
+	if ( (ring_channel->recs = (char *) __get_free_pages(GFP_KERNEL, order)) == NULL ) {
+		pr_err("Xen deterministic time-travel buffers: memory allocationcd failed, "
+		       "size in pages:%lu, order:%lu\n", size_in_pages, order);
+		ret = -EINVAL; goto cleanup;
+	}
+
+
+	ring_channel->priv_metadata = (char *) (ring_channel->buf + 1);
+	ring_channel->buf->cons = ring_channel->buf->prod = 0;
+
+	ttd_ring_channel_reinit_stats(ring_channel->buf);