Merge pull request #4 from GMAP/v0.4.3-alpha

V0.4.3 alpha

Changelog.txt

@@ -1,3 +1,9 @@
+v0.4.3-alpha
+  - [CLI update] Exec command now allows users to set a range for nthreads. It will execute the benchmark in loop iterating throughout the range.
+  - [New benchmarks] SPBench now includes benchmarks using GrPPI.
+  - [CLI update] Exec command now have a "quiet" option. It will run the benchmarks showing less information on the screen.
+  - [New feature] If repetitions and nthreads range are used together in the exec command, a specific resulting log will be automatically generated inside the log folder. This log file will store average latency, throughput, and exec. time, with their respective std. deviation side by side.
+
 * v0.4.2-alpha
   - [New feature] Now, all the benchmark set can be moved to another place just by moving the "benchmarks" folder and replacing it in the new place.
   - [New feature] SPBench now keeps a persistent execution log in a log.csv file inside the 'logs' folder. This general log contains some results and parameters of each execution.

README.md

@@ -48,7 +48,7 @@ The main goal of SPBench is to enable users to easily create custom benchmarks f
  - Intel TBB
  - FastFlow
  - SPar
- - GrPPI (forthcoming)
+ - GrPPI (backends: Intel TBB, FastFlow, OpenMP, and C++11 threads)
  - Standard C++ Threads (forthcoming)
  - OpenMP (forthcoming)
 

benchmarks/benchmarks_registry.json

@@ -12,6 +12,9 @@
         },
         "fastflow": {
             "bzip2_ff_farm": "simple"
+        },
+        "grppi": {
+            "bzip2_grppi_farm": "single"
         }
     },
     "lane_detection": {
@@ -27,6 +30,9 @@
         },
         "fastflow": {
             "lane_ff_farm": "simple"
+        },
+        "grppi": {
+            "lane_grppi_farm": "simple"
         }
     },
     "ferret": {
@@ -48,6 +54,12 @@
             "ferret_ff_farm": "simple",
             "ferret_ff_pipeline": "simple",
             "ferret_ff_a2a_pipe-farm": "single"
+        },
+        "grppi": {
+            "ferret_grppi_pipe-farm": "single",
+            "ferret_grppi_pipeline": "single",
+            "ferret_grppi_farm": "simple",
+            "ferret_grppi_farm-pipe": "single"
         }
     },
     "person_recognition": {
@@ -64,6 +76,8 @@
         "fastflow": {
             "person_ff_farm": "simple"
         },
-        "mpi": {}
+        "grppi": {
+            "person_grppi_farm": "single"
+        }
     }
 }

benchmarks/bzip2/fastflow/bzip2_ff_farm/bzip2_ff_farm.cpp

@@ -22,6 +22,7 @@ struct Worker_comp: ff::ff_node_t<spb::Item>{
 struct Collector_comp: ff::ff_node_t<spb::Item>{
 	spb::Item * svc(spb::Item * item){
 		spb::Sink::op(*item);
+		delete item;
 		return GO_ON;
 	}
 }Collector_comp;

benchmarks/bzip2/grppi/bzip2_grppi_farm/bzip2.hpp

@@ -0,0 +1,36 @@
+#ifndef BZIP2_H
+#define BZIP2_H
+
+#include <bzip2_utils.hpp>
+
+namespace spb{
+class Compress;
+class Decompress;
+
+class Compress{
+private:
+	static inline void compress_op(spb::item_data &item);
+public:
+	static void op(spb::Item &item);
+	Compress(spb::Item &item){
+		op(item);
+	}
+    Compress(){};
+
+	virtual ~Compress(){}
+};
+
+class Decompress{
+private:
+	static inline void decompress_op(spb::item_data &item);
+public:
+	static void op(spb::Item &item);
+	Decompress(spb::Item &item){
+		op(item);
+	}
+    Decompress(){};
+	virtual ~Decompress(){}
+};
+
+} // end of namespace spb
+#endif

benchmarks/bzip2/grppi/bzip2_grppi_farm/bzip2_grppi_farm.cpp

@@ -0,0 +1,87 @@
+#include <bzip2.hpp>
+#include "grppi/grppi.h"
+#include "dyn/dynamic_execution.h"
+
+using namespace std;
+using namespace experimental;
+
+void run_compress(grppi::dynamic_execution & ex) {
+	tbb::task_scheduler_init init(spb::nthreads);
+	grppi::pipeline(ex,
+		[]() mutable -> optional<spb::Item> {
+			spb::Item item;
+			if(!spb::Source::op(item)) {
+					return {};
+			}
+			return item;
+		},
+		grppi::farm(spb::nthreads,
+			[](spb::Item item) {
+				spb::Compress::op(item);                 
+				return item;
+		}),
+		[](spb::Item item) {spb::Sink::op(item); }
+	);
+}
+
+void run_decompress(grppi::dynamic_execution & ex) {
+	tbb::task_scheduler_init init(spb::nthreads);
+	grppi::pipeline(ex,
+		[]() mutable -> optional<spb::Item> {
+			spb::Item item;
+			if(!spb::Source_d::op(item)) {
+					return {};
+			} else {
+					return item;
+			}
+		},
+		grppi::farm(spb::nthreads,
+			[](spb::Item item) {
+				spb::Decompress::op(item);                 
+				return item;
+		}),
+		[](spb::Item item) {spb::Sink_d::op(item); }
+	);
+}
+
+grppi::dynamic_execution execution_mode(){
+	string backend = spb::SPBench::getArg(0);
+	if(backend == "tbb"){
+		auto ex = grppi::parallel_execution_tbb(spb::nthreads, true);
+		ex.set_queue_attributes(1, grppi::queue_mode::blocking, spb::nthreads*10);
+		return ex;
+	} else if (backend == "ff"){
+		auto ex = grppi::parallel_execution_ff(spb::nthreads, true);
+		return ex;
+	} else if (backend == "omp"){
+		auto ex = grppi::parallel_execution_omp(spb::nthreads, true);
+		ex.set_queue_attributes(1, grppi::queue_mode::blocking);
+		return ex;
+	} else if (backend == "thr"){
+		auto ex = grppi::parallel_execution_native(spb::nthreads, true);
+		ex.set_queue_attributes(1, grppi::queue_mode::blocking);
+		return ex;
+	} else {
+		cout << "No backend selected" << endl;
+		exit(1);
+	}
+}
+
+void compress(){
+	spb::Metrics::init();
+	auto ex = execution_mode();
+	run_compress(ex);
+	spb::Metrics::stop();
+}
+
+void decompress(){
+	spb::Metrics::init();
+	auto ex = execution_mode();
+	run_decompress(ex);
+	spb::Metrics::stop();
+}
+
+int main (int argc, char* argv[]){
+	spb::bzip2_main(argc, argv);
+	return 0;
+}

benchmarks/bzip2/grppi/bzip2_grppi_farm/config.json

@@ -0,0 +1,32 @@
+{
+    "CXX": "g++ -std=c++1y",
+    "CXX_FLAGS":"-O3 -finline-functions",
+    "PPI_CXX": "g++ -std=c++1y",
+    "PPI_CXX_FLAGS":"-O3 -finline-functions",
+    "PRE_SRC_CMD": "",
+    "POST_SRC_CMD": "",
+    "MACROS": "-DNO_CMAKE_CONFIG -D_LARGEFILE64_SOURCE -D_FILE_OFFSET_BITS=64 -DGRPPI_TBB -DGRPPI_FF -DGRPPI_OMP",
+    "PKG-CONFIG": {
+            "myPKG_1": "",
+            "myPKG_2": "",
+            "myPKG_N": ""
+    },
+    "INCLUDES": {
+            "myINC_1": "",
+            "myINC_2": "",
+            "myINC_N": "",
+            "ff":"-I $SPB_HOME/ppis/grppi/fastflow-2.2.0",
+            "grppi":"-I $SPB_HOME/ppis/grppi/grppi-0.4.0/include/",
+            "grppi_":"-I $SPB_HOME/ppis/grppi/grppi-0.4.0/grppi/include/",
+            "tbb": "-I $SPB_HOME/ppis/tbb/tbb/include/",
+            "bzlib":"-I $SPB_HOME/libs/bzlib/include/"
+    },
+    "LIBS": {
+            "myLIB_1": "",
+            "myLIB_2": "",
+            "myLIB_N": "",
+            "tbb"    : "-L $SPB_HOME/ppis/tbb/tbb/ -ltbb",
+            "bzlib"  : "-L $SPB_HOME/libs/bzlib/lib/ -lbz2"
+    },
+    "LDFLAGS": "-lpthread -fopenmp"
+}

benchmarks/bzip2/grppi/bzip2_grppi_farm/operators/include/compress_op.hpp

@@ -0,0 +1,24 @@
+#include <bzip2.hpp>
+
+namespace spb{
+
+void Compress::op(Item &item){	Metrics metrics;
+	volatile unsigned long latency_op;
+	if(metrics.latency_is_enabled()){
+		latency_op = current_time_usecs();
+	}
+	unsigned int num_item = 0;
+
+	while(num_item < item.batch_size){ //batch loop
+
+		compress_op(item.item_batch[num_item]);
+
+		num_item++;
+	}
+
+	if(metrics.latency_is_enabled()){
+		item.latency_op.push_back(current_time_usecs() - latency_op);
+	}
+}
+
+}

benchmarks/bzip2/grppi/bzip2_grppi_farm/operators/include/decompress_op.hpp

@@ -0,0 +1,24 @@
+#include <bzip2.hpp>
+
+namespace spb{
+
+void Decompress::op(Item &item){	Metrics metrics;
+	volatile unsigned long latency_op;
+	if(metrics.latency_is_enabled()){
+		latency_op = current_time_usecs();
+	}
+	unsigned int num_item = 0;
+
+	while(num_item < item.batch_size){ //batch loop
+
+		decompress_op(item.item_batch[num_item]);
+
+		num_item++;
+	}
+
+	if(metrics.latency_is_enabled()){
+		item.latency_op.push_back(current_time_usecs() - latency_op);
+	}
+}
+
+}

benchmarks/bzip2/grppi/bzip2_grppi_farm/operators/operators.json

@@ -0,0 +1,4 @@
+{
+    "compress" : "",
+    "decompress" : ""
+}

benchmarks/bzip2/grppi/bzip2_grppi_farm/operators/src/compress_op.cpp

@@ -0,0 +1,26 @@
+#include <../include/compress_op.hpp>
+
+inline void spb::Compress::compress_op(spb::item_data &item){
+
+    unsigned int outSize = (int) ((item.buffSize*1.01)+600);
+
+    // allocate memory for compressed data
+    item.CompDecompData == NULL;
+    item.CompDecompData = new char[outSize];
+
+    // make sure memory was allocated properly
+    if (item.CompDecompData == NULL)
+    {
+        fprintf(stderr, "Bzip2: *ERROR: Could not allocate memory (CompressedData)!  Skipping...\n");
+        exit(-1);	
+    }
+
+    // compress the memory buffer (blocksize=9*100k, verbose=0, worklevel=30)
+    int ret = BZ2_bzBuffToBuffCompress(item.CompDecompData, &outSize, item.FileData, item.buffSize, BWTblockSize, Verbosity, 30);
+
+    if (ret != BZ_OK)
+        fprintf(stderr, "Bzip2: *ERROR during compression: %d\n", ret);
+
+    item.buffSize = outSize;
+
+}

benchmarks/bzip2/grppi/bzip2_grppi_farm/operators/src/decompress_op.cpp

@@ -0,0 +1,64 @@
+#include <../include/decompress_op.hpp>
+
+inline void spb::Decompress::decompress_op(spb::item_data &item){
+
+    //int blockNum = 0;
+#ifdef PBZIP_DEBUG
+    fprintf(stderr, "consumer:  Buffer: %x  Size: %u   Block: %d\n", item.FileData, item.buffSize, blockNum);
+#endif
+
+#ifdef PBZIP_DEBUG
+    printf ("consumer: recieved %d.\n", blockNum);
+#endif
+
+    unsigned int outSize = 900000;
+
+    // allocate memory for decompressed data (start with default 900k block size)
+    item.CompDecompData = new char[outSize];
+    // make sure memory was allocated properly
+    if (item.CompDecompData == NULL)
+    {
+        fprintf(stderr, " *ERROR: Could not allocate memory (DecompressedData)!  Skipping...\n");
+        exit(-1);
+    }
+
+    // decompress the memory buffer (verbose=0)
+    int ret = BZ2_bzBuffToBuffDecompress(item.CompDecompData, &outSize, item.FileData, item.buffSize, 0, Verbosity);
+
+    while (ret == BZ_OUTBUFF_FULL)
+    {
+#ifdef PBZIP_DEBUG
+        fprintf(stderr, "Increasing DecompressedData buffer size: %d -> %d\n", outSize, outSize*4);
+#endif
+
+        if (item.CompDecompData != NULL)
+            delete [] item.CompDecompData;
+        item.CompDecompData = NULL;
+        // increase buffer space
+        outSize = outSize * 4;
+        // allocate memory for decompressed data (start with default 900k block size)
+        item.CompDecompData = new char[outSize];
+        // make sure memory was allocated properly
+        if (item.CompDecompData == NULL)
+        {
+            fprintf(stderr, "Bzip2: *ERROR: Could not allocate memory (DecompressedData)!  Skipping...\n");
+            exit(-1);
+        }
+
+        // decompress the memory buffer (verbose=0)
+        ret = BZ2_bzBuffToBuffDecompress(item.CompDecompData, &outSize, item.FileData, item.buffSize, 0, Verbosity);
+
+    } // while
+
+    if ((ret != BZ_OK) && (ret != BZ_OUTBUFF_FULL))
+        fprintf(stderr, "Bzip2: *ERROR during decompression: %d\n", ret);
+
+#ifdef PBZIP_DEBUG
+    fprintf(stderr, "\n Compressed Block Size: %u\n", item.buffSize);
+    fprintf(stderr, "   Original Block Size: %u\n", outSize);
+#endif
+
+    blockNum++;
+    item.buffSize = outSize;
+
+}

benchmarks/bzip2/tbb/bzip2_tbb_farm/bzip2_tbb_farm.cpp

@@ -31,6 +31,7 @@ class stage3_comp : public tbb::filter{
 	void* operator() (void* new_item){
 		spb::Item * item = static_cast <spb::Item*> (new_item);
 		spb::Sink::op(*item);
+		delete item;
 		return NULL;
 	}
 };

benchmarks/ferret/fastflow/ferret_ff_a2a_pipe-farm/ferret_ff_a2a_pipe-farm.cpp

@@ -61,6 +61,7 @@ struct Sink: ff::ff_minode_t<spb::Item>{
     Sink(){}
     spb::Item * svc(spb::Item * item){
         spb::Sink::op(*item);
+		delete item;
         return GO_ON;
     }
 };