Load from hdfs to hdfs

.gitignore

@@ -1,17 +1,23 @@
 !thirdparty/
+.idea
+.vscode/*
 *.Po
 *.a
-*.doxyfile
 *.dat*
 *.doxyfile
 *.o
+*sql.output
 .cproject
 .gitignore.swp
 .history
 .project
 .settings
 /test
 Build
+CLAIMS.config
+CLAIMS.creator*
+CLAIMS.files
+CLAIMS.includes
 Config.cpp
 Debug.h
 Default
@@ -23,6 +29,7 @@ autom4te.cache/
 bin/
 build
 claims.log
+claims_log/
 claimsserver
 client
 compile
@@ -38,47 +45,29 @@ config.status
 config.sub
 configure
 depcomp
+doc
 include/
 install-sh
 install/
 libtool
 log
 ltmain.sh
 missing
-stamp-h1
-Makefile.in
-*.Po
-Debug.h
-install/
-*.a
-*.o
-Makefile
-config.h
-config.h.in
-config.log
-configure
-claims.log
-doc
-conf
-conf/config
-build
-Build
-*.dat*
-include/
-.history
-compile
-thirdparty/
-ar-lib
-缺省值
-CLAIMS.config
-CLAIMS.creator*
-CLAIMS.files
-CLAIMS.includes
-*sql.output
-*.doxyfile
 sbin/.claimssserver.pid
+<<<<<<< HEAD
+sbin/2-claims-conf/claims_log/
 sbin/2-claims-conf/config-*
+sbin/2-claims-conf/test_for_DI_GetTuple
+=======
+sbin/2-claims-conf/
+>>>>>>> f5005b1ecfdcfb429ed630a0dfd573e1129a592b
 sbin/claims-test/claims_log/
 sbin/claims-test/test_for_DI_GetTuple
+sbin/claims-test/testresult/
 sbin/claims-test/testresult/*.result
+sbin/claims_log/
 sbin/logs/
+stamp-h1
+test_for_DI_GetTuple
+thirdparty/
+缺省值

Client.cpp

@@ -51,7 +51,10 @@ void submit_command(Client& client, std::string& command) {
       printf("%s", message.c_str());
       break;
     case Client::error:
-      printf("%s", message.c_str());
+      printf(
+          "\e[0;31m"
+          "%s\033[0m\n",
+          message.c_str());
       break;
     default:
       assert(false);
@@ -73,15 +76,45 @@ void submit_command_repeated(Client& client, std::string& command,
         printf("%s", message.c_str());
         break;
       case Client::error:
-        printf("%s", message.c_str());
+        printf(
+            "\e[0;31m"
+            "%s\033[0m\n",
+            message.c_str());
         break;
       default:
         assert(false);
         break;
     }
   }
 }
+void PrintUsage() {
+  cout << "Welcome to use CLAIMS. " << endl;
+  cout << "Type: " << endl;
+  cout << "\t help; for usage of CLAIMS." << endl;
+  cout << "\t copyright; for distribution terms." << endl;
+  cout << "\t exit; or shutdown; for exiting CLAIMS." << endl;
+}
 
+void PrintCopyright() {
+  cout << "Copyright [2012-2015] DaSE@ECNU " << endl
+       << " Licensed to the Apache Software Foundation (ASF) under one or more"
+       << " contributor license agreements.  See the NOTICE file distributed "
+          "with"
+       << " this work for additional information regarding copyright ownership."
+       << " The ASF licenses this file to You under the Apache License, "
+          "Version 2.0"
+       << " (the \" License\"); you may not use this file except in compliance "
+          "with"
+       << " the License.  You may obtain a copy of the License at" << endl
+       << "     http://www.apache.org/licenses/LICENSE-2.0" << endl
+       << " Unless required by applicable law or agreed to in writing, software"
+       << " distributed under the License is distributed on an \" AS IS \" "
+          "BASIS,"
+       << " WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or "
+          "implied."
+       << " See the License for the specific language governing permissions and"
+       << " limitations under the License." << endl;
+}
 int main(int argc, char** argv) {
   /* Client */
 
@@ -98,6 +131,7 @@ int main(int argc, char** argv) {
 
   claims::common::Logging claims_logging(argv[0]);
   print_welcome();
+  PrintUsage();
 
   Client client;
   client.connection(argv[1], atoi(argv[2]));
@@ -115,6 +149,12 @@ int main(int argc, char** argv) {
       break;
     } else if (command.empty()) {
       continue;
+    } else if (command == "help;") {
+      PrintUsage();
+      continue;
+    } else if (command == "copyright;") {
+      PrintCopyright();
+      continue;
     }
 
     submit_command(client, command);

Client/ClaimsServer.cpp

@@ -71,6 +71,7 @@ int ClientListener::receiveRequest(const int fd, const char *cmd) {
   remote_command rcmd;
   rcmd.socket_fd = fd;
   //	rcmd.cmd.append(cmd);
+  cout << cmd << endl;
   rcmd.cmd = std::string(cmd);
 
   Daemon::getInstance()->addRemoteCommand(rcmd);
@@ -266,11 +267,10 @@ void *ClientListener::receiveHandler(void *para) {
               server->removeClient(server->m_clientFds[i]);
               continue;
             }
-
-            //					cout<<"nread:"<<nread<<endl;
             memset(buf, 0, sizeof(buf));
             int read_count = read(server->m_clientFds[i], buf, nread);
-            buf[read_count] = '\0';      // fix a bug
+            buf[read_count] = '\0';  // fix a bug
+
             int sql_type = buf[0] - 48;  // '1' - 48 = 1
             ClientLogging::log("sql_type is %d", sql_type);
             if (sql_type <= 9 && sql_type >= 0) {
@@ -763,16 +763,20 @@ void *ClientListener::sendHandler(void *para) {
         } else if (client_type::c == server->client_type_) {
           // query return true
           cliRes.setOk("Yes Ok");
-          ClientListenerLogging::log(
-              "to send data response-- status:%d  length:%d  content:%s  fd:%d",
-              cliRes.status, cliRes.length, cliRes.content.c_str(), result.fd_);
+          //          ClientListenerLogging::log(
+          //              "to send data response-- status:%d  length:%d
+          //              content:%s  fd:%d",
+          //              cliRes.status, cliRes.length, cliRes.content.c_str(),
+          //              result.fd_);
           server->write(result.fd_, cliRes);
           printf("send ok response packet ok\n");
 
           cliRes.setSchema(result.result_->schema_);
-          ClientListenerLogging::log(
-              "to send data response-- status:%d  length:%d  content:%s  fd:%d",
-              cliRes.status, cliRes.length, cliRes.content.c_str(), result.fd_);
+          //          ClientListenerLogging::log(
+          //              "to send data response-- status:%d  length:%d
+          //              content:%s  fd:%d",
+          //              cliRes.status, cliRes.length, cliRes.content.c_str(),
+          //              result.fd_);
           server->write(result.fd_, cliRes);
           printf("send schema response packet ok\n");
 
@@ -782,9 +786,11 @@ void *ClientListener::sendHandler(void *para) {
             header.add_header(list[i]);
           }
           cliRes.setAttributeName(header);
-          ClientListenerLogging::log(
-              "to send data response-- status:%d  length:%d  content:%s  fd:%d",
-              cliRes.status, cliRes.length, cliRes.content.c_str(), result.fd_);
+          //          ClientListenerLogging::log(
+          //              "to send data response-- status:%d  length:%d
+          //              content:%s  fd:%d",
+          //              cliRes.status, cliRes.length, cliRes.content.c_str(),
+          //              result.fd_);
           server->write(result.fd_, cliRes);
           printf("send head response packet ok\n");
 
@@ -797,11 +803,13 @@ void *ClientListener::sendHandler(void *para) {
           while (block = (BlockStreamBase *)it.atomicNextBlock()) {
             block->serialize(serialzed_block);
             cliRes.setDataBlock(serialzed_block);
-            ClientListenerLogging::log(
-                "to send data response-- status:%d  length:%d  content:%s  "
-                "fd:%d",
-                cliRes.status, cliRes.length, cliRes.content.c_str(),
-                result.fd_);
+            //            ClientListenerLogging::log(
+            //                "to send data response-- status:%d  length:%d
+            //                content:%s  "
+            //                "fd:%d",
+            //                cliRes.status, cliRes.length,
+            //                cliRes.content.c_str(),
+            //                result.fd_);
             server->write(result.fd_, cliRes);
           }
           printf("send data packet ok\n");
@@ -812,6 +820,11 @@ void *ClientListener::sendHandler(void *para) {
               cliRes.status, cliRes.length, cliRes.content.c_str());
           server->write(result.fd_, cliRes);
           printf("send end response packet ok\n");
+          if (NULL != result.result_) {
+            delete result.result_;
+            result.result_ = NULL;
+            //            cout << "delete result in memory" << endl;
+          }
         }
       }
     } else {
@@ -828,6 +841,11 @@ void *ClientListener::sendHandler(void *para) {
           result.warning_.c_str());
       server->write(result.fd_, cliRes);
       printf("send error packet ok\n");
+      if (NULL != result.result_) {
+        delete result.result_;
+        result.result_ = NULL;
+        cout << "delete result in memory" << endl;
+      }
     }
   }
   return NULL;
@@ -1035,8 +1053,9 @@ int ClientListener::write(const int fd, const ClientResponse &res) const {
 
   //	ret = ::write(fd, buffer, length);
   ret = send(fd, buffer, length, MSG_WAITALL);
-  LOG(INFO) << "Server: " << res.status << " bytes:" << ret << "\t"
-            << res.length << "\t" << res.content.c_str() << " is send!" << endl;
+  //  LOG(INFO) << "Server: " << res.status << " bytes:" << ret << "\t"
+  //            << res.length << "\t" << res.content.c_str() << " is send!" <<
+  //            endl;
   //	ret = ::write(fd,buffer,length);
   //	if (ret < 0) {
   //		ClientLogging::elog("when send to fd %d, send buffer failed.%s",

Client/ClientResponse.h

@@ -196,12 +196,14 @@ struct ClientResponse {
     void* content_start = buffer + sizeof(int) + sizeof(int);
     memcpy(content_start, content.data(), content.length());
 
-    ClientLogging::log("buffer to send is : %d-%d-%d-%d-%d-%d-%d-%d|	%d=%d",
-                       (unsigned int)buffer[0], (unsigned int)buffer[1],
-                       (unsigned int)buffer[2], (unsigned int)buffer[3],
-                       (unsigned int)buffer[4], (unsigned int)buffer[5],
-                       (unsigned int)buffer[6], (unsigned int)buffer[7], status,
-                       length);
+    //    ClientLogging::log("buffer to send is : %d-%d-%d-%d-%d-%d-%d-%d|
+    //    %d=%d",
+    //                       (unsigned int)buffer[0], (unsigned int)buffer[1],
+    //                       (unsigned int)buffer[2], (unsigned int)buffer[3],
+    //                       (unsigned int)buffer[4], (unsigned int)buffer[5],
+    //                       (unsigned int)buffer[6], (unsigned int)buffer[7],
+    //                       status,
+    //                       length);
     return ret;
   }
 

Client/Test/Makefile.am

@@ -1,11 +1,9 @@
-
 AM_CPPFLAGS= -fPIC -fpermissive \
 -I${BOOST_HOME} \
 -I${BOOST_HOME}/boost/serialization \
--I${HADOOP_HOME}/src/c++/libhdfs\
+-I${HADOOP_HOME}/include\
 -I${JAVA_HOME}/include\
--I${JAVA_HOME}/include/linux\
--I${THERON_HOME}/Include
+-I${JAVA_HOME}/include/linux
 
 #<boost/unordered_map.hpp>应该加什么编译选项
 AM_LDFLAGS=-lc -lm -lrt lboost_serialization -lxs
@@ -17,10 +15,9 @@ endif
 
 LDADD = 	../../catalog/libcatalog.a \
 		../../common/libcommon.a \
-		${HADOOP_HOME}/c++/Linux-amd64-64/lib/libhdfs.a \
+		${HADOOP_HOME}/lib/native/libhdfs.a\
 		${JAVA_HOME}/jre/lib/amd64/server/libjvm.so\
-		${HADOOP_HOME}/c++/Linux-amd64-64/lib/libhdfs.so \
-		${THERON_HOME}/Lib/libtherond.a \
+		${HADOOP_HOME}/lib/native/libhdfs.so\
 		${BOOST_HOME}/stage/lib/libboost_serialization.so \
 		${BOOST_HOME}/stage/lib/libboost_serialization.a
 

Config.cpp

@@ -20,7 +20,7 @@ using namespace std;
 string gete() {
   char *p = getenv("CLAIMS_HOME");
   stringstream sp;
-  sp << string(p).c_str() << "/conf/config";
+  sp << string(p).c_str() << "conf/config";
   return sp.str();
   //	return "/home/imdb/config/wangli/config";
 }
@@ -89,6 +89,7 @@ std::string Config::catalog_file;
 int Config::thread_pool_init_thread_num;
 
 int Config::load_thread_num;
+int Config::memory_utilization;
 
 Config *Config::getInstance() {
   if (instance_ == 0) {
@@ -151,6 +152,8 @@ void Config::initialize() {
 
   load_thread_num = getInt("load_thread_num", sysconf(_SC_NPROCESSORS_CONF));
 
+  memory_utilization = getInt("memory_utilization", 100);
+
 #ifdef DEBUG_Config
   print_configure();
 #endif

Config.h

@@ -77,6 +77,7 @@ class Config {
   static bool enable_codegen;
   static std::string catalog_file;
   static int thread_pool_init_thread_num;
+  static int memory_utilization;
 
   static int load_thread_num;
 

Daemon/Daemon.h

@@ -21,6 +21,8 @@
 using claims::utility::LockGuard;
 
 struct remote_command {
+  remote_command() {}
+  remote_command(std::string s, int fd) : cmd(s), socket_fd(fd) {}
   std::string cmd;
   int socket_fd;
 };