For merge

Client/Client.cpp

@@ -58,7 +58,8 @@ Client::query_result Client::submit(std::string command, std::string &message,
 
   command = "#" + command;
 
-  write(m_clientFd, command.c_str(), command.length() + 1);
+  int bytes = write(m_clientFd, command.c_str(), command.length() + 1);
+  if (bytes != command.length() + 1) perror("failed to send SQL to claims");
   ClientLogging::log("Client: message from server!\n");
   const int maxBytes = 75536 + sizeof(int) * 2;
   char *buf = new char[maxBytes];

Client/Makefile.am

@@ -17,8 +17,7 @@ LDADD = 	../catalog/libcatalog.a \
 		../common/Block/libblock.a \
 		../common/Schema/libschema.a \
 		${BOOST_HOME}/stage/lib/libboost_serialization.a \
-		${BOOST_HOME}/stage/lib/libboost_serialization.so \
-		${THERON_HOME}/Lib/libtherond.a		
+		${BOOST_HOME}/stage/lib/libboost_serialization.so 
 
 noinst_LIBRARIES=libclient.a
 libclient_a_SOURCES = \
@@ -27,5 +26,5 @@ libclient_a_SOURCES = \
 	ClientResponse.cpp 		ClientResponse.h \
 	jsoncpp.cpp
 
-SUBDIRS = json Test
-DIST_SUBDIRS = json Test
+#SUBDIRS = json Test
+#DIST_SUBDIRS = json Test

Client/Test/TestSeverClient.cpp

@@ -123,7 +123,7 @@ static int loadData() {
   catalog->add_table(table_1);
 
   ////////////////////////////////////Create table
-  ///right//////////////////////////
+  /// right//////////////////////////
   TableDescriptor* table_2 = new TableDescriptor(
       "sb",
       Environment::getInstance()->getCatalog()->allocate_unique_table_id());
@@ -212,20 +212,20 @@ static int loadData() {
   }
   // partitioned by row_id
   //		for(unsigned
-  //i=0;i<table_1->getProjectoin(14)->getPartitioner()->getNumberOfPartitions();i++){
+  // i=0;i<table_1->getProjectoin(14)->getPartitioner()->getNumberOfPartitions();i++){
   //
   //			catalog->getTable(0)->getProjectoin(14)->getPartitioner()->RegisterPartition(i,2);
   //		}
   //
   //		// 8 partitions
   //	for(unsigned
-  //i=0;i<table_1->getProjectoin(2)->getPartitioner()->getNumberOfPartitions();i++){
+  // i=0;i<table_1->getProjectoin(2)->getPartitioner()->getNumberOfPartitions();i++){
   //
   //		catalog->getTable(0)->getProjectoin(2)->getPartitioner()->RegisterPartition(i,1);
   //	}
   //
   //	for(unsigned
-  //i=0;i<table_1->getProjectoin(3)->getPartitioner()->getNumberOfPartitions();i++){
+  // i=0;i<table_1->getProjectoin(3)->getPartitioner()->getNumberOfPartitions();i++){
   //
   //		catalog->getTable(0)->getProjectoin(3)->getPartitioner()->RegisterPartition(i,3);
   //	}
@@ -245,100 +245,100 @@ static int loadData() {
         i, 6);
   }
   //	for(unsigned
-  //i=0;i<table_2->getProjectoin(2)->getPartitioner()->getNumberOfPartitions();i++){
+  // i=0;i<table_2->getProjectoin(2)->getPartitioner()->getNumberOfPartitions();i++){
   //
   //		catalog->getTable(1)->getProjectoin(2)->getPartitioner()->RegisterPartition(i,1);
   //	}
 
   //	for(unsigned
-  //i=0;i<table_2->getProjectoin(3)->getPartitioner()->getNumberOfPartitions();i++){
+  // i=0;i<table_2->getProjectoin(3)->getPartitioner()->getNumberOfPartitions();i++){
   //
   //		catalog->getTable(1)->getProjectoin(3)->getPartitioner()->RegisterPartition(i,3);
   //	}
   //
   //		//partitioned by row_id
   //		for(unsigned
-  //i=0;i<table_2->getProjectoin(14)->getPartitioner()->getNumberOfPartitions();i++){
+  // i=0;i<table_2->getProjectoin(14)->getPartitioner()->getNumberOfPartitions();i++){
   //
   //			catalog->getTable(1)->getProjectoin(14)->getPartitioner()->RegisterPartition(i,2);
   //		}
   //
   //		////////////////////////////////////////
   //
   //		///////////////////ONE
-  //MONTH/////////////////////////////////////////////////////////////
+  // MONTH/////////////////////////////////////////////////////////////
   //		//CJ
   //		// 4 partition
   //	for(unsigned
-  //i=0;i<table_1->getProjectoin(4)->getPartitioner()->getNumberOfPartitions();i++){
+  // i=0;i<table_1->getProjectoin(4)->getPartitioner()->getNumberOfPartitions();i++){
   //
   //		catalog->getTable(0)->getProjectoin(4)->getPartitioner()->RegisterPartition(i,40);
   //	}
   //
   //	for(unsigned
-  //i=0;i<table_1->getProjectoin(5)->getPartitioner()->getNumberOfPartitions();i++){
+  // i=0;i<table_1->getProjectoin(5)->getPartitioner()->getNumberOfPartitions();i++){
   //
   //		catalog->getTable(0)->getProjectoin(5)->getPartitioner()->RegisterPartition(i,104);
   //	}
   //		//8 partitions
   //		for(unsigned
-  //i=0;i<table_1->getProjectoin(10)->getPartitioner()->getNumberOfPartitions();i++){
+  // i=0;i<table_1->getProjectoin(10)->getPartitioner()->getNumberOfPartitions();i++){
   //
   //			catalog->getTable(0)->getProjectoin(10)->getPartitioner()->RegisterPartition(i,20);
   //		}
   //
   //		for(unsigned
-  //i=0;i<table_1->getProjectoin(11)->getPartitioner()->getNumberOfPartitions();i++){
+  // i=0;i<table_1->getProjectoin(11)->getPartitioner()->getNumberOfPartitions();i++){
   //
   //			catalog->getTable(0)->getProjectoin(11)->getPartitioner()->RegisterPartition(i,52);
   //		}
   //		// 18 partitions
   //	for(unsigned
-  //i=0;i<table_1->getProjectoin(6)->getPartitioner()->getNumberOfPartitions();i++){
+  // i=0;i<table_1->getProjectoin(6)->getPartitioner()->getNumberOfPartitions();i++){
   //
   //		catalog->getTable(0)->getProjectoin(6)->getPartitioner()->RegisterPartition(i,10);
   //	}
   //
   //	for(unsigned
-  //i=0;i<table_1->getProjectoin(7)->getPartitioner()->getNumberOfPartitions();i++){
+  // i=0;i<table_1->getProjectoin(7)->getPartitioner()->getNumberOfPartitions();i++){
   //
   //		catalog->getTable(0)->getProjectoin(7)->getPartitioner()->RegisterPartition(i,24);
   //	}
   //
   //		//SB
   //		// 4 partition
   //	for(unsigned
-  //i=0;i<table_2->getProjectoin(4)->getPartitioner()->getNumberOfPartitions();i++){
+  // i=0;i<table_2->getProjectoin(4)->getPartitioner()->getNumberOfPartitions();i++){
   //
   //		catalog->getTable(1)->getProjectoin(4)->getPartitioner()->RegisterPartition(i,39);
   //	}
   //
   //	for(unsigned
-  //i=0;i<table_2->getProjectoin(5)->getPartitioner()->getNumberOfPartitions();i++){
+  // i=0;i<table_2->getProjectoin(5)->getPartitioner()->getNumberOfPartitions();i++){
   //
   //		catalog->getTable(1)->getProjectoin(5)->getPartitioner()->RegisterPartition(i,131);
   //	}
   //		// 8 partitions
   //		for(unsigned
-  //i=0;i<table_2->getProjectoin(10)->getPartitioner()->getNumberOfPartitions();i++){
+  // i=0;i<table_2->getProjectoin(10)->getPartitioner()->getNumberOfPartitions();i++){
   //
   //			catalog->getTable(1)->getProjectoin(10)->getPartitioner()->RegisterPartition(i,20);
   //		}
   //
   //		for(unsigned
-  //i=0;i<table_2->getProjectoin(11)->getPartitioner()->getNumberOfPartitions();i++){
+  // i=0;i<table_2->getProjectoin(11)->getPartitioner()->getNumberOfPartitions();i++){
   //
   //			catalog->getTable(1)->getProjectoin(11)->getPartitioner()->RegisterPartition(i,66);
   //		}
   //		// 18 partitions
   //	for(unsigned
-  //i=0;i<table_2->getProjectoin(6)->getPartitioner()->getNumberOfPartitions();i++){
+  // i=0;i<table_2->getProjectoin(6)->getPartitioner()->getNumberOfPartitions();i++){
   //
   //		catalog->getTable(1)->getProjectoin(6)->getPartitioner()->RegisterPartition(i,10);
   //	}
   //
   //	for(unsigned
-  //i=0;i<table_2->getProjectoin(7)->getPartitioner()->getNumberOfPartitions();i++){
+  // i=0;i<table_2->getProjectoin(7)->getPartitioner()->getNumberOfPartitions();i++){
   //
   //		catalog->getTable(1)->getProjectoin(7)->getPartitioner()->RegisterPartition(i,30);
   //	}
@@ -349,51 +349,51 @@ static int loadData() {
   //		//// cj////
   //		// 4 partitions
   //		for(unsigned
-  //i=0;i<table_1->getProjectoin(8)->getPartitioner()->getNumberOfPartitions();i++){
+  // i=0;i<table_1->getProjectoin(8)->getPartitioner()->getNumberOfPartitions();i++){
   //
   //			catalog->getTable(0)->getProjectoin(8)->getPartitioner()->RegisterPartition(i,14);
   //		}
   //
   //		for(unsigned
-  //i=0;i<table_1->getProjectoin(9)->getPartitioner()->getNumberOfPartitions();i++){
+  // i=0;i<table_1->getProjectoin(9)->getPartitioner()->getNumberOfPartitions();i++){
   //
   //			catalog->getTable(0)->getProjectoin(9)->getPartitioner()->RegisterPartition(i,36);
   //		}
   //		// 8 partitions
   //		for(unsigned
-  //i=0;i<table_1->getProjectoin(12)->getPartitioner()->getNumberOfPartitions();i++){
+  // i=0;i<table_1->getProjectoin(12)->getPartitioner()->getNumberOfPartitions();i++){
   //
   //			catalog->getTable(0)->getProjectoin(12)->getPartitioner()->RegisterPartition(i,7);
   //		}
   //
   //		for(unsigned
-  //i=0;i<table_1->getProjectoin(13)->getPartitioner()->getNumberOfPartitions();i++){
+  // i=0;i<table_1->getProjectoin(13)->getPartitioner()->getNumberOfPartitions();i++){
   //
   //			catalog->getTable(0)->getProjectoin(13)->getPartitioner()->RegisterPartition(i,19);
   //		}
   //
   //		//// sb ////
   //		// 4 partitions//
   //		for(unsigned
-  //i=0;i<table_2->getProjectoin(8)->getPartitioner()->getNumberOfPartitions();i++){
+  // i=0;i<table_2->getProjectoin(8)->getPartitioner()->getNumberOfPartitions();i++){
   //
   //			catalog->getTable(1)->getProjectoin(8)->getPartitioner()->RegisterPartition(i,14);
   //		}
   //
   //		for(unsigned
-  //i=0;i<table_2->getProjectoin(9)->getPartitioner()->getNumberOfPartitions();i++){
+  // i=0;i<table_2->getProjectoin(9)->getPartitioner()->getNumberOfPartitions();i++){
   //
   //			catalog->getTable(1)->getProjectoin(9)->getPartitioner()->RegisterPartition(i,131);
   //		}
   //		// 8 partitions//
   //		for(unsigned
-  //i=0;i<table_2->getProjectoin(12)->getPartitioner()->getNumberOfPartitions();i++){
+  // i=0;i<table_2->getProjectoin(12)->getPartitioner()->getNumberOfPartitions();i++){
   //
   //			catalog->getTable(1)->getProjectoin(12)->getPartitioner()->RegisterPartition(i,7);
   //		}
   //
   //		for(unsigned
-  //i=0;i<table_2->getProjectoin(13)->getPartitioner()->getNumberOfPartitions();i++){
+  // i=0;i<table_2->getProjectoin(13)->getPartitioner()->getNumberOfPartitions();i++){
   //
   //			catalog->getTable(1)->getProjectoin(13)->getPartitioner()->RegisterPartition(i,23);
   //		}

Config.cpp

@@ -91,6 +91,23 @@ int Config::thread_pool_init_thread_num;
 int Config::load_thread_num;
 int Config::memory_utilization;
 
+bool Config::is_master_loader;
+std::string Config::master_loader_ip;
+int Config::master_loader_port;
+std::string Config::amq_url;
+std::string Config::amq_topic;
+
+bool Config::enable_txn_server;
+int Config::txn_server_cores;
+std::string Config::txn_server_ip;
+int Config::txn_server_port;
+
+bool Config::enable_txn_log = false;
+std::string Config::txn_log_path;
+
+int Config::master_loader_thread_num;
+int Config::slave_loader_thread_num;
+
 Config *Config::getInstance() {
   if (instance_ == 0) {
     instance_ = new Config();
@@ -152,6 +169,33 @@ void Config::initialize() {
 
   load_thread_num = getInt("load_thread_num", sysconf(_SC_NPROCESSORS_CONF));
 
+  is_master_loader = getBoolean("is_master_loader", true);
+
+  master_loader_ip = getString("master_loader_ip", "10.11.1.193");
+
+  master_loader_port = getInt("master_loader_port", 9001);
+
+  amq_url = getString("amq_url", "58.198.176.92:61616");
+
+  amq_topic = getString("amq_topic", "claims");
+
+  // txn manager
+  enable_txn_server = getBoolean("txn_server", true);
+
+  txn_server_cores = getInt("txn_server_cores", 4);
+
+  txn_server_ip = getString("txn_server_ip", "127.0.0.1");
+
+  txn_server_port = getInt("txn_server_port", 9100);
+
+  // txn log
+  enable_txn_log = getBoolean("txn_log", true);
+
+  txn_log_path = getString("txn_log_path", ".");
+
+  master_loader_thread_num = getInt("master_loader_thread_num", 4);
+  slave_loader_thread_num = getInt("slave_loader_thread_num", 4);
+
   memory_utilization = getInt("memory_utilization", 100);
 
 #ifdef DEBUG_Config
@@ -212,6 +256,16 @@ void Config::print_configure() const {
   std::cout << "catalog_file:" << catalog_file << std::endl;
   std::cout << "codegen:" << enable_codegen << std::endl;
   std::cout << "load_thread_num:" << load_thread_num << std::endl;
+  std::cout << "amq_url:" << amq_url << std::endl;
+  std::cout << "amq_topic:" << amq_topic << std::endl;
+
+  std::cout << "enable_txn_serverr:" << enable_txn_server << std::endl;
+  std::cout << "txn_server_cores:" << txn_server_cores << std::endl;
+  std::cout << "txn_server_ip:" << txn_server_ip << std::endl;
+  std::cout << "txn_server_port:" << txn_server_port << std::endl;
+
+  std::cout << "enable_txn_log:" << enable_txn_log << std::endl;
+  std::cout << "txn_log_path:" << txn_log_path << std::endl;
 }
 
 void Config::setConfigFile(std::string file_name) { config_file = file_name; }

Config.h

@@ -80,6 +80,22 @@ class Config {
   static int memory_utilization;
 
   static int load_thread_num;
+  static bool is_master_loader;
+  static std::string master_loader_ip;
+  static int master_loader_port;
+  static std::string amq_url;
+  static std::string amq_topic;
+
+  static bool enable_txn_server;
+  static int txn_server_cores;
+  static std::string txn_server_ip;
+  static int txn_server_port;
+
+  static bool enable_txn_log;
+  static std::string txn_log_path;
+
+  static int master_loader_thread_num;
+  static int slave_loader_thread_num;
 
  private:
   static Config* instance_;

Debug.h

@@ -25,13 +25,13 @@
 #define BLOCK_SIZE (64 * 1024)
 #define SUCHUNK_SIZE (64 * 1024)
 #define CHUNK_SIZE (64 * 1024 * 1024)
-#define CHUNK_SIZE_IN_MB 64
-#define HEARTBEAT_MESSAGE_LEN 64
-#define REGISTER_MESSAGE_LEN 64
-#define BLOCK_STATUS_MESSAGE_LEN 256
-#define MATCHER_MESSAGE_FILENAME_LEN 256
-#define MATCHER_MESSAGE_BMI_LEN 256
-#define MATCHER_MESSAGE_PROJECT_LEN 256
+#define CHUNK_SIZE_IN_MB (64)
+#define HEARTBEAT_MESSAGE_LEN (64)
+#define REGISTER_MESSAGE_LEN (64)
+#define BLOCK_STATUS_MESSAGE_LEN (256)
+#define MATCHER_MESSAGE_FILENAME_LEN (256)
+#define MATCHER_MESSAGE_BMI_LEN (256)
+#define MATCHER_MESSAGE_PROJECT_LEN (256)
 // 分布式文件系统的主节点
 #define HDFS_N "10.11.1.190"
 // 磁盘目录