#include "RPCCallbackServer.h"
#include "NFS4Defs.h"
#include "RPCCallback.h"
#include "RPCCallbackReply.h"
#include "RPCCallbackRequest.h"
#include "RPCServer.h"
using namespace RPC;
CallbackServer* gRPCCallbackServer = NULL;
CallbackServer* gRPCCallbackServer6 = NULL;
CallbackServer::CallbackServer(int networkFamily)
:
fConnectionList(NULL),
fListener(NULL),
fThreadRunning(false),
fCallbackArray(NULL),
fArraySize(0),
fFreeSlot(-1),
fNetworkFamily(networkFamily)
{
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);
}
CallbackServer*
CallbackServer::Get(Server* server)
{
ASSERT(server != NULL);
int family = server->ID().Family();
ASSERT(family == AF_INET || family == AF_INET6);
int idx;
switch (family) {
case AF_INET:
idx = 0;
break;
case AF_INET6:
idx = 1;
break;
default:
return NULL;
}
MutexLocker _(fServerCreationLock);
if (fServers[idx] == NULL)
fServers[idx] = new CallbackServer(family);
return fServers[idx];
}
void
CallbackServer::ShutdownAll()
{
MutexLocker _(fServerCreationLock);
for (unsigned int i = 0; i < sizeof(fServers) / sizeof(fServers[0]); i++)
delete fServers[i];
memset(&fServers, 0, sizeof(fServers));
}
mutex CallbackServer::fServerCreationLock = MUTEX_INITIALIZER(NULL);
CallbackServer* CallbackServer::fServers[2] = { NULL, NULL };
status_t
CallbackServer::RegisterCallback(Callback* callback)
{
ASSERT(callback != NULL);
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[newSize - 1].fNext = -1;
fCallbackArray = array;
fFreeSlot = fArraySize;
fArraySize = newSize;
}
int32 id = fFreeSlot;
fFreeSlot = fCallbackArray[id].fNext;
fCallbackArray[id].fCallback = callback;
callback->SetID(id);
callback->SetCBServer(this);
return B_OK;
}
status_t
CallbackServer::UnregisterCallback(Callback* callback)
{
ASSERT(callback != NULL);
ASSERT(callback->CBServer() == this);
int32 id = callback->ID();
WriteLocker _(fArrayLock);
fCallbackArray[id].fNext = fFreeSlot;
fFreeSlot = id;
callback->SetCBServer(NULL);
return B_OK;
}
status_t
CallbackServer::StartServer()
{
MutexLocker _(fThreadLock);
if (fThreadRunning)
return B_OK;
status_t result = ConnectionListener::Listen(&fListener, fNetworkFamily);
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();
status_t result;
wait_for_thread(entry->fThread, &result);
delete entry->fConnection;
delete entry;
}
delete fListener;
fThreadRunning = false;
return B_OK;
}
status_t
CallbackServer::NewConnection(Connection* connection)
{
ASSERT(connection != NULL);
ConnectionEntry* entry = new ConnectionEntry;
entry->fConnection = connection;
entry->fPrev = NULL;
MutexLocker locker(fConnectionLock);
entry->fNext = fConnectionList;
if (fConnectionList != NULL)
fConnectionList->fPrev = entry;
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] = entry;
thread_id thread;
thread = spawn_kernel_thread(&CallbackServer::ConnectionThreadLauncher,
"NFSv4 Callback Connection", B_NORMAL_PRIORITY, arguments);
if (thread < B_OK) {
ReleaseConnection(entry);
free(arguments);
return thread;
}
entry->fThread = thread;
status_t result = resume_thread(thread);
if (result != B_OK) {
kill_thread(thread);
ReleaseConnection(entry);
free(arguments);
return result;
}
return B_OK;
}
status_t
CallbackServer::ReleaseConnection(ConnectionEntry* entry)
{
ASSERT(entry != NULL);
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)
{
ASSERT(object != NULL);
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)
{
ASSERT(entry != NULL);
Connection* connection = entry->fConnection;
CallbackReply* reply;
while (fThreadRunning) {
uint32 size;
void* buffer;
status_t result = connection->Receive(&buffer, &size);
if (result != B_OK) {
if (result != ECONNABORTED)
ReleaseConnection(entry);
return result;
}
CallbackRequest* request
= new(std::nothrow) CallbackRequest(buffer, size);
if (request == NULL) {
free(buffer);
continue;
} else if (request->Error() != B_OK) {
reply = CallbackReply::Create(request->XID(), request->RPCError());
if (reply != NULL) {
connection->Send(reply->Stream().Buffer(),
reply->Stream().Size());
delete reply;
}
delete request;
continue;
}
switch (request->Procedure()) {
case CallbackProcCompound:
GetCallback(request->ID())->EnqueueRequest(request, connection);
break;
case CallbackProcNull:
reply = CallbackReply::Create(request->XID());
if (reply != NULL) {
connection->Send(reply->Stream().Buffer(),
reply->Stream().Size());
delete reply;
}
default:
delete request;
}
}
return B_OK;
}
status_t
CallbackServer::ListenerThreadLauncher(void* object)
{
ASSERT(object != NULL);
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;
}
