PurpleI2P
/
i2pd
mirror of https://github.com/PurpleI2P/i2pd.git
2
0
Fork 0
Code Issues Releases Activity
Browse Source

shared pointers for SAM sockets

pull/113/head
orignal 10 years ago
parent
dcefe7d221
commit
81c63b0c30
2 changed files with 38 additions and 50 deletions
Whitespace
Show all changes Ignore whitespace when comparing lines Ignore changes in amount of whitespace Ignore changes in whitespace at EOL
Unified View
Diff Options
Show Stats Download Patch File Download Diff File
  1. 78
      SAM.cpp
  2. 10
      SAM.h

78
SAM.cpp
Unescape View File

@ -26,21 +26,18 @@ namespace client
SAMSocket::~SAMSocket () SAMSocket::~SAMSocket ()
{ {
if (m_Stream) if (m_Stream)
{
m_Stream->Close ();
i2p::stream::DeleteStream (m_Stream); i2p::stream::DeleteStream (m_Stream);
m_Stream = nullptr;
}
} }
void SAMSocket::Terminate () void SAMSocket::Terminate ()
{ {
if (m_Stream) if (m_Stream)
{
m_Stream->Close (); m_Stream->Close ();
i2p::stream::DeleteStream (m_Stream);
m_Stream = nullptr; // TODO: make this swap atomic
} auto session = m_Session;
m_Session = nullptr;
switch (m_SocketType) switch (m_SocketType)
{ {
case eSAMSocketTypeSession: case eSAMSocketTypeSession:
@ -48,16 +45,16 @@ namespace client
break; break;
case eSAMSocketTypeStream: case eSAMSocketTypeStream:
{ {
if (m_Session) if (session)
m_Session->sockets.remove (this); session->sockets.remove (shared_from_this ());
break; break;
} }
case eSAMSocketTypeAcceptor: case eSAMSocketTypeAcceptor:
{ {
if (m_Session) if (session)
{ {
m_Session->sockets.remove (this); session->sockets.remove (shared_from_this ());
m_Session->localDestination->StopAcceptingStreams (); session->localDestination->StopAcceptingStreams ();
} }
break; break;
} }
@ -65,13 +62,12 @@ namespace client
; ;
} }
m_Socket.close (); m_Socket.close ();
// delete this;
} }
void SAMSocket::ReceiveHandshake () void SAMSocket::ReceiveHandshake ()
{ {
m_Socket.async_read_some (boost::asio::buffer(m_Buffer, SAM_SOCKET_BUFFER_SIZE), m_Socket.async_read_some (boost::asio::buffer(m_Buffer, SAM_SOCKET_BUFFER_SIZE),
boost::bind(&SAMSocket::HandleHandshakeReceived, this, boost::bind(&SAMSocket::HandleHandshakeReceived, shared_from_this (),
boost::asio::placeholders::error, boost::asio::placeholders::bytes_transferred)); boost::asio::placeholders::error, boost::asio::placeholders::bytes_transferred));
} }
@ -91,7 +87,7 @@ namespace client
{ {
// TODO: check version // TODO: check version
boost::asio::async_write (m_Socket, boost::asio::buffer (SAM_HANDSHAKE_REPLY, strlen (SAM_HANDSHAKE_REPLY)), boost::asio::transfer_all (), boost::asio::async_write (m_Socket, boost::asio::buffer (SAM_HANDSHAKE_REPLY, strlen (SAM_HANDSHAKE_REPLY)), boost::asio::transfer_all (),
boost::bind(&SAMSocket::HandleHandshakeReplySent, this, boost::bind(&SAMSocket::HandleHandshakeReplySent, shared_from_this (),
boost::asio::placeholders::error, boost::asio::placeholders::bytes_transferred)); boost::asio::placeholders::error, boost::asio::placeholders::bytes_transferred));
} }
else else
@ -113,7 +109,7 @@ namespace client
else else
{ {
m_Socket.async_read_some (boost::asio::buffer(m_Buffer, SAM_SOCKET_BUFFER_SIZE), m_Socket.async_read_some (boost::asio::buffer(m_Buffer, SAM_SOCKET_BUFFER_SIZE),
boost::bind(&SAMSocket::HandleMessage, this, boost::bind(&SAMSocket::HandleMessage, shared_from_this (),
boost::asio::placeholders::error, boost::asio::placeholders::bytes_transferred)); boost::asio::placeholders::error, boost::asio::placeholders::bytes_transferred));
} }
} }
@ -122,7 +118,7 @@ namespace client
{ {
if (!m_IsSilent || m_SocketType == eSAMSocketTypeAcceptor) if (!m_IsSilent || m_SocketType == eSAMSocketTypeAcceptor)
boost::asio::async_write (m_Socket, boost::asio::buffer (msg, len), boost::asio::transfer_all (), boost::asio::async_write (m_Socket, boost::asio::buffer (msg, len), boost::asio::transfer_all (),
boost::bind(&SAMSocket::HandleMessageReplySent, this, boost::bind(&SAMSocket::HandleMessageReplySent, shared_from_this (),
boost::asio::placeholders::error, boost::asio::placeholders::bytes_transferred, close)); boost::asio::placeholders::error, boost::asio::placeholders::bytes_transferred, close));
else else
{ {
@ -228,7 +224,7 @@ namespace client
if (style == SAM_VALUE_DATAGRAM) if (style == SAM_VALUE_DATAGRAM)
{ {
auto dest = m_Session->localDestination->CreateDatagramDestination (); auto dest = m_Session->localDestination->CreateDatagramDestination ();
dest->SetReceiver (std::bind (&SAMSocket::HandleI2PDatagramReceive, this, dest->SetReceiver (std::bind (&SAMSocket::HandleI2PDatagramReceive, shared_from_this (),
std::placeholders::_1, std::placeholders::_2, std::placeholders::_3)); std::placeholders::_1, std::placeholders::_2, std::placeholders::_3));
} }
SendSessionCreateReplyOk (); SendSessionCreateReplyOk ();
@ -237,7 +233,7 @@ namespace client
{ {
m_Timer.expires_from_now (boost::posix_time::seconds(SAM_SESSION_READINESS_CHECK_INTERVAL)); m_Timer.expires_from_now (boost::posix_time::seconds(SAM_SESSION_READINESS_CHECK_INTERVAL));
m_Timer.async_wait (boost::bind (&SAMSocket::HandleSessionReadinessCheckTimer, m_Timer.async_wait (boost::bind (&SAMSocket::HandleSessionReadinessCheckTimer,
this, boost::asio::placeholders::error)); shared_from_this (), boost::asio::placeholders::error));
} }
} }
else else
@ -254,7 +250,7 @@ namespace client
{ {
m_Timer.expires_from_now (boost::posix_time::seconds(SAM_SESSION_READINESS_CHECK_INTERVAL)); m_Timer.expires_from_now (boost::posix_time::seconds(SAM_SESSION_READINESS_CHECK_INTERVAL));
m_Timer.async_wait (boost::bind (&SAMSocket::HandleSessionReadinessCheckTimer, m_Timer.async_wait (boost::bind (&SAMSocket::HandleSessionReadinessCheckTimer,
this, boost::asio::placeholders::error)); shared_from_this (), boost::asio::placeholders::error));
} }
} }
} }
@ -299,7 +295,7 @@ namespace client
i2p::data::netdb.RequestDestination (dest.GetIdentHash (), true, m_Session->localDestination->GetTunnelPool ()); i2p::data::netdb.RequestDestination (dest.GetIdentHash (), true, m_Session->localDestination->GetTunnelPool ());
m_Timer.expires_from_now (boost::posix_time::seconds(SAM_CONNECT_TIMEOUT)); m_Timer.expires_from_now (boost::posix_time::seconds(SAM_CONNECT_TIMEOUT));
m_Timer.async_wait (boost::bind (&SAMSocket::HandleStreamDestinationRequestTimer, m_Timer.async_wait (boost::bind (&SAMSocket::HandleStreamDestinationRequestTimer,
this, boost::asio::placeholders::error, dest.GetIdentHash ())); shared_from_this (), boost::asio::placeholders::error, dest.GetIdentHash ()));
} }
} }
else else
@ -309,7 +305,7 @@ namespace client
void SAMSocket::Connect (const i2p::data::LeaseSet& remote) void SAMSocket::Connect (const i2p::data::LeaseSet& remote)
{ {
m_SocketType = eSAMSocketTypeStream; m_SocketType = eSAMSocketTypeStream;
m_Session->sockets.push_back (this); m_Session->sockets.push_back (shared_from_this ());
m_Stream = m_Session->localDestination->CreateStream (remote); m_Stream = m_Session->localDestination->CreateStream (remote);
m_Stream->Send ((uint8_t *)m_Buffer, 0); // connect m_Stream->Send ((uint8_t *)m_Buffer, 0); // connect
I2PReceive (); I2PReceive ();
@ -366,8 +362,8 @@ namespace client
if (!m_Session->localDestination->IsAcceptingStreams ()) if (!m_Session->localDestination->IsAcceptingStreams ())
{ {
m_SocketType = eSAMSocketTypeAcceptor; m_SocketType = eSAMSocketTypeAcceptor;
m_Session->sockets.push_back (this); m_Session->sockets.push_back (shared_from_this ());
m_Session->localDestination->AcceptStreams (std::bind (&SAMSocket::HandleI2PAccept, this, std::placeholders::_1)); 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); SendMessageReply (SAM_STREAM_STATUS_OK, strlen(SAM_STREAM_STATUS_OK), false);
} }
else else
@ -422,7 +418,7 @@ namespace client
i2p::data::netdb.RequestDestination (ident, true, m_Session->localDestination->GetTunnelPool ()); i2p::data::netdb.RequestDestination (ident, true, m_Session->localDestination->GetTunnelPool ());
m_Timer.expires_from_now (boost::posix_time::seconds(SAM_NAMING_LOOKUP_TIMEOUT)); m_Timer.expires_from_now (boost::posix_time::seconds(SAM_NAMING_LOOKUP_TIMEOUT));
m_Timer.async_wait (boost::bind (&SAMSocket::HandleNamingLookupDestinationRequestTimer, m_Timer.async_wait (boost::bind (&SAMSocket::HandleNamingLookupDestinationRequestTimer,
this, boost::asio::placeholders::error, ident)); shared_from_this (), boost::asio::placeholders::error, ident));
} }
} }
else else
@ -475,7 +471,7 @@ namespace client
{ {
m_Socket.async_read_some (boost::asio::buffer(m_Buffer, SAM_SOCKET_BUFFER_SIZE), m_Socket.async_read_some (boost::asio::buffer(m_Buffer, SAM_SOCKET_BUFFER_SIZE),
boost::bind((m_SocketType == eSAMSocketTypeSession) ? &SAMSocket::HandleMessage : &SAMSocket::HandleReceived, boost::bind((m_SocketType == eSAMSocketTypeSession) ? &SAMSocket::HandleMessage : &SAMSocket::HandleReceived,
this, boost::asio::placeholders::error, boost::asio::placeholders::bytes_transferred)); shared_from_this (), boost::asio::placeholders::error, boost::asio::placeholders::bytes_transferred));
} }
void SAMSocket::HandleReceived (const boost::system::error_code& ecode, std::size_t bytes_transferred) void SAMSocket::HandleReceived (const boost::system::error_code& ecode, std::size_t bytes_transferred)
@ -498,7 +494,7 @@ namespace client
{ {
if (m_Stream) if (m_Stream)
m_Stream->AsyncReceive (boost::asio::buffer (m_StreamBuffer, SAM_SOCKET_BUFFER_SIZE), m_Stream->AsyncReceive (boost::asio::buffer (m_StreamBuffer, SAM_SOCKET_BUFFER_SIZE),
boost::bind (&SAMSocket::HandleI2PReceive, this, boost::bind (&SAMSocket::HandleI2PReceive, shared_from_this (),
boost::asio::placeholders::error, boost::asio::placeholders::bytes_transferred), boost::asio::placeholders::error, boost::asio::placeholders::bytes_transferred),
SAM_SOCKET_CONNECTION_MAX_IDLE); SAM_SOCKET_CONNECTION_MAX_IDLE);
} }
@ -514,7 +510,7 @@ namespace client
else else
{ {
boost::asio::async_write (m_Socket, boost::asio::buffer (m_StreamBuffer, bytes_transferred), boost::asio::async_write (m_Socket, boost::asio::buffer (m_StreamBuffer, bytes_transferred),
boost::bind (&SAMSocket::HandleWriteI2PData, this, boost::asio::placeholders::error)); boost::bind (&SAMSocket::HandleWriteI2PData, shared_from_this (), boost::asio::placeholders::error));
} }
} }
@ -567,7 +563,7 @@ namespace client
{ {
memcpy (m_StreamBuffer + l2, buf, len); memcpy (m_StreamBuffer + l2, buf, len);
boost::asio::async_write (m_Socket, boost::asio::buffer (m_StreamBuffer, len + l2), boost::asio::async_write (m_Socket, boost::asio::buffer (m_StreamBuffer, len + l2),
boost::bind (&SAMSocket::HandleWriteI2PData, this, boost::asio::placeholders::error)); boost::bind (&SAMSocket::HandleWriteI2PData, shared_from_this (), boost::asio::placeholders::error));
} }
else else
LogPrint (eLogWarning, "Datagram size ", len," exceeds buffer"); LogPrint (eLogWarning, "Datagram size ", len," exceeds buffer");
@ -576,15 +572,13 @@ namespace client
SAMBridge::SAMBridge (int port): SAMBridge::SAMBridge (int port):
m_IsRunning (false), m_Thread (nullptr), m_IsRunning (false), m_Thread (nullptr),
m_Acceptor (m_Service, boost::asio::ip::tcp::endpoint(boost::asio::ip::tcp::v4(), port)), m_Acceptor (m_Service, boost::asio::ip::tcp::endpoint(boost::asio::ip::tcp::v4(), port)),
m_DatagramEndpoint (boost::asio::ip::udp::v4 (), port-1), m_DatagramSocket (m_Service, m_DatagramEndpoint), m_DatagramEndpoint (boost::asio::ip::udp::v4 (), port-1), m_DatagramSocket (m_Service, m_DatagramEndpoint)
m_NewSocket (nullptr)
{ {
} }
SAMBridge::~SAMBridge () SAMBridge::~SAMBridge ()
{ {
Stop (); Stop ();
delete m_NewSocket;
} }
void SAMBridge::Start () void SAMBridge::Start ()
@ -624,24 +618,20 @@ namespace client
void SAMBridge::Accept () void SAMBridge::Accept ()
{ {
m_NewSocket = new SAMSocket (*this); auto newSocket = std::make_shared<SAMSocket> (*this);
m_Acceptor.async_accept (m_NewSocket->GetSocket (), boost::bind (&SAMBridge::HandleAccept, this, m_Acceptor.async_accept (newSocket->GetSocket (), boost::bind (&SAMBridge::HandleAccept, this,
boost::asio::placeholders::error)); boost::asio::placeholders::error, newSocket));
} }
void SAMBridge::HandleAccept(const boost::system::error_code& ecode) void SAMBridge::HandleAccept(const boost::system::error_code& ecode, std::shared_ptr<SAMSocket> socket)
{ {
if (!ecode) if (!ecode)
{ {
LogPrint ("New SAM connection from ", m_NewSocket->GetSocket ().remote_endpoint ()); LogPrint ("New SAM connection from ", socket->GetSocket ().remote_endpoint ());
m_NewSocket->ReceiveHandshake (); socket->ReceiveHandshake ();
} }
else else
{
LogPrint ("SAM accept error: ", ecode.message ()); LogPrint ("SAM accept error: ", ecode.message ());
delete m_NewSocket;
m_NewSocket = nullptr;
}
if (ecode != boost::asio::error::operation_aborted) if (ecode != boost::asio::error::operation_aborted)
Accept (); Accept ();
@ -680,8 +670,6 @@ namespace client
auto it = m_Sessions.find (id); auto it = m_Sessions.find (id);
if (it != m_Sessions.end ()) if (it != m_Sessions.end ())
{ {
for (auto it1 : it->second.sockets)
delete it1;
it->second.sockets.clear (); it->second.sockets.clear ();
it->second.localDestination->Stop (); it->second.localDestination->Stop ();
m_Sessions.erase (it); m_Sessions.erase (it);

10
SAM.h
Unescape View File

@ -7,6 +7,7 @@
#include <list> #include <list>
#include <thread> #include <thread>
#include <mutex> #include <mutex>
#include <memory>
#include <boost/asio.hpp> #include <boost/asio.hpp>
#include "Identity.h" #include "Identity.h"
#include "LeaseSet.h" #include "LeaseSet.h"
@ -39,7 +40,7 @@ namespace client
const char SAM_DEST_REPLY_I2P_ERROR[] = "DEST REPLY RESULT=I2P_ERROR\n"; const char SAM_DEST_REPLY_I2P_ERROR[] = "DEST REPLY RESULT=I2P_ERROR\n";
const char SAM_NAMING_LOOKUP[] = "NAMING LOOKUP"; const char SAM_NAMING_LOOKUP[] = "NAMING LOOKUP";
const char SAM_NAMING_REPLY[] = "NAMING REPLY RESULT=OK NAME=ME VALUE=%s\n"; const char SAM_NAMING_REPLY[] = "NAMING REPLY RESULT=OK NAME=ME VALUE=%s\n";
const char SAM_DATAGRAM_RECEIVED[] = "DATAGRAM_RECEIVED DESTINATION=%s SIZE=%i\n"; const char SAM_DATAGRAM_RECEIVED[] = "DATAGRAM_RECEIVED DESTINATION=%s SIZE=%lu\n";
const char SAM_NAMING_REPLY_INVALID_KEY[] = "NAMING REPLY RESULT=INVALID_KEY NAME=%s\n"; const char SAM_NAMING_REPLY_INVALID_KEY[] = "NAMING REPLY RESULT=INVALID_KEY NAME=%s\n";
const char SAM_NAMING_REPLY_KEY_NOT_FOUND[] = "NAMING REPLY RESULT=INVALID_KEY_NOT_FOUND NAME=%s\n"; const char SAM_NAMING_REPLY_KEY_NOT_FOUND[] = "NAMING REPLY RESULT=INVALID_KEY_NOT_FOUND NAME=%s\n";
const char SAM_PARAM_STYLE[] = "STYLE"; const char SAM_PARAM_STYLE[] = "STYLE";
@ -64,7 +65,7 @@ namespace client
class SAMBridge; class SAMBridge;
class SAMSession; class SAMSession;
class SAMSocket class SAMSocket: public std::enable_shared_from_this<SAMSocket>
{ {
public: public:
@ -122,7 +123,7 @@ namespace client
struct SAMSession struct SAMSession
{ {
ClientDestination * localDestination; ClientDestination * localDestination;
std::list<SAMSocket *> sockets; std::list<std::shared_ptr<SAMSocket> > sockets;
}; };
class SAMBridge class SAMBridge
@ -145,7 +146,7 @@ namespace client
void Run (); void Run ();
void Accept (); void Accept ();
void HandleAccept(const boost::system::error_code& ecode); void HandleAccept(const boost::system::error_code& ecode, std::shared_ptr<SAMSocket> socket);
void ReceiveDatagram (); void ReceiveDatagram ();
void HandleReceivedDatagram (const boost::system::error_code& ecode, std::size_t bytes_transferred); void HandleReceivedDatagram (const boost::system::error_code& ecode, std::size_t bytes_transferred);
@ -158,7 +159,6 @@ namespace client
boost::asio::ip::tcp::acceptor m_Acceptor; boost::asio::ip::tcp::acceptor m_Acceptor;
boost::asio::ip::udp::endpoint m_DatagramEndpoint, m_SenderEndpoint; boost::asio::ip::udp::endpoint m_DatagramEndpoint, m_SenderEndpoint;
boost::asio::ip::udp::socket m_DatagramSocket; boost::asio::ip::udp::socket m_DatagramSocket;
SAMSocket * m_NewSocket;
std::mutex m_SessionsMutex; std::mutex m_SessionsMutex;
std::map<std::string, SAMSession> m_Sessions; std::map<std::string, SAMSession> m_Sessions;
uint8_t m_DatagramReceiveBuffer[i2p::datagram::MAX_DATAGRAM_SIZE+1]; uint8_t m_DatagramReceiveBuffer[i2p::datagram::MAX_DATAGRAM_SIZE+1];

Write Preview
Loading…
Cancel
Save