Skip to content
GitLab
Commit 3319a980 authored by Jason Wu's avatar Jason Wu
Browse files

leap mode and ska-mpg 10 bit unpacker first commit

parent be610572
Pipeline #92139 failed with stages
in 16 seconds
Hide whitespace changes
Inline Side-by-side
psrdada_cpp/CMakeLists.txt
View file @ 3319a980
......@@ -9,6 +9,7 @@ set(psrdada_cpp_src
src/dada_client_base.cpp
src/dada_db.cpp
src/dada_disk_sink.cpp
src/dada_disk_sink_leap.cpp
src/dada_null_sink.cpp
src/dada_output_stream.cpp
src/dada_read_client.cpp
......@@ -28,6 +29,7 @@ set(psrdada_cpp_inc
dada_client_base.hpp
dada_db.hpp
dada_disk_sink.hpp
dada_disk_sink_leap.hpp
dada_input_stream.hpp
dada_junk_source.hpp
dada_null_sink.hpp
......@@ -75,11 +77,16 @@ target_link_libraries (dbnull ${PSRDADA_CPP_LIBRARIES})
add_executable(dbreset examples/dbreset.cpp)
target_link_libraries (dbreset ${PSRDADA_CPP_LIBRARIES})
#dbisk
#add_executable(dbdisk examples/dbdisk.cpp)
#target_link_libraries (dbisk ${PSRDADA_CPP_LIBRARIES})
#dbdisk
add_executable(dbdisk examples/dbdisk.cpp)
target_link_libraries (dbdisk ${PSRDADA_CPP_LIBRARIES})
install (TARGETS junkdb dbnull syncdb dbreset fbfuse_output_db DESTINATION bin)
#dbdiskleap
add_executable(dbdiskleap dbdisk_leap.cpp)
target_link_libraries (dbdiskleap ${PSRDADA_CPP_LIBRARIES})
install (TARGETS junkdb dbnull syncdb dbreset fbfuse_output_db dbdisk dbdiskleap DESTINATION bin)
install (TARGETS ${CMAKE_PROJECT_NAME}
RUNTIME DESTINATION bin
LIBRARY DESTINATION lib
......
psrdada_cpp/effelsberg/edd/CMakeLists.txt
View file @ 3319a980
......@@ -13,11 +13,11 @@ set(psrdada_cpp_effelsberg_edd_src
src/DetectorAccumulator.cu
src/GatedSpectrometer.cu
src/EDDPolnMerge.cpp
src/EDDPolnMerge10to8.cpp
src/EDDRoach.cpp
src/EDDRoach_merge.cpp
src/EDDRoach_merge_leap.cpp
src/ScaledTransposeTFtoTFT.cu
src/SKRfiReplacementCuda.cu
src/SpectralKurtosisCuda.cu
src/Tools.cu
src/Unpacker.cu
src/VLBI.cu
......@@ -28,8 +28,10 @@ set(psrdada_cpp_effelsberg_edd_inc
DadaBufferLayout.hpp
DetectorAccumulator.cuh
EDDPolnMerge.hpp
EDDPolnMerge10to8.hpp
EDDRoach.hpp
EDDRoach_merge.hpp
EDDRoach_merge_leap.hpp
FftSpectrometer.cuh
GatedSpectrometer.cuh
Packer.cuh
......@@ -76,6 +78,10 @@ add_executable(edd_merge src/EDDPolnMerge_cli.cpp)
target_link_libraries(edd_merge ${PSRDADA_CPP_EFFELSBERG_EDD_LIBRARIES})
install(TARGETS edd_merge DESTINATION bin)
add_executable(edd_merge_10to8 src/EDDPolnMerge10to8_cli.cpp)
target_link_libraries(edd_merge_10to8 ${PSRDADA_CPP_EFFELSBERG_EDD_LIBRARIES})
install(TARGETS edd_merge_10to8 DESTINATION bin)
add_executable(edd_roach src/EDDRoach_cli.cpp)
target_link_libraries(edd_roach ${PSRDADA_CPP_EFFELSBERG_EDD_LIBRARIES})
install(TARGETS edd_roach DESTINATION bin)
......@@ -84,5 +90,10 @@ add_executable(edd_roach_merge src/EDDRoach_merge_cli.cpp)
target_link_libraries(edd_roach_merge ${PSRDADA_CPP_EFFELSBERG_EDD_LIBRARIES})
install(TARGETS edd_roach_merge DESTINATION bin)
add_executable(edd_roach_merge_leap src/EDDRoach_merge_leap_cli.cpp)
target_link_libraries(edd_roach_merge_leap ${PSRDADA_CPP_EFFELSBERG_EDD_LIBRARIES})
install(TARGETS edd_roach_merge_leap DESTINATION bin)
add_subdirectory(test)
endif(ENABLE_CUDA)
psrdada_cpp/effelsberg/edd/EDDPolnMerge10to8.hpp 0 → 100644
View file @ 3319a980
#ifndef PSRDADA_CPP_EFFELSBERG_EDD_EDDPOLNMERGE10TO8_HPP
#define PSRDADA_CPP_EFFELSBERG_EDD_EDDPOLNMERGE10TO8_HPP
#include "psrdada_cpp/dada_write_client.hpp"
#include "psrdada_cpp/raw_bytes.hpp"
#include "psrdada_cpp/common.hpp"
#include <vector>
namespace psrdada_cpp {
namespace effelsberg {
namespace edd {
class EDDPolnMerge10to8
{
public:
EDDPolnMerge10to8(std::size_t nsamps_per_heap, std::size_t npol, DadaWriteClient& writer);
~EDDPolnMerge10to8();
/**
* @brief A callback to be called on connection
* to a ring buffer.
*
* @detail The first available header block in the
* in the ring buffer is provided as an argument.
* It is here that header parameters could be read
* if desired.
*
* @param block A RawBytes object wrapping a DADA header buffer
*/
void init(RawBytes& block);
/**
* @brief A callback to be called on acqusition of a new
* data block.
*
* @param block A RawBytes object wrapping a DADA data buffer
*/
bool operator()(RawBytes& block);
private:
std::size_t _nsamps_per_heap;
std::size_t _npol;
DadaWriteClient& _writer;
};
} // edd
} // effelsberg
} // psrdada_cpp
#endif //PSRDADA_CPP_EFFELSBERG_EDD_EDDPOLNMERGE10TO8_HPP
psrdada_cpp/effelsberg/edd/EDDRoach_merge_leap.hpp 0 → 100644
View file @ 3319a980
#ifndef PSRDADA_CPP_EFFELSBERG_EDD_EDDROACH_MERGE_LEAP_HPP
#define PSRDADA_CPP_EFFELSBERG_EDD_EDDROACH_MERGE_LEAP_HPP
#include "psrdada_cpp/dada_write_client.hpp"
#include "psrdada_cpp/raw_bytes.hpp"
#include "psrdada_cpp/common.hpp"
#include <vector>
namespace psrdada_cpp {
namespace effelsberg {
namespace edd {
class EDDRoach_merge_leap
{
public:
EDDRoach_merge_leap(std::size_t nsamps_per_heap, std::size_t nchunck, DadaWriteClient& writer);
~EDDRoach_merge_leap();
/**
* @brief A callback to be called on connection
* to a ring buffer.
*
* @detail The first available header block in the
* in the ring buffer is provided as an argument.
* It is here that header parameters could be read
* if desired.
*
* @param block A RawBytes object wrapping a DADA header buffer
*/
void init(RawBytes& block);
/**
* @brief A callback to be called on acqusition of a new
* data block.
*
* @param block A RawBytes object wrapping a DADA data buffer
*/
bool operator()(RawBytes& block);
private:
std::size_t _nsamps_per_heap;
std::size_t _nchunck;
DadaWriteClient& _writer;
};
} // edd
} // effelsberg
} // psrdada_cpp
#endif //PSRDADA_CPP_EFFELSBERG_EDD_EDDROACH_MERGE_LEAP_HPP
psrdada_cpp/effelsberg/edd/dada_disk_sink_leap.hpp 0 → 100644
View file @ 3319a980
#ifndef PSRDADA_CPP_EFFELSBERG_EDD_DADA_DISK_SINK_LEAP_HPP
#define PSRDADA_CPP_EFFELSBERG_EDD_DADA_DISK_SINK_LEAP_HPP
#include "psrdada_cpp/raw_bytes.hpp"
#include "psrdada_cpp/common.hpp"
#include <fstream>
namespace psrdada_cpp{
namespace effelsberg {
namespace edd {
{
class DiskSinkLeap
{
public:
DiskSinkLeap(std::string prefix, DadaWriteClient& writer);
~DiskSinkLeap();
void init(RawBytes&);
bool operator()(RawBytes&);
public:
std::string _prefix;
std::size_t _counter;
DadaWriteClient& _writer;
std::ofstream _current_file;
std::ofstream _current_file_1;
std::ofstream _current_file_2;
std::ofstream _current_file_3;
std::ofstream _current_file_4;
std::ofstream _current_file_5;
std::ofstream _current_file_6;
std::ofstream _current_file_7;
};
} // edd
} // effelsberg
} //namespace psrdada_cpp
#endif //PSRDADA_CPP__EFFELSBERG_EDD_DADA_DISK_SINK_LEAP_HPP
\ No newline at end of file
psrdada_cpp/effelsberg/edd/dbdisk_leap.cpp 0 → 100644
View file @ 3319a980
#include "psrdada_cpp/multilog.hpp"
#include "psrdada_cpp/raw_bytes.hpp"
#include "psrdada_cpp/dada_input_stream.hpp"
#include "psrdada_cpp/dada_disk_sink.hpp"
#include "psrdada_cpp/cli_utils.hpp"
#include "boost/program_options.hpp"
#include <sys/types.h>
#include <iostream>
#include <string>
#include <sstream>
#include <ios>
#include <algorithm>
using namespace psrdada_cpp;
namespace
{
const size_t ERROR_IN_COMMAND_LINE = 1;
const size_t SUCCESS = 0;
const size_t ERROR_UNHANDLED_EXCEPTION = 2;
} // namespace
int main(int argc, char** argv)
{
try
{
key_t key;
std::string prefix;
/** Define and parse the program options
*/
namespace po = boost::program_options;
po::options_description desc("Options");
desc.add_options()
("help,h", "Print help messages")
("key,k", po::value<std::string>()
->default_value("dada")
->notifier([&key](std::string in)
{
key = string_to_key(in);
}),
"The shared memory key for the dada buffer to connect to (hex string)")
("prefix,p", po::value<std::string>(&prefix)
->default_value("dbdisk_dump"),
"Prefix for the filename to write to")
("log_level", po::value<std::string>()
->default_value("info")
->notifier([](std::string level)
{
set_log_level(level);
}),
"The logging level to use (debug, info, warning, error)");
po::variables_map vm;
try
{
po::store(po::parse_command_line(argc, argv, desc), vm);
if ( vm.count("help") )
{
std::cout << "DbDiskLeap -- read from DADA ring buffer and write to disk in LEAP spec" << std::endl
<< desc << std::endl;
return SUCCESS;
}
po::notify(vm);
}
catch(po::error& e)
{
std::cerr << "ERROR: " << e.what() << std::endl << std::endl;
std::cerr << desc << std::endl;
return ERROR_IN_COMMAND_LINE;
}
/**
* All the application code goes here
*/
MultiLog log("dbdixk");
DadaWriteClient output(key, log);
DiskSinkLeap sink(prefix, output);
DadaInputStream<decltype(sink)> stream(key, log, sink);
stream.start();
/**
* End of application code
*/
}
catch(std::exception& e)
{
std::cerr << "Unhandled Exception reached the top of main: "
<< e.what() << ", application will now exit" << std::endl;
return ERROR_UNHANDLED_EXCEPTION;
}
return SUCCESS;
}
\ No newline at end of file
psrdada_cpp/effelsberg/edd/src/EDDPolnMerge10to8.cpp 0 → 100644
View file @ 3319a980
#include "psrdada_cpp/effelsberg/edd/EDDPolnMerge10to8.hpp"
#include "ascii_header.h"
#include <immintrin.h>
#include <time.h>
#include <iomanip>
#include <cmath>
namespace psrdada_cpp {
namespace effelsberg {
namespace edd {
uint64_t interleave(uint32_t x, uint32_t y) {
__m128i xvec = _mm_cvtsi32_si128(x);
__m128i yvec = _mm_cvtsi32_si128(y);
__m128i interleaved = _mm_unpacklo_epi8(yvec, xvec);
return _mm_cvtsi128_si64(interleaved);
}
void handle_packet_numbers_4096x10_s(char const *buf, char *out)
{ // Print 4096 numbers of 10 bit signed integers.
int i;
int16_t adc;
uint64_t val,rest;
uint64_t *qword;
qword = (uint64_t*)(buf);
for(i=0;i<640/5;i++)
{
// 1st:
val = be64toh(*qword);
//printf("0x%016lX\n",val);
qword++;
out[i] = ((int64_t)(( 0xFFC0000000000000 & val) << 0) >> 54) & 0x3FC;
out[i+1] = ((int64_t)(( 0x003FF00000000000 & val) << 10) >> 54) & 0x3FC;
out[i+2] = ((int64_t)(( 0x00000FFC00000000 & val) << 20) >> 54) & 0x3FC;
out[i+3] = ((int64_t)(( 0x00000003FF000000 & val) << 30) >> 54) & 0x3FC;
out[i+4] = ((int64_t)(( 0x0000000000FFC000 & val) << 40) >> 54) & 0x3FC;
out[i+5] = ((int64_t)(( 0x0000000000003FF0 & val) << 50) >> 54) & 0x3FC;
rest = ( 0x000000000000000F & val) << 60; // 4 bits rest.
// 2nd:
val = be64toh(*qword);
//printf("0x%016lX\n",val);
qword++;
out[i+6] = ((int64_t)(((0xFC00000000000000 & val) >> 4) | rest) >> 54) & 0x3FC;
out[i+7] = ((int64_t)(( 0x03FF000000000000 & val) << 6) >> 54) & 0x3FC;
out[i+8] = ((int64_t)(( 0x0000FFC000000000 & val) << 16) >> 54) & 0x3FC;
out[i+9] = ((int64_t)(( 0x0000003FF0000000 & val) << 26) >> 54) & 0x3FC;
out[i+10] =((int64_t)(( 0x000000000FFC0000 & val) << 36) >> 54) & 0x3FC;
out[i+11] = ((int64_t)(( 0x000000000003FF00 & val) << 46) >> 54) & 0x3FC;
rest = ( 0x00000000000000FF & val) << 56; // 8 bits rest.
// 3rd:
val = be64toh(*qword);
//printf("0x%016lX\n",val);
qword++;
out[i+12] = ((int64_t)(((0xC000000000000000 & val) >> 8) | rest) >> 54) & 0x3FC;
out[i+13] = ((int64_t)(( 0x3FF0000000000000 & val) << 2) >> 54) & 0x3FC;
out[i+14] = ((int64_t)(( 0x000FFC0000000000 & val) << 12) >> 54) & 0x3FC;
out[i+15] = ((int64_t)(( 0x000003FF00000000 & val) << 22) >> 54) & 0x3FC;
out[i+16] = ((int64_t)(( 0x00000000FFC00000 & val) << 32) >> 54) & 0x3FC;
out[i+17] = ((int64_t)(( 0x00000000003FF000 & val) << 42) >> 54) & 0x3FC;
out[i+18] = ((int64_t)(( 0x0000000000000FFC & val) << 52) >> 54) & 0x3FC;
rest = ( 0x0000000000000003 & val) << 62; // 2 bits rest.
// 4th:
val = be64toh(*qword);
//printf("0x%016lX\n",val);
qword++;
out[i+19] = ((int64_t)(((0xFF00000000000000 & val) >> 2) | rest) >> 54) & 0x3FC;
out[i+20] = ((int64_t)(( 0x00FFC00000000000 & val) << 8) >> 54) & 0x3FC;
out[i+21] = ((int64_t)(( 0x00003FF000000000 & val) << 18) >> 54) & 0x3FC;
out[i+22] = ((int64_t)(( 0x0000000FFC000000 & val) << 28) >> 54) & 0x3FC;
out[i+23] = ((int64_t)(( 0x0000000003FF0000 & val) << 38) >> 54) & 0x3FC;
out[i+24] = ((int64_t)(( 0x000000000000FFC0 & val) << 48) >> 54) & 0x3FC;
rest = ( 0x000000000000003F & val) << 58; // 6 bits rest.
// 5th:
val = be64toh(*qword);
//printf("0x%016lX\n",val);
qword++;
out[i+25] = ((int64_t)(((0xF000000000000000 & val) >> 6) | rest) >> 54) & 0x3FC;
out[i+26] = ((int64_t)(( 0x0FFC000000000000 & val) << 4) >> 54) & 0x3FC;
out[i+27] = ((int64_t)(( 0x0003FF0000000000 & val) << 14) >> 54) & 0x3FC;
out[i+28] = ((int64_t)(( 0x000000FFC0000000 & val) << 24) >> 54) & 0x3FC;
out[i+29] = ((int64_t)(( 0x000000003FF00000 & val) << 34) >> 54) & 0x3FC;
out[i+30] = ((int64_t)(( 0x00000000000FFC00 & val) << 44) >> 54) & 0x3FC;
out[i+31] = ((int64_t)(( 0x00000000000003FF & val) << 54) >> 54) & 0x3FC;
rest = 0; // No rest.
}
}
EDDPolnMerge10to8::EDDPolnMerge10to8(std::size_t nsamps_per_heap, std::size_t npol, DadaWriteClient& writer)
: _nsamps_per_heap(nsamps_per_heap)
, _npol(npol)
, _writer(writer)
{
}
EDDPolnMerge10to8::~EDDPolnMerge10to8()
{
}
void EDDPolnMerge10to8::init(RawBytes& block)
{
RawBytes& oblock = _writer.header_stream().next();
if (block.used_bytes() > oblock.total_bytes())
{
_writer.header_stream().release();
throw std::runtime_error("Output DADA buffer does not have enough space for header");
}
std::memcpy(oblock.ptr(), block.ptr(), block.used_bytes());
char buffer[1024];
ascii_header_get(block.ptr(), "SAMPLE_CLOCK_START", "%s", buffer);
std::size_t sample_clock_start = std::strtoul(buffer, NULL, 0);
ascii_header_get(block.ptr(), "CLOCK_SAMPLE", "%s", buffer);
long double sample_clock = std::strtold(buffer, NULL);
ascii_header_get(block.ptr(), "SYNC_TIME", "%s", buffer);
long double sync_time = std::strtold(buffer, NULL);
BOOST_LOG_TRIVIAL(debug) << "this is sample_clock_start "<< sample_clock_start;
BOOST_LOG_TRIVIAL(debug)<< "this is sample_clock "<< sample_clock;
BOOST_LOG_TRIVIAL(debug) << "this is sync_time "<< sync_time;
BOOST_LOG_TRIVIAL(debug) << "this is sample_clock_start / sample_clock "<< sample_clock_start / sample_clock;
long double unix_time = sync_time + (sample_clock_start / sample_clock);
long double mjd_time = unix_time / 86400 - 40587.5;
char time_buffer[80];
std::time_t unix_time_int;
struct std::tm * timeinfo;
double fractpart, intpart;
fractpart = std::modf (static_cast<double>(unix_time) , &intpart);
unix_time_int = static_cast<std::time_t>(intpart);
timeinfo = std::gmtime (&unix_time_int);
std::strftime(time_buffer, 80, "%Y-%m-%d-%H:%M:%S", timeinfo);
std::stringstream utc_time_stamp;
BOOST_LOG_TRIVIAL(debug) << "unix_time" << unix_time;
BOOST_LOG_TRIVIAL(debug) << "fractional part " << fractpart;
utc_time_stamp<< time_buffer << "." << std::setw(10) << std::setfill('0') << std::size_t(fractpart*10000000000) << std::setfill(' ');
BOOST_LOG_TRIVIAL(debug) << "this is start time in utc "<< utc_time_stamp.str().c_str()<< "\n";
ascii_header_set(oblock.ptr(), "UTC_START", "%s", utc_time_stamp.str().c_str());
ascii_header_set(oblock.ptr(), "UNIX_TIME", "%Lf", unix_time);
oblock.used_bytes(oblock.total_bytes());
_writer.header_stream().release();
BOOST_LOG_TRIVIAL(info) << "Output header released"<< "\n";
}
bool EDDPolnMerge10to8::operator()(RawBytes& block)
{
std::cout << "Beginning of the operator"<< std::endl;
std:size_t nheap_groups = 0.8*block.used_bytes()/_npol/_nsamps_per_heap;
RawBytes& oblock = _writer.data_stream().next();
BOOST_LOG_TRIVIAL(debug) << "block.used_bytes() = "<<block.used_bytes();
BOOST_LOG_TRIVIAL(debug) << "Entering unpack loop";
#pragma omp parallel for schedule(dynamic, 4) num_threads(4)
for (std::size_t kk = 0; kk < block.used_bytes()/ 5120/2 ; ++kk)
{
char *buffer_pol0 = block.ptr() + 5120 * kk * 2;
handle_packet_numbers_4096x10_s(buffer_pol0, oblock.ptr() + kk* 2 * 4096);
char *buffer_pol1 = block.ptr() + (5120 * kk * 2)+ 5120;
handle_packet_numbers_4096x10_s(buffer_pol1, oblock.ptr() + kk* 2 * 4096 + 4096);
}
oblock.used_bytes(block.used_bytes()*0.8);
_writer.data_stream().release();
return false;
}
}//edd
}//effelsberg
}//psrdada_cpp
psrdada_cpp/effelsberg/edd/src/EDDPolnMerge10to8_cli.cpp 0 → 100644
View file @ 3319a980
#include "psrdada_cpp/multilog.hpp"
#include "psrdada_cpp/cli_utils.hpp"
#include "psrdada_cpp/common.hpp"
#include "psrdada_cpp/dada_input_stream.hpp"
#include "psrdada_cpp/dada_write_client.hpp"
#include "psrdada_cpp/effelsberg/edd/EDDPolnMerge10to8.hpp"
#include "boost/program_options.hpp"
using namespace psrdada_cpp;
namespace
{
const size_t ERROR_IN_COMMAND_LINE = 1;
const size_t SUCCESS = 0;
const size_t ERROR_UNHANDLED_EXCEPTION = 2;
} // namespace
int main(int argc, char** argv)
{
try
{
key_t input_key;
key_t output_key;
std::size_t npol;
std::size_t nsamps_per_heap;
/** Define and parse the program options
*/
namespace po = boost::program_options;
po::options_description desc("Options");
desc.add_options()
("help,h", "Print help messages")
("input_key,i", po::value<std::string>()
->default_value("dada")
->notifier([&input_key](std::string in)
{
input_key = string_to_key(in);
}),
"The shared memory key for the dada buffer to connect to (hex string)")
("output_key,o", po::value<std::string>()
->default_value("dadc")
->notifier([&output_key](std::string out)
{
output_key = string_to_key(out);
}),
"The shared memory key for the dada buffer to connect to (hex string)")
("npol,p", po::value<std::size_t>(&npol)->default_value(2),
"Value of number of pol")
("nsamps_per_heap,n", po::value<std::size_t>(&nsamps_per_heap)->default_value(4096),
"Value of samples per heap")
("log_level", po::value<std::string>()
->default_value("info")
->notifier([](std::string level)
{
set_log_level(level);
}),
"The logging level to use (debug, info, warning, error)");
po::variables_map vm;
try
{
po::store(po::parse_command_line(argc, argv, desc), vm);
if ( vm.count("help") )
{
std::cout << "EDDPolnMerge10to8 -- Read EDD data from a DADA buffer and merge the polarizations"
<< std::endl << desc << std::endl;
return SUCCESS;
}
po::notify(vm);
}
catch(po::error& e)
{
std::cerr << "ERROR: " << e.what() << std::endl << std::endl;
std::cerr << desc << std::endl;
return ERROR_IN_COMMAND_LINE;
}
/**
* All the application code goes here
*/
MultiLog log("edd::EDDPolnMerge10to8");
DadaWriteClient output(output_key, log);
effelsberg::edd::EDDPolnMerge10to8 merger(nsamps_per_heap, npol, output);
DadaInputStream <decltype(merger)> input(input_key, log, merger);
input.start();
/**
* End of application code
*/
}
catch(std::exception& e)
{
std::cerr << "Unhandled Exception reached the top of main: "
<< e.what() << ", application will now exit" << std::endl;
return ERROR_UNHANDLED_EXCEPTION;
}
return SUCCESS;
}