nfs4: Add asynchronous work queue

src/add-ons/kernel/file_systems/nfs4/Connection.cpp

@@ -163,7 +163,7 @@ ServerAddress::InAddr() const
 status_t
 ServerAddress::ResolveName(const char* name, ServerAddress* address)
 {
-	address->fProtocol = IPPROTO_TCP;
+	address->fProtocol = IPPROTO_UDP;
 
 	// getaddrinfo() is very expensive when called from kernel, so we do not
 	// want to call it unless there is no other choice.

src/add-ons/kernel/file_systems/nfs4/Jamfile

@@ -35,6 +35,7 @@ KernelAddon nfs4 :
 	RPCCallbackServer.cpp
 	RPCReply.cpp
 	RPCServer.cpp
+	WorkQueue.cpp
 	XDR.cpp
 ;
 

src/add-ons/kernel/file_systems/nfs4/NFS4Server.cpp

@@ -11,6 +11,7 @@
 #include "Inode.h"
 #include "NFS4Server.h"
 #include "Request.h"
+#include "WorkQueue.h"
 
 
 NFS4Server::NFS4Server(RPC::Server* serv)
@@ -247,6 +248,9 @@ NFS4Server::_Renewal()
 		request.Send();
 
 		switch (request.Reply().NFS4Error()) {
+			case NFS4ERR_CB_PATH_DOWN:
+				RecallAll();
+				break;
 			case NFS4ERR_STALE_CLIENTID:
 				ServerRebooted(clientId);
 				break;
@@ -328,11 +332,40 @@ NFS4Server::CallbackRecall(RequestInterpreter* request, ReplyBuilder* reply)
 		return B_FILE_NOT_FOUND;
 	}
 
-	// TODO: should be asynchronous
+	DelegationRecallArgs* args = new(std::nothrow) DelegationRecallArgs;
-	delegation->GetInode()->RecallDelegation(truncate);
+	args->fDelegation = delegation;
+	args->fTruncate = truncate;
+	gWorkQueue->EnqueueJob(DelegationRecall, args);
 
 	reply->Recall(B_OK);
 
 	return B_OK;
 }
 
+
+status_t
+NFS4Server::RecallAll()
+{
+#if 0
+	MutexLocker locker(fFSLock);
+
+	Delegation* delegation = NULL;
+	FileSystem* current = fFileSystems;
+	while (current != NULL) {
+		delegation = current->GetDelegation(handle);
+		if (delegation != NULL)
+			break;
+
+		current = current->fNext;
+	}
+	locker.Unlock();
+
+	DelegationRecallArgs args = new(std::nothrow) DelegationRecallArgs;
+	DelegationRecallArgs* args;
+	args->fDelegation = delegation;
+	args->fTruncate = truncate;
+	gWorkQueue->EnqueueJob(DelegationRecall, args);
+#endif
+	return B_OK;
+}
+

src/add-ons/kernel/file_systems/nfs4/NFS4Server.h

@@ -42,6 +42,7 @@ public:
 
 			status_t		CallbackRecall(RequestInterpreter* request,
 								ReplyBuilder* reply);
+			status_t		RecallAll();
 private:
 			status_t		_GetLeaseTime();
 

src/add-ons/kernel/file_systems/nfs4/RPCCallbackServer.cpp

@@ -26,7 +26,7 @@ CallbackServer::CallbackServer()
 	fConnectionList(NULL),
 	fListener(NULL),
 	fThreadRunning(false),
-	//fCallbackArray(NULL),
+	fCallbackArray(NULL),
 	fArraySize(0),
 	fFreeSlot(-1)
 {
@@ -40,7 +40,7 @@ CallbackServer::~CallbackServer()
 {
 	StopServer();
 
-	//free(fCallbackArray);
+	free(fCallbackArray);
 	rw_lock_destroy(&fArrayLock);
 	mutex_destroy(&fThreadLock);
 	mutex_destroy(&fConnectionLock);

src/add-ons/kernel/file_systems/nfs4/WorkQueue.cpp

@@ -0,0 +1,184 @@
+/*
+ * Copyright 2012 Haiku, Inc. All rights reserved.
+ * Distributed under the terms of the MIT License.
+ *
+ * Authors:
+ *		Paweł Dziepak, pdziepak@quarnos.org
+ */
+
+
+#include "WorkQueue.h"
+
+#include <io_requests.h>
+
+
+WorkQueue*		gWorkQueue		= NULL;
+
+
+WorkQueue::WorkQueue()
+	:
+	fQueueSemaphore(create_sem(0, NULL)),
+	fThreadCancel(create_sem(0, NULL))
+{
+	mutex_init(&fQueueLock, NULL);
+
+	fThread = spawn_kernel_thread(&WorkQueue::LaunchWorkingThread,
+		"NFSv4 Work Queue", B_NORMAL_PRIORITY, this);
+	if (fThread < B_OK) {
+		fInitError = fThread;
+		return;
+	}
+
+	status_t result = resume_thread(fThread);
+	if (result != B_OK) {
+		kill_thread(fThread);
+		fInitError = result;
+		return;
+	}
+
+	fInitError = B_OK;
+}
+
+
+WorkQueue::~WorkQueue()
+{
+	release_sem(fThreadCancel);
+
+	status_t result;
+	wait_for_thread(fThread, &result);
+
+	mutex_destroy(&fQueueLock);
+	delete_sem(fThreadCancel);
+	delete_sem(fQueueSemaphore);
+}
+
+
+status_t
+WorkQueue::EnqueueJob(JobType type, void* args)
+{
+	WorkQueueEntry* entry = new(std::nothrow) WorkQueueEntry;
+	if (entry == NULL)
+		return B_NO_MEMORY;
+
+	entry->fType = type;
+	entry->fArguments = args;
+	MutexLocker locker(fQueueLock);
+	fQueue.InsertAfter(fQueue.Tail(), entry);
+	locker.Unlock();
+
+	release_sem(fQueueSemaphore);
+	return B_OK;
+}
+
+
+status_t
+WorkQueue::LaunchWorkingThread(void* object)
+{
+	WorkQueue* queue = reinterpret_cast<WorkQueue*>(object);
+	return queue->WorkingThread();
+}
+
+
+status_t
+WorkQueue::WorkingThread()
+{
+	while (true) {
+		object_wait_info object[2];
+		object[0].object = fThreadCancel;
+		object[0].type = B_OBJECT_TYPE_SEMAPHORE;
+		object[0].events = B_EVENT_ACQUIRE_SEMAPHORE;
+
+		object[1].object = fQueueSemaphore;
+		object[1].type = B_OBJECT_TYPE_SEMAPHORE;
+		object[1].events = B_EVENT_ACQUIRE_SEMAPHORE;
+
+		status_t result = wait_for_objects(object, 2);
+
+		if (result < B_OK ||
+			(object[0].events & B_EVENT_ACQUIRE_SEMAPHORE) != 0) {
+			return result;
+		} else if ((object[1].events & B_EVENT_ACQUIRE_SEMAPHORE) == 0)
+			continue;
+
+		acquire_sem(fQueueSemaphore);
+
+		DequeueJob();
+	}
+
+	return B_OK;
+}
+
+
+void
+WorkQueue::DequeueJob()
+{
+	MutexLocker locker(fQueueLock);
+
+	WorkQueueEntry* entry = fQueue.RemoveHead();
+
+	void* args = entry->fArguments;
+	switch (entry->fType) {
+		case DelegationRecall:
+			JobRecall(reinterpret_cast<DelegationRecallArgs*>(args));
+			break;
+		case IORequest:
+			JobIO(reinterpret_cast<IORequestArgs*>(args));
+			break;
+	}
+
+	delete entry;
+}
+
+
+void
+WorkQueue::JobRecall(DelegationRecallArgs* args)
+{
+	args->fDelegation->GetInode()->RecallDelegation(args->fTruncate);
+}
+
+
+void
+WorkQueue::JobIO(IORequestArgs* args)
+{
+	uint64 offset = io_request_offset(args->fRequest);
+	uint64 length = io_request_length(args->fRequest);
+
+	char* buffer = reinterpret_cast<char*>(malloc(length));
+	if (buffer == NULL) {
+		notify_io_request(args->fRequest, B_NO_MEMORY);
+		return;
+	}
+
+	bool eof = false;
+	uint64 size = 0;
+	status_t result;
+	if (io_request_is_write(args->fRequest)) {
+		result = read_from_io_request(args->fRequest, buffer, length);
+
+		do {
+			size_t bytesWritten = length - size;
+			result = args->fInode->WriteDirect(NULL, offset + size,
+				buffer + size, &bytesWritten);
+			size += bytesWritten;
+		} while (size < length && result == B_OK);
+	} else {
+		do {
+			size_t bytesRead = length - size;
+			result = args->fInode->ReadDirect(NULL, offset + size, buffer,
+				&bytesRead, &eof);
+			if (result != B_OK)
+				break;
+
+			result = write_to_io_request(args->fRequest, buffer, bytesRead);
+			if (result != B_OK)
+				break;
+
+			size += bytesRead;
+		} while (size < length && result == B_OK && !eof);
+		
+	}
+	free(buffer);
+
+	notify_io_request(args->fRequest, result);
+}
+

src/add-ons/kernel/file_systems/nfs4/WorkQueue.h

@@ -0,0 +1,82 @@
+/*
+ * Copyright 2012 Haiku, Inc. All rights reserved.
+ * Distributed under the terms of the MIT License.
+ *
+ * Authors:
+ *		Paweł Dziepak, pdziepak@quarnos.org
+ */
+#ifndef WORKQUEUE_H
+#define WORKQUEUE_H
+
+
+#include <io_requests.h>
+#include <lock.h>
+#include <SupportDefs.h>
+#include <util/DoublyLinkedList.h>
+
+#include "Delegation.h"
+#include "Inode.h"
+
+
+enum JobType {
+	DelegationRecall,
+	IORequest
+};
+
+struct DelegationRecallArgs {
+	Delegation*		fDelegation;
+	bool			fTruncate;
+};
+
+struct IORequestArgs {
+	io_request*		fRequest;
+	Inode*			fInode;
+};
+
+struct WorkQueueEntry : public DoublyLinkedListLinkImpl<WorkQueueEntry> {
+	JobType			fType;
+	void*			fArguments;
+};
+
+class WorkQueue {
+public:
+						WorkQueue();
+						~WorkQueue();
+
+	inline	status_t	InitStatus();
+
+			status_t	EnqueueJob(JobType type, void* args);
+
+protected:
+	static	status_t	LaunchWorkingThread(void* object);
+			status_t	WorkingThread();
+
+			void		DequeueJob();
+
+			void		JobRecall(DelegationRecallArgs* args);
+			void		JobIO(IORequestArgs* args);
+
+private:
+			status_t	fInitError;
+
+			sem_id		fQueueSemaphore;
+			mutex		fQueueLock;
+			DoublyLinkedList<WorkQueueEntry>	fQueue;
+
+			sem_id		fThreadCancel;
+			thread_id	fThread;	
+};
+
+
+inline status_t
+WorkQueue::InitStatus()
+{
+	return fInitError;
+}
+
+
+extern WorkQueue*		gWorkQueue;
+
+
+#endif	// WORKQUEUE_H
+

src/add-ons/kernel/file_systems/nfs4/kernel_interface.cpp

@@ -22,6 +22,7 @@
 #include "RootInode.h"
 #include "RPCCallbackServer.h"
 #include "RPCServer.h"
+#include "WorkQueue.h"
 
 
 extern fs_volume_ops gNFSv4VolumeOps;
@@ -282,8 +283,21 @@ nfs4_write_pages(fs_volume* _volume, fs_vnode* vnode, void* _cookie, off_t pos,
 static status_t
 nfs4_io(fs_volume* volume, fs_vnode* vnode, void* cookie, io_request* request)
 {
-	// no asynchronous calls yet
+	Inode* inode = reinterpret_cast<Inode*>(vnode->private_node);
-	return B_UNSUPPORTED;
+
+	IORequestArgs* args = new(std::nothrow) IORequestArgs;
+	if (args == NULL) {
+		notify_io_request(request, B_NO_MEMORY);
+		return B_NO_MEMORY;
+	}
+	args->fRequest = request;
+	args->fInode = inode;
+
+	status_t result = gWorkQueue->EnqueueJob(IORequest, args);
+	if (result != B_OK)
+		notify_io_request(request, result);
+
+	return result;
 }
 
 
@@ -651,10 +665,19 @@ nfs4_init()
 		return B_NO_MEMORY;
 	}
 
+	gWorkQueue = new(std::nothrow) WorkQueue;
+	if (gWorkQueue == NULL || gWorkQueue->InitStatus() != B_OK) {
+		delete gWorkQueue;
+		delete gIdMapper;
+		delete gRPCServerManager;
+		return B_NO_MEMORY;
+	}
+
 	gRPCCallbackServer = new(std::nothrow) RPC::CallbackServer;
 	if (gRPCCallbackServer == NULL) {
-		delete gRPCServerManager;
 		delete gIdMapper;
+		delete gWorkQueue;
+		delete gRPCServerManager;
 		return B_NO_MEMORY;
 	}
 
@@ -669,6 +692,7 @@ nfs4_uninit()
 
 	delete gRPCCallbackServer;
 	delete gIdMapper;
+	delete gWorkQueue;
 	delete gRPCServerManager;
 
 	return B_OK;