diff --git a/src/FileDmaInputFilter.cc b/src/FileDmaInputFilter.cc
new file mode 100644
index 0000000000000000000000000000000000000000..8e1e76064cef309317f7ad9fe777bb6e75030819
--- /dev/null
+++ b/src/FileDmaInputFilter.cc
@@ -0,0 +1,91 @@
+#include <cassert>
+#include <cstdio>
+#include <cerrno>
+#include <system_error>
+#include <iostream>
+#include <sstream>
+
+#include <sys/types.h>
+#include <sys/stat.h>
+#include <fcntl.h>
+
+#include "FileDmaInputFilter.h"
+
+
+FileDmaInputFilter::FileDmaInputFilter( const std::string& fileName, size_t packetBufferSize, size_t nbPacketBuffers, ctrl& control ) :
+ InputFilter( packetBufferSize, nbPacketBuffers, control )
+{
+ inputFile = fopen( fileName.c_str(), "r" );
+ if ( !inputFile ) {
+ throw std::invalid_argument( "Invalid input file name: " + fileName );
+ }
+ std::cerr << "Created file input filter\n";
+}
+
+FileDmaInputFilter::~FileDmaInputFilter() {
+ fclose( inputFile );
+ std::cerr << "Destroyed file input filter\n";
+}
+
+/*
+ * This function reads packet by packet from a file
+ * in order to simulate DMA reads.
+ *
+ * TODO: Better would be mmap directly the file
+ */
+static inline ssize_t read_dma_packet_from_file(FILE *inputFile, char *buffer, uint64_t size, uint64_t nbReads)
+{
+ static constexpr uint64_t deadbeaf = 0xdeadbeefdeadbeefL;
+ size_t bytesRead = 0;
+ size_t rc;
+
+ while ( !feof(inputFile) && bytesRead < size) {
+ // Expecting 32 byte alignment
+ rc = fread( buffer, 1, 32, inputFile );
+
+ if (ferror(inputFile)) {
+ throw std::system_error(errno, std::system_category(), "File error");
+ }
+
+ if (rc != 32) {
+ if ( feof(inputFile) && rc == 0 && bytesRead == 0) {
+ // We have reached the perfect end of the file, let's start again
+ fseek(inputFile, 0, SEEK_SET);
+ continue;
+ }
+
+ // Misaligned data
+ throw std::runtime_error("#" + std::to_string(nbReads) + ": File read ends prematurely, missing " + std::to_string(32-rc) + " bytes. Something is probably misaligned.");
+ }
+
+ bytesRead += 32;
+
+ if ( *(uint64_t*)(buffer) == deadbeaf ) {
+ // End of the packet was found
+ return bytesRead;
+ }
+
+ buffer += 32;
+ }
+
+ // We are here either
+ // because we found EOF earlier than the end of the packet
+ // or the packet cannot fit into the buffer
+ // We can deal with both conditions but for the moment we throw an error
+
+ if (feof(inputFile)) {
+ throw std::runtime_error("EOF reached but no end of the packet found.");
+ }
+
+ throw std::runtime_error("Packet is too big and cannot fit into preallocated memory");
+ // TODO: Read and discard data until deadbeef is found
+
+ return -1;
+}
+
+
+ssize_t FileDmaInputFilter::readInput(char **buffer, size_t bufferSize)
+{
+ return read_dma_packet_from_file(inputFile, *buffer, bufferSize, nbReads() );
+}
+
diff --git a/src/FileDmaInputFilter.h b/src/FileDmaInputFilter.h
new file mode 100644
index 0000000000000000000000000000000000000000..18395e1f8772d5149dfea189a463e9134cc76d9d
--- /dev/null
+++ b/src/FileDmaInputFilter.h
@@ -0,0 +1,29 @@
+#ifndef FILE_DMA_INPUT_FILTER_H
+#define FILE_DMA_INPUT_FILTER_H
+
+#include <memory>
+#include <string>
+#include "tbb/pipeline.h"
+#include "tbb/tick_count.h"
+
+#include "InputFilter.h"
+
+class FileDmaInputFilter: public InputFilter {
+public:
+ //FileDmaInputFilter( const std::string&, size_t, size_t);
+ FileDmaInputFilter( const std::string& fileName, size_t packetBufferSize, size_t nbPacketBuffers, ctrl& control );
+ virtual ~FileDmaInputFilter();
+
+protected:
+ ssize_t readInput(char **buffer, size_t bufferSize); // Override
+
+ // When reading from file this method does nothing
+ void readComplete(char *buffer) { (void)(buffer); }
+
+private:
+ FILE* inputFile;
+};
+
+typedef std::shared_ptr<FileDmaInputFilter> FileDmaInputFilterPtr;
+
+#endif // FILE_DMA_INPUT_FILTER_H
diff --git a/src/InputFilter.cc b/src/InputFilter.cc
new file mode 100644
index 0000000000000000000000000000000000000000..26e4d9751cd8d1b6795647a5962ed16fa486598d
--- /dev/null
+++ b/src/InputFilter.cc
@@ -0,0 +1,109 @@
+#include <cassert>
+#include <iostream>
+#include <system_error>
+
+#include "InputFilter.h"
+#include "slice.h"
+#include "controls.h"
+
+InputFilter::InputFilter(size_t packetBufferSize, size_t nbPacketBuffers, ctrl& control) :
+ filter(serial_in_order),
+ control_(control),
+ nextSlice_(Slice::preAllocate( packetBufferSize, nbPacketBuffers )),
+ nbReads_(0),
+ nbBytesRead_(0),
+ nbErrors_(0),
+ nbOversized_(0),
+ previousNbBytesRead_(0),
+ previousStartTime_( tbb::tick_count::now() )
+{
+ std::cerr << "Created input filter and allocated at " << static_cast<void*>(nextSlice_) << "\n";
+}
+
+InputFilter::~InputFilter() {
+ std::cerr << "Destroy input filter and delete at " << static_cast<void*>(nextSlice_) << "\n";
+
+ Slice::giveAllocated(nextSlice_);
+ std::cerr << "Input operator performed " << nbReads_ << " read\n";
+}
+
+inline ssize_t InputFilter::readHelper(char **buffer, size_t bufferSize)
+{
+ // Read from DMA
+ int skip = 0;
+ ssize_t bytesRead = readInput( buffer, bufferSize );
+
+ // If large packet returned, skip and read again
+ while ( bytesRead > (ssize_t)bufferSize ) {
+ nbOversized_++;
+ skip++;
+ std::cerr
+ << "#" << nbReads_ << ": ERROR: Read returned " << bytesRead << " > buffer size " << bufferSize
+ << ". Skipping packet #" << skip << ".\n";
+ if (skip >= 100) {
+ throw std::runtime_error("FATAL: Read is still returning large packets.");
+ }
+ bytesRead = readInput( buffer, bufferSize );
+ }
+
+ return bytesRead;
+}
+
+void* InputFilter::operator()(void*) {
+ // Prepare destination buffer
+ char *buffer = nextSlice_->begin();
+ // Available buffer size
+ size_t bufferSize = nextSlice_->avail();
+ ssize_t bytesRead = 0;
+
+ // We need at least 1MB buffer
+ assert( bufferSize >= 1024*1024 );
+
+ nbReads_++;
+
+ // It is optional to use the provided buffer
+ bytesRead = readHelper( &buffer, bufferSize );
+
+ // This should really not happen
+ assert( bytesRead != 0);
+
+ if (buffer != nextSlice_->begin()) {
+ // If read returned a different buffer, then it didn't use our buffer and we have to copy data
+ // FIXME: It is a bit stupid to copy buffer, better would be to use zero copy approach
+ memcpy( nextSlice_->begin(), buffer, bytesRead );
+ }
+
+ // Notify that we processed the given buffer
+ readComplete(buffer);
+
+ nbBytesRead_ += bytesRead;
+
+ // TODO: Make this configurable
+ if (control_.packets_per_report && (nbReads_ % control_.packets_per_report == 0)) {
+ // Calculate DMA bandwidth
+ tbb::tick_count now = tbb::tick_count::now();
+ double time_diff = (double)((now - previousStartTime_).seconds());
+ previousStartTime_ = now;
+
+ uint64_t nbBytesReadDiff = nbBytesRead_ - previousNbBytesRead_;
+ previousNbBytesRead_ = nbBytesRead_;
+
+ double bwd = nbBytesReadDiff / ( time_diff * 1024.0 * 1024.0 );
+
+ std::cout
+ << "#" << nbReads_ << ": Read(s) returned: " << nbBytesReadDiff
+ << ", bandwidth " << bwd << "MBytes/sec, read errors " << nbErrors_
+ << ", read returned oversized packet " << nbOversized_ << ".\n";
+ }
+
+ // Have more data to process.
+ Slice* thisSlice = nextSlice_;
+ nextSlice_ = Slice::getAllocated();
+
+ // Adjust the end of this buffer
+ thisSlice->set_end( thisSlice->end() + bytesRead );
+
+ return thisSlice;
+
+}
+
diff --git a/src/InputFilter.h b/src/InputFilter.h
new file mode 100644
index 0000000000000000000000000000000000000000..e0a21ed86bc3ce8793353899e311c668f65ebc33
--- /dev/null
+++ b/src/InputFilter.h
@@ -0,0 +1,64 @@
+#ifndef INPUT_FILTER_H
+#define INPUT_FILTER_H
+
+#include <cstddef>
+
+#include "tbb/pipeline.h"
+#include "tbb/tick_count.h"
+
+#include "controls.h"
+
+class Slice;
+
+/*
+ * This is an abstract class.
+ * A dervide class has to implement methods readInput and readComplete
+ */
+class InputFilter: public tbb::filter {
+public:
+ //InputFilter( FILE*, size_t, size_t);
+ InputFilter(size_t packet_buffer_size, size_t number_of_packet_buffers, ctrl& control);
+ virtual ~InputFilter();
+
+ uint64_t nbReads() { return nbReads_; }
+
+protected:
+ // Read input to a provided buffer or return a different buffer
+ virtual ssize_t readInput(char **buffer, size_t bufferSize) = 0;
+
+ // Notify the read that the buffer returned by readInput is processed (can be freed or reused)
+ virtual void readComplete(char *buffer) = 0;
+
+private:
+ void* operator()(void* item); // Override
+
+ inline ssize_t readHelper(char **buffer, size_t buffer_size);
+
+private:
+ ctrl& control_;
+ //FILE* input_file;
+ Slice* nextSlice_;
+ //uint32_t counts;
+
+ // Number of successfull reads
+ uint64_t nbReads_;
+
+ // Number of byted read
+ uint64_t nbBytesRead_;
+
+ // Number of read errors detected
+ uint64_t nbErrors_;
+
+ // Number of oversized packets returned ()
+ uint64_t nbOversized_;
+
+ // Performance monitoring
+
+ // Snapshot of nbBytesRead_
+ uint64_t previousNbBytesRead_;
+
+ // Remember timestamp for performance monitoring
+ tbb::tick_count previousStartTime_;
+};
+
+#endif // INPUT_FILTER_H
diff --git a/src/Makefile b/src/Makefile
index 5b21051a23559c531f2e5d1e1a2b2f22068f8b0c..45fb35ea696deb1b8937413a74656d46889cf51f 100644
--- a/src/Makefile
+++ b/src/Makefile
@@ -12,7 +12,7 @@
TARGET = scdaq
# source files
-SOURCES = config.cc dma_input.cc elastico.cc file_dma_input.cc file_input.cc input.cc output.cc processor.cc scdaq.cc session.cc slice.cc wzdma_input.cc
+SOURCES = config.cc dma_input.cc elastico.cc FileDmaInputFilter.cc file_input.cc InputFilter.cc output.cc processor.cc scdaq.cc session.cc slice.cc wzdma_input.cc
C_SOURCES = wz_dma.c
# work out names of object files from sources
@@ -24,7 +24,7 @@ OBJECTS += $(C_SOURCES:.c=.o)
# invocation for both compilation (where -c is needed) and linking
# (where it's not.)
CXXFLAGS = -std=c++11 -Wall -Wextra -O0 -g -rdynamic
-#CXXFLAGS = -std=c++11 -Wall -Wextra
+#CXXFLAGS = -std=c++11 -Wall -Wextra -g -rdynamic
CFLAGS = $(CXXFLAGS)
LDFLAGS = -ltbb -lboost_thread -lcurl
@@ -62,8 +62,9 @@ config.o: config.h
dma_input.o: dma_input.h slice.h
elastico.o: elastico.h format.h slice.h controls.h
file_dma_input.o: file_dma_input.h
+FileDmaInputFilter.o: FileDmaInputFilter.h
file_input.o: file_input.h slice.h utility.h
-input.o: input.h slice.h
+InputFilter.o: InputFilter.h
output.o: output.h slice.h
processor.o: processor.h slice.h format.h
session.o: session.h
diff --git a/src/file_dma_input.cc b/src/file_dma_input.cc
deleted file mode 100644
index 107246eabccf3d810e8923c7fcf3a0b88ab7c7a0..0000000000000000000000000000000000000000
--- a/src/file_dma_input.cc
+++ /dev/null
@@ -1,164 +0,0 @@
-#include <cassert>
-#include <cstdio>
-#include <cerrno>
-#include <system_error>
-#include <iostream>
-#include <sstream>
-
-#include <sys/types.h>
-#include <sys/stat.h>
-#include <fcntl.h>
-
-#include "file_dma_input.h"
-#include "slice.h"
-
-
-FileDmaInputFilter::FileDmaInputFilter( const std::string& file_name_, size_t packet_buffer_size_,
- size_t number_of_packet_buffers_) :
-
- filter(serial_in_order),
- next_slice(Slice::preAllocate( packet_buffer_size_, number_of_packet_buffers_) ),
- counts(0),
- ncalls(0),
- lastStartTime(tbb::tick_count::now()),
- last_count(0)
-{
- input_file = fopen( file_name_.c_str(), "r" );
- if ( !input_file ) {
- throw std::invalid_argument( "Invalid input file name: " + file_name_ );
- }
- fprintf(stderr,"Created input dma filter and allocated at 0x%llx \n",(unsigned long long)next_slice);
-}
-
-FileDmaInputFilter::~FileDmaInputFilter() {
- fprintf(stderr,"Destroy input dma filter and delete at 0x%llx \n",(unsigned long long)next_slice);
- Slice::giveAllocated(next_slice);
- fprintf(stderr,"input operator total %lu read \n",counts);
- // fclose( output_file );
- fclose( input_file );
-}
-
-/*
- * This function reads packet by packet from a file
- * in order to simulate DMA reads.
- *
- * TODO: Better would be mmap directly the file
- */
-static inline ssize_t read_dma_packet_from_file(FILE *input_file, char *buffer, uint64_t size, uint64_t pkt_count)
-{
- static constexpr uint64_t deadbeaf = 0xdeadbeefdeadbeefL;
- size_t bytes_read = 0;
- size_t rc;
-
- while ( !feof(input_file) && bytes_read < size) {
- // Expecting 32 byte alignment
- rc = fread( buffer, 1, 32, input_file );
-
- if (ferror(input_file)) {
- throw std::system_error(errno, std::system_category(), "File error");
- }
-
- if (rc != 32) {
- if ( feof(input_file) && rc == 0 && bytes_read == 0) {
- // We have reached the perfect end of the file, let's start again
- fseek(input_file, 0, SEEK_SET);
- continue;
- }
-
- // Misaligned data
- throw std::runtime_error("#" + std::to_string(pkt_count) + ": File read ends prematurely, missing " + std::to_string(32-rc) + " bytes. Something is probably misaligned.");
- }
-
- bytes_read += 32;
-
- if ( *(uint64_t*)(buffer) == deadbeaf ) {
- // End of the packet was found
- return bytes_read;
- }
-
- buffer += 32;
- }
-
- // We are here either
- // because we found EOF earlier than the end of the packet
- // or the packet cannot fit into the buffer
- // We can deal with both conditions but for the moment we throw an error
-
- if (feof(input_file)) {
- throw std::runtime_error("EOF reached but no end of the packet found.");
- }
-
- throw std::runtime_error("Packet is too big and cannot fit into preallocated memory");
- // TODO: Read and discard data until deadbeef is found
-
- return -1;
-}
-
-
-void* FileDmaInputFilter::operator()(void*) {
- size_t buffer_size = next_slice->avail();
- ssize_t bytes_read = 0;
-
- // We need at least 1MB buffer
- assert( buffer_size >= 1024*1024 );
-
- while (true) {
- // Count reads
- ncalls++;
-
- // Read from DMA
- bytes_read = read_dma_packet_from_file(input_file, next_slice->begin(), buffer_size, ncalls);
-
- if (bytes_read < 0) {
- // Check for errors we can skip
- if (errno == EIO || errno == EMSGSIZE) {
- std::cerr << "Iteration: " << ncalls;
- if (errno == EIO) {
- std::cerr << " DMA ERROR: I/O ERROR, skipping packet...\n";
- } else {
- std::cerr << " DMA ERROR: Packet too long, skipping packet...\n";
- }
- continue;
- }
-
- // Some fatal error occured
- std::ostringstream os;
- os << "Iteration: " << ncalls
- << " ERROR: DMA read failed.";
- throw std::system_error(errno, std::system_category(), os.str() );
- }
-
- // We have some data
- break;
- }
-
- // This should not happen
- if (bytes_read > (ssize_t)buffer_size ){
- std::ostringstream os;
- os << "Iteration: " << ncalls
- << " ERROR: DMA read returned " << bytes_read
- << " > buffer size " << buffer_size;
- throw std::runtime_error( os.str() );
- }
-
- // This should really not happen
- assert( bytes_read != 0);
-
- // Calculate DMA bandwidth
- tbb::tick_count now = tbb::tick_count::now();
- double time_diff = (double)((now - lastStartTime).seconds());
- lastStartTime = now;
- double bwd = bytes_read / ( time_diff * 1024.0 * 1024.0 );
-
- std::cout << "#" << ncalls << ": Read returned: " << bytes_read << ", DMA bandwidth " << bwd << "MBytes/sec\n";
-
- // Have more data to process.
- Slice* this_slice = next_slice;
- next_slice = Slice::getAllocated();
-
- // Adjust the end of this buffer
- this_slice->set_end( this_slice->end() + bytes_read );
-
- return this_slice;
-
-}
diff --git a/src/file_dma_input.h b/src/file_dma_input.h
deleted file mode 100644
index 1094796baab5f90bd75314dc78ff338ee3317181..0000000000000000000000000000000000000000
--- a/src/file_dma_input.h
+++ /dev/null
@@ -1,27 +0,0 @@
-#ifndef FILE_DMA_INPUT_H
-#define FILE_DMA_INPUT_H
-
-#include <memory>
-#include <string>
-#include "tbb/pipeline.h"
-#include "tbb/tick_count.h"
-
-class Slice;
-
-class FileDmaInputFilter: public tbb::filter {
- public:
- FileDmaInputFilter( const std::string&, size_t, size_t);
- ~FileDmaInputFilter();
- private:
- FILE* input_file;
- Slice* next_slice;
- void* operator()(void*) /*override*/;
- uint64_t counts;
- uint64_t ncalls;
- tbb::tick_count lastStartTime;
- uint64_t last_count;
-};
-
-typedef std::shared_ptr<FileDmaInputFilter> FileDmaInputFilterPtr;
-
-#endif
diff --git a/src/format.h b/src/format.h
index b7b53a31df3325a6ff8bd503690b900ce46b2b5e..ef70b49e6b356ed0ce966ae29ae6a377d30694bd 100644
--- a/src/format.h
+++ b/src/format.h
@@ -13,6 +13,16 @@ struct block1{
uint32_t mu2s[8];
};
+// struct out_block1{
+// uint32_t header;
+// uint32_t bx;
+// uint32_t orbit;
+
+// // In the header is information of how many of muons are there
+// uint32_t mu_w1;
+// uint32_t mu_w2;
+// };
+
struct masks{
static constexpr uint32_t phiext = 0x1ff;
static constexpr uint32_t pt = 0x1ff;
diff --git a/src/input.cc b/src/input.cc
deleted file mode 100644
index 2bf2cb87a27bf9e4e06400893a1baf12baa64e84..0000000000000000000000000000000000000000
--- a/src/input.cc
+++ /dev/null
@@ -1,46 +0,0 @@
-#include <cstdio>
-#include "input.h"
-#include "slice.h"
-
-InputFilter::InputFilter( FILE* input_file_ , size_t max_size_,
- size_t nslices_) :
-
- filter(serial_in_order),
- input_file(input_file_),
- next_slice(Slice::preAllocate( max_size_,nslices_) ),
- counts(0)
-{
- fprintf(stderr,"Created input filter and allocated at 0x%llx \n",(unsigned long long)next_slice);
-}
-
-InputFilter::~InputFilter() {
- fprintf(stderr,"Destroy input filter and delete at 0x%llx \n",(unsigned long long)next_slice);
- Slice::giveAllocated(next_slice);
- fprintf(stderr,"input operator total %u read \n",counts);
-}
-
-void* InputFilter::operator()(void*) {
-
- size_t m = next_slice->avail();
- size_t n = fread( next_slice->begin(), 1, m, input_file );
-
- if(n<m){
- fprintf(stderr,"input operator read less %d \n",n);
-
- }
- counts+=n/192;
-
- if( n==0 ) {
- return NULL;
- } else {
- // Have more data to process.
- Slice& t = *next_slice;
- next_slice = Slice::getAllocated();
-
- char* p = t.end()+n;
- t.set_end(p);
-
- return &t;
- }
-
-}
diff --git a/src/input.h b/src/input.h
deleted file mode 100644
index a9c0f82403e19ad879eeb35e76358f45b7044752..0000000000000000000000000000000000000000
--- a/src/input.h
+++ /dev/null
@@ -1,19 +0,0 @@
-#ifndef INPUT_H
-#define INPUT_H
-
-#include "tbb/pipeline.h"
-
-class Slice;
-
-class InputFilter: public tbb::filter {
- public:
- InputFilter( FILE*, size_t, size_t);
- ~InputFilter();
- private:
- FILE* input_file;
- Slice* next_slice;
- void* operator()(void*) /*override*/;
- uint32_t counts;
-};
-
-#endif
diff --git a/src/scdaq.cc b/src/scdaq.cc
index 68a033b17d625161e0872ca46eb849c9df261f33..8b5f195a0009bc9faa394357cf276b351c632bd2 100644
--- a/src/scdaq.cc
+++ b/src/scdaq.cc
@@ -15,9 +15,10 @@
#include <boost/thread.hpp>
+#include "InputFilter.h"
+#include "FileDmaInputFilter.h"
#include "wzdma_input.h"
#include "dma_input.h"
-#include "file_dma_input.h"
#include "file_input.h"
#include "processor.h"
#include "elastico.h"
@@ -37,7 +38,7 @@ using namespace std;
bool silent = false;
-int run_pipeline( int nthreads, ctrl *control, config *conf)
+int run_pipeline( int nthreads, ctrl& control, config *conf)
{
config::InputType input = conf->getInput();
@@ -45,7 +46,7 @@ int run_pipeline( int nthreads, ctrl *control, config *conf)
size_t TOTAL_SLICES = 0;
// Create empty input reader, will assing later when we know what is the data source
- std::shared_ptr<tbb::filter> input_filter;
+ std::shared_ptr<InputFilter> input_filter;
// Create the pipeline
tbb::pipeline pipeline;
@@ -56,7 +57,8 @@ int run_pipeline( int nthreads, ctrl *control, config *conf)
TOTAL_SLICES = conf->getNumInputBuffers();
// Create file-reading writing stage and add it to the pipeline
- input_filter = std::make_shared<FileInputFilter>( conf->getInputFile(), MAX_BYTES_PER_INPUT_SLICE, TOTAL_SLICES );
+ //input_filter = std::make_shared<FileInputFilter>( conf->getInputFile(), MAX_BYTES_PER_INPUT_SLICE, TOTAL_SLICES );
+ throw std::runtime_error("input FILE is not supported");
} else if (input == config::InputType::DMA) {
// Prepare reading from DMA
@@ -64,15 +66,17 @@ int run_pipeline( int nthreads, ctrl *control, config *conf)
TOTAL_SLICES = conf->getNumberOfDmaPacketBuffers();
// Create DMA reader
- input_filter = std::make_shared<DmaInputFilter>( conf->getDmaDevice(), MAX_BYTES_PER_INPUT_SLICE, TOTAL_SLICES );
+ //input_filter = std::make_shared<DmaInputFilter>( conf->getDmaDevice(), MAX_BYTES_PER_INPUT_SLICE, TOTAL_SLICES );
+ throw std::runtime_error("input DMA is not supported");
+
} else if (input == config::InputType::FILEDMA) {
// Prepare reading from FILE and simulating DMA
MAX_BYTES_PER_INPUT_SLICE = conf->getDmaPacketBufferSize();
TOTAL_SLICES = conf->getNumberOfDmaPacketBuffers();
- // Create DMA reader
- input_filter = std::make_shared<FileDmaInputFilter>( conf->getInputFile(), MAX_BYTES_PER_INPUT_SLICE, TOTAL_SLICES );
+ // Create FILE DMA reader
+ input_filter = std::make_shared<FileDmaInputFilter>( conf->getInputFile(), MAX_BYTES_PER_INPUT_SLICE, TOTAL_SLICES, control );
} else if (input == config::InputType::WZDMA ) {
// Prepare reading from WZ DMA
@@ -80,7 +84,8 @@ int run_pipeline( int nthreads, ctrl *control, config *conf)
TOTAL_SLICES = conf->getNumberOfDmaPacketBuffers();
// Create WZ DMA reader
- input_filter = std::make_shared<WZDmaInputFilter>( MAX_BYTES_PER_INPUT_SLICE, TOTAL_SLICES, control );
+ //input_filter = std::make_shared<WZDmaInputFilter>( MAX_BYTES_PER_INPUT_SLICE, TOTAL_SLICES, &control );
+ throw std::runtime_error("input WZDMA is not supported");
} else {
throw std::invalid_argument("Configuration error: Unknown input type was specified");
@@ -104,7 +109,7 @@ int run_pipeline( int nthreads, ctrl *control, config *conf)
std::string url = conf->getElasticUrl();
// TODO: Created here so we are not subject of scoping, fix later...
ElasticProcessor elastic_processor(MAX_BYTES_PER_INPUT_SLICE,
- control,
+ &control,
url,
conf->getPtCut(),
conf->getQualCut());
@@ -115,7 +120,7 @@ int run_pipeline( int nthreads, ctrl *control, config *conf)
std::string output_file_base = conf->getOutputFilenameBase();
// Create file-writing stage and add it to the pipeline
- OutputStream output_stream( output_file_base.c_str() , control);
+ OutputStream output_stream( output_file_base.c_str(), &control);
pipeline.add_filter( output_stream );
// Run the pipeline
@@ -155,7 +160,7 @@ int main( int argc, char* argv[] ) {
int p = conf.getNumThreads();
tbb::task_scheduler_init init(p);
- if(!run_pipeline (p,&control, &conf))
+ if(!run_pipeline (p, control, &conf))
return 1;
// utility::report_elapsed_time((tbb::tick_count::now() - mainStartTime).seconds());
diff --git a/src/scdaq.conf b/src/scdaq.conf
index 08c5ba815a055fbe03bd6f39d79b0d82c4231a49..613161e94b8efdc0285094d4e679e23b0368021a 100644
--- a/src/scdaq.conf
+++ b/src/scdaq.conf
@@ -3,8 +3,8 @@
# "dma" for XILINX DMA driver
# "filedma" for reading from file and simulating DMA
# "file" for reading from file
-input:wzdma
-#input:filedma
+#input:wzdma
+input:filedma
## Settings for DMA input
@@ -19,11 +19,14 @@ dma_packet_buffer_size:1048576
dma_number_of_packet_buffers:1000
# Print report each N packets, use 0 to disable
-packets_per_report:5000
+#packets_per_report:5000
+packets_per_report:1
## Settings for file input
#input_file:/dev/shm/testdata.bin
input_file:testdata.bin
+#input_file:../dumps/dump-empty-run.bin
+
input_buffers:10
blocks_buffer:1000