Merge pull request #411 from p12tic/rewrite-memory-management

Use explicit memory ownership in SocketMultiplexer
This commit is contained in:
Adrian Lucrèce Céleste 2019-08-22 15:01:21 -04:00 committed by GitHub
commit 7bb541ea91
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
14 changed files with 156 additions and 154 deletions

View File

@ -200,7 +200,7 @@ App::initApp(int argc, const char** argv)
void void
App::initIpcClient() App::initIpcClient()
{ {
m_ipcClient = new IpcClient(m_events, m_socketMultiplexer); m_ipcClient = new IpcClient(m_events, m_socketMultiplexer.get());
m_ipcClient->connect(); m_ipcClient->connect();
m_events->adoptHandler( m_events->adoptHandler(

View File

@ -23,7 +23,9 @@
#include "base/String.h" #include "base/String.h"
#include "base/Log.h" #include "base/Log.h"
#include "base/EventQueue.h" #include "base/EventQueue.h"
#include "net/SocketMultiplexer.h"
#include "common/common.h" #include "common/common.h"
#include <memory>
#if SYSAPI_WIN32 #if SYSAPI_WIN32
#include "barrier/win32/AppUtilWindows.h" #include "barrier/win32/AppUtilWindows.h"
@ -95,8 +97,8 @@ public:
virtual IEventQueue* getEvents() const { return m_events; } virtual IEventQueue* getEvents() const { return m_events; }
void setSocketMultiplexer(SocketMultiplexer* sm) { m_socketMultiplexer = sm; } void setSocketMultiplexer(std::unique_ptr<SocketMultiplexer>&& sm) { m_socketMultiplexer = std::move(sm); }
SocketMultiplexer* getSocketMultiplexer() const { return m_socketMultiplexer; } SocketMultiplexer* getSocketMultiplexer() const { return m_socketMultiplexer.get(); }
void setEvents(EventQueue& events) { m_events = &events; } void setEvents(EventQueue& events) { m_events = &events; }
@ -119,7 +121,7 @@ private:
CreateTaskBarReceiverFunc m_createTaskBarReceiver; CreateTaskBarReceiverFunc m_createTaskBarReceiver;
ARCH_APP_UTIL m_appUtil; ARCH_APP_UTIL m_appUtil;
IpcClient* m_ipcClient; IpcClient* m_ipcClient;
SocketMultiplexer* m_socketMultiplexer; std::unique_ptr<SocketMultiplexer> m_socketMultiplexer;
}; };
class MinimalApp : public App { class MinimalApp : public App {

View File

@ -443,8 +443,7 @@ ClientApp::mainLoop()
{ {
// create socket multiplexer. this must happen after daemonization // create socket multiplexer. this must happen after daemonization
// on unix because threads evaporate across a fork(). // on unix because threads evaporate across a fork().
SocketMultiplexer multiplexer; setSocketMultiplexer(std::make_unique<SocketMultiplexer>());
setSocketMultiplexer(&multiplexer);
// start client, etc // start client, etc
appUtil().startNode(); appUtil().startNode();

View File

@ -713,8 +713,7 @@ ServerApp::mainLoop()
{ {
// create socket multiplexer. this must happen after daemonization // create socket multiplexer. this must happen after daemonization
// on unix because threads evaporate across a fork(). // on unix because threads evaporate across a fork().
SocketMultiplexer multiplexer; setSocketMultiplexer(std::make_unique<SocketMultiplexer>());
setSocketMultiplexer(&multiplexer);
// if configuration has no screens then add this system // if configuration has no screens then add this system
// as the default // as the default

View File

@ -20,6 +20,19 @@
#include "arch/IArchNetwork.h" #include "arch/IArchNetwork.h"
#include "common/IInterface.h" #include "common/IInterface.h"
#include <memory>
class ISocketMultiplexerJob;
struct MultiplexerJobStatus
{
MultiplexerJobStatus(bool cont, std::unique_ptr<ISocketMultiplexerJob>&& nj) :
continue_servicing(cont), new_job(std::move(nj))
{}
bool continue_servicing = false;
std::unique_ptr<ISocketMultiplexerJob> new_job;
};
//! Socket multiplexer job //! Socket multiplexer job
/*! /*!
@ -32,21 +45,20 @@ public:
//! Handle socket event //! Handle socket event
/*! /*!
Called by a socket multiplexer when the socket becomes readable, Called by a socket multiplexer when the socket becomes readable, writable, or has an error.
writable, or has an error. It should return itself if the same The socket is readable if \p readable is true, writable if \p writable is true, and in error
job can continue to service events, a new job if the socket must if \p error is true.
be serviced differently, or NULL if the socket should no longer
be serviced. The socket is readable if \p readable is true, The method returns false as the continue_servicing member of the returned struct if the socket
writable if \p writable is true, and in error if \p error is should no longer be served and true otherwise. Additionally, if the new_job member of the
true. returned pair is not empty, the socket should be serviced differently with the specified job.
This call must not attempt to directly change the job for this This call must not attempt to directly change the job for this
socket by calling \c addSocket() or \c removeSocket() on the socket by calling \c addSocket() or \c removeSocket() on the
multiplexer. It must instead return the new job. It can, multiplexer. It must instead return the new job. It can,
however, add or remove jobs for other sockets. however, add or remove jobs for other sockets.
*/ */
virtual ISocketMultiplexerJob* virtual MultiplexerJobStatus run(bool readable, bool writable, bool error) = 0;
run(bool readable, bool writable, bool error) = 0;
//@} //@}
//! @name accessors //! @name accessors
@ -72,5 +84,6 @@ public:
*/ */
virtual bool isWritable() const = 0; virtual bool isWritable() const = 0;
virtual bool isCursor() const { return false; }
//@} //@}
}; };

View File

@ -83,7 +83,7 @@ SecureSocket::~SecureSocket()
// take socket from multiplexer ASAP otherwise the race condition // take socket from multiplexer ASAP otherwise the race condition
// could cause events to get called on a dead object. TCPSocket // could cause events to get called on a dead object. TCPSocket
// will do this, too, but the double-call is harmless // will do this, too, but the double-call is harmless
setJob(NULL); removeJob();
freeSSLResources(); freeSSLResources();
// removing sleep() because I have no idea why you would want to do it // removing sleep() because I have no idea why you would want to do it
@ -125,13 +125,12 @@ SecureSocket::connect(const NetworkAddress& addr)
TCPSocket::connect(addr); TCPSocket::connect(addr);
} }
ISocketMultiplexerJob* std::unique_ptr<ISocketMultiplexerJob> SecureSocket::newJob()
SecureSocket::newJob()
{ {
// after TCP connection is established, SecureSocket will pick up // after TCP connection is established, SecureSocket will pick up
// connected event and do secureConnect // connected event and do secureConnect
if (m_connected && !m_secureReady) { if (m_connected && !m_secureReady) {
return NULL; return {};
} }
return TCPSocket::newJob(); return TCPSocket::newJob();
@ -140,7 +139,7 @@ SecureSocket::newJob()
void void
SecureSocket::secureConnect() SecureSocket::secureConnect()
{ {
setJob(new TSocketMultiplexerMethodJob<SecureSocket>( setJob(std::make_unique<TSocketMultiplexerMethodJob<SecureSocket>>(
this, &SecureSocket::serviceConnect, this, &SecureSocket::serviceConnect,
getSocket(), isReadable(), isWritable())); getSocket(), isReadable(), isWritable()));
} }
@ -148,7 +147,7 @@ SecureSocket::secureConnect()
void void
SecureSocket::secureAccept() SecureSocket::secureAccept()
{ {
setJob(new TSocketMultiplexerMethodJob<SecureSocket>( setJob(std::make_unique<TSocketMultiplexerMethodJob<SecureSocket>>(
this, &SecureSocket::serviceAccept, this, &SecureSocket::serviceAccept,
getSocket(), isReadable(), isWritable())); getSocket(), isReadable(), isWritable()));
} }
@ -740,10 +739,11 @@ SecureSocket::verifyCertFingerprint()
return isValid; return isValid;
} }
ISocketMultiplexerJob* MultiplexerJobStatus SecureSocket::serviceConnect(ISocketMultiplexerJob* job,
SecureSocket::serviceConnect(ISocketMultiplexerJob* job, bool read, bool write, bool error)
bool, bool write, bool error)
{ {
(void) read;
Lock lock(&getMutex()); Lock lock(&getMutex());
int status = 0; int status = 0;
@ -755,25 +755,28 @@ SecureSocket::serviceConnect(ISocketMultiplexerJob* job,
// If status < 0, error happened // If status < 0, error happened
if (status < 0) { if (status < 0) {
return NULL; return {false, {}};
} }
// If status > 0, success // If status > 0, success
if (status > 0) { if (status > 0) {
sendEvent(m_events->forIDataSocket().secureConnected()); sendEvent(m_events->forIDataSocket().secureConnected());
return newJob(); return {true, newJob()};
} }
// Retry case // Retry case
return new TSocketMultiplexerMethodJob<SecureSocket>( return {
true,
std::make_unique<TSocketMultiplexerMethodJob<SecureSocket>>(
this, &SecureSocket::serviceConnect, this, &SecureSocket::serviceConnect,
getSocket(), isReadable(), isWritable()); getSocket(), isReadable(), isWritable())
};
} }
ISocketMultiplexerJob* MultiplexerJobStatus SecureSocket::serviceAccept(ISocketMultiplexerJob* job,
SecureSocket::serviceAccept(ISocketMultiplexerJob* job, bool read, bool write, bool error)
bool, bool write, bool error)
{ {
(void) read;
Lock lock(&getMutex()); Lock lock(&getMutex());
int status = 0; int status = 0;
@ -784,19 +787,19 @@ SecureSocket::serviceAccept(ISocketMultiplexerJob* job,
#endif #endif
// If status < 0, error happened // If status < 0, error happened
if (status < 0) { if (status < 0) {
return NULL; return {false, {}};
} }
// If status > 0, success // If status > 0, success
if (status > 0) { if (status > 0) {
sendEvent(m_events->forClientListener().accepted()); sendEvent(m_events->forClientListener().accepted());
return newJob(); return {true, newJob()};
} }
// Retry case // Retry case
return new TSocketMultiplexerMethodJob<SecureSocket>( return {true, std::make_unique<TSocketMultiplexerMethodJob<SecureSocket>>(
this, &SecureSocket::serviceAccept, this, &SecureSocket::serviceAccept,
getSocket(), isReadable(), isWritable()); getSocket(), isReadable(), isWritable())};
} }
void void

View File

@ -44,8 +44,7 @@ public:
// IDataSocket overrides // IDataSocket overrides
virtual void connect(const NetworkAddress&); virtual void connect(const NetworkAddress&);
ISocketMultiplexerJob* std::unique_ptr<ISocketMultiplexerJob> newJob() override;
newJob();
bool isFatal() const { return m_fatal; } bool isFatal() const { return m_fatal; }
void isFatal(bool b) { m_fatal = b; } void isFatal(bool b) { m_fatal = b; }
bool isSecureReady(); bool isSecureReady();
@ -74,13 +73,8 @@ private:
bool separator = true); bool separator = true);
bool verifyCertFingerprint(); bool verifyCertFingerprint();
ISocketMultiplexerJob* MultiplexerJobStatus serviceConnect(ISocketMultiplexerJob*, bool, bool, bool);
serviceConnect(ISocketMultiplexerJob*, MultiplexerJobStatus serviceAccept(ISocketMultiplexerJob*, bool, bool, bool);
bool, bool, bool);
ISocketMultiplexerJob*
serviceAccept(ISocketMultiplexerJob*,
bool, bool, bool);
void showSecureConnectInfo(); void showSecureConnectInfo();
void showSecureLibInfo(); void showSecureLibInfo();

View File

@ -33,6 +33,20 @@
// SocketMultiplexer // SocketMultiplexer
// //
class CursorMultiplexerJob : public ISocketMultiplexerJob {
public:
MultiplexerJobStatus run(bool readable, bool writable, bool error) override
{
return {false, {}};
}
ArchSocket getSocket() const override { return {}; }
bool isReadable() const override { return false; }
bool isWritable() const override { return false; }
bool isCursor() const override { return true; }
};
SocketMultiplexer::SocketMultiplexer() : SocketMultiplexer::SocketMultiplexer() :
m_mutex(new Mutex), m_mutex(new Mutex),
m_thread(NULL), m_thread(NULL),
@ -43,12 +57,6 @@ SocketMultiplexer::SocketMultiplexer() :
m_jobListLocker(NULL), m_jobListLocker(NULL),
m_jobListLockLocker(NULL) m_jobListLockLocker(NULL)
{ {
// this pointer just has to be unique and not NULL. it will
// never be dereferenced. it's used to identify cursor nodes
// in the jobs list.
// TODO: Remove this evilness
m_cursorMark = reinterpret_cast<ISocketMultiplexerJob*>(this);
// start thread // start thread
m_thread = new Thread(new TMethodJob<SocketMultiplexer>( m_thread = new Thread(new TMethodJob<SocketMultiplexer>(
this, &SocketMultiplexer::serviceThread)); this, &SocketMultiplexer::serviceThread));
@ -66,16 +74,9 @@ SocketMultiplexer::~SocketMultiplexer()
delete m_jobListLocker; delete m_jobListLocker;
delete m_jobListLockLocker; delete m_jobListLockLocker;
delete m_mutex; delete m_mutex;
// clean up jobs
for (SocketJobMap::iterator i = m_socketJobMap.begin();
i != m_socketJobMap.end(); ++i) {
delete *(i->second);
}
} }
void void SocketMultiplexer::addSocket(ISocket* socket, std::unique_ptr<ISocketMultiplexerJob>&& job)
SocketMultiplexer::addSocket(ISocket* socket, ISocketMultiplexerJob* job)
{ {
assert(socket != NULL); assert(socket != NULL);
assert(job != NULL); assert(job != NULL);
@ -95,16 +96,12 @@ SocketMultiplexer::addSocket(ISocket* socket, ISocketMultiplexerJob* job)
// we *must* put the job at the end so the order of jobs in // we *must* put the job at the end so the order of jobs in
// the list continue to match the order of jobs in pfds in // the list continue to match the order of jobs in pfds in
// serviceThread(). // serviceThread().
JobCursor j = m_socketJobs.insert(m_socketJobs.end(), job); JobCursor j = m_socketJobs.insert(m_socketJobs.end(), std::move(job));
m_update = true; m_update = true;
m_socketJobMap.insert(std::make_pair(socket, j)); m_socketJobMap.insert(std::make_pair(socket, j));
} }
else { else {
JobCursor j = i->second; *(i->second) = std::move(job);
if (*j != job) {
delete *j;
*j = job;
}
m_update = true; m_update = true;
} }
@ -131,10 +128,9 @@ SocketMultiplexer::removeSocket(ISocket* socket)
// to match the order of jobs in pfds in serviceThread(). // to match the order of jobs in pfds in serviceThread().
SocketJobMap::iterator i = m_socketJobMap.find(socket); SocketJobMap::iterator i = m_socketJobMap.find(socket);
if (i != m_socketJobMap.end()) { if (i != m_socketJobMap.end()) {
if (*(i->second) != NULL) { if (*(i->second)) {
delete *(i->second); i->second->reset();
*(i->second) = NULL; m_update = true;
m_update = true;
} }
} }
@ -173,14 +169,13 @@ SocketMultiplexer::serviceThread(void*)
JobCursor cursor = newCursor(); JobCursor cursor = newCursor();
JobCursor jobCursor = nextCursor(cursor); JobCursor jobCursor = nextCursor(cursor);
while (jobCursor != m_socketJobs.end()) { while (jobCursor != m_socketJobs.end()) {
ISocketMultiplexerJob* job = *jobCursor; if (*jobCursor) {
if (job != NULL) { pfd.m_socket = (*jobCursor)->getSocket();
pfd.m_socket = job->getSocket();
pfd.m_events = 0; pfd.m_events = 0;
if (job->isReadable()) { if ((*jobCursor)->isReadable()) {
pfd.m_events |= IArchNetwork::kPOLLIN; pfd.m_events |= IArchNetwork::kPOLLIN;
} }
if (job->isWritable()) { if ((*jobCursor)->isWritable()) {
pfd.m_events |= IArchNetwork::kPOLLOUT; pfd.m_events |= IArchNetwork::kPOLLOUT;
} }
pfds.push_back(pfd); pfds.push_back(pfd);
@ -221,15 +216,16 @@ SocketMultiplexer::serviceThread(void*)
IArchNetwork::kPOLLNVAL)) != 0); IArchNetwork::kPOLLNVAL)) != 0);
// run job // run job
ISocketMultiplexerJob* job = *jobCursor; MultiplexerJobStatus status = (*jobCursor)->run(read, write, error);
ISocketMultiplexerJob* newJob = job->run(read, write, error);
// save job, if different if (!status.continue_servicing) {
if (newJob != job) {
Lock lock(m_mutex); Lock lock(m_mutex);
delete job; jobCursor->reset();
*jobCursor = newJob; m_update = true;
m_update = true; } else if (status.new_job) {
Lock lock(m_mutex);
*jobCursor = std::move(status.new_job);
m_update = true;
} }
++i; ++i;
} }
@ -262,7 +258,7 @@ SocketMultiplexer::JobCursor
SocketMultiplexer::newCursor() SocketMultiplexer::newCursor()
{ {
Lock lock(m_mutex); Lock lock(m_mutex);
return m_socketJobs.insert(m_socketJobs.begin(), m_cursorMark); return m_socketJobs.insert(m_socketJobs.begin(), std::make_unique<CursorMultiplexerJob>());
} }
SocketMultiplexer::JobCursor SocketMultiplexer::JobCursor
@ -272,7 +268,7 @@ SocketMultiplexer::nextCursor(JobCursor cursor)
JobCursor j = m_socketJobs.end(); JobCursor j = m_socketJobs.end();
JobCursor i = cursor; JobCursor i = cursor;
while (++i != m_socketJobs.end()) { while (++i != m_socketJobs.end()) {
if (*i != m_cursorMark) { if (*i && !(*i)->isCursor()) {
// found a real job (as opposed to a cursor) // found a real job (as opposed to a cursor)
j = i; j = i;

View File

@ -21,6 +21,7 @@
#include "arch/IArchNetwork.h" #include "arch/IArchNetwork.h"
#include "common/stdlist.h" #include "common/stdlist.h"
#include "common/stdmap.h" #include "common/stdmap.h"
#include <memory>
template <class T> template <class T>
class CondVar; class CondVar;
@ -41,7 +42,7 @@ public:
//! @name manipulators //! @name manipulators
//@{ //@{
void addSocket(ISocket*, ISocketMultiplexerJob*); void addSocket(ISocket*, std::unique_ptr<ISocketMultiplexerJob>&& job);
void removeSocket(ISocket*); void removeSocket(ISocket*);
@ -58,7 +59,7 @@ public:
private: private:
// list of jobs. we use a list so we can safely iterate over it // list of jobs. we use a list so we can safely iterate over it
// while other threads modify it. // while other threads modify it.
typedef std::list<ISocketMultiplexerJob*> SocketJobs; using SocketJobs = std::list<std::unique_ptr<ISocketMultiplexerJob>>;
typedef SocketJobs::iterator JobCursor; typedef SocketJobs::iterator JobCursor;
typedef std::map<ISocket*, JobCursor> SocketJobMap; typedef std::map<ISocket*, JobCursor> SocketJobMap;
@ -106,6 +107,4 @@ private:
SocketJobs m_socketJobs; SocketJobs m_socketJobs;
SocketJobMap m_socketJobMap; SocketJobMap m_socketJobMap;
ISocketMultiplexerJob*
m_cursorMark;
}; };

View File

@ -69,10 +69,11 @@ TCPListenSocket::bind(const NetworkAddress& addr)
ARCH->setReuseAddrOnSocket(m_socket, true); ARCH->setReuseAddrOnSocket(m_socket, true);
ARCH->bindSocket(m_socket, addr.getAddress()); ARCH->bindSocket(m_socket, addr.getAddress());
ARCH->listenOnSocket(m_socket); ARCH->listenOnSocket(m_socket);
m_socketMultiplexer->addSocket(this,
new TSocketMultiplexerMethodJob<TCPListenSocket>( auto new_job = std::make_unique<TSocketMultiplexerMethodJob<TCPListenSocket>>(
this, &TCPListenSocket::serviceListening, this, &TCPListenSocket::serviceListening, m_socket, true, false);
m_socket, true, false));
m_socketMultiplexer->addSocket(this, std::move(new_job));
} }
catch (XArchNetworkAddressInUse& e) { catch (XArchNetworkAddressInUse& e) {
throw XSocketAddressInUse(e.what()); throw XSocketAddressInUse(e.what());
@ -135,24 +136,22 @@ TCPListenSocket::accept()
void void
TCPListenSocket::setListeningJob() TCPListenSocket::setListeningJob()
{ {
m_socketMultiplexer->addSocket(this, auto new_job = std::make_unique<TSocketMultiplexerMethodJob<TCPListenSocket>>(
new TSocketMultiplexerMethodJob<TCPListenSocket>( this, &TCPListenSocket::serviceListening, m_socket, true, false);
this, &TCPListenSocket::serviceListening, m_socketMultiplexer->addSocket(this, std::move(new_job));
m_socket, true, false));
} }
ISocketMultiplexerJob* MultiplexerJobStatus TCPListenSocket::serviceListening(ISocketMultiplexerJob* job,
TCPListenSocket::serviceListening(ISocketMultiplexerJob* job, bool read, bool, bool error)
bool read, bool, bool error)
{ {
if (error) { if (error) {
close(); close();
return NULL; return {false, {}};
} }
if (read) { if (read) {
m_events->addEvent(Event(m_events->forIListenSocket().connecting(), this, NULL)); m_events->addEvent(Event(m_events->forIListenSocket().connecting(), this, NULL));
// stop polling on this socket until the client accepts // stop polling on this socket until the client accepts
return NULL; return {false, {}};
} }
return job; return {true, {}};
} }

View File

@ -19,10 +19,10 @@
#pragma once #pragma once
#include "net/IListenSocket.h" #include "net/IListenSocket.h"
#include "net/ISocketMultiplexerJob.h"
#include "arch/IArchNetwork.h" #include "arch/IArchNetwork.h"
class Mutex; class Mutex;
class ISocketMultiplexerJob;
class IEventQueue; class IEventQueue;
class SocketMultiplexer; class SocketMultiplexer;
@ -48,9 +48,7 @@ protected:
void setListeningJob(); void setListeningJob();
public: public:
ISocketMultiplexerJob* MultiplexerJobStatus serviceListening(ISocketMultiplexerJob*, bool, bool, bool);
serviceListening(ISocketMultiplexerJob*,
bool, bool, bool);
protected: protected:
ArchSocket m_socket; ArchSocket m_socket;

View File

@ -388,40 +388,42 @@ TCPSocket::doWrite()
return kRetry; return kRetry;
} }
void void TCPSocket::removeJob()
TCPSocket::setJob(ISocketMultiplexerJob* job)
{ {
// multiplexer will delete the old job // multiplexer will delete the old job
if (job == NULL) { m_socketMultiplexer->removeSocket(this);
m_socketMultiplexer->removeSocket(this); }
}
else { void TCPSocket::setJob(std::unique_ptr<ISocketMultiplexerJob>&& job)
m_socketMultiplexer->addSocket(this, job); {
if (job.get() == nullptr) {
removeJob();
} else {
m_socketMultiplexer->addSocket(this, std::move(job));
} }
} }
ISocketMultiplexerJob* std::unique_ptr<ISocketMultiplexerJob> TCPSocket::newJob()
TCPSocket::newJob()
{ {
// note -- must have m_mutex locked on entry // note -- must have m_mutex locked on entry
if (m_socket == NULL) { if (m_socket == NULL) {
return NULL; return {};
} }
else if (!m_connected) { else if (!m_connected) {
assert(!m_readable); assert(!m_readable);
if (!(m_readable || m_writable)) { if (!(m_readable || m_writable)) {
return NULL; return {};
} }
return new TSocketMultiplexerMethodJob<TCPSocket>( return std::make_unique<TSocketMultiplexerMethodJob<TCPSocket>>(
this, &TCPSocket::serviceConnecting, this, &TCPSocket::serviceConnecting,
m_socket, m_readable, m_writable); m_socket, m_readable, m_writable);
} }
else { else {
if (!(m_readable || (m_writable && (m_outputBuffer.getSize() > 0)))) { if (!(m_readable || (m_writable && (m_outputBuffer.getSize() > 0)))) {
return NULL; return {};
} }
return new TSocketMultiplexerMethodJob<TCPSocket>( return std::make_unique<TSocketMultiplexerMethodJob<TCPSocket>>(
this, &TCPSocket::serviceConnected, this, &TCPSocket::serviceConnected,
m_socket, m_readable, m_socket, m_readable,
m_writable && (m_outputBuffer.getSize() > 0)); m_writable && (m_outputBuffer.getSize() > 0));
@ -488,9 +490,7 @@ TCPSocket::onDisconnected()
m_connected = false; m_connected = false;
} }
ISocketMultiplexerJob* MultiplexerJobStatus TCPSocket::serviceConnecting(ISocketMultiplexerJob* job, bool, bool write, bool error)
TCPSocket::serviceConnecting(ISocketMultiplexerJob* job,
bool, bool write, bool error)
{ {
Lock lock(&m_mutex); Lock lock(&m_mutex);
@ -519,29 +519,36 @@ TCPSocket::serviceConnecting(ISocketMultiplexerJob* job,
catch (XArchNetwork& e) { catch (XArchNetwork& e) {
sendConnectionFailedEvent(e.what()); sendConnectionFailedEvent(e.what());
onDisconnected(); onDisconnected();
return newJob(); auto new_job = newJob();
if (new_job)
return {true, std::move(new_job)};
else
return {false, {}};
} }
} }
if (write) { if (write) {
sendEvent(m_events->forIDataSocket().connected()); sendEvent(m_events->forIDataSocket().connected());
onConnected(); onConnected();
return newJob(); auto new_job = newJob();
if (new_job)
return {true, std::move(new_job)};
else
return {false, {}};
} }
return job; return {true, {}};
} }
ISocketMultiplexerJob* MultiplexerJobStatus TCPSocket::serviceConnected(ISocketMultiplexerJob* job,
TCPSocket::serviceConnected(ISocketMultiplexerJob* job, bool read, bool write, bool error)
bool read, bool write, bool error)
{ {
Lock lock(&m_mutex); Lock lock(&m_mutex);
if (error) { if (error) {
sendEvent(m_events->forISocket().disconnected()); sendEvent(m_events->forISocket().disconnected());
onDisconnected(); onDisconnected();
return newJob(); return {true, newJob()};
} }
EJobResult writeResult = kRetry; EJobResult writeResult = kRetry;
@ -594,10 +601,10 @@ TCPSocket::serviceConnected(ISocketMultiplexerJob* job,
} }
if (writeResult == kBreak || readResult == kBreak) { if (writeResult == kBreak || readResult == kBreak) {
return NULL; return {false, {}};
} else if (writeResult == kNew || readResult == kNew) { } else if (writeResult == kNew || readResult == kNew) {
return newJob(); return {true, newJob()};
} else { } else {
return job; return {true, {}};
} }
} }

View File

@ -19,14 +19,15 @@
#pragma once #pragma once
#include "net/IDataSocket.h" #include "net/IDataSocket.h"
#include "net/ISocketMultiplexerJob.h"
#include "io/StreamBuffer.h" #include "io/StreamBuffer.h"
#include "mt/CondVar.h" #include "mt/CondVar.h"
#include "mt/Mutex.h" #include "mt/Mutex.h"
#include "arch/IArchNetwork.h" #include "arch/IArchNetwork.h"
#include <memory>
class Mutex; class Mutex;
class Thread; class Thread;
class ISocketMultiplexerJob;
class IEventQueue; class IEventQueue;
class SocketMultiplexer; class SocketMultiplexer;
@ -59,8 +60,7 @@ public:
virtual void connect(const NetworkAddress&); virtual void connect(const NetworkAddress&);
virtual ISocketMultiplexerJob* virtual std::unique_ptr<ISocketMultiplexerJob> newJob();
newJob();
protected: protected:
enum EJobResult { enum EJobResult {
@ -74,7 +74,8 @@ protected:
virtual EJobResult doRead(); virtual EJobResult doRead();
virtual EJobResult doWrite(); virtual EJobResult doWrite();
void setJob(ISocketMultiplexerJob*); void removeJob();
void setJob(std::unique_ptr<ISocketMultiplexerJob>&& job);
bool isReadable() { return m_readable; } bool isReadable() { return m_readable; }
bool isWritable() { return m_writable; } bool isWritable() { return m_writable; }
@ -93,12 +94,8 @@ private:
void onOutputShutdown(); void onOutputShutdown();
void onDisconnected(); void onDisconnected();
ISocketMultiplexerJob* MultiplexerJobStatus serviceConnecting(ISocketMultiplexerJob*, bool, bool, bool);
serviceConnecting(ISocketMultiplexerJob*, MultiplexerJobStatus serviceConnected(ISocketMultiplexerJob*, bool, bool, bool);
bool, bool, bool);
ISocketMultiplexerJob*
serviceConnected(ISocketMultiplexerJob*,
bool, bool, bool);
protected: protected:
bool m_readable; bool m_readable;

View File

@ -28,8 +28,7 @@ A socket multiplexer job class that invokes a member function.
template <class T> template <class T>
class TSocketMultiplexerMethodJob : public ISocketMultiplexerJob { class TSocketMultiplexerMethodJob : public ISocketMultiplexerJob {
public: public:
typedef ISocketMultiplexerJob* using Method = MultiplexerJobStatus (T::*)(ISocketMultiplexerJob*, bool, bool, bool);
(T::*Method)(ISocketMultiplexerJob*, bool, bool, bool);
//! run() invokes \c object->method(arg) //! run() invokes \c object->method(arg)
TSocketMultiplexerMethodJob(T* object, Method method, TSocketMultiplexerMethodJob(T* object, Method method,
@ -37,11 +36,10 @@ public:
virtual ~TSocketMultiplexerMethodJob(); virtual ~TSocketMultiplexerMethodJob();
// IJob overrides // IJob overrides
virtual ISocketMultiplexerJob* virtual MultiplexerJobStatus run(bool readable, bool writable, bool error) override;
run(bool readable, bool writable, bool error); virtual ArchSocket getSocket() const override;
virtual ArchSocket getSocket() const; virtual bool isReadable() const override;
virtual bool isReadable() const; virtual bool isWritable() const override;
virtual bool isWritable() const;
private: private:
T* m_object; T* m_object;
@ -74,14 +72,12 @@ TSocketMultiplexerMethodJob<T>::~TSocketMultiplexerMethodJob()
} }
template <class T> template <class T>
inline inline MultiplexerJobStatus TSocketMultiplexerMethodJob<T>::run(bool read, bool write, bool error)
ISocketMultiplexerJob*
TSocketMultiplexerMethodJob<T>::run(bool read, bool write, bool error)
{ {
if (m_object != NULL) { if (m_object != NULL) {
return (m_object->*m_method)(this, read, write, error); return (m_object->*m_method)(this, read, write, error);
} }
return NULL; return {false, {}};
} }
template <class T> template <class T>