VLBI: Allow for varying number fo frames in output block.

psrdada_cpp/effelsberg/edd/VLBI.cuh

@@ -34,10 +34,12 @@ class VDIFHeaderView
 		uint32_t getReferenceEpoch() const;
     size_t getTimestamp() const;
 		uint32_t getDataFrameNumber() const;
+		// Length of the data frame including the header, in units of 8 bytes
 		uint32_t getDataFrameLength() const;
 		uint32_t getNumberOfChannels() const;
     bool isRealDataType() const;
 		bool isComplexDataType() const;
+		// Number of bits per sample -1 (max 32/bits per sample)
 		uint32_t getBitsPerSample() const;
 		uint32_t getThreadId() const;
 		uint32_t getStationId() const;

psrdada_cpp/effelsberg/edd/detail/VLBI.cu

@@ -36,8 +36,8 @@ VLBI<HandlerType>::VLBI(std::size_t buffer_bytes, std::size_t input_bitDepth,
   BOOST_LOG_TRIVIAL(info) << "Creating new VLBI instance";
   BOOST_LOG_TRIVIAL(info) << "   Output data in VDIF format with "
                           << vlbiHeaderSize << "bytes header info and "
-                          << _vdifHeader.getDataFrameLength()
-                          << " bytes payload";
+                          << _vdifHeader.getDataFrameLength() * 8
+                          << " bytes data frame length";
   BOOST_LOG_TRIVIAL(debug) << "   Expecting speadheaps of size "
                            << speadHeapSize << "   byte";
 
@@ -49,14 +49,23 @@ VLBI<HandlerType>::VLBI(std::size_t buffer_bytes, std::size_t input_bitDepth,
   BOOST_LOG_TRIVIAL(debug) << "   Input voltages size : "
                            << _raw_voltage_db.size() << " 64-bit words,"
                            << _raw_voltage_db.size() * 64 / 8 << " bytes";
-
   _unpacked_voltage.resize(n64bit_words * 64 / input_bitDepth );
   _packed_voltage.resize(n64bit_words * 64 / input_bitDepth * _output_bitDepth /
                          8);
-
-  _spillOver.reserve(vdifHeader.getDataFrameLength() * 8);
   BOOST_LOG_TRIVIAL(debug) << "   Output voltages size: "
                            << _packed_voltage.size() << " byte";
+  _spillOver.reserve(vdifHeader.getDataFrameLength() * 8 - vlbiHeaderSize);
+
+	// number of vlbi frames per input block
+	size_t nSamplesPerInputBlock = _packed_voltage.size() * 8 / _output_bitDepth;
+	size_t frames_per_block = _packed_voltage.size() / (vdifHeader.getDataFrameLength() * 8 - vlbiHeaderSize); 
+  BOOST_LOG_TRIVIAL(debug) << "   this correspoonds to " << frames_per_block << " - " << frames_per_block + 1 << " frames";
+
+	_outputBuffer.resize((frames_per_block+1) * vdifHeader.getDataFrameLength() * 8 );
+	// potetnitally invalidating the last frame
+  BOOST_LOG_TRIVIAL(info) << "   Output data in VDIF format with " << _outputBuffer.size() << " bytes per buffer";
+
+
 
   CUDA_ERROR_CHECK(cudaStreamCreate(&_h2d_stream));
   CUDA_ERROR_CHECK(cudaStreamCreate(&_proc_stream));
@@ -228,8 +237,8 @@ bool VLBI<HandlerType>::operator()(RawBytes &block) {
   BOOST_LOG_TRIVIAL(debug) << "   Number of blocks in output "
                            << numberOfBlocksInOutput;
 
-  _outputBuffer.a().resize(numberOfBlocksInOutput *
-                           (outputBlockSize + vlbiHeaderSize));
+  //_outputBuffer.a().resize(numberOfBlocksInOutput *
+   //                        (outputBlockSize + vlbiHeaderSize));
 
   BOOST_LOG_TRIVIAL(debug) << "   Copying " << _spillOver.size()
                            << " bytes spill over";
@@ -279,12 +288,26 @@ bool VLBI<HandlerType>::operator()(RawBytes &block) {
   BOOST_LOG_TRIVIAL(debug) << " Samples per data frame: " << samplesPerDataFrame; 
   BOOST_LOG_TRIVIAL(debug) << " Dataframes per second: " << dataFramesPerSecond; 
 
-  for (uint32_t i = 0; i < numberOfBlocksInOutput; i++)
+  for (uint32_t ib = 0; ib < _outputBuffer.a().size(); ib += _vdifHeader.getDataFrameLength() * 8)
   {
      // copy header to correct position
     std::copy(reinterpret_cast<uint8_t *>(_vdifHeader.getData()),
         reinterpret_cast<uint8_t *>(_vdifHeader.getData()) + vlbiHeaderSize,
-        _outputBuffer.a().begin() + i * (outputBlockSize + vlbiHeaderSize));
+        _outputBuffer.a().begin() + ib);
+		size_t i = ib / _vdifHeader.getDataFrameLength() / 8;
+
+		// invalidate rest of data so it can be dropped later. 
+		// Needed so that the outpuitbuffer can have always the same size
+		if (i < numberOfBlocksInOutput)
+		{
+			_vdifHeader.setValid();
+		}
+		else
+		{
+			_vdifHeader.setInvalid();
+			continue;
+		}
+
     // update header
     uint32_t dataFrame = _vdifHeader.getDataFrameNumber();
     if (i < 5)

psrdada_cpp/effelsberg/edd/src/vdif_send.cu

@@ -88,7 +88,7 @@ class VDIF_Sender
     if (block.used_bytes() == 0)
     {
       BOOST_LOG_TRIVIAL(info) << "Received empty block, exiting.";
-      return true;
+      return false;
     }
     boost::system::error_code err;
     VDIFHeaderView vdifHeader(reinterpret_cast<uint32_t*>(block.ptr()));
@@ -102,6 +102,10 @@ class VDIF_Sender
     for(char* frame_start = block.ptr(); frame_start < block.ptr() + blockSize; frame_start += vdifHeader.getDataFrameLength() * 8)
     {
       vdifHeader.setDataLocation(reinterpret_cast<uint32_t*>(frame_start));
+			// skip invalid blocks
+			if (!vdifHeader.isValid())
+				continue;
+
       uint32_t frameLength = vdifHeader.getDataFrameLength() * 8; // in units of 8 bytes
 
       socket.send_to(boost::asio::buffer(frame_start, frameLength), remote_endpoint, 0, err);