Make ownership of SocketMultiplexerJob explicit

src/lib/net/ISocketMultiplexerJob.h

@@ -20,6 +20,19 @@
 
 #include "arch/IArchNetwork.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
 /*!
@@ -32,21 +45,20 @@ public:
 
     //! 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
     socket by calling \c addSocket() or \c removeSocket() on the
     multiplexer.  It must instead return the new job.  It can,
     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
@@ -72,5 +84,6 @@ public:
     */
     virtual bool        isWritable() const = 0;
 
+    virtual bool isCursor() const { return false; }
     //@}
 };

src/lib/net/SecureSocket.cpp

@@ -83,7 +83,7 @@ SecureSocket::~SecureSocket()
     // take socket from multiplexer ASAP otherwise the race condition
     // could cause events to get called on a dead object. TCPSocket
     // will do this, too, but the double-call is harmless
-    setJob(NULL);
+    removeJob();
     freeSSLResources();
 
     // 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);
 }
 
-ISocketMultiplexerJob*
+std::unique_ptr<ISocketMultiplexerJob> SecureSocket::newJob()
-SecureSocket::newJob()
 {
     // after TCP connection is established, SecureSocket will pick up
     // connected event and do secureConnect
     if (m_connected && !m_secureReady) {
-        return NULL;
+        return {};
     }
     
     return TCPSocket::newJob();
@@ -140,7 +139,7 @@ SecureSocket::newJob()
 void
 SecureSocket::secureConnect()
 {
-    setJob(new TSocketMultiplexerMethodJob<SecureSocket>(
+    setJob(std::make_unique<TSocketMultiplexerMethodJob<SecureSocket>>(
                     this, &SecureSocket::serviceConnect,
                     getSocket(), isReadable(), isWritable()));
 }
@@ -148,7 +147,7 @@ SecureSocket::secureConnect()
 void
 SecureSocket::secureAccept()
 {
-    setJob(new TSocketMultiplexerMethodJob<SecureSocket>(
+    setJob(std::make_unique<TSocketMultiplexerMethodJob<SecureSocket>>(
                     this, &SecureSocket::serviceAccept,
                     getSocket(), isReadable(), isWritable()));
 }
@@ -740,10 +739,11 @@ SecureSocket::verifyCertFingerprint()
     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());
 
     int status = 0;
@@ -755,25 +755,28 @@ SecureSocket::serviceConnect(ISocketMultiplexerJob* job,
 
     // If status < 0, error happened
     if (status < 0) {
-        return NULL;
+        return {false, {}};
     }
 
     // If status > 0, success
     if (status > 0) {
         sendEvent(m_events->forIDataSocket().secureConnected());
-        return newJob();
+        return {true, newJob()};
     }
 
     // Retry case
-    return new TSocketMultiplexerMethodJob<SecureSocket>(
+    return {
+        true,
+        std::make_unique<TSocketMultiplexerMethodJob<SecureSocket>>(
             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());
 
     int status = 0;
@@ -784,19 +787,19 @@ SecureSocket::serviceAccept(ISocketMultiplexerJob* job,
 #endif
         // If status < 0, error happened
     if (status < 0) {
-        return NULL;
+        return {false, {}};
     }
 
     // If status > 0, success
     if (status > 0) {
         sendEvent(m_events->forClientListener().accepted());
-        return newJob();
+        return {true, newJob()};
     }
 
     // Retry case
-    return new TSocketMultiplexerMethodJob<SecureSocket>(
+    return {true, std::make_unique<TSocketMultiplexerMethodJob<SecureSocket>>(
             this, &SecureSocket::serviceAccept,
-            getSocket(), isReadable(), isWritable());
+            getSocket(), isReadable(), isWritable())};
 }
 
 void

src/lib/net/SecureSocket.h

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

src/lib/net/SocketMultiplexer.cpp

@@ -33,6 +33,20 @@
 // 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() :
     m_mutex(new Mutex),
     m_thread(NULL),
@@ -43,12 +57,6 @@ SocketMultiplexer::SocketMultiplexer() :
     m_jobListLocker(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
     m_thread = new Thread(new TMethodJob<SocketMultiplexer>(
                                 this, &SocketMultiplexer::serviceThread));
@@ -66,16 +74,9 @@ SocketMultiplexer::~SocketMultiplexer()
     delete m_jobListLocker;
     delete m_jobListLockLocker;
     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(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
         // the list continue to match the order of jobs in pfds in
         // 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_socketJobMap.insert(std::make_pair(socket, j));
     }
     else {
-        JobCursor j = i->second;
+        *(i->second) = std::move(job);
-        if (*j != job) {
-            delete *j;
-            *j = job;
-        }
         m_update = true;
     }
 
@@ -131,9 +128,8 @@ SocketMultiplexer::removeSocket(ISocket* socket)
     // to match the order of jobs in pfds in serviceThread().
     SocketJobMap::iterator i = m_socketJobMap.find(socket);
     if (i != m_socketJobMap.end()) {
-        if (*(i->second) != NULL) {
+        if (*(i->second)) {
-            delete *(i->second);
+            i->second->reset();
-            *(i->second) = NULL;
             m_update = true;
         }
     }
@@ -173,14 +169,13 @@ SocketMultiplexer::serviceThread(void*)
             JobCursor cursor    = newCursor();
             JobCursor jobCursor = nextCursor(cursor);
             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;
-                    if (job->isReadable()) {
+                    if ((*jobCursor)->isReadable()) {
                         pfd.m_events |= IArchNetwork::kPOLLIN;
                     }
-                    if (job->isWritable()) {
+                    if ((*jobCursor)->isWritable()) {
                         pfd.m_events |= IArchNetwork::kPOLLOUT;
                     }
                     pfds.push_back(pfd);
@@ -221,14 +216,15 @@ SocketMultiplexer::serviceThread(void*)
                                               IArchNetwork::kPOLLNVAL)) != 0);
 
                     // 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);
-                        delete job;
+                        jobCursor->reset();
-                        *jobCursor = newJob;
+                        m_update = true;
+                    } else if (status.new_job) {
+                        Lock lock(m_mutex);
+                        *jobCursor = std::move(status.new_job);
                         m_update = true;
                     }
                     ++i;
@@ -262,7 +258,7 @@ SocketMultiplexer::JobCursor
 SocketMultiplexer::newCursor()
 {
     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
@@ -272,7 +268,7 @@ SocketMultiplexer::nextCursor(JobCursor cursor)
     JobCursor j = m_socketJobs.end();
     JobCursor i = cursor;
     while (++i != m_socketJobs.end()) {
-        if (*i != m_cursorMark) {
+        if (*i && !(*i)->isCursor()) {
             // found a real job (as opposed to a cursor)
             j = i;
 

src/lib/net/SocketMultiplexer.h

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

src/lib/net/TCPListenSocket.cpp

@@ -69,10 +69,11 @@ TCPListenSocket::bind(const NetworkAddress& addr)
         ARCH->setReuseAddrOnSocket(m_socket, true);
         ARCH->bindSocket(m_socket, addr.getAddress());
         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) {
         throw XSocketAddressInUse(e.what());
@@ -135,24 +136,22 @@ TCPListenSocket::accept()
 void
 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)
 {
     if (error) {
         close();
-        return NULL;
+        return {false, {}};
     }
     if (read) {
         m_events->addEvent(Event(m_events->forIListenSocket().connecting(), this, NULL));
         // stop polling on this socket until the client accepts
-        return NULL;
+        return {false, {}};
     }
-    return job;
+    return {true, {}};
 }

src/lib/net/TCPListenSocket.h

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

src/lib/net/TCPSocket.cpp

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

src/lib/net/TCPSocket.h

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

src/lib/net/TSocketMultiplexerMethodJob.h

@@ -28,8 +28,7 @@ A socket multiplexer job class that invokes a member function.
 template <class T>
 class TSocketMultiplexerMethodJob : public ISocketMultiplexerJob {
 public:
-    typedef ISocketMultiplexerJob*
+    using Method = MultiplexerJobStatus (T::*)(ISocketMultiplexerJob*, bool, bool, bool);
-                (T::*Method)(ISocketMultiplexerJob*, bool, bool, bool);
 
     //! run() invokes \c object->method(arg)
     TSocketMultiplexerMethodJob(T* object, Method method,
@@ -37,11 +36,10 @@ public:
     virtual ~TSocketMultiplexerMethodJob();
 
     // 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:
     T*                    m_object;
@@ -74,14 +72,12 @@ TSocketMultiplexerMethodJob<T>::~TSocketMultiplexerMethodJob()
 }
 
 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) {
         return (m_object->*m_method)(this, read, write, error);
     }
-    return NULL;
+    return {false, {}};
 }
 
 template <class T>