Repository: intellistream/Sesame (language: python)

Issue:
Code Reformat to include more evaluations



Discussion / hints:

Fix patch:
diff --git a/benchmark/src/Benchmark.cpp b/benchmark/src/Benchmark.cpp
index b7c8f09d..7c35cab9 100644
--- a/benchmark/src/Benchmark.cpp
+++ b/benchmark/src/Benchmark.cpp
@@ -1,4 +1,3 @@
-
 // Copyright (C) 2021 by the IntelliStream team (https://github.com/intellistream)
 
 /**
@@ -28,6 +27,7 @@ int main(int argc, char **argv) {
 
   //Create Spout.
   SESAME::DataSourcePtr sourcePtr = SESAME::DataSourceFactory::create();
+
   //Directly load data from file. TODO: configure it to load from external sensors, e.g., HTTP.
   BenchmarkUtils::loadData(cmd_params, sourcePtr);
 
diff --git a/include/Algorithm/Algorithm.hpp b/include/Algorithm/Algorithm.hpp
index b095e893..7a2f7a8e 100644
--- a/include/Algorithm/Algorithm.hpp
+++ b/include/Algorithm/Algorithm.hpp
@@ -14,6 +14,8 @@
 using namespace std;
 namespace SESAME {
 
+enum algoType { BirchType, StreamKMeansType, CluStreamType, DenStreamType };
+
 class Algorithm;
 typedef std::shared_ptr<Algorithm> AlgorithmPtr;
 
diff --git a/include/Engine/SimpleEngine.hpp b/include/Engine/SimpleEngine.hpp
index 255d8a3a..ef6085bb 100644
--- a/include/Engine/SimpleEngine.hpp
+++ b/include/Engine/SimpleEngine.hpp
@@ -26,6 +26,7 @@ class SimpleEngine : SESAME::Engine {
   AlgorithmPtr algoPtr;
   SingleThreadPtr threadPtr;//SimpleEngine has only one thread to run algorithm.
   atomic_int threadID;
+  TimeMeter overallMeter;
  public:
   BarrierPtr barrierPtr;
   SimpleEngine(DataSourcePtr sourcePtr, DataSinkPtr sinkPtr, AlgorithmPtr algoPtr);
@@ -39,6 +40,7 @@ class SimpleEngine : SESAME::Engine {
              int id);//start the algorithm thread.
   bool stop();
   int assignID();
+  void printTime();
 };
 }
 #endif //SESAME_INCLUDE_Engine_SINGLETHREADENGINE_H_
diff --git a/include/Evaluation/Evaluation.hpp b/include/Evaluation/Evaluation.hpp
index ebcaad48..6f8671d7 100644
--- a/include/Evaluation/Evaluation.hpp
+++ b/include/Evaluation/Evaluation.hpp
@@ -10,7 +10,7 @@
 #include <vector>
 namespace SESAME {
 
-enum evaluateType { euclideanCost };
+enum evaluateType { euclideanCostType };
 
 class Evaluation {
  public:
diff --git a/include/Sinks/DataSink.hpp b/include/Sinks/DataSink.hpp
index ef543da6..f95ade94 100644
--- a/include/Sinks/DataSink.hpp
+++ b/include/Sinks/DataSink.hpp
@@ -32,7 +32,7 @@ class DataSink {
   void runningRoutine();
   bool start(int id);
   bool stop();
-  void sourceEnded();
+  void Ended();
   bool isFinished();
   std::vector<PointPtr> getResults();
   void setBarrier(BarrierPtr barrierPtr);
diff --git a/include/Sources/DataSource.hpp b/include/Sources/DataSource.hpp
index 11f378ed..7931626a 100644
--- a/include/Sources/DataSource.hpp
+++ b/include/Sources/DataSource.hpp
@@ -14,6 +14,7 @@
 #include <Utils/UtilityFunctions.hpp>
 #include <Utils/SPSCQueue.hpp>
 #include <Engine/SingleThread.hpp>
+#include <Timer/TimeMeter.hpp>
 
 using namespace std;
 
@@ -27,8 +28,10 @@ class DataSource {
   std::shared_ptr<rigtorp::SPSCQueue<PointPtr>> inputQueue;
   SingleThreadPtr threadPtr;
   BarrierPtr barrierPtr;
+  TimeMeter overallMeter;
+  bool sourceEnd;
  public:
-  void load(int point_number, int dimension, vector <string> input);
+  void load(int point_number, int dimension, vector<string> input);
   bool empty();
   PointPtr get();
   std::vector<PointPtr> getInputs();
@@ -38,7 +41,8 @@ class DataSource {
   bool start(int i);
   bool stop();
   void setBarrier(BarrierPtr barrierPtr);
-
+  void printTime();
+  bool sourceEnded();
 };
 }
 #endif //SESAME_SRC_SOURCES_DATASOURCE_HPP_
diff --git a/include/Timer/TimeMeter.hpp b/include/Timer/TimeMeter.hpp
new file mode 100644
index 00000000..eb4e30b8
--- /dev/null
+++ b/include/Timer/TimeMeter.hpp
@@ -0,0 +1,20 @@
+// Copyright (C) 2021 by the IntelliStream team (https://github.com/intellistream)
+
+#ifndef SESAME_SRC_TIMER_CYCLEMETER_HPP_
+#define SESAME_SRC_TIMER_CYCLEMETER_HPP_
+
+#include <time.h>           /* gettimeofday */
+#include <cstddef>
+#include <cstdint>
+namespace SESAME {
+class TimeMeter {
+ private:
+  struct timespec start, stop;
+ public:
+  void START_MEASURE();
+  void END_MEASURE();
+  long MeterUSEC();//return the meter result in micro second unit.
+};
+}
+
+#endif //SESAME_SRC_TIMER_CYCLEMETER_HPP_
diff --git a/include/Utils/BenchmarkUtils.hpp b/include/Utils/BenchmarkUtils.hpp
index b4deaa52..77ed62b3 100644
--- a/include/Utils/BenchmarkUtils.hpp
+++ b/include/Utils/BenchmarkUtils.hpp
@@ -46,7 +46,7 @@ struct param_t {
   double beta;
   std::string inputPath;
   std::string outputPath;
-  std::string algoName;
+  SESAME::algoType algoType;
   SESAME::evaluateType evaluateType;
 };
 
diff --git a/src/Algorithm/AlgorithmFactory.cpp b/src/Algorithm/AlgorithmFactory.cpp
index 1027e04f..95fe6541 100644
--- a/src/Algorithm/AlgorithmFactory.cpp
+++ b/src/Algorithm/AlgorithmFactory.cpp
@@ -11,19 +11,19 @@
 #include <Algorithm/AlgorithmFactory.hpp>
 
 SESAME::AlgorithmPtr SESAME::AlgorithmFactory::create(param_t &cmd_params) {
-  if (cmd_params.algoName == "StreamKMeans") {
+  if (cmd_params.algoType == SESAME::StreamKMeansType) {
     shared_ptr<StreamKM> streamkm = std::make_shared<StreamKM>(cmd_params);;
     return (SESAME::AlgorithmPtr) streamkm;
   }
-  if (cmd_params.algoName == "CluStream") {
+  if (cmd_params.algoType == SESAME::CluStreamType) {
     shared_ptr<CluStream> cluStream = std::make_shared<CluStream>(cmd_params);
     return (SESAME::AlgorithmPtr) cluStream;
   }
-  if (cmd_params.algoName == "Birch") {
+  if (cmd_params.algoType == SESAME::BirchType) {
     shared_ptr<Birch> birch = std::make_shared<Birch>(cmd_params);
     return (SESAME::AlgorithmPtr) birch;
   }
-  if (cmd_params.algoName == "DenStream") {
+  if (cmd_params.algoType == SESAME::DenStreamType) {
     shared_ptr<DenStream> denStream = std::make_shared<DenStream>(cmd_params);
     return (SESAME::AlgorithmPtr) denStream;
   }
diff --git a/src/Algorithm/DenStream.cpp b/src/Algorithm/DenStream.cpp
index 37ec2f89..6ae30758 100644
--- a/src/Algorithm/DenStream.cpp
+++ b/src/Algorithm/DenStream.cpp
@@ -1,10 +1,12 @@
+// Copyright (C) 2021 by the IntelliStream team (https://github.com/intellistream)
+
 //
 // Created by 1124a on 2021/8/23.
 //
 #include <Algorithm/DenStream.hpp>
 #include <Algorithm/WindowModel/WindowFactory.hpp>
 #include <Algorithm/DataStructure/DataStructureFactory.hpp>
-SESAME::DenStream::DenStream(param_t &cmd_params){
+SESAME::DenStream::DenStream(param_t &cmd_params) {
   this->denStreamParams.pointNumber = cmd_params.pointNumber;
   this->denStreamParams.dimension = cmd_params.dimension;
   this->denStreamParams.minPoints = cmd_params.minPoints;
@@ -16,63 +18,60 @@ SESAME::DenStream::DenStream(param_t &cmd_params){
   this->denStreamParams.initBufferSize = cmd_params.initBuffer;
 
 }
-SESAME::DenStream:: ~DenStream()
-{
+SESAME::DenStream::~DenStream() {
 
 }
-void SESAME::DenStream::init(vector <PointPtr> &initData){
-  SESAME_INFO("tp is "<<this->Tp);
-  this->pMicroClusterIndex=-1;
-  this->oMicroClusterIndex=-1;
-  for(int i=0;i<denStreamParams.initBufferSize;i++)
-  {
-    if(initData.at(i)->getClusteringCenter()==noVisited)
-    {
+void SESAME::DenStream::init(vector<PointPtr> &initData) {
+  SESAME_INFO("tp is " << this->Tp);
+  this->pMicroClusterIndex = -1;
+  this->oMicroClusterIndex = -1;
+  for (int i = 0; i < denStreamParams.initBufferSize; i++) {
+    if (initData.at(i)->getClusteringCenter() == noVisited) {
       std::vector<int> pointIndex;
       pMicroClusterIndex++;
-      MicroClusterPtr newMicroCluster=SESAME::DataStructureFactory::createMicroCluster(denStreamParams.dimension,pMicroClusterIndex);
+      MicroClusterPtr newMicroCluster =
+          SESAME::DataStructureFactory::createMicroCluster(denStreamParams.dimension, pMicroClusterIndex);
       newMicroCluster->init(initData.at(i), 0);
-      pointsNearCorePoint(initData,pointIndex,newMicroCluster);
-      if(newMicroCluster->weight<=this->minWeight)//TODO need to change minweight
+      pointsNearCorePoint(initData, pointIndex, newMicroCluster);
+      if (newMicroCluster->weight <= this->minWeight)//TODO need to change minweight
       {
         pMicroClusterIndex--;
-        for(vector<int>::size_type a=0;a<pointIndex.size();a++)
-        {
-          int index=pointIndex.at(a);
+        for (vector<int>::size_type a = 0; a < pointIndex.size(); a++) {
+          int index = pointIndex.at(a);
           initData.at(index)->setClusteringCenter(noVisited);
         }
-      }
-      else
+      } else
         pMicroClusters.push_back(newMicroCluster);
     }
   }
-  SESAME_INFO("NOW PMC number is: "<<this->pMicroClusterIndex<<" , Init succeed!");
+  SESAME_INFO("NOW PMC number is: " << this->pMicroClusterIndex << " , Init succeed!");
 }
-void SESAME::DenStream::pointsNearCorePoint(vector <PointPtr> &initData,std::vector<int> pointIndex,MicroClusterPtr microCluster){
+void SESAME::DenStream::pointsNearCorePoint(vector<PointPtr> &initData,
+                                            std::vector<int> pointIndex,
+                                            MicroClusterPtr microCluster) {
   int size = denStreamParams.initBufferSize;
-  for (int i = 0; i < size; i++)
-  {
-    if(initData.at(i)->getClusteringCenter()==noVisited)
-    {
-      double dist=microCluster->calCentroidDistance(initData.at(i));
-      if (dist<denStreamParams.epsilon)
-      {
+  for (int i = 0; i < size; i++) {
+    if (initData.at(i)->getClusteringCenter() == noVisited) {
+      double dist = microCluster->calCentroidDistance(initData.at(i));
+      if (dist < denStreamParams.epsilon) {
         initData[i]->setClusteringCenter(microCluster->id.front());
-        microCluster->insert(initData.at(i),0);
+        microCluster->insert(initData.at(i), 0);
         pointIndex.push_back(i);
       }
     }
   }
 }
-void SESAME::DenStream::Initilize(){
-  this->dampedWindow = WindowFactory::createDampedWindow(denStreamParams.base,  denStreamParams.lambda);
-  this->dbscan=std::make_shared<DBSCAN>(denStreamParams.minPoints,denStreamParams.epsilon,denStreamParams.initBufferSize);
+void SESAME::DenStream::Initilize() {
+  this->dampedWindow = WindowFactory::createDampedWindow(denStreamParams.base, denStreamParams.lambda);
+  this->dbscan =
+      std::make_shared<DBSCAN>(denStreamParams.minPoints, denStreamParams.epsilon, denStreamParams.initBufferSize);
   this->startTime = clock();
-  this->pointArrivingTime=this->startTime;
-  this->minWeight=denStreamParams.beta*denStreamParams.mu;
-  this->Tp=(long)(1/denStreamParams.lambda)*(log(minWeight/(minWeight-1))/log(denStreamParams.base));
+  SESAME_INFO("Start time: " << this->startTime);
+  this->pointArrivingTime = this->startTime;
+  this->minWeight = denStreamParams.beta * denStreamParams.mu;
+  this->Tp = (long) (1 / denStreamParams.lambda) * (log(minWeight / (minWeight - 1)) / log(denStreamParams.base));
 }
-void  SESAME::DenStream::runOnlineClustering(PointPtr input) {
+void SESAME::DenStream::runOnlineClustering(PointPtr input) {
 
   if (!this->isInitial) {
     Initilize();
@@ -82,103 +81,92 @@ void  SESAME::DenStream::runOnlineClustering(PointPtr input) {
       init(this->initialBuffer);
       this->isInitial = true;
     }
-  }
-  else {
+  } else {
 
-    this->pointArrivingTime=clock();
+    this->pointArrivingTime = clock();
+    SESAME_INFO("pointArrivingTime time: " << this->pointArrivingTime);
     merge(input);
 
-    long elapsedTime=(long)(this->pointArrivingTime-this->startTime)/CLOCKS_PER_SEC;
-    //SESAME_INFO("Merge! "<<elapsedTime<<","<<elapsedTime%this->Tp);
+    long elapsedTime = (long) (this->pointArrivingTime - this->startTime) / CLOCKS_PER_SEC;
+    SESAME_INFO("Merge! " << elapsedTime << "," << elapsedTime % this->Tp);
 
-    if(elapsedTime%this->Tp==0)
-    {//SESAME_INFO("Check");
-      for(int iter=0;iter<pMicroClusters.size();iter++)
-      {
-        if(pMicroClusters.at(iter)->weight<minWeight)
-        {
-          pMicroClusters.erase(pMicroClusters.begin()+iter);
-           SESAME_INFO("NOW PMC number is: "<<this->pMicroClusterIndex);
+    if (elapsedTime % this->Tp == 0) {//SESAME_INFO("Check");
+      for (int iter = 0; iter < pMicroClusters.size(); iter++) {
+        if (pMicroClusters.at(iter)->weight < minWeight) {
+          pMicroClusters.erase(pMicroClusters.begin() + iter);
+          SESAME_INFO("NOW PMC number is: " << this->pMicroClusterIndex);
         }
       }
-      if(oMicroClusters.size()!=0)
-      {
-        for(int iter=0;iter<oMicroClusters.size();iter++)
-        {
-          double a=(-(denStreamParams.lambda)*floor(((pointArrivingTime-oMicroClusters.at(iter)->createTime)
-              /CLOCKS_PER_SEC+this->Tp)));
-          double b=(-denStreamParams.lambda*this->Tp);
-          double Xi=(pow(denStreamParams.base,a )-1)/(pow(denStreamParams.base,b)-1);
-         // SESAME_INFO("NOW Xi  "<<Xi);
-          if(oMicroClusters.at(iter)->weight<Xi)
-          {
-            oMicroClusters.erase(oMicroClusters.begin()+iter);
-           SESAME_INFO("NOW oMicroClusterIndex number is: "<<oMicroClusters.size());
+      if (oMicroClusters.size() != 0) {
+        for (int iter = 0; iter < oMicroClusters.size(); iter++) {
+          double a = (-(denStreamParams.lambda) * floor(((pointArrivingTime - oMicroClusters.at(iter)->createTime)
+              / CLOCKS_PER_SEC + this->Tp)));
+          double b = (-denStreamParams.lambda * this->Tp);
+          double Xi = (pow(denStreamParams.base, a) - 1) / (pow(denStreamParams.base, b) - 1);
+          // SESAME_INFO("NOW Xi  "<<Xi);
+          if (oMicroClusters.at(iter)->weight < Xi) {
+            oMicroClusters.erase(oMicroClusters.begin() + iter);
+            SESAME_INFO("NOW oMicroClusterIndex number is: " << oMicroClusters.size());
           }
         }
       }
     }
-   // SESAME_INFO("Insert Succeed "<<iterpoint);
+    // SESAME_INFO("Insert Succeed "<<iterpoint);
   }
   iterpoint++;
 }
 
-void SESAME::DenStream::merge(PointPtr dataPoint){
-  int index=-1;
-  if(!this->pMicroClusters.empty())
-  {
-    index=mergeToMicroCluster(dataPoint,this->pMicroClusters);
+void SESAME::DenStream::merge(PointPtr dataPoint) {
+  int index = -1;
+  if (!this->pMicroClusters.empty()) {
+    index = mergeToMicroCluster(dataPoint, this->pMicroClusters);
     //SESAME_INFO("Merge into PMC! "<<iterpoint<<","<< index<<",");
   }
 
-  if(index<0&&!this->oMicroClusters.empty())
-  {
-    index=mergeToMicroCluster(dataPoint,this->oMicroClusters);
-   // SESAME_INFO("Merge into OMC! "<<iterpoint<<","<< index<<",");
-    if(index>=0)
-    { double decayFactor= this->dampedWindow->decayFunction(this->oMicroClusters.at(index)->lastUpdateTime,pointArrivingTime);
+  if (index < 0 && !this->oMicroClusters.empty()) {
+    index = mergeToMicroCluster(dataPoint, this->oMicroClusters);
+    // SESAME_INFO("Merge into OMC! "<<iterpoint<<","<< index<<",");
+    if (index >= 0) {
+      double decayFactor =
+          this->dampedWindow->decayFunction(this->oMicroClusters.at(index)->lastUpdateTime, pointArrivingTime);
       //SESAME_INFO("Merge INTO OMC! "<<iterpoint<<","<<index);
-      if((this->oMicroClusters.at(index)->weight)*decayFactor>minWeight)
-      {
-       // SESAME_INFO("erase OMC and turn into PMC! ");
+      if ((this->oMicroClusters.at(index)->weight) * decayFactor > minWeight) {
+        // SESAME_INFO("erase OMC and turn into PMC! ");
         pMicroClusterIndex++;
         oMicroClusters.at(index)->resetID(pMicroClusterIndex);
-        pMicroClusters.push_back( oMicroClusters.at(index)->copy());
-        oMicroClusters.erase(oMicroClusters.begin()+index);
+        pMicroClusters.push_back(oMicroClusters.at(index)->copy());
+        oMicroClusters.erase(oMicroClusters.begin() + index);
 
       }
     }
 
   }
-  if(index<0)
-  {
+  if (index < 0) {
     oMicroClusterIndex++;
-    MicroClusterPtr newOMicroCluster=DataStructureFactory::createMicroCluster(denStreamParams.dimension, oMicroClusterIndex);
+    MicroClusterPtr
+        newOMicroCluster = DataStructureFactory::createMicroCluster(denStreamParams.dimension, oMicroClusterIndex);
     newOMicroCluster->init(dataPoint, 0);
     oMicroClusters.push_back(newOMicroCluster->copy());
-   // SESAME_INFO("Create new OMC! "<<iterpoint<<","<<oMicroClusterIndex);
+    // SESAME_INFO("Create new OMC! "<<iterpoint<<","<<oMicroClusterIndex);
   }
 }
-int SESAME::DenStream::mergeToMicroCluster(PointPtr dataPoint,std::vector <MicroClusterPtr> microClusters){
-  int index=nearestNeighbor(dataPoint,microClusters);
-  if(index!=-1)
-  {
-    double decayFactor= this->dampedWindow->decayFunction(pointArrivingTime,clock());
-    if(!microClusters.at(index)->insert(dataPoint,decayFactor,denStreamParams.epsilon))
-      index=-1;
+int SESAME::DenStream::mergeToMicroCluster(PointPtr dataPoint, std::vector<MicroClusterPtr> microClusters) {
+  int index = nearestNeighbor(dataPoint, microClusters);
+  if (index != -1) {
+    double decayFactor = this->dampedWindow->decayFunction(pointArrivingTime, clock());
+    if (!microClusters.at(index)->insert(dataPoint, decayFactor, denStreamParams.epsilon))
+      index = -1;
   }
   return index;
 }
-int SESAME::DenStream::nearestNeighbor(PointPtr dataPoint,std::vector <MicroClusterPtr> microClusters){
-  int index=-1;
-  double dist=0,minDist= std::numeric_limits<double>::max();
-  for(vector<MicroClusterPtr>::size_type i=0;i<microClusters.size();i++)
-  {
-    dist=microClusters.at(i)->calCentroidDistance(dataPoint);
-    if(dist<minDist)
-    {
-      minDist=dist;
-      index=i;
+int SESAME::DenStream::nearestNeighbor(PointPtr dataPoint, std::vector<MicroClusterPtr> microClusters) {
+  int index = -1;
+  double dist = 0, minDist = std::numeric_limits<double>::max();
+  for (vector<MicroClusterPtr>::size_type i = 0; i < microClusters.size(); i++) {
+    dist = microClusters.at(i)->calCentroidDistance(dataPoint);
+    if (dist < minDist) {
+      minDist = dist;
+      index = i;
     }
   }
 
@@ -187,19 +175,20 @@ int SESAME::DenStream::nearestNeighbor(PointPtr dataPoint,std::vector <MicroClus
 
 void SESAME::DenStream::runOfflineClustering(DataSinkPtr sinkPtr) {
   SESAME_INFO("Finish online");
-  vector <PointPtr> transformedPoints;
+  vector<PointPtr> transformedPoints;
   std::vector<std::vector<PointPtr>> oldGroups;
-  microClusterToPoint(pMicroClusters,transformedPoints);
+  microClusterToPoint(pMicroClusters, transformedPoints);
   this->dbscan->run(transformedPoints);
 
-  this->dbscan->produceResult(transformedPoints,sinkPtr);
+  this->dbscan->produceResult(transformedPoints, sinkPtr);
 
 }
 void SESAME::DenStream::microClusterToPoint(std::vector<MicroClusterPtr> &microClusters,
-                                            vector <PointPtr> &points){
-  for (std::vector<MicroClusterPtr>::size_type i=0; i < microClusters.size(); i++) {
+                                            vector<PointPtr> &points) {
+  for (std::vector<MicroClusterPtr>::size_type i = 0; i < microClusters.size(); i++) {
     PointPtr
-    point = DataStructureFactory::createPoint(i, microClusters.at(i)->weight, microClusters.at(i)->centroid.size(), 0);
+        point =
+        DataStructureFactory::createPoint(i, microClusters.at(i)->weight, microClusters.at(i)->centroid.size(), 0);
     for (int j = 0; j < microClusters.at(i)->centroid.size(); j++)
       point->setFeatureItem(microClusters[i]->centroid[j], j);
     points.push_back(point);
diff --git a/src/CMakeLists.txt b/src/CMakeLists.txt
index b21818e2..2d8729a1 100644
--- a/src/CMakeLists.txt
+++ b/src/CMakeLists.txt
@@ -4,4 +4,5 @@ add_subdirectory(Engine)
 add_subdirectory(Sources)
 add_subdirectory(Sinks)
 add_subdirectory(Evaluation)
-add_subdirectory(Utils)
\ No newline at end of file
+add_subdirectory(Utils)
+add_subdirectory(Timer)
\ No newline at end of file
diff --git a/src/Engine/SimpleEngine.cpp b/src/Engine/SimpleEngine.cpp
index 3e4b1364..346e7116 100644
--- a/src/Engine/SimpleEngine.cpp
+++ b/src/Engine/SimpleEngine.cpp
@@ -62,20 +62,30 @@ void SESAME::SimpleEngine::runningRoutine(DataSourcePtr sourcePtr,
 
   barrierPtr->arrive_and_wait();//wait for source and sink.
   SESAME_INFO("Algorithm start to process data");
+  overallMeter.START_MEASURE();
   //initialization
   algoPtr->Initilize();
+
   // run online clustering
-  while (!sourcePtr->empty()) {
+  while (!sourcePtr->sourceEnded()) {//continuously processing infinite incoming data streams.
+    if (!sourcePtr->empty()) {
+      auto item = sourcePtr->get();
+      algoPtr->runOnlineClustering(item);
+    }
+  }
+  while (!sourcePtr->empty()) {//process the remaining data streams after source stops.
     auto item = sourcePtr->get();
     algoPtr->runOnlineClustering(item);
   }
   // run offline clustering
   algoPtr->runOfflineClustering(sinkPtr);
   SESAME_INFO("Engine sourceEnd process data");
-  sinkPtr->sourceEnded();//Let sink knows that there won't be any more data coming.
+  overallMeter.END_MEASURE();
+  sinkPtr->Ended();//Let sink knows that there won't be any more data coming.
   SESAME_INFO("Engine sourceEnd emit data");
   barrierPtr->arrive_and_wait();//wait for source and sink.
   SESAME_DEBUG("Engine sourceEnd wait for source and sink.");
+  printTime();
 }
 
 bool SESAME::SimpleEngine::stop() {
@@ -92,3 +102,6 @@ bool SESAME::SimpleEngine::stop() {
 int SESAME::SimpleEngine::assignID() {
   return threadID++;
 }
+void SESAME::SimpleEngine::printTime() {
+  SESAME_INFO("Engine takes " << overallMeter.MeterUSEC() << " useconds to finish.");
+}
diff --git a/src/Sinks/DataSink.cpp b/src/Sinks/DataSink.cpp
index 3ddf6f8f..9d66d459 100644
--- a/src/Sinks/DataSink.cpp
+++ b/src/Sinks/DataSink.cpp
@@ -63,7 +63,7 @@ void SESAME::DataSink::setBarrier(SESAME::BarrierPtr barrierPtr) {
 SESAME::DataSink::~DataSink() {
   stop();
 }
-void SESAME::DataSink::sourceEnded() {
+void SESAME::DataSink::Ended() {
   this->sourceEnd = true;
 }
 bool SESAME::DataSink::isFinished() {
diff --git a/src/Sources/DataSource.cpp b/src/Sources/DataSource.cpp
index 5a6b1d55..bf958379 100644
--- a/src/Sources/DataSource.cpp
+++ b/src/Sources/DataSource.cpp
@@ -43,17 +43,22 @@ void SESAME::DataSource::load(int point_number, int dimension, vector<string> in
 SESAME::DataSource::DataSource() {
   inputQueue = std::make_shared<rigtorp::SPSCQueue<PointPtr>>(1000);//TODO: remove hard-coded queue initialization.
   threadPtr = std::make_shared<SingleThread>();
+  sourceEnd = false;
 }
 
 //TODO: we can control the source speed here.
 void SESAME::DataSource::runningRoutine() {
   barrierPtr->arrive_and_wait();
+  overallMeter.START_MEASURE();
   SESAME_INFO("DataSource start to emit data");
   for (PointPtr p: this->input) {
     inputQueue->push(p);
   }
+  sourceEnd = true;//Let engine knows that there won't be any more data coming.
   barrierPtr->arrive_and_wait();
+  overallMeter.END_MEASURE();
   SESAME_INFO("DataSource sourceEnd emit data");
+  printTime();
 }
 
 bool SESAME::DataSource::start(int id) {
@@ -66,7 +71,7 @@ bool SESAME::DataSource::start(int id) {
 }
 bool SESAME::DataSource::stop() {
   if (threadPtr) {
-    SESAME_INFO("DataSource::stop try to join threads="<< threadPtr->getID());
+    SESAME_INFO("DataSource::stop try to join threads=" << threadPtr->getID());
     threadPtr->join();
     threadPtr.reset();
   } else {
@@ -78,11 +83,13 @@ bool SESAME::DataSource::stop() {
 bool SESAME::DataSource::empty() {
   return inputQueue->empty();
 }
+
 SESAME::PointPtr SESAME::DataSource::get() {
   auto rt = *inputQueue->front();
   inputQueue->pop();
   return rt;
 }
+
 void SESAME::DataSource::setBarrier(SESAME::BarrierPtr barrierPtr) {
   this->barrierPtr = barrierPtr;
 }
@@ -92,4 +99,10 @@ SESAME::DataSource::~DataSource() {
 vector<SESAME::PointPtr> SESAME::DataSource::getInputs() {
   return input;
 }
+void SESAME::DataSource::printTime() {
+  SESAME_INFO("DataSource takes " << overallMeter.MeterUSEC() << " useconds to finish.");
+}
+bool SESAME::DataSource::sourceEnded() {
+  return sourceEnd;
+}
 
diff --git a/src/Timer/CMakeLists.txt b/src/Timer/CMakeLists.txt
new file mode 100644
index 00000000..fa0cbf92
--- /dev/null
+++ b/src/Timer/CMakeLists.txt
@@ -0,0 +1,1 @@
+add_source_sesame(TimeMeter.cpp)
\ No newline at end of file
diff --git a/src/Timer/TimeMeter.cpp b/src/Timer/TimeMeter.cpp
new file mode 100644
index 00000000..3955df2e
--- /dev/null
+++ b/src/Timer/TimeMeter.cpp
@@ -0,0 +1,13 @@
+// Copyright (C) 2021 by the IntelliStream team (https://github.com/intellistream)
+
+#include <Timer/TimeMeter.hpp>
+void SESAME::TimeMeter::START_MEASURE() {
+  clock_gettime(CLOCK_REALTIME, &start);
+}
+void SESAME::TimeMeter::END_MEASURE() {
+  clock_gettime(CLOCK_REALTIME, &stop);
+}
+long SESAME::TimeMeter::MeterUSEC() {
+  return (((this->stop).tv_sec * 1000000L + (this->stop).tv_nsec / 1000L)
+      - ((this->start).tv_sec * 1000000L + (this->start).tv_nsec / 1000L));
+}
diff --git a/src/Utils/BenchmarkUtils.cpp b/src/Utils/BenchmarkUtils.cpp
index ff5c7022..e6ab0a27 100644
--- a/src/Utils/BenchmarkUtils.cpp
+++ b/src/Utils/BenchmarkUtils.cpp
@@ -149,8 +149,8 @@ void BenchmarkUtils::defaultParam(param_t &cmd_params) {
   cmd_params.inputPath = std::filesystem::current_path().generic_string() + "/datasets/CoverType.txt";
   SESAME_INFO("Default Input Data Directory: " + cmd_params.inputPath);
   cmd_params.outputPath = "results.txt";
-  cmd_params.algoName = "Birch";//StreamKMeans  CluStream Birch
-  cmd_params.evaluateType = SESAME::euclideanCost;
+  cmd_params.algoType = SESAME::BirchType;
+  cmd_params.evaluateType = SESAME::euclideanCostType;
 }
 
 /* command line handling functions */
@@ -204,7 +204,7 @@ void BenchmarkUtils::runBenchmark(param_t &cmd_params,
   SESAME_INFO("Finished store results: " << sinkPtr->getResults().size());
 
   switch (cmd_params.evaluateType) {
-    case SESAME::euclideanCost:
+    case SESAME::euclideanCostType:
       SESAME::Evaluation::euclideanCost(cmd_params.pointNumber,
                                         sinkPtr->getResults().size(),
                                         cmd_params.dimension,


Test patch:
diff --git a/test/SystemTest/BirchTest.cpp b/test/SystemTest/BirchTest.cpp
index 950919b9..e0c2730b 100644
--- a/test/SystemTest/BirchTest.cpp
+++ b/test/SystemTest/BirchTest.cpp
@@ -1,3 +1,5 @@
+// Copyright (C) 2021 by the IntelliStream team (https://github.com/intellistream)
+
 //
 // Created by tuidan on 2021/8/25.
 //
@@ -25,8 +27,7 @@ TEST(SystemTest, BirchTest) {
 
   cmd_params.inputPath = std::filesystem::current_path().generic_string() + "/datasets/Mock.txt";
   cmd_params.outputPath = "results.txt";
-  cmd_params.algoName = "Birch";
-
+  cmd_params.algoType = SESAME::BirchType;
 
   std::vector<SESAME::PointPtr> input;
   std::vector<SESAME::PointPtr> results;
diff --git a/test/SystemTest/CluStreamTest.cpp b/test/SystemTest/CluStreamTest.cpp
index e68343e8..89476f01 100644
--- a/test/SystemTest/CluStreamTest.cpp
+++ b/test/SystemTest/CluStreamTest.cpp
@@ -31,7 +31,7 @@ TEST(SystemTest, CluStreamTest) {
   cmd_params.offlineTimeWindow = 2;
   cmd_params.inputPath = std::filesystem::current_path().generic_string() + "/datasets/CoverType.txt";
   cmd_params.outputPath = "results.txt";
-  cmd_params.algoName = "CluStream";//StreamKMeans  CluStream
+  cmd_params.algoType = SESAME::CluStreamType;
 
   std::vector<SESAME::PointPtr> input;
   std::vector<SESAME::PointPtr> results;
diff --git a/test/SystemTest/DenStreamTest.cpp b/test/SystemTest/DenStreamTest.cpp
index ff584278..20ee088e 100644
--- a/test/SystemTest/DenStreamTest.cpp
+++ b/test/SystemTest/DenStreamTest.cpp
@@ -1,3 +1,5 @@
+// Copyright (C) 2021 by the IntelliStream team (https://github.com/intellistream)
+
 //
 // Created by Zhenyu on 2021/8/25.
 //
@@ -27,7 +29,7 @@ TEST(SystemTest, DenStreamTest) {
   cmd_params.initBuffer = 1000;
   cmd_params.inputPath = std::filesystem::current_path().generic_string() + "/datasets/CoverType.txt";
   cmd_params.outputPath = "results.txt";
-  cmd_params.algoName = "DenStream";
+  cmd_params.algoType = SESAME::DenStreamType;
 
   std::vector<SESAME::PointPtr> input;
   std::vector<SESAME::PointPtr> results;
diff --git a/test/SystemTest/StreamKMTest.cpp b/test/SystemTest/StreamKMTest.cpp
index fa3314ea..9bc4fcbc 100644
--- a/test/SystemTest/StreamKMTest.cpp
+++ b/test/SystemTest/StreamKMTest.cpp
@@ -26,7 +26,7 @@ TEST(SystemTest, StreamKMTest) {
   cmd_params.inputPath = std::filesystem::current_path().generic_string() + "/datasets/Mock.txt";
   SESAME_INFO("Default Input Data Directory: " + cmd_params.inputPath);
   cmd_params.outputPath = "results.txt";
-  cmd_params.algoName = "StreamKMeans";
+  cmd_params.algoType = SESAME::StreamKMeansType;
 
 
   std::vector<SESAME::PointPtr> input;