nfs4: Add RPC::Server release and repair code

This commit is contained in:
Pawel Dziepak
2012-06-29 02:14:12 +02:00
parent b35311f1ba
commit 5cdf02db09
2 changed files with 116 additions and 37 deletions
@@ -60,7 +60,6 @@ RequestManager::FindRequest(uint32 xid)
if (fQueueHead == req) if (fQueueHead == req)
fQueueHead = req->fNext; fQueueHead = req->fNext;
mutex_unlock(&fLock); mutex_unlock(&fLock);
dprintf("Found %x %x\n", (unsigned int)fQueueHead, (unsigned int)fQueueTail);
return req; return req;
} }
@@ -68,16 +67,16 @@ RequestManager::FindRequest(uint32 xid)
prev = req; prev = req;
req = req->fNext; req = req->fNext;
} }
dprintf("Nothing found\n");
mutex_unlock(&fLock); mutex_unlock(&fLock);
return NULL; return NULL;
} }
Server::Server(Connection* conn) Server::Server(Connection* conn, ServerAddress* addr)
: :
fConnection(conn), fConnection(conn),
fAddress(addr),
fXID(rand() << 1) fXID(rand() << 1)
{ {
_StartListening(); _StartListening();
@@ -98,8 +97,8 @@ Server::~Server()
status_t status_t
Server::_StartListening() Server::_StartListening()
{ {
dprintf("new thread\n");
fThreadCancel = false; fThreadCancel = false;
fThreadError = B_OK;
fThread = spawn_kernel_thread(&Server::_ListenerThreadStart, fThread = spawn_kernel_thread(&Server::_ListenerThreadStart,
"NFSv4 Listener", B_NORMAL_PRIORITY, this); "NFSv4 Listener", B_NORMAL_PRIORITY, this);
if (fThread < B_OK) if (fThread < B_OK)
@@ -120,6 +119,9 @@ Server::SendCall(Call* call, Reply** reply)
{ {
status_t result; status_t result;
if (fThreadError != B_OK)
return fThreadError;
Request* req = new(std::nothrow) Request; Request* req = new(std::nothrow) Request;
if (req == NULL) if (req == NULL)
return B_NO_MEMORY; return B_NO_MEMORY;
@@ -145,16 +147,21 @@ Server::SendCall(Call* call, Reply** reply)
return B_OK; return B_OK;
out_cancel: out_cancel:
CancelCall(xid); fRequests.FindRequest(xid);
delete req; delete req;
return result; return result;
} }
void status_t
Server::CancelCall(uint32 xid) Server::Repair()
{ {
fRequests.FindRequest(xid); fThreadCancel = true;
status_t result = fConnection->Reconnect();
if (result != B_OK)
return result;
wait_for_thread(fThread, &result);
return _StartListening();
} }
@@ -174,12 +181,15 @@ Server::_Listener()
while (!fThreadCancel) { while (!fThreadCancel) {
result = fConnection->Receive(&buffer, &size); result = fConnection->Receive(&buffer, &size);
if (result != B_OK) if (result != B_OK) {
fThreadError = result;
return result; return result;
}
Reply* reply = new(std::nothrow) Reply(buffer, size); Reply* reply = new(std::nothrow) Reply(buffer, size);
if (reply == NULL) { if (reply == NULL) {
free(buffer); free(buffer);
fThreadError = result;
return B_NO_MEMORY; return B_NO_MEMORY;
} }
@@ -212,6 +222,7 @@ ServerManager::ServerManager()
ServerManager::~ServerManager() ServerManager::~ServerManager()
{ {
mutex_destroy(&fLock);
} }
@@ -235,7 +246,6 @@ ServerManager::Acquire(Server** pserv, uint32 ip, uint16 port, Transport proto)
return B_OK; return B_OK;
} }
mutex_unlock(&fLock); mutex_unlock(&fLock);
dprintf("creating\n");
node = new(std::nothrow) ServerNode; node = new(std::nothrow) ServerNode;
if (node == NULL) if (node == NULL)
@@ -250,7 +260,7 @@ ServerManager::Acquire(Server** pserv, uint32 ip, uint16 port, Transport proto)
return result; return result;
} }
node->fServer = new Server(conn); node->fServer = new Server(conn, &node->fID);
if (node->fServer == NULL) { if (node->fServer == NULL) {
delete node; delete node;
delete conn; delete conn;
@@ -283,9 +293,8 @@ ServerManager::Acquire(Server** pserv, uint32 ip, uint16 port, Transport proto)
void void
ServerManager::Release(Server* serv) ServerManager::Release(Server* serv)
{ {
#if 0
mutex_lock(&fLock); mutex_lock(&fLock);
ServerNode* node = _Find(serv->GetID()); ServerNode* node = _Find(serv->ID());
if (node != NULL) { if (node != NULL) {
node->fRefCount--; node->fRefCount--;
@@ -298,16 +307,14 @@ ServerManager::Release(Server* serv)
delete node->fServer; delete node->fServer;
delete node; delete node;
} }
#endif
} }
ServerNode* ServerNode*
ServerManager::_Find(ServerAddress& id) ServerManager::_Find(const ServerAddress& id)
{ {
ServerNode* node = fRoot; ServerNode* node = fRoot;
while (node != NULL) { while (node != NULL) {
dprintf("passing addr: %x port %d proto: %d\n", (int)node->fID.fAddress, (int)node->fID.fPort, (int)node->fID.fProtocol);
if (node->fID == id) if (node->fID == id)
return node; return node;
if (node->fID < id) if (node->fID < id)
@@ -320,6 +327,67 @@ ServerManager::_Find(ServerAddress& id)
} }
void
ServerManager::_Delete(ServerNode* node)
{
bool found = false;
ServerNode* previous = NULL;
ServerNode* current = fRoot;
while (current != NULL) {
if (current->fID == node->fID) {
found = true;
break;
}
if (current->fID < node->fID) {
previous = current;
current = current->fRight;
} else {
previous = current;
current = current->fLeft;
}
}
if (!found)
return;
if (previous == NULL)
fRoot = NULL;
else if (current->fLeft == NULL && current->fRight == NULL) {
if (previous->fID < node->fID)
previous->fRight = NULL;
else
previous->fLeft = NULL;
} else if (current->fLeft != NULL && current->fRight == NULL) {
if (previous->fID < node->fID)
previous->fRight = current->fLeft;
else
previous->fLeft = current->fLeft;
} else if (current->fLeft == NULL && current->fRight != NULL) {
if (previous->fID < node->fID)
previous->fRight = current->fRight;
else
previous->fLeft = current->fRight;
} else {
ServerNode* left_prev = current;
ServerNode* left = current->fLeft;
while (left->fLeft != NULL) {
left_prev = left;
left = left->fLeft;
}
if (previous->fID < node->fID)
previous->fRight = left;
else
previous->fLeft = left;
left_prev->fLeft = NULL;
}
}
ServerNode* ServerNode*
ServerManager::_Insert(ServerNode* node) ServerManager::_Insert(ServerNode* node)
{ {
@@ -44,30 +44,42 @@ private:
class Server { class Server {
public: public:
Server(Connection* conn); Server(Connection* conn,
ServerAddress* addr);
virtual ~Server(); virtual ~Server();
status_t SendCall(Call* call, Reply** reply); status_t SendCall(Call* call, Reply** reply);
inline void CancelCall(uint32 xid);
protected: status_t Repair();
inline uint32 _GetXID();
status_t _StartListening(); inline const ServerAddress& ID() const;
private: private:
inline uint32 _GetXID();
status_t _StartListening();
status_t _Listener(); status_t _Listener();
static status_t _ListenerThreadStart(void* ptr); static status_t _ListenerThreadStart(void* ptr);
thread_id fThread; thread_id fThread;
bool fThreadCancel; bool fThreadCancel;
status_t fThreadError;
RequestManager fRequests; RequestManager fRequests;
Connection* fConnection; Connection* fConnection;
const ServerAddress* fAddress;
vint32 fXID; vint32 fXID;
static const bigtime_t kWaitTime = 1000000; static const bigtime_t kWaitTime = 1000000;
}; };
inline const ServerAddress&
Server::ID() const
{
return *fAddress;
}
struct ServerNode { struct ServerNode {
ServerAddress fID; ServerAddress fID;
Server* fServer; Server* fServer;
@@ -88,8 +100,7 @@ public:
private: private:
ServerNode* _Find(const ServerAddress& id);
ServerNode* _Find(ServerAddress& id);
void _Delete(ServerNode* node); void _Delete(ServerNode* node);
ServerNode* _Insert(ServerNode* node); ServerNode* _Insert(ServerNode* node);