X-Git-Url: http://git.indexdata.com/?a=blobdiff_plain;f=src%2Ffilter_frontend_net.cpp;h=2e2ada228645ba4bf1371bf8f92eaa870bc15bfa;hb=1fe3038c15740cf21a2eb7b00edad8d216384c2e;hp=c744a66511432556bdabca2e70fe6d3eaa50d60b;hpb=ffe725e290246877e53874354229cd340a02de8e;p=metaproxy-moved-to-github.git diff --git a/src/filter_frontend_net.cpp b/src/filter_frontend_net.cpp index c744a66..2e2ada2 100644 --- a/src/filter_frontend_net.cpp +++ b/src/filter_frontend_net.cpp @@ -18,6 +18,9 @@ Foundation, Inc., 51 Franklin St, Fifth Floor, Boston, MA 02110-1301 USA #include "config.hpp" +#if HAVE_GETRLIMIT +#include +#endif #include #include #include @@ -55,14 +58,18 @@ namespace metaproxy_1 { friend class FrontendNet; int m_no_threads; + int m_max_threads; + int m_stack_size; std::vector m_ports; int m_listen_duration; int m_session_timeout; int m_connect_max; + int m_http_req_max; std::string m_msg_config; std::string m_stat_req; yazpp_1::SocketManager mySocketManager; ZAssocServer **az; + yazpp_1::PDU_Assoc **pdu; int m_duration_freq[22]; double m_duration_lim[22]; double m_duration_max; @@ -89,10 +96,11 @@ namespace metaproxy_1 { ZAssocChild(yazpp_1::IPDU_Observable *the_PDU_Observable, mp::ThreadPoolSocketObserver *m_thread_pool_observer, const mp::Package *package, - std::string route, - Rep *rep); + Port *port, + Rep *rep, + yazpp_1::LimitConnect &limit_connect); int m_no_requests; - std::string m_route; + Port *m_port; private: yazpp_1::IPDU_Observer* sessionNotify( yazpp_1::IPDU_Observable *the_PDU_Observable, @@ -109,6 +117,7 @@ namespace metaproxy_1 { bool m_delete_flag; const mp::Package *m_package; Rep *m_p; + yazpp_1::LimitConnect &m_limit_connect; }; class FrontendNet::ThreadPoolPackage : public mp::IThreadPoolMsg { public: @@ -129,7 +138,7 @@ namespace metaproxy_1 { public: ~ZAssocServer(); ZAssocServer(yazpp_1::IPDU_Observable *PDU_Observable, - std::string route, + FrontendNet::Port *port, Rep *rep); void set_package(const mp::Package *package); void set_thread_pool(ThreadPoolSocketObserver *observer); @@ -146,7 +155,7 @@ namespace metaproxy_1 { mp::ThreadPoolSocketObserver *m_thread_pool_observer; const mp::Package *m_package; yazpp_1::LimitConnect limit_connect; - std::string m_route; + Port *m_port; Rep *m_p; }; } @@ -258,7 +267,7 @@ void yf::FrontendNet::ThreadPoolPackage::result(const char *t_info) mp::IThreadPoolMsg *yf::FrontendNet::ThreadPoolPackage::handle() { - m_package->move(m_assoc_child->m_route); + m_package->move(m_assoc_child->m_port->route); return this; } @@ -266,14 +275,15 @@ yf::FrontendNet::ZAssocChild::ZAssocChild( yazpp_1::IPDU_Observable *PDU_Observable, mp::ThreadPoolSocketObserver *my_thread_pool, const mp::Package *package, - std::string route, Rep *rep) - : Z_Assoc(PDU_Observable), m_p(rep) + Port *port, Rep *rep, + yazpp_1::LimitConnect &limit_connect) + : Z_Assoc(PDU_Observable), m_p(rep), m_limit_connect(limit_connect) { m_thread_pool_observer = my_thread_pool; m_no_requests = 0; m_delete_flag = false; m_package = package; - m_route = route; + m_port = port; const char *peername = PDU_Observable->getpeername(); if (!peername) peername = "unknown"; @@ -283,7 +293,11 @@ yf::FrontendNet::ZAssocChild::ZAssocChild( if (cp) peername = cp + 1; } - m_origin.set_tcpip_address(std::string(peername), m_session.id()); + std::string addr; + addr.append(peername); + addr.append(" "); + addr.append(port->port); + m_origin.set_tcpip_address(addr, m_session.id()); timeout(m_p->m_session_timeout); } @@ -382,6 +396,20 @@ void yf::FrontendNet::ZAssocChild::recv_GDU(Z_GDU *z_pdu, int len) report(hreq); return; } + std::string peername = p->origin().get_address(); + + m_limit_connect.add_connect(peername.c_str()); + m_limit_connect.cleanup(false); + int con_sz = m_limit_connect.get_total(peername.c_str()); + + if (m_p->m_http_req_max && con_sz >= m_p->m_http_req_max) + { + mp::odr o; + Z_GDU *gdu_res = o.create_HTTP_Response(m_session, hreq, 500); + int len; + send_GDU(gdu_res, &len); + return; + } } ThreadPoolPackage *tp = new ThreadPoolPackage(p, this, m_p); @@ -436,10 +464,10 @@ void yf::FrontendNet::ZAssocChild::connectNotify() yf::FrontendNet::ZAssocServer::ZAssocServer( yazpp_1::IPDU_Observable *PDU_Observable, - std::string route, + Port *port, Rep *rep) : - Z_Assoc(PDU_Observable), m_route(route), m_p(rep) + Z_Assoc(PDU_Observable), m_port(port), m_p(rep) { m_package = 0; } @@ -471,7 +499,7 @@ yazpp_1::IPDU_Observer *yf::FrontendNet::ZAssocServer::sessionNotify( } ZAssocChild *my = new ZAssocChild(the_PDU_Observable, m_thread_pool_observer, - m_package, m_route, m_p); + m_package, m_port, m_p, limit_connect); return my; } @@ -501,7 +529,8 @@ yf::FrontendNet::FrontendNet() : m_p(new Rep) yf::FrontendNet::Rep::Rep() { - m_no_threads = 5; + m_max_threads = m_no_threads = 5; + m_stack_size = 0; m_listen_duration = 0; m_session_timeout = 300; // 5 minutes m_connect_max = 0; @@ -545,6 +574,7 @@ yf::FrontendNet::Rep::~Rep() for (i = 0; i < m_ports.size(); i++) delete az[i]; delete [] az; + delete [] pdu; } az = 0; } @@ -558,6 +588,16 @@ void yf::FrontendNet::stop(int signo) const m_p->m_stop_signo = signo; } +void yf::FrontendNet::start() const +{ +#if HAVE_GETRLIMIT + struct rlimit limit_data; + getrlimit(RLIMIT_NOFILE, &limit_data); + yaz_log(YLOG_LOG, "getrlimit NOFILE cur=%ld max=%ld", + (long) limit_data.rlim_cur, (long) limit_data.rlim_max); +#endif +} + bool yf::FrontendNet::My_Timer_Thread::timeout() { return m_timeout; @@ -590,7 +630,9 @@ void yf::FrontendNet::process(mp::Package &package) const tt = new My_Timer_Thread(&m_p->mySocketManager, m_p->m_listen_duration); - ThreadPoolSocketObserver tp(&m_p->mySocketManager, m_p->m_no_threads); + ThreadPoolSocketObserver tp(&m_p->mySocketManager, m_p->m_no_threads, + m_p->m_max_threads, + m_p->m_stack_size); for (i = 0; im_ports.size(); i++) { @@ -602,6 +644,16 @@ void yf::FrontendNet::process(mp::Package &package) const if (m_p->m_stop_signo == SIGTERM) { yaz_log(YLOG_LOG, "metaproxy received SIGTERM"); + if (m_p->az) + { + size_t i; + for (i = 0; i < m_p->m_ports.size(); i++) + { + m_p->pdu[i]->shutdown(); + m_p->az[i]->server(""); + } + yaz_daemon_stop(); + } break; /* stop right away */ } #ifndef WIN32 @@ -669,6 +721,24 @@ void yf::FrontendNet::configure(const xmlNode * ptr, bool test_only, + threads_str); m_p->m_no_threads = threads; } + else if (!strcmp((const char *) ptr->name, "max-threads")) + { + std::string threads_str = mp::xml::get_text(ptr); + int threads = atoi(threads_str.c_str()); + if (threads < 1) + throw yf::FilterException("Bad value for max-threads: " + + threads_str); + m_p->m_max_threads = threads; + } + else if (!strcmp((const char *) ptr->name, "stack-size")) + { + std::string sz_str = mp::xml::get_text(ptr); + int sz = atoi(sz_str.c_str()); + if (sz < 0) + throw yf::FilterException("Bad value for stack-size: " + + sz_str); + m_p->m_stack_size = sz * 1024; + } else if (!strcmp((const char *) ptr->name, "timeout")) { std::string timeout_str = mp::xml::get_text(ptr); @@ -682,6 +752,10 @@ void yf::FrontendNet::configure(const xmlNode * ptr, bool test_only, { m_p->m_connect_max = mp::xml::get_int(ptr, 0); } + else if (!strcmp((const char *) ptr->name, "http-req-max")) + { + m_p->m_http_req_max = mp::xml::get_int(ptr, 0); + } else if (!strcmp((const char *) ptr->name, "message")) { m_p->m_msg_config = mp::xml::get_text(ptr); @@ -724,6 +798,7 @@ void yf::FrontendNet::set_ports(std::vector &ports) m_p->m_ports = ports; m_p->az = new yf::FrontendNet::ZAssocServer *[m_p->m_ports.size()]; + m_p->pdu = new yazpp_1::PDU_Assoc *[m_p->m_ports.size()]; // Create yf::FrontendNet::ZAssocServer for each port size_t i; @@ -737,8 +812,9 @@ void yf::FrontendNet::set_ports(std::vector &ports) if (m_p->m_ports[i].cert_fname.length()) as->set_cert_fname(m_p->m_ports[i].cert_fname.c_str()); // create ZAssoc with PDU Assoc + m_p->pdu[i] = as; m_p->az[i] = new yf::FrontendNet::ZAssocServer( - as, m_p->m_ports[i].route, m_p.get()); + as, &m_p->m_ports[i], m_p.get()); if (m_p->az[i]->server(m_p->m_ports[i].port.c_str())) { throw yf::FilterException("Unable to bind to address "