diff --git a/src/add-ons/kernel/file_systems/nfs4/Connection.cpp b/src/add-ons/kernel/file_systems/nfs4/Connection.cpp index cc4611b9d8..322ef47a7c 100644 --- a/src/add-ons/kernel/file_systems/nfs4/Connection.cpp +++ b/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. diff --git a/src/add-ons/kernel/file_systems/nfs4/Jamfile b/src/add-ons/kernel/file_systems/nfs4/Jamfile index a3a41946f2..246ba99d2a 100644 --- a/src/add-ons/kernel/file_systems/nfs4/Jamfile +++ b/src/add-ons/kernel/file_systems/nfs4/Jamfile @@ -35,6 +35,7 @@ KernelAddon nfs4 : RPCCallbackServer.cpp RPCReply.cpp RPCServer.cpp + WorkQueue.cpp XDR.cpp ; diff --git a/src/add-ons/kernel/file_systems/nfs4/NFS4Server.cpp b/src/add-ons/kernel/file_systems/nfs4/NFS4Server.cpp index 274ac6a3ce..9ea450b910 100644 --- a/src/add-ons/kernel/file_systems/nfs4/NFS4Server.cpp +++ b/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 - delegation->GetInode()->RecallDelegation(truncate); + DelegationRecallArgs* args = new(std::nothrow) DelegationRecallArgs; + 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; +} + diff --git a/src/add-ons/kernel/file_systems/nfs4/NFS4Server.h b/src/add-ons/kernel/file_systems/nfs4/NFS4Server.h index d312018e21..fe159b3b74 100644 --- a/src/add-ons/kernel/file_systems/nfs4/NFS4Server.h +++ b/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(); diff --git a/src/add-ons/kernel/file_systems/nfs4/RPCCallbackServer.cpp b/src/add-ons/kernel/file_systems/nfs4/RPCCallbackServer.cpp index b007a3ba99..54fd3d346f 100644 --- a/src/add-ons/kernel/file_systems/nfs4/RPCCallbackServer.cpp +++ b/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); diff --git a/src/add-ons/kernel/file_systems/nfs4/WorkQueue.cpp b/src/add-ons/kernel/file_systems/nfs4/WorkQueue.cpp new file mode 100644 index 0000000000..9d387248cf --- /dev/null +++ b/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 + + +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(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(args)); + break; + case IORequest: + JobIO(reinterpret_cast(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(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); +} + diff --git a/src/add-ons/kernel/file_systems/nfs4/WorkQueue.h b/src/add-ons/kernel/file_systems/nfs4/WorkQueue.h new file mode 100644 index 0000000000..954d011388 --- /dev/null +++ b/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 +#include +#include +#include + +#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 { + 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 fQueue; + + sem_id fThreadCancel; + thread_id fThread; +}; + + +inline status_t +WorkQueue::InitStatus() +{ + return fInitError; +} + + +extern WorkQueue* gWorkQueue; + + +#endif // WORKQUEUE_H + diff --git a/src/add-ons/kernel/file_systems/nfs4/kernel_interface.cpp b/src/add-ons/kernel/file_systems/nfs4/kernel_interface.cpp index 9a9f8b9c23..f396be2df2 100644 --- a/src/add-ons/kernel/file_systems/nfs4/kernel_interface.cpp +++ b/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 - return B_UNSUPPORTED; + Inode* inode = reinterpret_cast(vnode->private_node); + + 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;