Merge pull request #464 from majestrate/master

Add locking to SAM

HTTPServer.cpp

@@ -749,7 +749,7 @@ namespace util
 				s << "&" << HTTP_PARAM_BASE32_ADDRESS << "=" << ident.ToBase32 () << ">"; 
 				s << i2p::client::context.GetAddressBook ().ToAddress(ident) << "</a><br>\r\n" << std::endl;
 				s << "<b>Streams:</b><br>\r\n";
-				for (auto it: session->sockets)
+				for (auto it: session->ListSockets())
 				{
 					switch (it->GetSocketType ())
 					{

SAM.cpp

@@ -47,16 +47,19 @@ namespace client
 			break;
 			case eSAMSocketTypeStream:
 			{
-				if (m_Session)
+				if (m_Session) {
-					m_Session->sockets.remove (shared_from_this ());
+					m_Session->DelSocket (shared_from_this ());
+					m_Session = nullptr;
+				}
 				break;
 			}
 			case eSAMSocketTypeAcceptor:
 			{
 				if (m_Session)
 				{
-					m_Session->sockets.remove (shared_from_this ());
+					m_Session->DelSocket (shared_from_this ());
 					m_Session->localDestination->StopAcceptingStreams ();
+					m_Session = nullptr;
 				}
 				break;
 			}
@@ -64,7 +67,7 @@ namespace client
 				;
 		}
 		m_SocketType = eSAMSocketTypeTerminated;
-		m_Socket.close ();
+		if (m_Socket.is_open()) m_Socket.close ();
 	}
 
 	void SAMSocket::ReceiveHandshake ()
@@ -369,7 +372,7 @@ namespace client
 	void SAMSocket::Connect (std::shared_ptr<const i2p::data::LeaseSet> remote)
 	{
 		m_SocketType = eSAMSocketTypeStream;
-		m_Session->sockets.push_back (shared_from_this ());
+		m_Session->AddSocket (shared_from_this ());
 		m_Stream = m_Session->localDestination->CreateStream (remote);
 		m_Stream->Send ((uint8_t *)m_Buffer, 0); // connect
 		I2PReceive ();			
@@ -402,7 +405,7 @@ namespace client
 			if (!m_Session->localDestination->IsAcceptingStreams ())
 			{
 				m_SocketType = eSAMSocketTypeAcceptor;
-				m_Session->sockets.push_back (shared_from_this ());
+				m_Session->AddSocket (shared_from_this ());
 				m_Session->localDestination->AcceptStreams (std::bind (&SAMSocket::HandleI2PAccept, shared_from_this (), std::placeholders::_1));
 				SendMessageReply (SAM_STREAM_STATUS_OK, strlen(SAM_STREAM_STATUS_OK), false);
 			}
@@ -676,19 +679,20 @@ namespace client
 		
 	SAMSession::~SAMSession ()
 	{
-		for (auto it: sockets)
+		CloseStreams();
-			it->SetSocketType (eSAMSocketTypeTerminated);
 		i2p::client::context.DeleteLocalDestination (localDestination);
 	}
 
 	void SAMSession::CloseStreams ()
 	{
-		for (auto it: sockets)
 		{
-			it->CloseStream ();
+			std::lock_guard<std::mutex> lock(m_SocketsMutex);
-			it->SetSocketType (eSAMSocketTypeTerminated);
+			for (auto sock : m_Sockets) {
+				sock->CloseStream();
+			}
 		}
-		sockets.clear ();
+		// XXX: should this be done inside locked parts?
+		m_Sockets.clear();
 	}
 
 	SAMBridge::SAMBridge (const std::string& address, int port):

SAM.h

@@ -134,7 +134,30 @@ namespace client
 	struct SAMSession
 	{
 		std::shared_ptr<ClientDestination> localDestination;
-		std::list<std::shared_ptr<SAMSocket> > sockets;
+		std::list<std::shared_ptr<SAMSocket> > m_Sockets;
+		std::mutex m_SocketsMutex;
+
+		/** safely add a socket to this session */
+		void AddSocket(std::shared_ptr<SAMSocket> sock) {
+			std::lock_guard<std::mutex> lock(m_SocketsMutex);
+			m_Sockets.push_back(sock);
+		}
+
+		/** safely remove a socket from this session */
+		void DelSocket(std::shared_ptr<SAMSocket> sock) {
+			std::lock_guard<std::mutex> lock(m_SocketsMutex);
+			m_Sockets.remove(sock);
+		}
+
+		/** get a list holding a copy of all sam sockets from this session */
+		std::list<std::shared_ptr<SAMSocket> > ListSockets() {
+			std::list<std::shared_ptr<SAMSocket> > l;
+			{
+				std::lock_guard<std::mutex> lock(m_SocketsMutex);
+				for( auto & sock : m_Sockets ) l.push_back(sock);
+			}
+			return l;
+		}
 		
 		SAMSession (std::shared_ptr<ClientDestination> dest);		
 		~SAMSession ();