Fix code 0.1.6.7

main.cpp

@@ -8,20 +8,50 @@
 #include <syslog.h>
 #include <vector>
 #include <string>
+#include <string.h>
+#include "splaytree.h"
+#include "utiltools.h"
+#include <sys/time.h>
+
 using namespace std;
 
 void *lookUpRedis(void *arg) {
-   LogUtil::debug("main [print] : %s", (char *)arg);
+   LogUtil::debug("main [lookUpRedis] : %s", (char *)arg);
    return arg;
 }
 
 void *searchCache(void *arg) {
-   LogUtil::debug("main [print] : %s", (char *)arg);
+   LogUtil::debug("main [searchCache] : %s", (char *)arg);
+   test_msg tm;
+   memcpy(&tm, arg, sizeof(tm));
+   LogUtil::debug("main [searchCache] : name=%s,passwd=%s", tm.name, tm.passwd);
+
+   void *cache = NULL;
+   cache = stage::getPersistData("cache");
+   if (!cache) {
+      stage::setPersistData("cache");
+   }
+   splay_tree *splay_tree_cache = (splay_tree *)cache;
+   splay_tree *r = splaytree_splay(splay_tree_cache, utiltool::hash((char *)arg, strlen((char *)arg)));
+   if (r) { 
+      LogUtil::debug("main [searchCache] : cache hit");
+      struct timeval time;
+      gettimeofday(&time, NULL);
+      tm.end_time = time.tv_sec % 1000 * 1000 + time.tv_usec / 1000;
+      LogUtil::debug("main [success] : time cost=%ld", tm.end_time - tm.begin_time);
+      return NULL;
+   }
+
    return arg;
 }
 
 void *receiveAndResponse(void *arg) {
-   LogUtil::debug("main [print] : %s", (char *)arg);
+   LogUtil::debug("main [receiveAndResponse] : %s", (char *)arg);
+
+   test_msg tm;
+   memcpy(&tm, arg, sizeof(tm));
+   LogUtil::debug("main [receiveAndResponse] : name=%s,passwd=%s\n", tm.name, tm.passwd);
+
    return arg;
 }
 
@@ -97,21 +127,20 @@ void daemonize() {
    syslog(LOG_DEBUG, "daem ok ");    
 }
 
-void runStage(vector<string> &res, string &des, void *(*handler)(void *)) {
+void runStage(vector<string> &res, string &des, void *(*handler)(void *), const char *stage_name) {
    LogUtil::init();
    LogUtil::debug("main.cpp : [main] config success");
    Config config(CONFIG_FILE);
 
-   stage s;
+   stage s(stage_name);
    s.setResources(res);
    s.setDestination(des);
    s.init(config);
-   LogUtil::debug("main : [main] stage s1 init success");
+   LogUtil::debug("main : [main] stage %s init success", stage_name);
    Function fun;
    fun.setFunction(handler);
    s.getHandler()->setHandler(fun);
    s.run();
-   while (1) {}
 }
 
 int main(int argc, char **argv) { 
@@ -121,8 +150,8 @@ int main(int argc, char **argv) {
    vector<string> _des;
    vector<string> resources;
    string des;
-   resources.push_back("tcp://localhost:80");
    resources.push_back("tcp://localhost:5666");
+   //resources.push_back("tcp://localhost:5666");
    res.push_back(resources);
    resources.clear();
    resources.push_back("tcp://localhost:5667");
@@ -139,20 +168,20 @@ int main(int argc, char **argv) {
       /*
        * stage 1.
        */
-      runStage(res[0], _des[0], receiveAndResponse);
+      runStage(res[0], _des[0], receiveAndResponse, "receive_response");
    } else {
       if (pid < 0) {
          fprintf(stderr, "fork error\n");
          exit(-1);
       } else {
          if ((pid = fork())) {
-            runStage(res[1], _des[1], searchCache);
+            runStage(res[1], _des[1], searchCache, "search_cache");
          } else {
             if (pid < 0) {
                fprintf(stderr, "fork error\n");
                exit(-1);
             }
-            runStage(res[2], _des[2], lookUpRedis);
+            runStage(res[2], _des[2], lookUpRedis, "lookup_redis");
          }
       }
    }

marcos.h

@@ -7,6 +7,7 @@
 #define FORCE_CLOSE 0
 #define MAX_QUEUE_SIZE 51200
 #define USE_SENDFILE 1
+#define REG_MEM 16
 
 /*
  * For worker pool

receiver.cpp

@@ -45,7 +45,7 @@ void *receiver::run(void *arg) {
       //LogUtil::debug("receiver : [run] already recv");
       int len = (nrecv >= 256) ? 255 : nrecv;
       buf[255] = 0;
-      IElement ie(buf, len);
+      IElement ie(buf, len + 1);
       pthread_mutex_lock(&rc->lock);
       if (rc->elements.size() > MAX_QUEUE_SIZE) {
          LogUtil::debug("receiver [run], the rc->queue is less");

stage.cpp

@@ -6,6 +6,17 @@ stage::stage() {
 
 stage::stage(string stage_name) {
    this->stage_name = stage_name;
+   this->reg_mem = (void **)malloc(sizeof(void *) * REG_MEM);
+   memset(this->reg_mem, 0, sizeof(void *) * REG_MEM);
+}
+
+stage::~stage() {
+   for (int i = 0; i < REG_MEM; ++i) {
+      if (reg_mem[i]) {
+         free(reg_mem[i]);
+      }
+   }
+   free(reg_mem);
 }
 
 string stage::get_stage_name() {
@@ -62,6 +73,7 @@ bool stage::run() {
    }
    ec->run();
    */
+   while (true) {}
    return true;
 }
 

stage.h

@@ -16,6 +16,7 @@ using namespace std;
 class stage { 
 public:
    stage();
+   ~stage();
    stage(string stage_name);
    string get_stage_name();
    bool init(Config &config);
@@ -40,7 +41,7 @@ class stage {
       this->stage_name = name;
    }
 
-   static void setPersistData(string key, void *data);
+   static void setPersistData(string key);
    static void *getPersistData(string key);
    static map<string, void *> persistData;
 
@@ -57,6 +58,8 @@ class stage {
    stage_control *sc;
    event_core *ec;
 
+   void **reg_mem;
+
 };
 
 #endif /* _STAGE_H */

struct.h

@@ -45,4 +45,10 @@ struct user_info {
 */
 typedef void *(*function)(void *);
 
+struct test_msg {
+   char name[8];
+   char passwd[8];
+   long begin_time, end_time;
+};
+
 #endif ///

utiltools.cpp

@@ -10,3 +10,15 @@ std::list<char *> split_string(char *src, char *tail, char token) {
    }
    return _l;
 }
+
+unsigned int utiltool::hash(const char *_str, int len) {
+   unsigned int hash = 0;
+
+   char *str = (char *)_str;
+   while (len--) {
+      // equivalent to: hash = 65599*hash + (*str++);
+      hash = (*str++) + (hash << 6) + (hash << 16) - hash;
+   }
+
+   return (hash & 0x7FFFFFFF);
+}

utiltools.h

@@ -5,6 +5,7 @@
 class utiltool {
 public:
    static std::list<char *> split_string(char *src, char *tail, char token);
+   static unsigned int hash(const char *str, int len);
 };
 
 #endif //