Fixed multithread sync issue and duplicate thread on termination.

This commit is contained in:
Brad Arant committed 2023-03-11 21:55:33 -08:00
1 parent b0118cfc56
commit cfa70c5863
72 files changed
+395 -427

No files matched your search

+52 -18
View File
@@ -18,7 +18,7 @@ namespace core
Socket::~Socket() {
shutDown = true;
onUnregister();
ePoll.unregisterSocket(this);
disableSocket();
coreutils::Log(coreutils::LOG_DEBUG_4) << "Free on socket " << descriptor;
free(buffer);
if(descriptor == -1)
@@ -39,7 +39,7 @@ namespace core
throw coreutils::Exception("Descriptor out of range", __FILE__, __LINE__);
this->descriptor = descriptor;
onRegister();
ePoll.registerSocket(this);
enableSocket();
onRegistered();
}
@@ -68,29 +68,39 @@ namespace core
void Socket::onUnregistered() {}
bool Socket::eventReceived(struct epoll_event event, long long eventId) {
// lock.lock();
// coreutils::Log(coreutils::LOG_DEBUG_1) << "Event process beginning for socket " << getDescriptor();
if(inHandler)
// coreutils::Log(coreutils::LOG_DEBUG_2) << "inHandler was already true.";
inHandler = true;
if(event.events & EPOLLRDHUP) {
// coreutils::Log(coreutils::LOG_DEBUG_2) << "EPOLLRDHUP";
readHangup = true;
shutdown("hangup received");
}
if(event.events & EPOLLIN) {
// coreutils::Log(coreutils::LOG_DEBUG_2) << "EPOLLIN";
coreutils::ZString zbuffer(buffer, length);
lock.lock();
receiveData(zbuffer);
if(!shutDown) {
inHandler = false;
lock.unlock();
resetSocket();
}
}
if(event.events & EPOLLWRNORM) {
writeSocket();
// coreutils::Log(coreutils::LOG_DEBUG_2) << "EPOLLWRNORM";
writeSocket();
inHandler = false;
resetSocket();
}
if(event.events & EPOLLHUP) {
shutdown();
}
inHandler = false;
// lock.unlock();
inHandler = false;
// coreutils::Log(coreutils::LOG_DEBUG_1) << "Event process ending for socket " << getDescriptor();
return !shutDown;
}
void Socket::onDataReceived(std::string data)
{
{
throw coreutils::Exception("Need to override onDataReceived.", __FILE__, __LINE__, -1);
}
@@ -107,7 +117,7 @@ namespace core
int error = -1;
if((len = ::read(getDescriptor(), buffer.getData(), buffer.getLength())) >= 0) {
coreutils::ZString zbuffer(buffer.getData(), len);
coreutils::Log(coreutils::LOG_DEBUG_1) << zbuffer;
// coreutils::Log(coreutils::LOG_DEBUG_1) << zbuffer;
onDataReceived(zbuffer);
}
else
@@ -135,21 +145,23 @@ namespace core
}
void Socket::writeSocket() {
outlock.lock();
// coreutils::Log(coreutils::LOG_DEBUG_3) << "writing data to socket " << getDescriptor();
if(fifo.size() > 0) {
outlock.lock();
if(!shutDown)
int rc = ::write(descriptor, fifo.front().c_str(), fifo.front().length());
if(!shutDown)
int rc = ::write(descriptor, fifo.front().c_str(), fifo.front().length());
fifo.pop();
outlock.unlock();
}
outlock.unlock();
}
int Socket::write(std::string data) {
outlock.lock();
fifo.emplace(data);
outlock.unlock();
if(!inHandler) {
ePoll.resetSocket(this);
if(lock.try_lock()) {
resetSocket();
lock.unlock();
}
return 1;
}
@@ -170,4 +182,26 @@ namespace core
reset = false;
}
void Socket::enableSocket() {
struct epoll_event event;
event.data.ptr = this;
event.events = EPOLLIN | EPOLLRDHUP | EPOLLONESHOT | EPOLLET;
epoll_ctl(ePoll.getDescriptor(), EPOLL_CTL_ADD, getDescriptor(), &event);
}
void Socket::disableSocket() {
epoll_ctl(ePoll.getDescriptor(), EPOLL_CTL_DEL, getDescriptor(), NULL);
}
void Socket::resetSocket() {
struct epoll_event event;
event.data.ptr = this;
event.events = EPOLLIN | EPOLLRDHUP | EPOLLONESHOT | EPOLLET;
if(fifo.size() > 0)
event.events |= EPOLLWRNORM;
if(!shutDown)
epoll_ctl(ePoll.getDescriptor(), EPOLL_CTL_MOD, getDescriptor(), &event);
}
}