added basic skeleton to ipc_dispatch loop. awe_mapper propbably needs to be...

added basic skeleton to ipc_dispatch loop. awe_mapper propbably needs to be changed to ensure that each id that corresponds to awes in different threads is unique. Could be built into the kernel.

current/IPC/ipc.c

@@ -75,7 +75,7 @@ static int wait_for_tx_slot(struct ipc_message *imsg)
 	return 0;
 }
 
-static int wait_for_rx_slot(struct ipc_message *imsg, bool is_async)
+static int wait_for_rx_slot(struct ipc_message *imsg)
 {
 	while (check_rx_slot_available(imsg)) { //while a slot is not available
 			#if defined(USE_MWAIT)
@@ -192,11 +192,43 @@ struct ipc_message *recv(struct ttd_ring_channel *rx)
 
 	recv_msg = get_rx_rec(rx, sizeof(struct ipc_message));
 	inc_rx_slot(rx);
-	wait_for_rx_slot(recv_msg, false);
+	wait_for_rx_slot(recv_msg);
 	return recv_msg;
 }
 EXPORT_SYMBOL(recv);
 
+/*
+Takes an array of rx channels to iterate over. This function does one
+loop over the array and populates 'msg' with a received message and returns true
+if there is a message, else it returns false.
+start_ind: the index to start iterating and wrap around to
+NOTE: right now this just checks the first rx slot for each channel that previously didn't have a message.
+To make this check for everything where there could be a message, it would need to check the interval [rx, tx]
+*/
+bool poll_recv(struct ttd_ring_channel** rx_chans, int chans_num, int start_ind, struct ipc_message* msg)
+{
+    int curr_ind;
+    struct ttd_ring_channel* curr_chan;
+	struct ipc_message *recv_msg;
+
+    for( int i = 0; i < chans_num; i++ )
+    {
+        curr_ind  = (start_ind + i) % chans_num;
+        curr_chan = rx_chans[curr_ind];
+	    recv_msg  = get_rx_rec(curr_chan, sizeof(struct ipc_message));
+
+        if( !check_rx_slot_available(imsg) ) //if message exists
+        {
+            msg = recv_msg;
+	        inc_rx_slot(curr_chan);
+            return true;
+        }
+    }
+
+    return false;
+}
+EXPORT_SYMBOL(poll_recv);
+
 noinline struct ipc_message *async_recv(struct ttd_ring_channel *rx, unsigned long msg_id)
 {
 	struct ipc_message *recv_msg;
@@ -213,15 +245,15 @@ noinline struct ipc_message *async_recv(struct ttd_ring_channel *rx, unsigned lo
 			}		
 			else
 			{
-				printk(KERN_ERR "MESSAGE ID RECEIVED IS: %lx\n", recv_msg->msg_id);	
-				printk(KERN_ERR "MESSAGE ID FOR CONTEXT IS: %lx\n", msg_id);	
-				printk(KERN_ERR "CALLING YIELD TO\n");
+				//printk(KERN_ERR "MESSAGE ID RECEIVED IS: %lx\n", recv_msg->msg_id);	
+				//printk(KERN_ERR "MESSAGE ID FOR CONTEXT IS: %lx\n", msg_id);	
+				//printk(KERN_ERR "CALLING YIELD TO\n");
 				THCYieldToId((uint32_t) recv_msg->msg_id, (uint32_t) msg_id);
 			}
 		}
 		else
 		{
-			printk(KERN_ERR "spin yield\n");
+			//printk(KERN_ERR "spin yield\n");
 			THCYieldAndSave((uint32_t) msg_id);
 		}
 	}	

current/IPC/ipc.h

@@ -35,6 +35,7 @@ struct task_struct *attach_thread_to_channel(struct ttd_ring_channel *chan,
 void free_channel(struct ttd_ring_channel *channel);
 void send(struct ttd_ring_channel *tx, struct ipc_message *trans);
 struct ipc_message *recv(struct ttd_ring_channel *rx);
+bool poll_recv(struct ttd_ring_channel** rx_chans, int chans_num, int start_ind, struct ipc_message* msg);
 struct ipc_message *async_recv(struct ttd_ring_channel *rx, unsigned long msg_id);
 struct ipc_message *get_send_slot(struct ttd_ring_channel *tx);
 void transaction_complete(struct ipc_message *msg);

current/tests/dispatch-loop/ipc_dispatch.c

+#include <stdio.h>
+#include "ipc.h"
+#include "ipc_dispatch.h"
+
+static int foo(unsigned long* data, int length)
+{
+    for(int i = 0; i < length; i++)
+    {
+        printf("%lu\n", data[i]);
+    }
+}
+
+int ipc_dispatch_loop(ipc_local_fn_t* fns, int fns_length,struct ttd_ring_channel** rx_chans, int chans_num)
+    //ipc_channels, channel_length
+{
+    int start_ind = 0;
+    struct ipc_message* curr_msg;
+    DO_FINISH(
+    while( true )
+        {
+            if( poll_recv(rx_chans, chans_num, start_ind, curr_msg) )
+            {
+                //check if curr_msg corresponds to existing awe,
+                //else find corresponding function and execute.
+            }
+        }
+    );
+    return 0;
+}
+
+int main(void)
+{
+    ipc_local_fn_t test_fn = {1, foo};
+    unsigned long data[] = {0,1,2,3};
+    test_fn.local_fn(&data[1], 2);
+    return 0;
+}

current/tests/dispatch-loop/ipc_dispatch.h

+#ifndef IPC_DISPATCH_H
+#define IPC_DISPATCH_H
+
+typedef struct ipc_local_fn{
+    unsigned long fn_type;
+    int (*local_fn)(unsigned long*, int);
+} ipc_local_fn_t;
+
+#endif

current/tests/rpc/callee.c

@@ -85,7 +85,7 @@ static unsigned long add_nums_async(unsigned long trans, unsigned long res1, uns
 {
 	struct ipc_message *msg;
 	unsigned long result;
-	printk(KERN_ERR "MESSAGE ID IS: %lx\n", msg_id);	
+	//printk(KERN_ERR "MESSAGE ID IS: %lx\n", msg_id);	
 	msg = get_send_slot(channel);
 	msg->fn_type = ADD_NUMS;
 	msg->reg1    = trans;
@@ -97,7 +97,7 @@ static unsigned long add_nums_async(unsigned long trans, unsigned long res1, uns
 	msg->msg_id  = msg_id;
 	send(channel,msg);
 	msg = async_recv(channel, msg_id);
-	printk(KERN_ERR "result = %lx\n", msg->reg1);
+	//printk(KERN_ERR "result = %lx\n", msg->reg1);
 	result = msg->reg1;
 	transaction_complete(msg);
 	
@@ -209,23 +209,27 @@ void test_async_ipc(void* chan)
 	DO_FINISH(
 		uint32_t id_num;
 		while (num_transactions < TRANSACTIONS / 3) {
+		if((num_transactions * 3) % 300 == 0)
+		{
+			printk(KERN_ERR "num_transactions: %lu\n", num_transactions * 3);
+		}
 	//	start = RDTSC_START();
 
 		ASYNC(
 			id_num = awe_mapper_create_id();
-			printk(KERN_ERR "ID_NUM: %d\n", id_num);
+//			printk(KERN_ERR "ID_NUM: %d\n", id_num);
 			add_nums_async(num_transactions, 1,(unsigned long) id_num);
 		     );
 
 		ASYNC(
 			id_num = awe_mapper_create_id();
-			printk(KERN_ERR "ID_NUM: %d\n", id_num);
+//			printk(KERN_ERR "ID_NUM: %d\n", id_num);
 			add_nums_async(num_transactions, 2,(unsigned long) id_num);
 		     );
 
 		ASYNC(
 			id_num = awe_mapper_create_id();
-			printk(KERN_ERR "ID_NUM: %d\n", id_num);
+//			printk(KERN_ERR "ID_NUM: %d\n", id_num);
 			add_nums_async(num_transactions, 3,(unsigned long) id_num);
 		     );
 

current/tests/rpc/rpc.h

@@ -6,7 +6,7 @@ typedef enum {NULL_INVOCATION, ADD_CONSTANT, ADD_NUMS, ADD_3_NUMS,
 	      ADD_4_NUMS, ADD_5_NUMS, ADD_6_NUMS} fn_types;
 
 /* must be divisible by 6... because I call 6 functions in the Callee.c */
-#define TRANSACTIONS 6
+#define TRANSACTIONS 6000
 
 
 int callee(void *chan);

current/tests/timing/Makefile

+USE_ASYNC='y'
+
 obj-m := timing.o
 
-ccflags-y += -O2 -I/users/sbauer/ipc/ramstore/fast-ipc-module/current/IPC
+
+ASYNC_DIR=/../../../../libasync/src
+
+ccflags-y += -O2 -I$(PWD)/../../IPC -I$(PWD)/$(ASYNC_DIR)/include/ 
 
 INC_DIR = ../../IPC
 
@@ -9,12 +14,19 @@ CFLAGS_ring-channel.o = -I$(INC_DIR)
 DEPS= $(INC_DIR)/ipc.h
 
 
-timing-objs :=  ../../IPC/ipc.o ../../ring-chan/ring-channel.o timing_setup.o
+timing-objs = timing_util.o ../../IPC/ipc.o ../../ring-chan/ring-channel.o
+
+ifeq ($(USE_ASYNC),'y')
+	timing-objs += async_timing_setup.o $(ASYNC_DIR)/common/thc.o $(ASYNC_DIR)/common/thcsync.o $(ASYNC_DIR)/common/awe-mapper.o
+else
+	timing-objs += timing_setup.o
+endif
 
 KDIR := /lib/modules/`uname -r`/build
 
 default:
 	make -C $(KDIR) M=$(PWD) modules
+
 clean:
 	make -C $(KDIR) M=$(PWD) clean
 	rm ../../IPC/ipc.o

current/tests/timing/timing_setup.c

@@ -5,10 +5,11 @@
 #include <linux/string.h>
 #include <linux/slab.h>
 #include <linux/sched.h>
-#include <linux/sort.h>
 #include <asm/tsc.h>
 
 #include "ipc.h"
+#include "timing_util.h"
+
 
 static struct ttd_ring_channel *prod;
 static struct ttd_ring_channel *cons;
@@ -21,48 +22,13 @@ MODULE_LICENSE("GPL");
 #define TRANSACTIONS 10000
 #define _35_MILLION 35000000
 
-
-/* Timing is used via serializing instructios because this white paper says so:
- *
- *  http://www.intel.com/content/www/us/en/intelligent-systems/embedded-systems-training/ia-32-ia-64-benchmark-code-execution-paper.html
- */
-static unsigned long RDTSC_START(void)
-{
-
-	unsigned cycles_low, cycles_high;
-
-	asm volatile ("CPUID\n\t"
-		      "RDTSC\n\t"
-		      "mov %%edx, %0\n\t"
-		      "mov %%eax, %1\n\t": "=r" (cycles_high), "=r" (cycles_low)::
-		      "%rax", "%rbx", "%rcx", "%rdx");
-	return ((unsigned long) cycles_high << 32) | cycles_low;
-
-}
-
-
-static unsigned long RDTSCP(void)
-{
-	unsigned long tsc;
-	__asm__ __volatile__(
-        "rdtscp;"
-        "shl $32, %%rdx;"
-        "or %%rdx, %%rax"
-        : "=a"(tsc)
-        :
-        : "%rcx", "%rdx");
-
-	return tsc;
-}
-
-
 static void  producer_iops(struct ttd_ring_channel *chan)
 {
 	unsigned long count = 0;
 	unsigned long start,end;
 	struct ipc_message *msg;
 
-	start = RDTSC_START();
+	start = timing_util_RDTSC_START();
 	while (count < _35_MILLION) {
 		msg = get_send_slot(chan);
 		msg->fn_type = 0x31337;
@@ -72,13 +38,13 @@ static void  producer_iops(struct ttd_ring_channel *chan)
 		msg->reg4 = 0x8888888888888888;
 		msg->reg5 = 0x7777777777777777;
 		msg->reg6 = 0x6666666666666666;
-		msg->reg7 = 0x5555555555555555;
+		msg->msg_id = 0x5555555555555555;
 		send(chan,msg);
 		msg = recv(chan);
 		transaction_complete(msg);
 		count++;
 	}
-	end = RDTSCP();
+	end = timing_util_RDTSCP();
 	pr_err("35 million ping pongs in %lu cyles\n", end-start);
 }
 
@@ -98,7 +64,7 @@ static void consumer_iops(struct ttd_ring_channel *chan)
 		msg->reg4 = 0x8888888888888888;
 		msg->reg5 = 0x7777777777777777;
 		msg->reg6 = 0x6666666666666666;
-		msg->reg7 = 0x5555555555555555;
+		msg->msg_id = 0x5555555555555555;
 		send(chan,msg);
 		count++;
 	}
@@ -112,7 +78,7 @@ static void producer(struct ttd_ring_channel *chan)
 	struct ipc_message *msg;
 
 	while (count < TRANSACTIONS) {
-		start = RDTSC_START();
+		start = timing_util_RDTSC_START();
 
 		msg = get_send_slot(chan);
 		msg->fn_type = 0x31337;
@@ -122,12 +88,12 @@ static void producer(struct ttd_ring_channel *chan)
 		msg->reg4 = 0x8888888888888888;
 		msg->reg5 = 0x7777777777777777;
 		msg->reg6 = 0x6666666666666666;
-		msg->reg7 = 0x5555555555555555;
+		msg->msg_id = 0x5555555555555555;
 		send(chan,msg);
 		msg = recv(chan);
 		transaction_complete(msg);
 
-		end = RDTSCP();
+		end = timing_util_RDTSCP();
 		time[count] = end-start;
 		count++;
 	}
@@ -149,53 +115,18 @@ static void consumer(struct ttd_ring_channel *chan)
 		msg->reg4 = 0x8888888888888888;
 		msg->reg5 = 0x7777777777777777;
 		msg->reg6 = 0x6666666666666666;
-		msg->reg7 = 0x5555555555555555;
+		msg->msg_id = 0x5555555555555555;
 		send(chan,msg);
 		count++;
 	}
 }
 
-static int compare(const void *_a, const void *_b){
-
-	u64 a = *((u64 *)_a);
-	u64 b = *((u64 *)_b);
-
-	if(a < b)
-		return -1;
-	if(a > b)
-		return 1;
-	return 0;
-}
-
-
-static void dump_time(void)
-{
-
-	int i;
-	unsigned long long counter = 0;
-        unsigned long min;
-	unsigned long max;
-
-	for (i = 0; i < TRANSACTIONS; i++) {
-		counter+= time[i];
-	}
-
-	sort(time, TRANSACTIONS, sizeof(unsigned long), compare, NULL);
-
-	min = time[0];
-	max = time[TRANSACTIONS-1];
-	counter = min;
-
-	pr_err("MIN\tMAX\tAVG\tMEDIAN\n");
-	pr_err("%lu & %lu & %llu & %lu \n", min, max, counter/TRANSACTIONS, time[4999]);
-
-}
 
 int dispatch(void *data)
 {
 	if(data == prod) {
 		producer(data);
-		dump_time();
+		timing_util_dump_time(time, (unsigned long)TRANSACTIONS);
 		kfree(time);
 	        producer_iops(data);
 	}