diff --git a/apps/examples/xmlrpc2di/Makefile b/apps/examples/xmlrpc2di/Makefile index 238e24cb..2110ef8f 100644 --- a/apps/examples/xmlrpc2di/Makefile +++ b/apps/examples/xmlrpc2di/Makefile @@ -3,7 +3,7 @@ XMLRPCPP_DIR = xmlrpc++0.7 plug_in_name = xmlrpc2di module_ldflags = -lxmlrpc++ -module_cflags = +module_cflags = -DHAVE_XMLRPCPP_SSL # for gentoo ebuild or cvs-20040713 version: # module_cflags = -DHAVE_XMLRPCPP_SSL # diff --git a/apps/examples/xmlrpc2di/MultithreadXmlRpcServer.cpp b/apps/examples/xmlrpc2di/MultithreadXmlRpcServer.cpp new file mode 100644 index 00000000..0a55a9a7 --- /dev/null +++ b/apps/examples/xmlrpc2di/MultithreadXmlRpcServer.cpp @@ -0,0 +1,122 @@ +/* + * Written by Jae An (jae_young_an@yahoo.com) + * + * adapted for SEMS by Stefan Sayer stefan.sayer at iptego.com + */ +#include "XmlRpcServer.h" +#include "XmlRpcServerConnection.h" +#include "XmlRpcServerMethod.h" +#include "XmlRpcSocket.h" +#include "XmlRpcUtil.h" +#include "XmlRpcException.h" +#include "MultithreadXmlRpcServer.h" + +#include "log.h" + +using namespace XmlRpc; +WorkerThread::WorkerThread(MultithreadXmlRpcServer* chief) + : runcond(false), chief(chief) { +} + +// call this method before calling run +void WorkerThread::addXmlRpcSource(XmlRpcSource* source,unsigned eventMask) +{ + dispatcher.addSource(source,eventMask); + runcond.set(true); +} + +void WorkerThread::run() +{ + chief->reportBack(this); + + while (!is_stopped()) { + runcond.wait_for(); + dispatcher.work(-1.0); + + dispatcher.clear(); // close socket and others ... + runcond.set(false); + /* tell chief we can work again */ + chief->reportBack(this); + } + DBG("WorkerThread stopped.\n"); +} + +void WorkerThread::on_stop() { +} + +MultithreadXmlRpcServer::MultithreadXmlRpcServer() + : XmlRpcServer(), have_waiting(false) { +} + +// Wait until all the worker threads are done. +MultithreadXmlRpcServer::~MultithreadXmlRpcServer() +{ + for (std::vector::iterator it= + workers.begin(); it != workers.end();it++) { + (*it)->stop(); + (*it)->join(); + delete *it; + } +} + +// Accept a client connection request and create a connection to +// handle method calls from the client. +void MultithreadXmlRpcServer::acceptConnection() +{ + int s = XmlRpcSocket::accept(this->getfd()); + if (s < 0) + { + ERROR("MultithreadXmlRpcServer::acceptConnection: Could not accept connection (%s).", + XmlRpcSocket::getErrorMsg().c_str()); + + } + else if ( ! XmlRpcSocket::setNonBlocking(s)) + { + XmlRpcSocket::close(s); + ERROR("XmlRpcServer::acceptConnection: Could not set socket " + "to non-blocking input mode (%s).\n", + XmlRpcSocket::getErrorMsg().c_str()); + } + else // Notify the dispatcher to listen for input on this source when we are in work() + { + WorkerThread* thr = NULL; + while (!thr) { + if (!have_waiting.get()) + have_waiting.wait_for(); + thr = getIdleThread(); + } + thr->addXmlRpcSource(this->createConnection(s), XmlRpcDispatch::ReadableEvent); + + // // Uh oh.. all threads are busy... + // // Just close the connection so that the rejected client would try again. + // XmlRpcSocket::close(s); + // XmlRpcUtil::error("MultithreadXmlRpcServer::acceptConnection: All threads are busy. Rejected a client"); + } +} + +void MultithreadXmlRpcServer::reportBack(WorkerThread* thr) { + waiting_mut.lock(); + waiting.push(thr); + have_waiting.set(true); + waiting_mut.unlock(); +} + +WorkerThread* MultithreadXmlRpcServer::getIdleThread() { + WorkerThread* res = NULL; + waiting_mut.lock(); + if (!waiting.empty()) { + res = waiting.front(); + waiting.pop(); + } + have_waiting.set(!waiting.empty()); + waiting_mut.unlock(); + return res; +} + +void MultithreadXmlRpcServer::createThreads(unsigned int n) { + for (unsigned int i=0;istart(); + } +} diff --git a/apps/examples/xmlrpc2di/MultithreadXmlRpcServer.h b/apps/examples/xmlrpc2di/MultithreadXmlRpcServer.h new file mode 100644 index 00000000..e38588c8 --- /dev/null +++ b/apps/examples/xmlrpc2di/MultithreadXmlRpcServer.h @@ -0,0 +1,66 @@ +/** + * Written by Jae An (jae_young_an@yahoo.com) + * http://sourceforge.net/forum/forum.php?thread_id=910224&forum_id=240495 + * + * adapted for SEMS by Stefan Sayer stefan.sayer at iptego.com + */ + +#ifndef _MULTI_THREAD_XMLRPCSERVER_H_ +#define _MULTI_THREAD_XMLRPCSERVER_H_ + +#include +#include "XmlRpcServer.h" +#include "XmlRpcDispatch.h" + +#include +#include + +namespace XmlRpc { + class MultithreadXmlRpcServer; + + class WorkerThread : public AmThread { + MultithreadXmlRpcServer* chief; + AmCondition runcond; + + public: + WorkerThread(MultithreadXmlRpcServer* chief); + + void addXmlRpcSource(XmlRpcSource* source,unsigned eventMask); // call this method to make it run + + void run(); + void on_stop(); + private: + XmlRpcDispatch dispatcher; + }; + + +#define MAX_THREAD_SIZE 8 + + //! Multi-threaded sever class to handle XML RPC requests + class MultithreadXmlRpcServer : public XmlRpcServer { + public: + //! Create a server object. + MultithreadXmlRpcServer(); + //! Destructor. + virtual ~MultithreadXmlRpcServer(); + + /* report back from work */ + void reportBack(WorkerThread* thr); + void createThreads(unsigned int n); + + protected: + + //! Accept a client connection request + virtual void acceptConnection(); + + private: + AmMutex waiting_mut; + std::queue waiting; + AmCondition have_waiting; + std::vector workers; + WorkerThread* getIdleThread(); + }; + +} // namespace XmlRpc + +#endif diff --git a/apps/examples/xmlrpc2di/XMLRPC2DI.cpp b/apps/examples/xmlrpc2di/XMLRPC2DI.cpp index d4c966ab..f07921cf 100644 --- a/apps/examples/xmlrpc2di/XMLRPC2DI.cpp +++ b/apps/examples/xmlrpc2di/XMLRPC2DI.cpp @@ -65,10 +65,32 @@ int XMLRPC2DI::load() { return 0; configured = true; + AmConfigReader cfg; if(cfg.loadFile(AmConfig::ModConfigPath + string(MOD_NAME ".conf"))) return -1; + string multithreaded = cfg.getParameter("multithreaded", "yes"); + + XmlRpcServer* s; + bool multi_threaded = false; + unsigned int threads = 0; + if (multithreaded == "yes") { + multi_threaded = true; + if (!cfg.getParameter("threads").length()) + threads = 5; + else + threads = cfg.getParameterInt("threads", 5); + + DBG("Running multi-threaded XMLRPC server with %u threads\n", threads); + MultithreadXmlRpcServer* mt_s = new MultithreadXmlRpcServer(); + mt_s->createThreads(threads); + s = mt_s; + } else { + DBG("Running single-threaded XMLRPC server\n"); + s = new XmlRpcServer(); + } + ServerRetryAfter = cfg.getParameterInt("server_retry_after", 10); DBG("retrying failed server after %u seconds\n", ServerRetryAfter); @@ -106,8 +128,7 @@ int XMLRPC2DI::load() { DBG("XMLRPC Server: %snabling builtin method 'di'.\n", export_di?"E":"Not e"); - server = new XMLRPC2DIServer(XMLRPCPort, export_di, direct_export); - + server = new XMLRPC2DIServer(XMLRPCPort, export_di, direct_export, s); server->start(); return 0; } @@ -228,14 +249,16 @@ void XMLRPC2DI::invoke(const string& method, XMLRPC2DIServer::XMLRPC2DIServer(unsigned int port, bool di_export, - string direct_export) + string direct_export, + XmlRpcServer* s) : port(port), + s(s), // register method 'calls' - calls_method(&s), + calls_method(s), // register method 'get_loglevel' - setloglevel_method(&s), + setloglevel_method(s), // register method 'set_loglevel' - getloglevel_method(&s) + getloglevel_method(s) { DBG("XMLRPC Server: enabled builtin method 'calls'\n"); DBG("XMLRPC Server: enabled builtin method 'get_loglevel'\n"); @@ -244,7 +267,7 @@ XMLRPC2DIServer::XMLRPC2DIServer(unsigned int port, // export all methods via 'di' function? if (di_export) { // register method 'di' - di_method = new XMLRPC2DIServerDIMethod(&s); + di_method = new XMLRPC2DIServerDIMethod(s); } vector export_ifaces = explode(direct_export, ";"); @@ -281,7 +304,7 @@ void XMLRPC2DIServer::registerMethods(const std::string& iface) { for (unsigned int i=0;ifindMethod(method)); if (has_method) { ERROR("name conflict for method '%s' from interface '%s', " "method already exported!\n", @@ -294,14 +317,14 @@ void XMLRPC2DIServer::registerMethods(const std::string& iface) { DBG("XMLRPC Server: adding method '%s'\n", method.c_str()); DIMethodProxy* mp = new DIMethodProxy(method, method, di_f); - s.addMethod(mp); + s->addMethod(mp); } DBG("XMLRPC Server: adding method '%s.%s'\n", iface.c_str(), method.c_str()); DIMethodProxy* mp = new DIMethodProxy(iface + "." + method, method, di_f); - s.addMethod(mp); + s->addMethod(mp); } } catch (AmDynInvoke::NotImplemented& e) { ERROR("Not implemented in interface '%s': '%s'\n", @@ -318,9 +341,9 @@ void XMLRPC2DIServer::registerMethods(const std::string& iface) { void XMLRPC2DIServer::run() { DBG("Binding XMLRPC2DIServer to port %u \n", port); - s.bindAndListen(port); + s->bindAndListen(port); DBG("starting XMLRPC2DIServer...\n"); - s.work(-1.0); + s->work(-1.0); } void XMLRPC2DIServer::on_stop() { DBG("sorry, don't know how to stop the server.\n"); diff --git a/apps/examples/xmlrpc2di/XMLRPC2DI.h b/apps/examples/xmlrpc2di/XMLRPC2DI.h index bfd27d9d..e6657b24 100644 --- a/apps/examples/xmlrpc2di/XMLRPC2DI.h +++ b/apps/examples/xmlrpc2di/XMLRPC2DI.h @@ -28,6 +28,7 @@ #define XMLRPC2DISERVER_H #include "XmlRpc.h" +#include "MultithreadXmlRpcServer.h" using namespace XmlRpc; #include "AmThread.h" @@ -83,7 +84,8 @@ struct DIMethodProxy : public XmlRpcServerMethod }; class XMLRPC2DIServer : public AmThread { - XmlRpcServer s; + XmlRpcServer* s; + unsigned int port; XMLRPC2DIServerCallsMethod calls_method; XMLRPC2DIServerSetLoglevelMethod setloglevel_method; @@ -94,7 +96,8 @@ class XMLRPC2DIServer : public AmThread { public: XMLRPC2DIServer(unsigned int port, bool di_export, - string direct_export); + string direct_export, + XmlRpcServer* s); void run(); void on_stop(); diff --git a/apps/examples/xmlrpc2di/etc/xmlrpc2di.conf b/apps/examples/xmlrpc2di/etc/xmlrpc2di.conf index 1e129675..68750d55 100644 --- a/apps/examples/xmlrpc2di/etc/xmlrpc2di.conf +++ b/apps/examples/xmlrpc2di/etc/xmlrpc2di.conf @@ -1,6 +1,17 @@ # port to bind XMLRPC server to xmlrpc_port=8090 + +# run multi-threaded server? +# Default: yes +# +# multithreaded = yes + +# threads to run - this many requests can be processed in parallel +# Default: 5 +# +# threads=5 + # export all DI functions with the function call 'di'? # defaults to: yes # export_di=yes @@ -11,3 +22,6 @@ xmlrpc_port=8090 # defaults to: none # direct_export=di_dial;registrar_client +# run the XMLRPC server at all (default: yes) +# +# run_server=yes