[jboss-cvs] JBoss Messaging SVN: r3718 - projects/jaio/trunk/jaio/native/src.

[jboss-cvs-commits at lists.jboss.org]

Author: clebert.suconic at jboss.com
Date: 2008-02-13 16:58:05 -0500 (Wed, 13 Feb 2008)
New Revision: 3718

Modified:
   projects/jaio/trunk/jaio/native/src/AIOException.h
   projects/jaio/trunk/jaio/native/src/FileOutput.cpp
   projects/jaio/trunk/jaio/native/src/FileOutput.h
Log:
Aligning data

Modified: projects/jaio/trunk/jaio/native/src/AIOException.h
===================================================================
--- projects/jaio/trunk/jaio/native/src/AIOException.h	2008-02-13 16:03:16 UTC (rev 3717)
+++ projects/jaio/trunk/jaio/native/src/AIOException.h	2008-02-13 21:58:05 UTC (rev 3718)
@@ -10,7 +10,7 @@
 	int errorCode;
 	std::string message;
 public:
-	AIOException(int _errorCode, std::string & _message) throw() : errorCode(_errorCode), message(_message)
+	AIOException(int _errorCode, std::string  _message) throw() : errorCode(_errorCode), message(_message)
 	{
 		errorCode = _errorCode;
 		message = _message;

Modified: projects/jaio/trunk/jaio/native/src/FileOutput.cpp
===================================================================
--- projects/jaio/trunk/jaio/native/src/FileOutput.cpp	2008-02-13 16:03:16 UTC (rev 3717)
+++ projects/jaio/trunk/jaio/native/src/FileOutput.cpp	2008-02-13 21:58:05 UTC (rev 3718)
@@ -1,15 +1,40 @@
+#include <stdlib.h>
 #include <list>
 #include <iostream>
+#include <sstream>
 #include <memory.h>
+#include <errno.h>
+#include <libaio.h>
 #include <fcntl.h>
 #include "FileOutput.h"
 #include "AIOException.h"
 
-FileOutput::FileOutput(std::string & _fileName)
+
+std::string io_error(int rc)
 {
+	std::stringstream buffer;
+	
+	if (rc == -ENOSYS)
+		buffer << "AIO not in this kernel";
+	else 
+		buffer << "Error:= " << strerror(-rc);
+	
+	return buffer.str();
+}
+
+
+FileOutput::FileOutput(std::string & _fileName) : aioContext(0), filePointer(0)
+{
 	fileName = _fileName;
+	std::cout << "Initializing FileOutput " << aioContext << "\n";
+	if (io_queue_init(MAX_IO, &aioContext))
+	{
+		throw AIOException(1, "Can't initialize aio"); 
+	}
+	std::cout << "Initialized " << aioContext << "\n";
 	
-	fileHandle = open(fileName.data(),  O_WRONLY | O_CREAT, 0666);
+	fileHandle = open(fileName.data(),  O_WRONLY | O_CREAT | O_DIRECT, 0666);
+	//fileHandle = open(fileName.data(),  O_WRONLY | O_CREAT, 0666);
 	if (fileHandle < 0)
 	{
 		throw AIOException(1, "Can't open file"); 
@@ -19,6 +44,10 @@
 
 FileOutput::~FileOutput()
 {
+	if (io_queue_release(aioContext))
+	{
+		throw AIOException(2,"Can't release aio");
+	}
 	if (close(fileHandle))
 	{
 		throw AIOException(2,"Can't close file");
@@ -38,15 +67,54 @@
 	delete adapters;
 }
 
+void FileOutput::callbackDirect(io_context_t ctx, struct iocb *iocb, long res, long res2)
+{
+	if (res<0 || res2<0)
+	{
+		std::cout << io_error(res);
+		std::cout <<"res = " << res << " res2 = " << res2 << "\n";
+		std::cout.flush();
+	}
+		
+	BufferAdapter * buffer = 0;
+	
+	memcpy (&buffer, ((char *) iocb->u.c.buf)+iocb->u.c.nbytes, sizeof (BufferAdapter *));
+	
+	buffer->completeBlock();
+	
+	free (iocb->u.c.buf);
+	free (iocb);
+	buffer->deleteRef();
+	
+	
+}
 
+
 // Used only when Paging is disabled (as the user have this option to opt out for Paging)
 void FileOutput::addData(BufferAdapter * adapter)
 {
 	int size = adapter->blockSize();
-	void * buffer = malloc(size);
+	void * buffer;
+	if (posix_memalign(&buffer, 512, size + 512))
+	{
+		throw AIOException(10, "Error on posix_memalign");
+	}
+	
+	//void * buffer = malloc(size + sizeof (BufferAdapter *));
 	adapter->encode(size, buffer);
-	::write(fileHandle, buffer, size);
-	adapter->completeBlock();
-	adapter->deleteRef();
-	free(buffer);
+	memcpy (((char *)buffer) + size, &adapter, sizeof (adapter));
+	
+	struct iocb * iocb = new struct iocb();
+	::io_prep_pwrite(iocb, fileHandle, buffer, size, filePointer);
+	::io_set_callback(iocb, &callbackDirect);
+	filePointer+=size;
+	
+	int result = ::io_submit(aioContext, 1, &iocb);
+	if (result<0)
+	{
+		std::stringstream str;
+		str<< "Problem on submit block, errorCode=" << result;
+		throw AIOException (6, str.str());
+	}
 }
+

Modified: projects/jaio/trunk/jaio/native/src/FileOutput.h
===================================================================
--- projects/jaio/trunk/jaio/native/src/FileOutput.h	2008-02-13 16:03:16 UTC (rev 3717)
+++ projects/jaio/trunk/jaio/native/src/FileOutput.h	2008-02-13 21:58:05 UTC (rev 3718)
@@ -4,12 +4,17 @@
 #include "PageObserver.h"
 #include "DataManager.h"
 #include <string>
+#include <libaio.h>
 
+#define MAX_IO 3000
+
 class FileOutput : public PageObserver, DataManager
 {
 private:
 	int fileHandle;
 	std::string fileName;
+	io_context_t aioContext;
+	off_t filePointer;
 public:
 	FileOutput(std::string & _fileName);
 	virtual ~FileOutput();
@@ -24,9 +29,12 @@
 	// Nothing to be done on FileOutput 
 	void flushMemory()
 	{
-		
+		io_queue_run(aioContext);
 	}
+	
+	static void callbackDirect(io_context_t ctx, struct iocb *iocb, long res, long res2);
 
+
 };
 
 #endif /*FILEOUTPUT_H_*/