nfs4: Add basic RPC level support for callbacks

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<sockaddr*>(&fAddress)->sa_family) {
+		case AF_INET:
+			reinterpret_cast<sockaddr_in*>(&fAddress)->sin_port = port;
+			break;
+		case AF_INET6:
+			reinterpret_cast<sockaddr_in6*>(&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<sockaddr*>(&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;
+}
+

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
 

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

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,

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;
+}
+

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
+

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 <stdlib.h>
+
+#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);
+}
+
+

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
+

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<CallbackSlot*>(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<void**>(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<void**>(object);
+	CallbackServer* server = reinterpret_cast<CallbackServer*>(objects[0]);
+	ConnectionEntry* entry = reinterpret_cast<ConnectionEntry*>(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<CallbackServer*>(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;
+}
+

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 <util/AutoLock.h>
+
+#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<uint32>(id) < fArraySize)
+		return fCallbackArray[id].fCallback;
+	return NULL;
+}
+
+
+}		// namespace RPC
+
+
+extern RPC::CallbackServer* gRPCCallbackServer;
+
+
+#endif	// RPCCALLBACKSERVER_H
+

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

@@ -13,6 +13,7 @@
 
 #include <util/AutoLock.h>
 
+#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()
 {

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;
 };

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

@@ -13,6 +13,8 @@
 #include <string.h>
 
 #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++;
 

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,

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;