Mirrors/VILLASnode
1
0
Fork 0
mirror of https://git.rwth-aachen.de/acs/public/villas/node/ synced 2025-03-09 00:00:00 +01:00
Code Issues Releases Wiki Activity

api: improve handling of socket descriptors

Browse source
This commit is contained in:
Steffen Vogel 2018-12-02 03:21:27 +01:00
parent eab8f1cad7
commit b882617243
3 changed files with 41 additions and 26 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

3
include/villas/api/server.hpp
View file

@ -25,6 +25,7 @@
#include <string>
#include <vector>
#include <map>
#include <poll.h>
@ -54,7 +55,7 @@ protected:
int sd;
std::vector<pollfd> pfds;
std::vector<sessions::Socket *> sessions;
std::map<int, sessions::Socket *> sessions;
void acceptNewSession();
void closeSession(sessions::Socket *s);

2
include/villas/api/sessions/socket.hpp
View file

@ -47,6 +47,8 @@ public:
int write();
virtual std::string getName();
int getSd() const { return sd; }
};
} // sessions

62
lib/api/server.cpp
View file

@ -36,8 +36,11 @@
#include <villas/api/server.hpp>
#include <villas/api/sessions/socket.hpp>
using namespace villas;
using namespace villas::node::api;
static Logger logger = logging.get("api");
Server::Server(Api *a) :
state(STATE_INITIALIZED),
api(a)
@ -66,7 +69,6 @@ void Server::start()
};
pfds.push_back(pfd);
sessions.push_back(nullptr);
struct sockaddr_un sun = { .sun_family = AF_UNIX };
@ -127,30 +129,40 @@ void Server::run(int timeout)
assert(state == STATE_STARTED);
ret = poll(pfds.data(), pfds.size(), timeout);
logger->info("pfds.size() = {}", pfds.size());
auto len = pfds.size();
ret = poll(pfds.data(), len, timeout);
if (ret < 0)
throw SystemError("Failed to poll on API socket");;
throw SystemError("Failed to poll on API socket");
for (unsigned i = 0; i < pfds.size(); i++) {
std::vector<sessions::Socket *> closing;
for (unsigned i = 1; i < len; i++) {
auto &pfd = pfds[i];
auto s = sessions[i];
if (pfd.revents & POLLOUT) {
if (s)
s->write();
}
/* pfds[0] is the server socket */
auto s = sessions[pfd.fd];
if (pfd.revents & POLLIN) {
/* New connection */
if (s) {
ret = s->read();
if (ret < 0)
closeSession(s);
}
else
acceptNewSession();
ret = s->read();
if (ret < 0)
closing.push_back(s);
}
if (pfd.revents & POLLOUT) {
s->write();
}
}
/* Destroy closed sessions */
for (auto *s : closing)
closeSession(s);
/* Accept new connections */
if (pfds[0].revents & POLLIN)
acceptNewSession();
}
void Server::acceptNewSession() {
@ -164,21 +176,21 @@ void Server::acceptNewSession() {
};
pfds.push_back(pfd);
sessions.push_back(s);
sessions[fd] = s;
api->sessions.push_back(s);
}
void Server::closeSession(sessions::Socket *s)
{
int sd = s->getSd();
sessions.erase(sd);
api->sessions.remove(s);
ptrdiff_t pos = std::find(sessions.begin(), sessions.end(), s) - sessions.begin();
pfds.erase(std::remove_if(pfds.begin(), pfds.end(),
[sd](const pollfd &p){ return p.fd == sd; })
);
if (pos < (ptrdiff_t) sessions.size()) {
pfds.erase(pfds.begin() + pos);
sessions.erase(sessions.begin() + pos);
delete s;
}
delete s;
}