From 21eb5adadbe3110a8708f2570185566e1f137a49 Mon Sep 17 00:00:00 2001 From: Gavin Andresen Date: Wed, 6 Mar 2013 22:31:26 -0500 Subject: Port Thread* methods to boost::thread_group --- src/bitcoinrpc.cpp | 136 ++++++++++++++++------------------------------------- 1 file changed, 40 insertions(+), 96 deletions(-) (limited to 'src/bitcoinrpc.cpp') diff --git a/src/bitcoinrpc.cpp b/src/bitcoinrpc.cpp index c7219920ca..53b3635753 100644 --- a/src/bitcoinrpc.cpp +++ b/src/bitcoinrpc.cpp @@ -30,13 +30,12 @@ using namespace boost; using namespace boost::asio; using namespace json_spirit; -void ThreadRPCServer2(void* parg); - static std::string strRPCUserColonPass; -const Object emptyobj; - -void ThreadRPCServer3(void* parg); +// These are created by StartRPCThreads, destroyed in StopRPCThreads +static asio::io_service* rpc_io_service = NULL; +static ssl::context* rpc_ssl_context = NULL; +static boost::thread_group* rpc_worker_group = NULL; static inline unsigned short GetDefaultRPCPort() { @@ -650,26 +649,7 @@ private: iostreams::stream< SSLIOStreamDevice > _stream; }; -void ThreadRPCServer(void* parg) -{ - // Make this thread recognisable as the RPC listener - RenameThread("bitcoin-rpclist"); - - try - { - vnThreadsRunning[THREAD_RPCLISTENER]++; - ThreadRPCServer2(parg); - vnThreadsRunning[THREAD_RPCLISTENER]--; - } - catch (std::exception& e) { - vnThreadsRunning[THREAD_RPCLISTENER]--; - PrintException(&e, "ThreadRPCServer()"); - } catch (...) { - vnThreadsRunning[THREAD_RPCLISTENER]--; - PrintException(NULL, "ThreadRPCServer()"); - } - printf("ThreadRPCServer exited\n"); -} +void ServiceConnection(AcceptedConnection *conn); // Forward declaration required for RPCListen template @@ -711,11 +691,8 @@ static void RPCAcceptHandler(boost::shared_ptr< basic_socket_acceptoris_open()) + if (error != asio::error::operation_aborted && acceptor->is_open()) RPCListen(acceptor, context, fUseSSL); AcceptedConnectionImpl* tcp_conn = dynamic_cast< AcceptedConnectionImpl* >(conn); @@ -729,28 +706,22 @@ static void RPCAcceptHandler(boost::shared_ptr< basic_socket_acceptorpeer.address())) + else if (tcp_conn && !ClientAllowed(tcp_conn->peer.address())) { // Only send a 403 if we're not using SSL to prevent a DoS during the SSL handshake. if (!fUseSSL) conn->stream() << HTTPReply(HTTP_FORBIDDEN, "", false) << std::flush; delete conn; } - - // start HTTP client thread - else if (!NewThread(ThreadRPCServer3, conn)) { - printf("Failed to create RPC server client thread\n"); + else { + ServiceConnection(conn); + conn->close(); delete conn; } - - vnThreadsRunning[THREAD_RPCLISTENER]--; } -void ThreadRPCServer2(void* parg) +void StartRPCThreads() { - printf("ThreadRPCServer started\n"); - strRPCUserColonPass = mapArgs["-rpcuser"] + ":" + mapArgs["-rpcpassword"]; if ((mapArgs["-rpcpassword"] == "") || (mapArgs["-rpcuser"] == mapArgs["-rpcpassword"])) @@ -781,27 +752,28 @@ void ThreadRPCServer2(void* parg) return; } - const bool fUseSSL = GetBoolArg("-rpcssl"); + assert(rpc_io_service == NULL); + rpc_io_service = new asio::io_service(); + rpc_ssl_context = new ssl::context(*rpc_io_service, ssl::context::sslv23); - asio::io_service io_service; + const bool fUseSSL = GetBoolArg("-rpcssl"); - ssl::context context(io_service, ssl::context::sslv23); if (fUseSSL) { - context.set_options(ssl::context::no_sslv2); + rpc_ssl_context->set_options(ssl::context::no_sslv2); filesystem::path pathCertFile(GetArg("-rpcsslcertificatechainfile", "server.cert")); if (!pathCertFile.is_complete()) pathCertFile = filesystem::path(GetDataDir()) / pathCertFile; - if (filesystem::exists(pathCertFile)) context.use_certificate_chain_file(pathCertFile.string()); + if (filesystem::exists(pathCertFile)) rpc_ssl_context->use_certificate_chain_file(pathCertFile.string()); else printf("ThreadRPCServer ERROR: missing server certificate file %s\n", pathCertFile.string().c_str()); filesystem::path pathPKFile(GetArg("-rpcsslprivatekeyfile", "server.pem")); if (!pathPKFile.is_complete()) pathPKFile = filesystem::path(GetDataDir()) / pathPKFile; - if (filesystem::exists(pathPKFile)) context.use_private_key_file(pathPKFile.string(), ssl::context::pem); + if (filesystem::exists(pathPKFile)) rpc_ssl_context->use_private_key_file(pathPKFile.string(), ssl::context::pem); else printf("ThreadRPCServer ERROR: missing server private key file %s\n", pathPKFile.string().c_str()); string strCiphers = GetArg("-rpcsslciphers", "TLSv1+HIGH:!SSLv2:!aNULL:!eNULL:!AH:!3DES:@STRENGTH"); - SSL_CTX_set_cipher_list(context.impl(), strCiphers.c_str()); + SSL_CTX_set_cipher_list(rpc_ssl_context->impl(), strCiphers.c_str()); } // Try a dual IPv6/IPv4 socket, falling back to separate IPv4 and IPv6 sockets @@ -809,9 +781,7 @@ void ThreadRPCServer2(void* parg) asio::ip::address bindAddress = loopback ? asio::ip::address_v6::loopback() : asio::ip::address_v6::any(); ip::tcp::endpoint endpoint(bindAddress, GetArg("-rpcport", GetDefaultRPCPort())); boost::system::error_code v6_only_error; - boost::shared_ptr acceptor(new ip::tcp::acceptor(io_service)); - - boost::signals2::signal StopRequests; + boost::shared_ptr acceptor(new ip::tcp::acceptor(*rpc_io_service)); bool fListening = false; std::string strerr; @@ -826,11 +796,7 @@ void ThreadRPCServer2(void* parg) acceptor->bind(endpoint); acceptor->listen(socket_base::max_connections); - RPCListen(acceptor, context, fUseSSL); - // Cancel outstanding listen-requests for this acceptor when shutting down - StopRequests.connect(signals2::slot( - static_cast(&ip::tcp::acceptor::close), acceptor.get()) - .track(acceptor)); + RPCListen(acceptor, *rpc_ssl_context, fUseSSL); fListening = true; } @@ -846,17 +812,13 @@ void ThreadRPCServer2(void* parg) bindAddress = loopback ? asio::ip::address_v4::loopback() : asio::ip::address_v4::any(); endpoint.address(bindAddress); - acceptor.reset(new ip::tcp::acceptor(io_service)); + acceptor.reset(new ip::tcp::acceptor(*rpc_io_service)); acceptor->open(endpoint.protocol()); acceptor->set_option(boost::asio::ip::tcp::acceptor::reuse_address(true)); acceptor->bind(endpoint); acceptor->listen(socket_base::max_connections); - RPCListen(acceptor, context, fUseSSL); - // Cancel outstanding listen-requests for this acceptor when shutting down - StopRequests.connect(signals2::slot( - static_cast(&ip::tcp::acceptor::close), acceptor.get()) - .track(acceptor)); + RPCListen(acceptor, *rpc_ssl_context, fUseSSL); fListening = true; } @@ -872,11 +834,20 @@ void ThreadRPCServer2(void* parg) return; } - vnThreadsRunning[THREAD_RPCLISTENER]--; - while (!fShutdown) - io_service.run_one(); - vnThreadsRunning[THREAD_RPCLISTENER]++; - StopRequests(); + rpc_worker_group = new boost::thread_group(); + for (int i = 0; i < GetArg("-rpcthreads", 4); i++) + rpc_worker_group->create_thread(boost::bind(&asio::io_service::run, rpc_io_service)); +} + +void StopRPCThreads() +{ + if (rpc_io_service == NULL) return; + + rpc_io_service->stop(); + rpc_worker_group->join_all(); + delete rpc_worker_group; rpc_worker_group = NULL; + delete rpc_ssl_context; rpc_ssl_context = NULL; + delete rpc_io_service; rpc_io_service = NULL; } class JSONRequest @@ -953,32 +924,11 @@ static string JSONRPCExecBatch(const Array& vReq) return write_string(Value(ret), false) + "\n"; } -static CCriticalSection cs_THREAD_RPCHANDLER; - -void ThreadRPCServer3(void* parg) +void ServiceConnection(AcceptedConnection *conn) { - // Make this thread recognisable as the RPC handler - RenameThread("bitcoin-rpchand"); - - { - LOCK(cs_THREAD_RPCHANDLER); - vnThreadsRunning[THREAD_RPCHANDLER]++; - } - AcceptedConnection *conn = (AcceptedConnection *) parg; - bool fRun = true; - loop { - if (fShutdown || !fRun) - { - conn->close(); - delete conn; - { - LOCK(cs_THREAD_RPCHANDLER); - --vnThreadsRunning[THREAD_RPCHANDLER]; - } - return; - } - + while (fRun) + { int nProto = 0; map mapHeaders; string strRequest, strMethod, strURI; @@ -1049,12 +999,6 @@ void ThreadRPCServer3(void* parg) break; } } - - delete conn; - { - LOCK(cs_THREAD_RPCHANDLER); - vnThreadsRunning[THREAD_RPCHANDLER]--; - } } json_spirit::Value CRPCTable::execute(const std::string &strMethod, const json_spirit::Array ¶ms) const -- cgit v1.2.3