Changeset 332

Timestamp:

03/13/08 01:10:04 (8 months ago)

Author:

jake

Message:

Incremental update, not functional yet...

Location:

trunk/throxy/src

Files:

• ServiceMonitor.cpp (modified) (10 diffs)
• ServiceMonitor.h (modified) (6 diffs)
• StaticServiceMonitor.cpp (modified) (2 diffs)
• ThrudocHandler.cpp (modified) (3 diffs)
• ThrudocHandler.h (modified) (1 diff)
• main.cpp (modified) (6 diffs)
• static.conf (modified) (1 diff)
• throxy.conf (modified) (1 diff)

trunk/throxy/src/ServiceMonitor.cpp

@@ -15 +15 @@
 
 ServiceNode::ServiceNode(const string name, const string host, const int port, const bool master)
-    : name(name), host(host), port(port), master(master), active(false), last_ping(-1)
+    : name(name), master(master), host(host), port(port), active(false), last_ping(-1)
 {
 
@@ -26 +26 @@
 
 ServiceNode::ServiceNode()
+    : master(false)
 {
 
@@ -40 +41 @@
 
         if( !socket->isOpen() || active == false )
-            socket->open();
+            transport->open();
+        active = true;
+        return true;
 
         //Call ping() this doesn't need to exist
         protocol->writeMessageBegin("ping", facebook::thrift::protocol::T_CALL, 0);
+
+        //protocol->writeStructBegin("Thrudoc_ping_args");
+        //protocol->writeFieldStop();
+        //protocol->writeStructEnd();
+
         protocol->writeMessageEnd();
         protocol->getTransport()->flush();
         protocol->getTransport()->writeEnd();
 
+
+
+        LOG4CXX_INFO(logger,"Connected to remote host");
         int32_t rseqid = 0;
         std::string fname;
@@ -58 +69 @@
         protocol->getTransport()->readEnd();
 
+
+
         if(mtype == facebook::thrift::protocol::T_EXCEPTION ||
            mtype == facebook::thrift::protocol::T_REPLY ){
@@ -101 +114 @@
 
 
+shared_ptr<ServiceNode> ServiceNode::clone()
+{
+    return shared_ptr<ServiceNode>( new ServiceNode(name,host,port,master) );
+}
+
 ////////////////////////////////////////////////////////////////////////////
 
 LoggerPtr ServicePartition::logger (Logger::getLogger ("ServicePartition"));
 
-ServicePartition::ServicePartition(const std::string name)
-    : name(name)
+ServicePartition::ServicePartition(const std::string name, unsigned int pool_size)
+    : name(name), pool_size(pool_size)
 {
 
@@ -120 +138 @@
     vector<string> names;
 
-    map<string, shared_ptr<ServiceNode> >::iterator it;
+    map<string, vector<shared_ptr<ServiceNode> > >::iterator it;
 
     for(it = service_nodes.begin(); it != service_nodes.end(); ++it){
@@ -131 +149 @@
 void ServicePartition::addServiceNode( shared_ptr<ServiceNode> node )
 {
-    service_nodes[node->name] = node;
+
+    vector< shared_ptr<ServiceNode> > nodes;
+
+    nodes.push_back( node );
+
+    for(unsigned int i=0; i<(pool_size-1); i++){
+        nodes.push_back( node->clone() );
+    }
+
+    service_nodes[node->name] = nodes;
 }
 
@@ -141 +168 @@
 shared_ptr<ServiceNode> ServicePartition::getServiceNode( std::string _name )
 {
-    map<string, shared_ptr<ServiceNode> >::iterator it = service_nodes.find( _name );
+    map<string, vector<shared_ptr<ServiceNode> > >::iterator it = service_nodes.find( _name );
 
     if(it == service_nodes.end())
         throw TException("Unknown node "+_name);
 
-    return it->second;
-}
+    //return a random connection to this node
+    return it->second[ (rand() * 10000 ) % pool_size];
+}
+
+shared_ptr<TProtocol> ServicePartition::getConnection( bool master )
+{
+
+    map<string, vector< shared_ptr<ServiceNode> > >::iterator it;
+
+    for(it=service_nodes.begin(); it!=service_nodes.end(); ++it){
+
+        if( (master && it->second[0]->master) || !master )
+            return it->second[(rand() * 10000 ) % pool_size]->getConnection();
+
+    }
+
+    throw TException("No connections available ");
+}
+
+
 
 ////////////////////////////////////////////////////////////////////////////
@@ -185 +230 @@
 void ServiceMonitor::delServicePartition( string name )
 {
-
     service_partitions.erase(name);
-
 }
 
@@ -201 +244 @@
     return it->second;
 }
+
+
+
+vector< shared_ptr<TProtocol> > ServiceMonitor::getConnections( bool masters )
+{
+
+    vector< shared_ptr<TProtocol> > nodes;
+
+    map<string, shared_ptr<ServicePartition> >::iterator it;
+
+    for(it=service_partitions.begin(); it!=service_partitions.end(); ++it){
+
+        nodes.push_back( it->second->getConnection( masters ) );
+    }
+
+    return nodes;
+}

trunk/throxy/src/ServiceMonitor.h

@@ -17 +17 @@
     bool    ping();
     boost::shared_ptr<facebook::thrift::protocol::TProtocol> getConnection();
+        boost::shared_ptr<ServiceNode>                       clone(); // makin copies
 
     const std::string name;
+    const bool        master;
+
  protected:
     ServiceNode();
@@ -25 +28 @@
 
     int         port;
-    bool        master;
-
     bool        active;
     uint64_t    last_ping;
@@ -42 +43 @@
 {
  public:
-    ServicePartition(const std::string name);
+    ServicePartition(const std::string name, unsigned int pool_size = 10 );
     ~ServicePartition();
 
@@ -51 +52 @@
     boost::shared_ptr<ServiceNode> getServiceNode( std::string name );
 
+    boost::shared_ptr<facebook::thrift::protocol::TProtocol>  getConnection(bool master);
+
     const std::string name;
 
@@ -56 +59 @@
     ServicePartition();
 
-    std::map< std::string, boost::shared_ptr<ServiceNode> > service_nodes;
+    const unsigned int pool_size;
 
-
+    std::map< std::string, std::vector< boost::shared_ptr<ServiceNode> > > service_nodes;
 
     static log4cxx::LoggerPtr logger;
@@ -76 +79 @@
     boost::shared_ptr<ServicePartition> getServicePartition( std::string name );
 
+    std::vector< boost::shared_ptr<facebook::thrift::protocol::TProtocol> > getConnections( bool masters = false );
+
  protected:
     std::map< std::string, boost::shared_ptr<ServicePartition> > service_partitions;

trunk/throxy/src/StaticServiceMonitor.cpp

@@ -48 +48 @@
                 break;
 
-
             LOG4CXX_INFO(logger, string("Partition node added: ")+n_name);
 
@@ -63 +62 @@
         this->addServicePartition(part);
     }
+
+    cerr<<"Done"<<endl;
 }
 
 
+StaticServiceMonitor::~StaticServiceMonitor()
+{
 
+}
+
+

trunk/throxy/src/ThrudocHandler.cpp

@@ -16 +16 @@
 using namespace log4cxx;
 using namespace std;
+using namespace facebook::thrift::protocol;
 
 LoggerPtr ThrudocHandler::logger (Logger::getLogger ("ThrudocHandler"));
@@ -21 +22 @@
 ThrudocHandler::ThrudocHandler ()
 {
+    //never invoked (private)
+}
 
+ThrudocHandler::ThrudocHandler(shared_ptr<ServiceMonitor> svc_mon)
+    : monitor(svc_mon)
+{
 
 }
@@ -27 +33 @@
 void ThrudocHandler::getBuckets (vector<string> &_return)
 {
+    vector< shared_ptr<TProtocol> >  protocols = monitor->getConnections( true );
 
+    shared_ptr<MyThrudocClient>  client( new MyThrudocClient() );
+
+    //Send data to each partition
+    for(size_t i=0; i<protocols.size(); i++){
+        client->setProtocol(protocols[i]);
+
+        client->send_getBuckets();
+    }
+
+    map<string, int> all_buckets;
+
+    //Recv data from each partition
+    for(size_t i=0; i<protocols.size(); i++){
+        client->setProtocol(protocols[i]);
+
+        vector<string> buckets;
+        client->recv_getBuckets(buckets);
+
+        //Count buckets from each partition
+        for(size_t j=0; j<buckets.size(); j++){
+            map<string, int>::iterator it = all_buckets.find( buckets[j] );
+
+            if( it == all_buckets.end() ){
+                all_buckets[ buckets[j] ] = 1;
+            }else{
+                all_buckets[ buckets[j] ]++;
+            }
+        }
+    }
+
+    //
+    map<string, int>::iterator it;
+
+    for(it = all_buckets.begin(); it != all_buckets.end(); ++it){
+
+        if(all_buckets.size() != protocols.size()){
+            LOG4CXX_ERROR(logger,"Parition missing bucket"+it->first);
+        }
+
+        _return.push_back(it->first);
+    }
 
 }

trunk/throxy/src/ThrudocHandler.h

@@ -11 +11 @@
 #include <log4cxx/logger.h>
 
+#include "ServiceMonitor.h"
+
 class ThrudocHandler : virtual public thrudoc::ThrudocIf {
-    public:
-        ThrudocHandler ();
+ public:
+    ThrudocHandler (boost::shared_ptr<ServiceMonitor> svc_mon);
 
-        void getBuckets (std::vector<std::string> &_return);
+    void getBuckets (std::vector<std::string> &_return);
 
-        void put (const std::string &bucket, const std::string &key,
-                  const std::string &value);
+    void put (const std::string &bucket, const std::string &key,
+              const std::string &value);
 
-        void putValue (std::string &_return, const std::string &bucket,
-                       const std::string &value);
+    void putValue (std::string &_return, const std::string &bucket,
+                   const std::string &value);
 
-        void get (std::string &_return, const std::string &bucket,
-                  const std::string &key);
+    void get (std::string &_return, const std::string &bucket,
+              const std::string &key);
 
-        void remove (const std::string &bucket, const std::string &key);
+    void remove (const std::string &bucket, const std::string &key);
 
-        void scan (thrudoc::ScanResponse &_return,
-                   const std::string &bucket,
-                   const std::string &seed, int32_t count);
+    void scan (thrudoc::ScanResponse &_return,
+               const std::string &bucket,
+               const std::string &seed, int32_t count);
 
-        void putList(std::vector<thrudoc::ThrudocException> &_return,
-                     const std::vector<thrudoc::Element> &elements);
+    void putList(std::vector<thrudoc::ThrudocException> &_return,
+                 const std::vector<thrudoc::Element> &elements);
 
-        void getList(std::vector<thrudoc::ListResponse> &_return,
-                     const std::vector<thrudoc::Element> &elements);
+    void getList(std::vector<thrudoc::ListResponse> &_return,
+                 const std::vector<thrudoc::Element> &elements);
 
-        void removeList(std::vector<thrudoc::ThrudocException> &_return,
-                        const std::vector<thrudoc::Element> &elements);
+    void removeList(std::vector<thrudoc::ThrudocException> &_return,
+                    const std::vector<thrudoc::Element> &elements);
 
-        void putValueList(std::vector<thrudoc::ListResponse> &_return,
-                          const std::vector<thrudoc::Element> &elements);
+    void putValueList(std::vector<thrudoc::ListResponse> &_return,
+                      const std::vector<thrudoc::Element> &elements);
 
-        void admin (std::string &_return, const std::string &op,
-                    const std::string &data);
+    void admin (std::string &_return, const std::string &op,
+                const std::string &data);
 
-    private:
-        static log4cxx::LoggerPtr logger;
+ private:
+    ThrudocHandler();
+
+    static log4cxx::LoggerPtr         logger;
+
+    boost::shared_ptr<ServiceMonitor> monitor;
+
+
+};
+
+
+class MyThrudocClient : public thrudoc::ThrudocClient
+{
+ public:
+ MyThrudocClient():
+    ThrudocClient(boost::shared_ptr<facebook::thrift::protocol::TProtocol>()){};
+
+    //Let me reuse this client for each connection
+    void setProtocol(boost::shared_ptr<facebook::thrift::protocol::TProtocol> p){
+        this->piprot_ = p;
+        poprot_ = p;
+        iprot_ = p.get();
+        oprot_ = p.get();
+    }
+
 };
 

trunk/throxy/src/main.cpp

@@ -1 @@
-#include <concurrency/ThreadManager.h>
-#include <concurrency/PosixThreadFactory.h>
-#include <protocol/TBinaryProtocol.h>
-#include <server/TNonblockingServer.h>
-#include <transport/TServerSocket.h>
-#include <transport/TTransportUtils.h>
+#include <thrift/TProcessor.h>
+#include <thrift/concurrency/ThreadManager.h>
+#include <thrift/concurrency/PosixThreadFactory.h>
+#include <thrift/protocol/TBinaryProtocol.h>
+#include <thrift/server/TNonblockingServer.h>
+#include <thrift/transport/TServerSocket.h>
+#include <thrift/transport/TTransportUtils.h>
 
 
@@ -16 +17 @@
 #include "utils.h"
 
-#include "Thrudoc.h"
+#include "ThrudocHandler.h"
+#include "StaticServiceMonitor.h"
 #include "Thrudex.h"
 
@@ -90 +92 @@
         shared_ptr<PosixThreadFactory> thread_factory(new PosixThreadFactory());
         shared_ptr<TProtocolFactory> protocol_factory(new TBinaryProtocolFactory());
+        thread_factory->setDetached(false);
 
         //collection of server to throxy
@@ -96 +99 @@
 
         //Thrudoc enabled
-        if( ConfigManager->read<int>("THRUDOC_PORT",-1) == -1 ) {
+        if( ConfigManager->read<int>("THRUDOC_PORT",-1) != -1 ) {
 
             int    thrudoc_port  = ConfigManager->read<int>("THRUDOC_PORT");
-            int    thread_count  = 30;
+            int    thread_count  = 3;
 
-            shared_ptr<ThrudocHandler>  handler          getThrudocHandler();
-            shared_ptr<TProcessor>      processor        (new ThrudocProcessor(handler));
+            shared_ptr<ThrudocHandler>  handler = getThrudocHandler();
+            shared_ptr<TProcessor>      processor(new thrudoc::ThrudocProcessor(handler));
 
             shared_ptr<ThreadManager> thread_manager =
@@ -114 +117 @@
 
 
+
             shared_ptr<Thread> t =
                 thread_factory->newThread(shared_ptr<TServer>(new TNonblockingServer(processor, protocol_factory,
                                                                                      thrudoc_port,thread_manager)));
 
-            server_thread.push_back(t);
+            server_threads.push_back(t);
         }
 
@@ -139 +143 @@
 
         //Start servers
-        for(int i=0; i<server_threads.size(); i++){
+        for(size_t i=0; i<server_threads.size(); i++){
             server_threads[i]->start();
         }
 
         //wait for server threads
-        for(int i=0; i<server_threads.size(); i++){
+        for(size_t i=0; i<server_threads.size(); i++){
             server_threads[i]->join();
         }
-
 
 

trunk/throxy/src/static.conf

@@ -13 +13 @@
 PARTITION_2 = p2
 
-PARTITION_2_NODE_1_NAME   = node1
+PARTITION_2_NODE_1_NAME   = node2
 PARTITION_2_NODE_1_HOST   = 127.0.0.1
 PARTITION_2_NODE_1_PORT   = 1978

trunk/throxy/src/throxy.conf

@@ -4 +4 @@
 THRUDOC_PORT         = 1998
 THRUDOC_MONITOR      = STATIC
-THRUDOC_MONITOR_ARG1 = thrudoc.conf
+THRUDOC_MONITOR_ARG1 = static.conf
+
+
+# Set root logger level to DEBUG and its only appender to A1.
+log4j.rootLogger=DEBUG, A1
+
+# A1 is set to be a ConsoleAppender.
+log4j.appender.A1=org.apache.log4j.ConsoleAppender
+
+# A1 uses PatternLayout.
+log4j.appender.A1.layout=org.apache.log4j.PatternLayout
+log4j.appender.A1.layout.ConversionPattern=%-4r [%t] %-5p %c %x - %m%n
 
 
 
-
-