Order DSM TCP Responses

arch/x86/include/asm/kvm_host.h

@@ -670,8 +670,8 @@ struct kvm_lpage_info {
 #define KVM_DSM_DEBUG
 
 /* TODO: Make TCP interfaces compatible. */
-//#define USE_KTCP_NETWORK
-#define USE_KRDMA_NETWORK
+#define USE_KTCP_NETWORK
+//#define USE_KRDMA_NETWORK
 
 #define IVY_KVM_DSM
 //#define TARDIS_KVM_DSM

arch/x86/kvm/dsm-util.c

@@ -215,6 +215,8 @@ void dsm_decode_diff(char *page, int resp_len,
 	memcpy(page, buffer, PAGE_SIZE);
 	kfree(buffer);
 #else
+	if(resp_len != 0 && resp_len != PAGE_SIZE)
+		printk(KERN_WARNING "%s: response length unexpected %d\n", __func__, resp_len);
 	BUG_ON(resp_len != 0 && resp_len != PAGE_SIZE);
 #endif
 }

arch/x86/kvm/dsm-util.h

@@ -1,6 +1,7 @@
 #ifndef ARCH_X86_KVM_DSM_UTIL_H
 #define ARCH_X86_KVM_DSM_UTIL_H
 
+#include <linux/jhash.h>
 #include <linux/kvm_host.h>
 
 #define DSM_INITIAL     0

arch/x86/kvm/ktcp.c

@@ -27,11 +27,14 @@
 #include <linux/socket.h>
 #include <linux/slab.h>
 #include <linux/delay.h>
+#include <linux/kvm_host.h>
 
 #include "ktcp.h"
 
+#define USE_CACHE
+
 struct ktcp_hdr {
-	extent_t extent;
+	tx_add_t extent;
 	uint16_t length;
 } __attribute__((packed));
 
@@ -81,19 +84,25 @@ static int __ktcp_send(struct socket *sock, const char *buffer, size_t length,
 }
 
 int ktcp_send(struct socket *sock, const char *buffer, size_t length,
-		unsigned long flags, extent_t extent)
+		unsigned long flags, const tx_add_t * tx_add)
 {
 	struct ktcp_hdr hdr = {
 		.length = length,
-		.extent = extent,
+		.extent = *tx_add,
 	};
 	int ret;
 	mm_segment_t oldmm;
-	char *local_buffer = kmalloc(KTCP_BUFFER_SIZE, GFP_KERNEL);
+	char *local_buffer;
+
+	printk(KERN_DEBUG "%s: txid %d\n", __func__, tx_add->txid);
+
+	local_buffer = kmalloc(KTCP_BUFFER_SIZE, GFP_KERNEL);
 	if (!local_buffer) {
 		return -ENOMEM;
 	}
 
+	BUG_ON((length  + sizeof(hdr)) > KTCP_BUFFER_SIZE);
+
 	// Get current address access limit
 	oldmm = get_fs();
 	set_fs(KERNEL_DS);
@@ -112,7 +121,7 @@ int ktcp_send(struct socket *sock, const char *buffer, size_t length,
 	return ret < 0 ? ret : hdr.length;
 }
 
-static int __ktcp_receive(struct socket *sock, char *buffer, size_t expected_size,
+static int __ktcp_receive__(struct socket *sock, char *buffer, size_t expected_size,
 		unsigned long flags)
 {
 	struct kvec vec;
@@ -162,38 +171,123 @@ static int __ktcp_receive(struct socket *sock, char *buffer, size_t expected_siz
 	return len;
 }
 
-int ktcp_receive(struct socket *sock, char *buffer, unsigned long flags,
-		extent_t *extent)
+//Allocate a buffer and store the any received message
+static struct ktcp_hdr*  __ktcp_receive_get(struct socket *sock, unsigned long flags)
 {
-	struct ktcp_hdr hdr;
 	int ret;
 	char *local_buffer = kmalloc(KTCP_BUFFER_SIZE, GFP_KERNEL);
 	if (!local_buffer) {
-		return -ENOMEM;
+		return NULL;
 	}
 
-	hdr.length = 0xDEAD;
-	ret = __ktcp_receive(sock, local_buffer, KTCP_BUFFER_SIZE, flags);
+	ret = __ktcp_receive__(sock, local_buffer, KTCP_BUFFER_SIZE, flags);
 	if (ret < 0) {
-		goto out;
+		return NULL;
 	}
 
-	memcpy(&hdr, local_buffer, sizeof(hdr));
+	return (struct ktcp_hdr*) local_buffer;
+}
+
+#ifdef USE_CACHE
+#define BLOCKED_HASH_BITS	7
+static DEFINE_HASHTABLE(ktcp_hash, BLOCKED_HASH_BITS);
+static DEFINE_SPINLOCK(ktcp_hash_lock);
+
+struct ktcp_cache_entry_s
+{
+	struct ktcp_hdr* hdr;
+	struct hlist_node hlink;
+};
+
+static void ktcp_cache_put(uint16_t txid, struct ktcp_hdr* hdr)
+{
+ 	struct ktcp_cache_entry_s *entry;
+	entry = kmalloc(sizeof(struct ktcp_cache_entry_s), GFP_KERNEL);
+	entry->hdr=hdr;
+
+	spin_lock(&ktcp_hash_lock);
+	hash_add(ktcp_hash, &entry->hlink, hdr->extent.txid);
+	spin_unlock(&ktcp_hash_lock);
+}
 
+static struct ktcp_hdr* ktcp_cache_pop(uint16_t txid)
+{
+	int found=0;
+	struct ktcp_hdr *hdr=NULL;
+	struct ktcp_cache_entry_s *entry=NULL;
+
+	spin_lock(&ktcp_hash_lock);
+	hash_for_each_possible(ktcp_hash, entry, hlink, txid) {
+		hdr=entry->hdr;
+		if(txid==hdr->extent.txid || txid==0xFF)
+		{
+			found=1;
+			break;
+		}
+	}
+	if(found)
+		hash_del(&entry->hlink);
+	else
+		hdr=NULL;
+	spin_unlock(&ktcp_hash_lock);
+
+	kfree(entry); entry=NULL;
+	return hdr;
+}
+#endif
+
+//and return the corresponding local_buffer
+int ktcp_receive(struct socket *sock, char* buffer, unsigned long flags, 
+			tx_add_t *tx_add)
+{
+	int ret=0;
+	struct ktcp_hdr *hdr;
+	uint16_t txid=tx_add->txid;
+	uint16_t length=0;
+
+	//printk(KERN_DEBUG "%s: txid %d\n", __func__, tx_add->txid);
+	//Execute receive_get and cache_get until the right transaction is found
+	do{
+		//Get from network
+		hdr=(struct ktcp_hdr*) __ktcp_receive_get(sock, flags);
+#ifdef USE_CACHE
+		if(hdr->extent.txid==txid || txid==0xFF)
+		{
+			//if found, we exit the loop
+			break;
+		}
+
+		//add local buffer to the cache
+		ktcp_cache_put(txid, hdr); hdr=NULL;
+
+		//check if not already in the cache (putted by another thread)
+		hdr=ktcp_cache_pop(txid);
+
+	}while(hdr==NULL);//What if we never receive the transaction?
+	BUG_ON(!hdr || (hdr->extent.txid!=txid && txid!=0xFF));
+#else
+	}while(0);
+#endif
+
+	printk(KERN_DEBUG "%s: txid requested %d found %d\n", __func__, txid, hdr->extent.txid);
+
+	length = hdr->length;
 	/* hdr.length is undetermined on process killed */
-	if (unlikely(hdr.length > PAGE_SIZE)) {
+	if (unlikely(length > PAGE_SIZE)) {
+		printk(KERN_WARNING "%s: buffer to small\n", __func__);
 		ret = -EFAULT;
 		goto out;
 	}
-	memcpy(buffer, local_buffer + sizeof(hdr), hdr.length);
 
-	if (extent) {
-		*extent = hdr.extent;
+	memcpy(buffer, (char*)hdr + sizeof(hdr), length);
+
+	if (tx_add) {
+		*tx_add = hdr->extent;
 	}
 
 out:
-	kfree(local_buffer);
-	return ret < 0 ? ret : hdr.length;
+	kfree(hdr); hdr=NULL;
+	return ret < 0 ? ret : length;
 }
 
 int ktcp_connect(const char *host, const char *port, struct socket **conn_socket)
@@ -208,7 +302,7 @@ int ktcp_connect(const char *host, const char *port, struct socket **conn_socket
 
 	ret = sock_create(PF_INET, SOCK_STREAM, IPPROTO_TCP, conn_socket);
 	if (ret < 0) {
-		printk("sock_create %d\n", ret);
+		printk(KERN_DEBUG "sock_create %d\n", ret);
 		return ret;
 	}
 
@@ -226,7 +320,7 @@ int ktcp_connect(const char *host, const char *port, struct socket **conn_socket
 	}
 
 	if (ret && (ret != -EINPROGRESS)) {
-		printk("connect %d\n", ret);
+		printk(KERN_DEBUG "connect %d\n", ret);
 		sock_release(*conn_socket);
 		return ret;
 	}
@@ -239,8 +333,7 @@ int ktcp_listen(const char *host, const char *port, struct socket **listen_socke
 	struct sockaddr_in saddr;
 	long portdec;
 
-	BUILD_BUG_ON((sizeof(struct ktcp_hdr)) != (sizeof(uint16_t) +
-				sizeof(extent_t)));
+	BUILD_BUG_ON((sizeof(struct ktcp_hdr)) != (sizeof(uint16_t) + sizeof(extent_t)));
 
 	ret = sock_create(PF_INET, SOCK_STREAM, IPPROTO_TCP, listen_socket);
 	if (ret != 0) {

arch/x86/kvm/ktcp.h

@@ -8,13 +8,16 @@
 // How many requests can be buffered in the listening queue
 #define DEFAULT_BACKLOG 16
 
-typedef uint32_t extent_t;
+
+struct tx_add;
+typedef struct tx_add tx_add_t;
+typedef tx_add_t extent_t;
 
 int ktcp_send(struct socket *sock, const char *buffer, size_t length,
-		unsigned long flags, extent_t extent);
+		unsigned long flags, const tx_add_t * tx_add);
 
 int ktcp_receive(struct socket *sock, char *buffer, unsigned long flags,
-		extent_t *extent);
+		tx_add_t *tx_add);
 
 int ktcp_connect(const char *host, const char *port, struct socket **conn_socket);