From 94c47dc6a85fa8c028382eb3b54e04f40d08590c Mon Sep 17 00:00:00 2001 From: Pawel Dziepak Date: Sun, 5 Aug 2012 02:31:27 +0200 Subject: [PATCH] nfs4: Add basic RPC level support for callbacks --- .../kernel/file_systems/nfs4/Connection.cpp | 172 +++++++++-- .../kernel/file_systems/nfs4/Connection.h | 46 ++- src/add-ons/kernel/file_systems/nfs4/Jamfile | 3 + .../kernel/file_systems/nfs4/NFS4Defs.h | 5 + .../kernel/file_systems/nfs4/RPCCallback.cpp | 24 ++ .../kernel/file_systems/nfs4/RPCCallback.h | 50 +++ .../file_systems/nfs4/RPCCallbackRequest.cpp | 71 +++++ .../file_systems/nfs4/RPCCallbackRequest.h | 82 +++++ .../file_systems/nfs4/RPCCallbackServer.cpp | 285 ++++++++++++++++++ .../file_systems/nfs4/RPCCallbackServer.h | 99 ++++++ .../kernel/file_systems/nfs4/RPCServer.cpp | 22 ++ .../kernel/file_systems/nfs4/RPCServer.h | 6 + .../file_systems/nfs4/RequestBuilder.cpp | 16 +- .../kernel/file_systems/nfs4/RequestBuilder.h | 2 +- .../file_systems/nfs4/kernel_interface.cpp | 9 + 15 files changed, 858 insertions(+), 34 deletions(-) create mode 100644 src/add-ons/kernel/file_systems/nfs4/RPCCallback.cpp create mode 100644 src/add-ons/kernel/file_systems/nfs4/RPCCallback.h create mode 100644 src/add-ons/kernel/file_systems/nfs4/RPCCallbackRequest.cpp create mode 100644 src/add-ons/kernel/file_systems/nfs4/RPCCallbackRequest.h create mode 100644 src/add-ons/kernel/file_systems/nfs4/RPCCallbackServer.cpp create mode 100644 src/add-ons/kernel/file_systems/nfs4/RPCCallbackServer.h diff --git a/src/add-ons/kernel/file_systems/nfs4/Connection.cpp b/src/add-ons/kernel/file_systems/nfs4/Connection.cpp index 6f14ed8e58..b9a5f0eb41 100644 --- a/src/add-ons/kernel/file_systems/nfs4/Connection.cpp +++ b/src/add-ons/kernel/file_systems/nfs4/Connection.cpp @@ -129,6 +129,22 @@ ServerAddress::Port() const } +void +ServerAddress::SetPort(uint16 port) +{ + port = htons(port); + + switch (reinterpret_cast(&fAddress)->sa_family) { + case AF_INET: + reinterpret_cast(&fAddress)->sin_port = port; + break; + case AF_INET6: + reinterpret_cast(&fAddress)->sin6_port = port; + break; + } +} + + const void* ServerAddress::InAddr() const { @@ -147,7 +163,7 @@ ServerAddress::InAddr() const status_t ServerAddress::ResolveName(const char* name, ServerAddress* address) { - address->fProtocol = IPPROTO_UDP; + address->fProtocol = IPPROTO_TCP; // getaddrinfo() is very expensive when called from kernel, so we do not // want to call it unless there is no other choice. @@ -193,6 +209,20 @@ ServerAddress::ResolveName(const char* name, ServerAddress* address) Connection::Connection(const ServerAddress& address) + : + ConnectionBase(address) +{ +} + + +ConnectionListener::ConnectionListener(const ServerAddress& address) + : + ConnectionBase(address) +{ +} + + +ConnectionBase::ConnectionBase(const ServerAddress& address) : fWaitCancel(create_sem(0, NULL)), fSocket(-1), @@ -202,6 +232,7 @@ Connection::Connection(const ServerAddress& address) } + ConnectionStream::ConnectionStream(const ServerAddress& address) : Connection(address) @@ -216,7 +247,7 @@ ConnectionPacket::ConnectionPacket(const ServerAddress& address) } -Connection::~Connection() +ConnectionBase::~ConnectionBase() { if (fSocket != -1) close(fSocket); @@ -226,13 +257,13 @@ Connection::~Connection() status_t -Connection::GetLocalAddress(ServerAddress* address) +ConnectionBase::GetLocalAddress(ServerAddress* address) { address->fProtocol = fServerAddress.fProtocol; - socklen_t addressSize = fServerAddress.AddressSize(); - return getsockname(fSocket, - (struct sockaddr*)&address->fAddress, &addressSize); + socklen_t addressSize = sizeof(address->fAddress); + return getsockname(fSocket, (struct sockaddr*)&address->fAddress, + &addressSize); } @@ -409,20 +440,24 @@ ConnectionPacket::Receive(void** _buffer, uint32* _size) } +Connection* +Connection::CreateObject(const ServerAddress& address) +{ + switch (address.fProtocol) { + case IPPROTO_TCP: + return new(std::nothrow) ConnectionStream(address); + case IPPROTO_UDP: + return new(std::nothrow) ConnectionPacket(address); + default: + return NULL; + } +} + + status_t Connection::Connect(Connection **_connection, const ServerAddress& address) { - Connection* conn; - switch (address.fProtocol) { - case IPPROTO_TCP: - conn = new(std::nothrow) ConnectionStream(address); - break; - case IPPROTO_UDP: - conn = new(std::nothrow) ConnectionPacket(address); - break; - default: - return B_BAD_VALUE; - } + Connection* conn = CreateObject(address); if (conn == NULL) return B_NO_MEMORY; @@ -438,6 +473,21 @@ Connection::Connect(Connection **_connection, const ServerAddress& address) } +status_t +Connection::SetTo(Connection **_connection, int socket, + const ServerAddress& address) +{ + Connection* conn = CreateObject(address); + if (conn == NULL) + return B_NO_MEMORY; + conn->fSocket = socket; + + *_connection = conn; + + return B_OK; +} + + status_t Connection::Connect() { @@ -491,7 +541,7 @@ Connection::Reconnect() void -Connection::Disconnect() +ConnectionBase::Disconnect() { release_sem(fWaitCancel); @@ -499,3 +549,89 @@ Connection::Disconnect() fSocket = -1; } + +status_t +ConnectionListener::Listen(ConnectionListener** _listener, uint16 port) +{ + int sock = socket(AF_INET, SOCK_STREAM, IPPROTO_TCP); + if (sock < 0) + return errno; + + sockaddr_in addr; + memset(&addr, 0, sizeof(addr)); + addr.sin_len = sizeof(addr); + addr.sin_family = AF_INET; + addr.sin_addr.s_addr = INADDR_ANY; + addr.sin_port = htons(port); + if (bind(sock, (struct sockaddr*)&addr, sizeof(addr)) != B_OK) { + close(sock); + return errno; + } + + if (listen(sock, 5) != B_OK) { + close(sock); + return errno; + } + + ServerAddress address; + address.fProtocol = IPPROTO_TCP; + memset(&address.fAddress, 0, sizeof(address.fAddress)); + + ConnectionListener* listener; + listener = new(std::nothrow) ConnectionListener(address); + if (listener == NULL) { + close(sock); + return B_NO_MEMORY; + } + + listener->fSocket = sock; + + *_listener = listener; + + return B_OK; +} + + +status_t +ConnectionListener::AcceptConnection(Connection** _connection) +{ + object_wait_info object[2]; + object[0].object = fWaitCancel; + object[0].type = B_OBJECT_TYPE_SEMAPHORE; + object[0].events = B_EVENT_ACQUIRE_SEMAPHORE; + + object[1].object = fSocket; + object[1].type = B_OBJECT_TYPE_FD; + object[1].events = B_EVENT_READ; + + do { + status_t result = wait_for_objects(object, 2); + if (result < B_OK || + (object[0].events & B_EVENT_ACQUIRE_SEMAPHORE) != 0) { + return ECONNABORTED; + } else if ((object[1].events & B_EVENT_READ) == 0) + continue; + break; + } while (true); + + sockaddr_storage addr; + socklen_t length = sizeof(addr); + int sock = accept(fSocket, reinterpret_cast(&addr), &length); + if (sock < 0) + return errno; + + ServerAddress address; + address.fProtocol = IPPROTO_TCP; + address.fAddress = addr; + Connection* connection; + + status_t result = Connection::SetTo(&connection, sock, address); + if (result != B_OK) { + close(sock); + return result; + } + + *_connection = connection; + return B_OK; +} + diff --git a/src/add-ons/kernel/file_systems/nfs4/Connection.h b/src/add-ons/kernel/file_systems/nfs4/Connection.h index e16e2cdd8d..7b21564ad6 100644 --- a/src/add-ons/kernel/file_systems/nfs4/Connection.h +++ b/src/add-ons/kernel/file_systems/nfs4/Connection.h @@ -31,31 +31,25 @@ struct ServerAddress { socklen_t AddressSize() const; + void SetPort(uint16 port); uint16 Port() const; + const void* InAddr() const; static status_t ResolveName(const char* name, ServerAddress* address); }; -class Connection { +class ConnectionBase { public: - static status_t Connect(Connection **connection, - const ServerAddress& address); - virtual ~Connection(); - - virtual status_t Send(const void* buffer, uint32 size) = 0; - virtual status_t Receive(void** buffer, uint32* size) = 0; + ConnectionBase(const ServerAddress& address); + virtual ~ConnectionBase(); status_t GetLocalAddress(ServerAddress* address); - status_t Reconnect(); void Disconnect(); protected: - Connection(const ServerAddress& address); - status_t Connect(); - sem_id fWaitCancel; int fSocket; mutex fSocketLock; @@ -63,6 +57,26 @@ protected: const ServerAddress fServerAddress; }; +class Connection : public ConnectionBase { +public: + static status_t Connect(Connection **connection, + const ServerAddress& address); + static status_t SetTo(Connection **connection, int socket, + const ServerAddress& address); + + virtual status_t Send(const void* buffer, uint32 size) = 0; + virtual status_t Receive(void** buffer, uint32* size) = 0; + + status_t Reconnect(); + +protected: + static Connection* CreateObject(const ServerAddress& address); + + Connection(const ServerAddress& address); + status_t Connect(); + +}; + class ConnectionStream : public Connection { public: ConnectionStream(const ServerAddress& address); @@ -79,5 +93,15 @@ public: virtual status_t Receive(void** buffer, uint32* size); }; +class ConnectionListener : public ConnectionBase { +public: + static status_t Listen(ConnectionListener** listener, uint16 port = 0); + + status_t AcceptConnection(Connection** connection); + +protected: + ConnectionListener(const ServerAddress& address); +}; + #endif // CONNECTION_H diff --git a/src/add-ons/kernel/file_systems/nfs4/Jamfile b/src/add-ons/kernel/file_systems/nfs4/Jamfile index 16a0306a88..abe0a46688 100644 --- a/src/add-ons/kernel/file_systems/nfs4/Jamfile +++ b/src/add-ons/kernel/file_systems/nfs4/Jamfile @@ -26,6 +26,9 @@ KernelAddon nfs4 : RootInode.cpp RPCAuth.cpp RPCCall.cpp + RPCCallback.cpp + RPCCallbackRequest.cpp + RPCCallbackServer.cpp RPCReply.cpp RPCServer.cpp XDR.cpp diff --git a/src/add-ons/kernel/file_systems/nfs4/NFS4Defs.h b/src/add-ons/kernel/file_systems/nfs4/NFS4Defs.h index e9775aaa8d..c0abd7b0fd 100644 --- a/src/add-ons/kernel/file_systems/nfs4/NFS4Defs.h +++ b/src/add-ons/kernel/file_systems/nfs4/NFS4Defs.h @@ -20,6 +20,11 @@ enum Procedure { ProcCompound = 1 }; +enum CallbackProcedure { + CallbackProcNull = 0, + CallbackProcCompound = 1 +}; + enum Opcode { OpAccess = 3, OpClose = 4, diff --git a/src/add-ons/kernel/file_systems/nfs4/RPCCallback.cpp b/src/add-ons/kernel/file_systems/nfs4/RPCCallback.cpp new file mode 100644 index 0000000000..70da61c58e --- /dev/null +++ b/src/add-ons/kernel/file_systems/nfs4/RPCCallback.cpp @@ -0,0 +1,24 @@ +/* + * Copyright 2012 Haiku, Inc. All rights reserved. + * Distributed under the terms of the MIT License. + * + * Authors: + * Paweł Dziepak, pdziepak@quarnos.org + */ + + +#include "RPCCallback.h" + +#include "RPCCallbackRequest.h" + + +using namespace RPC; + + +status_t +Callback::EnqueueRequest(CallbackRequest* request, Connection* connection) +{ + dprintf("GOT A CALLBACK REQUEST %x\n", (int)request->XID()); + return B_OK; +} + diff --git a/src/add-ons/kernel/file_systems/nfs4/RPCCallback.h b/src/add-ons/kernel/file_systems/nfs4/RPCCallback.h new file mode 100644 index 0000000000..66d60ac116 --- /dev/null +++ b/src/add-ons/kernel/file_systems/nfs4/RPCCallback.h @@ -0,0 +1,50 @@ +/* + * Copyright 2012 Haiku, Inc. All rights reserved. + * Distributed under the terms of the MIT License. + * + * Authors: + * Paweł Dziepak, pdziepak@quarnos.org + */ +#ifndef RPCCALLBACK_H +#define RPCCALLBACK_H + + +#include "Connection.h" + + +namespace RPC { + +class CallbackRequest; + +class Callback { +public: + inline void SetID(int32 id); + inline int32 ID(); + + status_t EnqueueRequest(CallbackRequest* request, + Connection* connection); + +private: + int32 fID; +}; + + +inline void +Callback::SetID(int32 id) +{ + fID = id; +} + + +inline int32 +Callback::ID() +{ + return fID; +} + + +} // namespace RPC + + +#endif // RPCCALLBACK_H + diff --git a/src/add-ons/kernel/file_systems/nfs4/RPCCallbackRequest.cpp b/src/add-ons/kernel/file_systems/nfs4/RPCCallbackRequest.cpp new file mode 100644 index 0000000000..39788b1ca0 --- /dev/null +++ b/src/add-ons/kernel/file_systems/nfs4/RPCCallbackRequest.cpp @@ -0,0 +1,71 @@ +/* + * Copyright 2012 Haiku, Inc. All rights reserved. + * Distributed under the terms of the MIT License. + * + * Authors: + * Paweł Dziepak, pdziepak@quarnos.org + */ + + +#include "RPCCallbackRequest.h" + +#include + +#include "NFS4Defs.h" + + +using namespace RPC; + +enum { + CALL = 0 +}; + +#define VERSION 2 + +enum { + PROGRAM_NFS_CB = 0x40000000 +}; + +#define NFS_VERSION 4 + + +CallbackRequest::CallbackRequest(void *buffer, int size) + : + fError(B_BAD_VALUE), + fStream(buffer, size), + fBuffer(buffer) +{ + fXID = fStream.GetUInt(); + + if (fStream.GetUInt() != CALL) + return; + + if (fStream.GetUInt() != VERSION) + return; + + if (fStream.GetUInt() != PROGRAM_NFS_CB) + return; + + fProcedure = fStream.GetUInt(); + + fStream.GetOpaque(NULL); + fStream.GetOpaque(NULL); + + if (fProcedure == CallbackProcCompound) { + fStream.GetOpaque(NULL); // TODO: tag may be important + if (fStream.GetUInt() != 0) + return; + + fID = fStream.GetUInt(); + } + + fError = B_OK; +} + + +CallbackRequest::~CallbackRequest() +{ + free(fBuffer); +} + + diff --git a/src/add-ons/kernel/file_systems/nfs4/RPCCallbackRequest.h b/src/add-ons/kernel/file_systems/nfs4/RPCCallbackRequest.h new file mode 100644 index 0000000000..28b822e234 --- /dev/null +++ b/src/add-ons/kernel/file_systems/nfs4/RPCCallbackRequest.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 RPCCALLBACKREQUEST_H +#define RPCCALLBACKREQUEST_H + + +#include "XDR.h" + + +namespace RPC { + +class CallbackRequest { +public: + CallbackRequest(void *buffer, int size); + ~CallbackRequest(); + + inline uint32 XID(); + inline uint32 ID(); + + inline uint32 Procedure(); + + inline status_t Error(); + inline XDR::ReadStream& Stream(); + +private: + uint32 fXID; + uint32 fID; + + uint32 fProcedure; + + status_t fError; + + XDR::ReadStream fStream; + void* fBuffer; +}; + + +inline uint32 +CallbackRequest::XID() +{ + return fXID; +} + + +inline uint32 +CallbackRequest::ID() +{ + return fID; +} + + +inline uint32 +CallbackRequest::Procedure() +{ + return fProcedure; +} + + +inline status_t +CallbackRequest::Error() +{ + return fError; +} + + +inline XDR::ReadStream& +CallbackRequest::Stream() +{ + return fStream; +} + + +} // namespace RPC + + +#endif // RPCCALLBACKREQUEST_H + diff --git a/src/add-ons/kernel/file_systems/nfs4/RPCCallbackServer.cpp b/src/add-ons/kernel/file_systems/nfs4/RPCCallbackServer.cpp new file mode 100644 index 0000000000..837557e947 --- /dev/null +++ b/src/add-ons/kernel/file_systems/nfs4/RPCCallbackServer.cpp @@ -0,0 +1,285 @@ +/* + * Copyright 2012 Haiku, Inc. All rights reserved. + * Distributed under the terms of the MIT License. + * + * Authors: + * Paweł Dziepak, pdziepak@quarnos.org + */ + + +#include "RPCCallbackServer.h" + +#include "NFS4Defs.h" +#include "RPCCallback.h" +#include "RPCCallbackRequest.h" + + +using namespace RPC; + + +CallbackServer* gRPCCallbackServer = NULL; + + +CallbackServer::CallbackServer() + : + fConnectionList(NULL), + fListener(NULL), + fThreadRunning(false), + fCallbackArray(NULL), + fArraySize(0), + fFreeSlot(-1) +{ + mutex_init(&fConnectionLock, NULL); + mutex_init(&fThreadLock, NULL); + rw_lock_init(&fArrayLock, NULL); +} + + +CallbackServer::~CallbackServer() +{ + StopServer(); + + free(fCallbackArray); + rw_lock_destroy(&fArrayLock); + mutex_destroy(&fThreadLock); + mutex_destroy(&fConnectionLock); +} + + +status_t +CallbackServer::RegisterCallback(Callback* callback) +{ + status_t result = StartServer(); + if (result != B_OK) + return result; + + WriteLocker _(fArrayLock); + if (fFreeSlot == -1) { + uint32 newSize = max_c(fArraySize * 2, 4); + uint32 size = newSize * sizeof(CallbackSlot); + CallbackSlot* array = reinterpret_cast(malloc(size)); + if (array == NULL) + return B_NO_MEMORY; + + if (fCallbackArray != NULL) + memcpy(array, fCallbackArray, fArraySize * sizeof(CallbackSlot)); + + for (uint32 i = fArraySize; i < newSize; i++) + array[i].fNext = i + 1; + + array[fArraySize * 2 - 1].fNext = -1; + + fCallbackArray = array; + fFreeSlot = fArraySize; + fArraySize = newSize; + } + + int32 id = fFreeSlot; + fFreeSlot = fCallbackArray[id].fNext; + + fCallbackArray[id].fCallback = callback; + callback->SetID(id); + + return B_OK; +} + + +status_t +CallbackServer::UnregisterCallback(Callback* callback) +{ + int32 id = callback->ID(); + + WriteLocker _(fArrayLock); + fCallbackArray[id].fNext = fFreeSlot; + fFreeSlot = id; + + return B_OK; +} + + +status_t +CallbackServer::StartServer() +{ + MutexLocker _(fThreadLock); + if (fThreadRunning) + return B_OK; + + status_t result = ConnectionListener::Listen(&fListener); + if (result != B_OK) + return result; + + fThread = spawn_kernel_thread(&CallbackServer::ListenerThreadLauncher, + "NFSv4 Callback Listener", B_NORMAL_PRIORITY, this); + if (fThread < B_OK) + return fThread; + + fThreadRunning = true; + + result = resume_thread(fThread); + if (result != B_OK) { + kill_thread(fThread); + fThreadRunning = false; + return result; + } + + return B_OK; +} + + +status_t +CallbackServer::StopServer() +{ + MutexLocker _(&fThreadLock); + if (!fThreadRunning) + return B_OK; + + fListener->Disconnect(); + status_t result; + wait_for_thread(fThread, &result); + + MutexLocker locker(fConnectionLock); + while (fConnectionList != NULL) { + ConnectionEntry* entry = fConnectionList; + fConnectionList = entry->fNext; + entry->fConnection->Disconnect(); + delete entry->fConnection; + delete entry; + } + + delete fListener; + + fThreadRunning = false; + return B_OK; +} + + +status_t +CallbackServer::NewConnection(Connection* connection) +{ + ConnectionEntry* entry = new ConnectionEntry; + entry->fConnection = connection; + entry->fPrev = NULL; + + MutexLocker locker(fConnectionLock); + entry->fNext = fConnectionList; + fConnectionList = entry; + locker.Unlock(); + + void** arguments = reinterpret_cast(malloc(sizeof(void*) * 2)); + if (arguments == NULL) + return B_NO_MEMORY; + + arguments[0] = this; + arguments[1] = connection; + + thread_id thread; + thread = spawn_kernel_thread(&CallbackServer::ConnectionThreadLauncher, + "NFSv4 Callback Connection", B_NORMAL_PRIORITY, arguments); + if (thread < B_OK) { + free(arguments); + return thread; + } + + status_t result = resume_thread(thread); + if (result != B_OK) { + kill_thread(thread); + free(arguments); + return result; + } + + return B_OK; +} + + +status_t +CallbackServer::ReleaseConnection(ConnectionEntry* entry) +{ + MutexLocker _(fConnectionLock); + if (entry->fNext != NULL) + entry->fNext->fPrev = entry->fPrev; + if (entry->fPrev != NULL) + entry->fPrev->fNext = entry->fNext; + else + fConnectionList = entry->fNext; + + delete entry->fConnection; + delete entry; + return B_OK; +} + + +status_t +CallbackServer::ConnectionThreadLauncher(void* object) +{ + void** objects = reinterpret_cast(object); + CallbackServer* server = reinterpret_cast(objects[0]); + ConnectionEntry* entry = reinterpret_cast(objects[1]); + free(objects); + + return server->ConnectionThread(entry); +} + + +status_t +CallbackServer::ConnectionThread(ConnectionEntry* entry) +{ + Connection* connection = entry->fConnection; + dprintf("NEW CONNECTION\n"); + while (fThreadRunning) { + uint32 size; + void* buffer; + status_t result = connection->Receive(&buffer, &size); + if (result != B_OK) { + ReleaseConnection(entry); + return result; + } + + CallbackRequest* request = new CallbackRequest(buffer, size); + if (request == NULL || request->Error() != B_OK) { + free(buffer); + continue; + } + + switch (request->Procedure()) { + case CallbackProcCompound: + GetCallback(request->ID())->EnqueueRequest(request, connection); + break; + + case CallbackProcNull: + dprintf("GOT CB_NULL %x\n", (int)request->XID()); + + default: + free(buffer); + } + } + + return B_OK; +} + + +status_t +CallbackServer::ListenerThreadLauncher(void* object) +{ + CallbackServer* server = reinterpret_cast(object); + return server->ListenerThread(); +} + + +status_t +CallbackServer::ListenerThread() +{ + while (fThreadRunning) { + Connection* connection; + status_t result = fListener->AcceptConnection(&connection); + if (result != B_OK) { + fThreadRunning = false; + return result; + } + result = NewConnection(connection); + if (result != B_OK) + delete connection; + } + + return B_OK; +} + diff --git a/src/add-ons/kernel/file_systems/nfs4/RPCCallbackServer.h b/src/add-ons/kernel/file_systems/nfs4/RPCCallbackServer.h new file mode 100644 index 0000000000..9b22f20e05 --- /dev/null +++ b/src/add-ons/kernel/file_systems/nfs4/RPCCallbackServer.h @@ -0,0 +1,99 @@ +/* + * Copyright 2012 Haiku, Inc. All rights reserved. + * Distributed under the terms of the MIT License. + * + * Authors: + * Paweł Dziepak, pdziepak@quarnos.org + */ +#ifndef RPCCALLBACKSERVER_H +#define RPCCALLBACKSERVER_H + + +#include + +#include "Connection.h" + + +namespace RPC { + +class Callback; + +struct ConnectionEntry { + Connection* fConnection; + ConnectionEntry* fNext; + ConnectionEntry* fPrev; +}; + +union CallbackSlot { + Callback* fCallback; + int32 fNext; +}; + +class CallbackServer { +public: + CallbackServer(); + ~CallbackServer(); + + status_t RegisterCallback(Callback* callback); + status_t UnregisterCallback(Callback* callback); + + inline ServerAddress LocalID(); + +protected: + status_t StartServer(); + status_t StopServer(); + + status_t NewConnection(Connection* connection); + status_t ReleaseConnection(ConnectionEntry* entry); + + static status_t ListenerThreadLauncher(void* object); + status_t ListenerThread(); + + static status_t ConnectionThreadLauncher(void* object); + status_t ConnectionThread(ConnectionEntry* entry); + + inline Callback* GetCallback(int32 id); + +private: + mutex fConnectionLock; + ConnectionEntry* fConnectionList; + ConnectionListener* fListener; + + mutex fThreadLock; + thread_id fThread; + bool fThreadRunning; + + rw_lock fArrayLock; + CallbackSlot* fCallbackArray; + uint32 fArraySize; + int32 fFreeSlot; +}; + + +inline ServerAddress +CallbackServer::LocalID() +{ + ServerAddress address; + fListener->GetLocalAddress(&address); + return address; +} + + +inline Callback* +CallbackServer::GetCallback(int32 id) +{ + ReadLocker _(fArrayLock); + if (id >= 0 && static_cast(id) < fArraySize) + return fCallbackArray[id].fCallback; + return NULL; +} + + +} // namespace RPC + + +extern RPC::CallbackServer* gRPCCallbackServer; + + +#endif // RPCCALLBACKSERVER_H + diff --git a/src/add-ons/kernel/file_systems/nfs4/RPCServer.cpp b/src/add-ons/kernel/file_systems/nfs4/RPCServer.cpp index e13a36e276..fce6f5aa3b 100644 --- a/src/add-ons/kernel/file_systems/nfs4/RPCServer.cpp +++ b/src/add-ons/kernel/file_systems/nfs4/RPCServer.cpp @@ -13,6 +13,7 @@ #include +#include "RPCCallbackServer.h" #include "RPCReply.h" @@ -77,14 +78,22 @@ Server::Server(Connection* connection, ServerAddress* address) fConnection(connection), fAddress(address), fPrivateData(NULL), + fCallback(NULL), fXID(rand() << 1) { + mutex_init(&fCallbackLock, NULL); + _StartListening(); } Server::~Server() { + if (fCallback != NULL) + gRPCCallbackServer->UnregisterCallback(fCallback); + delete fCallback; + mutex_destroy(&fCallbackLock); + delete fPrivateData; fThreadCancel = true; @@ -214,6 +223,19 @@ Server::Repair() } +Callback* +Server::GetCallback() +{ + MutexLocker _(fCallbackLock); + if (fCallback == NULL) { + fCallback = new Callback; + gRPCCallbackServer->RegisterCallback(fCallback); + } + + return fCallback; +} + + uint32 Server::_GetXID() { diff --git a/src/add-ons/kernel/file_systems/nfs4/RPCServer.h b/src/add-ons/kernel/file_systems/nfs4/RPCServer.h index 3a954b35cd..ddcb80e423 100644 --- a/src/add-ons/kernel/file_systems/nfs4/RPCServer.h +++ b/src/add-ons/kernel/file_systems/nfs4/RPCServer.h @@ -14,6 +14,7 @@ #include "Connection.h" #include "RPCCall.h" +#include "RPCCallback.h" #include "RPCReply.h" @@ -76,6 +77,8 @@ public: inline ProgramData* PrivateData(); inline void SetPrivateData(ProgramData* privateData); + Callback* GetCallback(); + private: inline uint32 _GetXID(); @@ -94,6 +97,9 @@ private: ProgramData* fPrivateData; + mutex fCallbackLock; + Callback* fCallback; + vint32 fXID; static const bigtime_t kWaitTime = 1000000; }; diff --git a/src/add-ons/kernel/file_systems/nfs4/RequestBuilder.cpp b/src/add-ons/kernel/file_systems/nfs4/RequestBuilder.cpp index 9c44aa2815..c9d2408503 100644 --- a/src/add-ons/kernel/file_systems/nfs4/RequestBuilder.cpp +++ b/src/add-ons/kernel/file_systems/nfs4/RequestBuilder.cpp @@ -13,6 +13,8 @@ #include #include "Cookie.h" +#include "RPCCallback.h" +#include "RPCCallbackServer.h" RequestBuilder::RequestBuilder(Procedure proc) @@ -620,7 +622,7 @@ RequestBuilder::SetAttr(const uint32* id, uint32 stateSeq, AttrValue* attr, status_t -RequestBuilder::SetClientID(const RPC::Server* server) +RequestBuilder::SetClientID(RPC::Server* server) { if (fProcedure != ProcCompound) return B_BAD_VALUE; @@ -638,15 +640,21 @@ RequestBuilder::SetClientID(const RPC::Server* server) fRequest->Stream().AddUInt(0x40000000); - ServerAddress local = server->LocalID(); + uint32 id = server->GetCallback()->ID(); + + ServerAddress local = gRPCCallbackServer->LocalID(); + ServerAddress servAddr = server->LocalID(); + servAddr.SetPort(local.Port()); + fRequest->Stream().AddString(local.ProtocolString()); - char* uAddr = local.UniversalAddress(); + + char* uAddr = servAddr.UniversalAddress(); if (uAddr == NULL) return B_NO_MEMORY; fRequest->Stream().AddString(uAddr); free(uAddr); - fRequest->Stream().AddUInt(0); + fRequest->Stream().AddUInt(id); fOpCount++; diff --git a/src/add-ons/kernel/file_systems/nfs4/RequestBuilder.h b/src/add-ons/kernel/file_systems/nfs4/RequestBuilder.h index b3b7998955..a8f2d7618d 100644 --- a/src/add-ons/kernel/file_systems/nfs4/RequestBuilder.h +++ b/src/add-ons/kernel/file_systems/nfs4/RequestBuilder.h @@ -69,7 +69,7 @@ public: status_t SaveFH(); status_t SetAttr(const uint32* id, uint32 stateSeq, AttrValue* attr, uint32 count); - status_t SetClientID(const RPC::Server* server); + status_t SetClientID(RPC::Server* server); status_t SetClientIDConfirm(uint64 id, uint64 ver); status_t Verify(AttrValue* attr, uint32 count); status_t Write(const uint32* id, uint32 stateSeq, 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 cc58c5593b..95b8bd87eb 100644 --- a/src/add-ons/kernel/file_systems/nfs4/kernel_interface.cpp +++ b/src/add-ons/kernel/file_systems/nfs4/kernel_interface.cpp @@ -20,6 +20,7 @@ #include "RequestBuilder.h" #include "ReplyInterpreter.h" #include "RootInode.h" +#include "RPCCallbackServer.h" #include "RPCServer.h" @@ -637,6 +638,13 @@ nfs4_init() return B_NO_MEMORY; } + gRPCCallbackServer = new(std::nothrow) RPC::CallbackServer; + if (gRPCCallbackServer == NULL) { + delete gRPCServerManager; + delete gIdMapper; + return B_NO_MEMORY; + } + return B_OK; } @@ -646,6 +654,7 @@ nfs4_uninit() { dprintf("NFS4 Uninit\n"); + delete gRPCCallbackServer; delete gIdMapper; delete gRPCServerManager;