sockethandler.cpp.svn-base

絲路server源碼 Silk Road server source

					else
#endif
					{
						p -> OnRead();
					}
				}
				else
				{
					LogError(NULL, "GetSocket/handler/1", (int)i, "Did not find expected socket using file descriptor", LOG_LEVEL_WARNING);
				}
				n--;
			}
			if (FD_ISSET(i, &wfds))
			{
				socket_m::iterator itmp = m_sockets.find(i);
				if (itmp != m_sockets.end()) // found
				{
					Socket *p = itmp -> second;
					// new SSL negotiate method
#ifdef HAVE_OPENSSL
					if (p -> IsSSLNegotiate())
					{
						p -> SSLNegotiate();
					}
					else
#endif
					{
						p -> OnWrite();
					}
				}
				else
				{
					LogError(NULL, "GetSocket/handler/2", (int)i, "Did not find expected socket using file descriptor", LOG_LEVEL_WARNING);
				}
				n--;
			}
			if (FD_ISSET(i, &efds))
			{
				socket_m::iterator itmp = m_sockets.find(i);
				if (itmp != m_sockets.end()) // found
				{
					Socket *p = itmp -> second;
					p -> OnException();
				}
				else
				{
					LogError(NULL, "GetSocket/handler/3", (int)i, "Did not find expected socket using file descriptor", LOG_LEVEL_WARNING);
				}
				n--;
			}
		} // m_fds loop
		m_preverror = -1;
	} // if (n > 0)

	// check CallOnConnect - EVENT
	if (!m_fds_callonconnect.empty())
	{
		socket_v tmp = m_fds_callonconnect;
		for (socket_v::iterator it = tmp.begin(); it != tmp.end(); it++)
		{
			Socket *p = NULL;
			{
				socket_m::iterator itmp = m_sockets.find(*it);
				if (itmp != m_sockets.end()) // found
				{
					p = itmp -> second;
				}
				else
				{
					LogError(NULL, "GetSocket/handler/4", (int)*it, "Did not find expected socket using file descriptor", LOG_LEVEL_WARNING);
				}
			}
			if (p)
			{
//				if (p -> CallOnConnect() && p -> Ready() )
				{
					p -> SetConnected(); // moved here from inside if (tcp) check below
#ifdef HAVE_OPENSSL
					if (p -> IsSSL()) // SSL Enabled socket
						p -> OnSSLConnect();
					else
#endif
#ifdef ENABLE_SOCKS4
					if (p -> Socks4())
						p -> OnSocks4Connect();
					else
#endif
					{
						TcpSocket *tcp = dynamic_cast<TcpSocket *>(p);
						if (tcp)
						{
							if (tcp -> GetOutputLength())
							{
								p -> OnWrite();
							}
						}
#ifdef ENABLE_RECONNECT
						if (tcp && tcp -> IsReconnect())
							p -> OnReconnect();
						else
#endif
						{
//							LogError(p, "Calling OnConnect", 0, "Because CallOnConnect", LOG_LEVEL_INFO);
							p -> OnConnect();
						}
					}
//					p -> SetCallOnConnect( false );
					AddList(p -> GetSocket(), LIST_CALLONCONNECT, false);
				}
			}
		}
	}
#ifdef ENABLE_DETACH
	// check detach of socket if master handler - EVENT
	if (!m_slave && !m_fds_detach.empty())
	{
		// %! why not using tmp list here??!?
		for (socket_v::iterator it = m_fds_detach.begin(); it != m_fds_detach.end(); it++)
		{
			Socket *p = NULL;
			{
				socket_m::iterator itmp = m_sockets.find(*it);
				if (itmp != m_sockets.end()) // found
				{
					p = itmp -> second;
				}
				else
				{
					LogError(NULL, "GetSocket/handler/5", (int)*it, "Did not find expected socket using file descriptor", LOG_LEVEL_WARNING);
				}
			}
			if (p)
			{
//				if (p -> IsDetach())
				{
					Set(p -> GetSocket(), false, false, false);
					// After DetachSocket(), all calls to Handler() will return a reference
					// to the new slave SocketHandler running in the new thread.
					p -> DetachSocket();
					// Adding the file descriptor to m_fds_erase will now also remove the
					// socket from the detach queue - tnx knightmad
					m_fds_erase.push_back(p -> GetSocket());
				}
			}
		}
	}
#endif
	// check Connecting - connection timeout - conditional event
	if (m_fds_timeout.size())
	{
		time_t tnow = time(NULL);
		if (tnow != m_tlast)
		{
			socket_v tmp = m_fds_timeout;
DEB(			fprintf(stderr, "Checking %d socket(s) for timeout\n", tmp.size());)
			for (socket_v::iterator it = tmp.begin(); it != tmp.end(); it++)
			{
				Socket *p = NULL;
				{
					socket_m::iterator itmp = m_sockets.find(*it);
					if (itmp != m_sockets.end()) // found
					{
						p = itmp -> second;
					}
					else
					{
						itmp = m_add.find(*it);
						if (itmp != m_add.end())
						{
							p = itmp -> second;
						}
						else
						{
							LogError(NULL, "GetSocket/handler/6", (int)*it, "Did not find expected socket using file descriptor", LOG_LEVEL_WARNING);
						}
					}
				}
				if (p)
				{
					if (p -> Timeout(tnow))
					{
						StreamSocket *scp = dynamic_cast<StreamSocket *>(p);
						if (scp && scp -> Connecting())
							p -> OnConnectTimeout();
						else
							p -> OnTimeout();
						p -> SetTimeout(0);
					}
				}
			}
			m_tlast = tnow;
		} // tnow != tlast
	}
	// check retry client connect - EVENT
	if (!m_fds_retry.empty())
	{
		socket_v tmp = m_fds_retry;
		for (socket_v::iterator it = tmp.begin(); it != tmp.end(); it++)
		{
			Socket *p = NULL;
			{
				socket_m::iterator itmp = m_sockets.find(*it);
				if (itmp != m_sockets.end()) // found
				{
					p = itmp -> second;
				}
				else
				{
					LogError(NULL, "GetSocket/handler/7", (int)*it, "Did not find expected socket using file descriptor", LOG_LEVEL_WARNING);
				}
			}
			if (p)
			{
//				if (p -> RetryClientConnect())
				{
					TcpSocket *tcp = dynamic_cast<TcpSocket *>(p);
					SOCKET nn = *it; //(*it3).first;
					tcp -> SetRetryClientConnect(false);
DEB(					fprintf(stderr, "Close() before retry client connect\n");)
					p -> Close(); // removes from m_fds_retry
					std::auto_ptr<SocketAddress> ad = p -> GetClientRemoteAddress();
					if (ad.get())
					{
						tcp -> Open(*ad);
					}
					else
					{
						LogError(p, "RetryClientConnect", 0, "no address", LOG_LEVEL_ERROR);
					}
					Add(p);
					m_fds_erase.push_back(nn);
				}
			}
		}
	}
	// check close and delete - conditional event
	if (!m_fds_close.empty())
	{
		socket_v tmp = m_fds_close;
DEB(		fprintf(stderr, "m_fds_close.size() == %d\n", (int)m_fds_close.size());)
		for (socket_v::iterator it = tmp.begin(); it != tmp.end(); it++)
		{
			Socket *p = NULL;
			{
				socket_m::iterator itmp = m_sockets.find(*it);
				if (itmp != m_sockets.end()) // found
				{
					p = itmp -> second;
				}
				else
				{
					itmp = m_add.find(*it);
					if (itmp != m_add.end())
					{
						p = itmp -> second;
					}
					else
					{
						LogError(NULL, "GetSocket/handler/8", (int)*it, "Did not find expected socket using file descriptor", LOG_LEVEL_WARNING);
					}
				}
			}
			if (p)
			{
//				if (p -> CloseAndDelete() )
				{
					TcpSocket *tcp = dynamic_cast<TcpSocket *>(p);
					// new graceful tcp - flush and close timeout 5s
					if (tcp && p -> IsConnected() && tcp -> GetFlushBeforeClose() && 
#ifdef HAVE_OPENSSL
						!tcp -> IsSSL() && 
#endif
						p -> TimeSinceClose() < 5)
					{
DEB(						fprintf(stderr, " close(1)\n");)
						if (tcp -> GetOutputLength())
						{
							LogError(p, "Closing", (int)tcp -> GetOutputLength(), "Sending all data before closing", LOG_LEVEL_INFO);
						}
						else // shutdown write when output buffer is empty
						if (!(tcp -> GetShutdown() & SHUT_WR))
						{
							SOCKET nn = *it;
							if (nn != INVALID_SOCKET && shutdown(nn, SHUT_WR) == -1)
							{
								LogError(p, "graceful shutdown", Errno, StrError(Errno), LOG_LEVEL_ERROR);
							}
							tcp -> SetShutdown(SHUT_WR);
						}
					}
					else
#ifdef ENABLE_RECONNECT
					if (tcp && p -> IsConnected() && tcp -> Reconnect())
					{
						SOCKET nn = *it; //(*it3).first;
DEB(						fprintf(stderr, " close(2) fd %d\n", nn);)
						p -> SetCloseAndDelete(false);
						tcp -> SetIsReconnect();
						p -> SetConnected(false);
DEB(						fprintf(stderr, "Close() before reconnect\n");)
						p -> Close(); // dispose of old file descriptor (Open creates a new)
						p -> OnDisconnect();
						std::auto_ptr<SocketAddress> ad = p -> GetClientRemoteAddress();
						if (ad.get())
						{
							tcp -> Open(*ad);
						}
						else
						{
							LogError(p, "Reconnect", 0, "no address", LOG_LEVEL_ERROR);
						}
						tcp -> ResetConnectionRetries();
						Add(p);
						m_fds_erase.push_back(nn);
					}
					else
#endif
					{
						SOCKET nn = *it; //(*it3).first;
DEB(						fprintf(stderr, " close(3) fd %d GetSocket() %d\n", nn, p -> GetSocket());)
						if (tcp && p -> IsConnected() && tcp -> GetOutputLength())
						{
							LogError(p, "Closing", (int)tcp -> GetOutputLength(), "Closing socket while data still left to send", LOG_LEVEL_WARNING);
						}
#ifdef ENABLE_POOL
						if (p -> Retain() && !p -> Lost())
						{
							PoolSocket *p2 = new PoolSocket(*this, p);
							p2 -> SetDeleteByHandler();
							Add(p2);
							//
							p -> SetCloseAndDelete(false); // added - remove from m_fds_close
						}
						else
#endif // ENABLE_POOL
						{
							Set(p -> GetSocket(),false,false,false);
DEB(							fprintf(stderr, "Close() before OnDelete\n");)
							p -> Close();
						}
						p -> OnDelete();
						if (p -> DeleteByHandler())
						{
							p -> SetErasedByHandler();
						}
						m_fds_erase.push_back(nn);
					}
				}
			}
		}
	}

	// check erased sockets
	bool check_max_fd = false;
	while (!m_fds_erase.empty())
	{
		socket_v::iterator it = m_fds_erase.begin();
		SOCKET nn = *it;
#ifdef ENABLE_DETACH
		{
			for (socket_v::iterator it = m_fds_detach.begin(); it != m_fds_detach.end(); it++)
			{
				if (*it == nn)
				{
					m_fds_detach.erase(it);
					break;
				}
			}
		}
#endif
		{
			for (socket_v::iterator it = m_fds.begin(); it != m_fds.end(); it++)
			{
				if (*it == nn)
				{
					m_fds.erase(it);
					break;
				}
			}
		}
		{
			socket_m::iterator it = m_sockets.find(nn);
			if (it != m_sockets.end())
			{
				Socket *p = it -> second;
				/* Sometimes a SocketThread class can finish its run before the master
				   sockethandler gets here. In that case, the SocketThread has set the
				   'ErasedByHandler' flag on the socket which will make us end up with a
				   double delete on the socket instance. 
				   The fix is to make sure that the master sockethandler only can delete
				   non-detached sockets, and a slave sockethandler only can delete
				   detach sockets. */
				if (p -> ErasedByHandler()
#ifdef ENABLE_DETACH
					&& !(m_slave ^ p -> IsDetached()) 
#endif
					)
				{
#ifdef ENABLE_TRIGGERS
					bool again = false;
					do
					{
						again = false;
						for (std::map<int, Socket *>::iterator it = m_trigger_src.begin(); it != m_trigger_src.end(); it++)
						{
							int id = it -> first;
							Socket *src = it -> second;
							if (src == p)
							{
								for (std::map<Socket *, bool>::iterator it = m_trigger_dst[id].begin(); it != m_trigger_dst[id].end(); it++)
								{
									Socket *dst = it -> first;
									if (Valid(dst))
									{
										dst -> OnCancelled(id);
									}
								}
								m_trigger_src.erase(m_trigger_src.find(id));
								m_trigger_dst.erase(m_trigger_dst.find(id));
								again = true;
								break;
							}
						}
					} while (again);
#endif
					delete p;
				}
				m_sockets.erase(it);
			}
		}
		m_fds_erase.erase(it);
		check_max_fd = true;
	}
	// calculate max file descriptor for select() call
	if (check_max_fd)
	{
		m_maxsock = 0;
		for (socket_v::iterator it = m_fds.begin(); it != m_fds.end(); it++)
		{
			SOCKET s = *it;
			m_maxsock = s > m_maxsock ? s : m_maxsock;
		}
	}
	// remove Add's that fizzed
	while (!m_delete.empty())
	{
		std::list<Socket *>::iterator it = m_delete.begin();
		Socket *p = *it;
		p -> OnDelete();
		m_delete.erase(it);
		if (p -> DeleteByHandler()
#ifdef ENABLE_DETACH
			&& !(m_slave ^ p -> IsDetached()) 
#endif
			)
		{
			p -> SetErasedByHandler();
#ifdef ENABLE_TRIGGERS
			bool again = false;
			do
			{
				again = false;
				for (std::map<int, Socket *>::iterator it = m_trigger_src.begin(); it != m_trigger_src.end(); it++)
				{
					int id = it -> first;
					Socket *src = it -> second;
					if (src == p)
					{
						for (std::map<Socket *, bool>::iterator it = m_trigger_dst[id].begin(); it != m_trigger_dst[id].end(); it++)
						{
							Socket *dst = it -> first;
							if (Valid(dst))
							{
								dst -> OnCancelled(id);
							}