Skip to content

Commit

Permalink
Switch Sequence and Tag indices to unsigned (#438)
Browse files Browse the repository at this point in the history 
* Switch to Sequence indices to unsigned (fixes signed integer overflow UB)
* Make tag index unsigned
  Make the normal tag index unsigned (std::size_t), to match the new sequence index type.
* Remove Sequence::index_type
* Remove Tag::index_type and replace gr::RelativeIndexTag by simple std::pair.

There's three cases where we need relative indices though:
 - Port::tags() provides the tags with indices relative to stream position.
 - DataSet::timing_events are relative to the first sample in the set, and can
   be negative (e.g. for DataSink's snapshot acquisition mode, the trigger is
   in the past.
 - DataSink callback and pollers use tags relative to the current data chunk.

For these, use RelativeIndexTag, which have a signed index type.

Signed-off-by: Frank Osterfeld <[email protected]>
  • Loading branch information
@frankosterfeld
frankosterfeld authored Nov 8, 2024
1 parent af92697 commit 3273d08
Show file tree
Hide file tree
Showing 27 changed files with 264 additions and 286 deletions.
7 changes: 2 additions & 5 deletions algorithm/include/gnuradio-4.0/algorithm/dataset/DataSetUtils.hpp
View file
Original file line number Diff line number Diff line change
Expand Up @@ -74,7 +74,7 @@ template<DataSetLike TDataSet>
std::ranges::fill(tagVector, std::numeric_limits<TValueType>::lowest());
}
for (const auto& [index, tag] : dataSet.timing_events[i]) {
if (index < 0 || index >= static_cast<Tag::signed_index_type>(tagVector.size())) {
if (index < 0 || index >= static_cast<std::ptrdiff_t>(tagVector.size())) {
continue;
}
tagVector[static_cast<std::size_t>(index)] = dataSet.signal_values[static_cast<std::size_t>(index)];
Expand Down Expand Up @@ -187,10 +187,7 @@ DataSet<T> waveform(WaveType waveType, size_t length, T samplingRate, T frequenc

// check for zero crossing by seeing if the signs of previous and current values differ
if ((previousValue < 0 && currentValue >= 0) || (previousValue > 0 && currentValue <= 0)) {
Tag tag;
tag.index = static_cast<Tag::signed_index_type>(i); // Position of the zero crossing
tag.map["type"] = "Zero Crossing";
dataSet.timing_events[0].emplace_back(tag);
dataSet.timing_events[0].emplace_back(static_cast<std::ptrdiff_t>(i), property_map{{"type", "Zero Crossing"}});
}
previousValue = currentValue;
}
Expand Down
23 changes: 11 additions & 12 deletions blocks/basic/include/gnuradio-4.0/basic/DataSink.hpp
View file
Original file line number Diff line number Diff line change
Expand Up @@ -71,13 +71,12 @@ struct StreamingPoller {

auto readData = reader.get(nProcess);
if constexpr (requires { fnc(std::span<const T>(), std::span<const Tag>()); }) {
auto tags = tag_reader.get();
const auto it = std::find_if_not(tags.begin(), tags.end(), [until = static_cast<int64_t>(samples_read + nProcess)](const auto& tag) { return tag.index < until; });
auto relevantTags = std::vector<Tag>(tags.begin(), it);
for (auto& t : relevantTags) {
t.index -= static_cast<int64_t>(samples_read);
}
fnc(readData, std::span<const Tag>(relevantTags));
auto tags = tag_reader.get();
const auto it = std::ranges::find_if_not(tags, [until = samples_read + nProcess](const auto& tag) { return tag.index < until; });
auto relevantTagsView = std::span(tags.begin(), it) | std::views::transform([this](const auto& v) { return Tag{v.index - samples_read, v.map}; });
auto relevantTags = std::vector(relevantTagsView.begin(), relevantTagsView.end());

fnc(readData, relevantTags);
std::ignore = tags.consume(relevantTags.size());
} else {
auto tags = tag_reader.get();
Expand Down Expand Up @@ -689,7 +688,7 @@ This block type is mean for non-data set input streams. For input streams of typ
std::ranges::copy(data.first(n), buffer.begin() + static_cast<std::ptrdiff_t>(buffer_fill));
if constexpr (callbackTakesTags) {
if (auto tag = detail::tagAndMetadata(tagData0, _pendingMetadata)) {
tag_buffer.emplace_back(static_cast<Tag::signed_index_type>(buffer_fill), std::move(*tag));
tag_buffer.emplace_back(static_cast<std::ptrdiff_t>(buffer_fill), std::move(*tag));
}
_pendingMetadata.reset();
tagData0.reset();
Expand Down Expand Up @@ -745,7 +744,7 @@ This block type is mean for non-data set input streams. For input streams of typ
if (toWrite > 0) {
if (auto tag = detail::tagAndMetadata(tagData0, _pendingMetadata)) {
auto tw = poller->tag_writer.reserve(1);
tw[0] = {static_cast<Tag::signed_index_type>(samples_written), std::move(*tag)};
tw[0] = {samples_written, std::move(*tag)};
tw.publish(1);
}
_pendingMetadata.reset();
Expand Down Expand Up @@ -839,7 +838,7 @@ This block type is mean for non-data set input streams. For input streams of typ
const auto preSampleView = history.subspan(0UZ, std::min(preSamples, history.size()));
dataset.signal_values.insert(dataset.signal_values.end(), preSampleView.rbegin(), preSampleView.rend());

dataset.timing_events = {{{static_cast<Tag::signed_index_type>(preSampleView.size()), *tagData0}}};
dataset.timing_events = {{{static_cast<std::ptrdiff_t>(preSampleView.size()), *tagData0}}};
pending_trigger_windows.push_back({.dataset = std::move(dataset), .pending_post_samples = postSamples});
}

Expand Down Expand Up @@ -932,7 +931,7 @@ This block type is mean for non-data set input streams. For input streams of typ
if (obsr == trigger::MatchResult::NotMatching || obsr == trigger::MatchResult::Matching) {
if (pending_dataset) {
if (obsr == trigger::MatchResult::NotMatching) {
pending_dataset->timing_events[0].push_back({static_cast<Tag::signed_index_type>(pending_dataset->signal_values.size()), *tagData0});
pending_dataset->timing_events[0].emplace_back(pending_dataset->signal_values.size(), *tagData0);
}
bool published = this->publishDataSet(std::move(*pending_dataset));
if (published) {
Expand Down Expand Up @@ -1050,7 +1049,7 @@ This block type is mean for non-data set input streams. For input streams of typ
}

DataSet<T> dataset = dataset_template;
dataset.timing_events = {{{-static_cast<Tag::signed_index_type>(it->delay), std::move(it->tag_data)}}};
dataset.timing_events = {{{-static_cast<std::ptrdiff_t>(it->delay), std::move(it->tag_data)}}};
dataset.signal_values = {inData[it->pending_samples]};
bool published = this->publishDataSet(std::move(dataset));
if (published) {
Expand Down
2 changes: 1 addition & 1 deletion blocks/basic/include/gnuradio-4.0/basic/StreamToDataSet.hpp
View file
Original file line number Diff line number Diff line change
Expand Up @@ -236,7 +236,7 @@ If multiple 'start' or 'stop' Tags arrive in a single merged tag, only one DataS
if (n_max.value > ds.signal_values.size()) { // do not add Tags if DataSet is full
const Tag& mergedTag = this->mergedInputTag();
if (!ds.timing_events.empty() && !mergedTag.map.empty() && accState.isActive) {
ds.timing_events[0].emplace_back(Tag{static_cast<Tag::signed_index_type>(ds.signal_values.size()), mergedTag.map});
ds.timing_events[0].emplace_back(static_cast<std::ptrdiff_t>(ds.signal_values.size()), mergedTag.map);
}
}

Expand Down
2 changes: 1 addition & 1 deletion blocks/basic/include/gnuradio-4.0/basic/clock_source.hpp
View file
Original file line number Diff line number Diff line change
Expand Up @@ -137,7 +137,7 @@ The 'tag_times[ns]:tag_value(string)' vectors control the emission of tags with

if (samplesToNextTag < samplesToNextTimeTag) {
if (_nextTagIndex < tags.size() && samplesToNextTag <= samplesToProduce) {
const auto tagDeltaIndex = tags[_nextTagIndex].index - static_cast<Tag::signed_index_type>(n_samples_produced); // position w.r.t. start of this chunk
const auto tagDeltaIndex = tags[_nextTagIndex].index - static_cast<std::size_t>(n_samples_produced); // position w.r.t. start of this chunk
if (verbose_console) {
gr::testing::print_tag(tags[_nextTagIndex], fmt::format("{}::processBulk(...)\t publish tag at {:6}", this->name, n_samples_produced + tagDeltaIndex));
}
Expand Down
71 changes: 39 additions & 32 deletions blocks/basic/test/qa_DataSink.cpp
View file
Original file line number Diff line number Diff line change
Expand Up @@ -105,9 +105,9 @@ struct Matcher {
}
};

static Tag makeTag(Tag::signed_index_type index, int year, int month, int day) { return Tag{index, {{"YEAR", year}, {"MONTH", month}, {"DAY", day}}}; }
static Tag makeTag(std::size_t index, int year, int month, int day) { return Tag{index, {{"YEAR", year}, {"MONTH", month}, {"DAY", day}}}; }

static std::vector<Tag> makeTestTags(Tag::signed_index_type firstIndex, Tag::signed_index_type interval) {
static std::vector<Tag> makeTestTags(std::size_t firstIndex, std::size_t interval) {
std::vector<Tag> tags;
for (int y = 1; y <= 3; ++y) {
for (int m = 1; m <= 2; ++m) {
Expand Down Expand Up @@ -139,6 +139,13 @@ static std::string toAsciiArt(std::span<trigger::MatchResult> states) {
return r;
}

std::size_t checkedSum(std::size_t a, std::ptrdiff_t b) {
using namespace boost::ut;
const auto signedA = static_cast<std::ptrdiff_t>(a);
expect(ge(signedA + b, 0));
return static_cast<std::size_t>(signedA + b);
}

template<trigger::Matcher TMatcher>
std::string runMatcherTest(std::span<const Tag> tags, TMatcher matcher) {
std::vector<trigger::MatchResult> result;
Expand Down Expand Up @@ -294,20 +301,21 @@ const boost::ut::suite DataSinkTests = [] {
std::size_t samplesSeen2 = 0;
std::size_t chunksSeen2 = 0;
std::vector<Tag> receivedTags;
auto callbackWithTags = [&samplesSeen2, &chunksSeen2, &m2, &receivedTags, &kChunkSize](std::span<const float> buffer, std::span<const Tag> tags) {

auto callbackWithTags = [&samplesSeen2, &chunksSeen2, &m2, &receivedTags, &kChunkSize](std::span<const float> buffer, std::span<const Tag> tags) {
for (std::size_t i = 0; i < buffer.size(); ++i) {
expect(eq(buffer[i], static_cast<float>(samplesSeen2 + i)));
}

for (const auto& tag : tags) {
expect(ge(tag.index, 0));
expect(lt(tag.index, static_cast<decltype(tag.index)>(buffer.size())));
expect(ge(tag.index, 0UZ));
expect(lt(tag.index, buffer.size()));
}

auto lg = std::lock_guard{m2};
std::vector<Tag> adjusted;
std::transform(tags.begin(), tags.end(), std::back_inserter(adjusted), [samplesSeen2](const auto& tag) { return Tag{static_cast<Tag::signed_index_type>(samplesSeen2) + tag.index, tag.map}; });
receivedTags.insert(receivedTags.end(), adjusted.begin(), adjusted.end());
auto lg = std::lock_guard{m2};

auto absolute = tags | std::views::transform([&samplesSeen2](const auto& t) { return gr::Tag{t.index + samplesSeen2, t.map}; });
receivedTags.insert(receivedTags.end(), absolute.begin(), absolute.end());
samplesSeen2 += buffer.size();
chunksSeen2++;
if (chunksSeen2 < 201) {
Expand Down Expand Up @@ -403,11 +411,8 @@ const boost::ut::suite DataSinkTests = [] {
while (!seenFinished) {
seenFinished = poller->finished;
while (poller->process([&received, &receivedTags](const auto& data, const auto& tags_) {
auto rtags = std::vector<Tag>(tags_.begin(), tags_.end());
for (auto& t : rtags) {
t.index += static_cast<int64_t>(received.size());
}
receivedTags.insert(receivedTags.end(), rtags.begin(), rtags.end());
auto absolute = tags_ | std::views::transform([&received](const auto& t) { return gr::Tag{t.index + received.size(), t.map}; });
receivedTags.insert(receivedTags.end(), absolute.begin(), absolute.end());
received.insert(received.end(), data.begin(), data.end());
})) {
}
Expand Down Expand Up @@ -441,7 +446,7 @@ const boost::ut::suite DataSinkTests = [] {
expect(eq(nonMetadataTags.size(), tags.size()));
expect(eq(indexesMatch(nonMetadataTags, tags), true)) << fmt::format("{} != {}", formatList(nonMetadataTags), formatList(tags));
expect(eq(metadataTags.size(), 1UZ));
expect(eq(metadataTags[0].index, 0));
expect(eq(metadataTags[0].index, 0UZ));
const auto metadata = latestMetadata(metadataTags);
expect(eq(metadata.signal_name.value_or("<unset>"), "test signal"s));
expect(eq(metadata.signal_unit.value_or("<unset>"), "test unit"s));
Expand Down Expand Up @@ -476,9 +481,9 @@ const boost::ut::suite DataSinkTests = [] {
poller = DataSinkRegistry::instance().getTriggerPoller<int32_t>(DataSinkQuery::signalName("test signal"), isTrigger, 3, 2, BlockingMode::Blocking);
return poller != nullptr;
})) << boost::ut::fatal;
std::vector<int32_t> receivedData;
std::vector<Tag> receivedTags;
bool seenFinished = false;
std::vector<int32_t> receivedData;
std::vector<std::pair<std::ptrdiff_t, property_map>> receivedTags;
bool seenFinished = false;
while (!seenFinished) {
seenFinished = poller->finished;
[[maybe_unused]] auto r = poller->process([&receivedData, &receivedTags](const auto& datasets) {
Expand All @@ -493,7 +498,7 @@ const boost::ut::suite DataSinkTests = [] {
expect(eq(dataset.signal_units[0], "none"s));
expect(eq(dataset.signal_ranges[0], std::vector<int32_t>{-2, +2}));
expect(eq(dataset.timing_events[0].size(), 1u));
expect(eq(dataset.timing_events[0][0].index, 3));
expect(eq(dataset.timing_events[0][0].first, 3));
receivedTags.insert(receivedTags.end(), dataset.timing_events[0].begin(), dataset.timing_events[0].end());
}
});
Expand Down Expand Up @@ -528,10 +533,11 @@ const boost::ut::suite DataSinkTests = [] {
expect(eq(ConnectionResult::SUCCESS, testGraph.connect<"out">(delay).to<"in">(sink)));

auto polling = std::async([] {
std::vector<int32_t> receivedData;
std::vector<Tag> receivedTags;
bool seenFinished = false;
auto isTrigger = [](std::string_view /* filterSpec */, const Tag& tag, const property_map& /* filter state */) {
std::vector<int32_t> receivedData;
std::vector<std::pair<std::ptrdiff_t, property_map>> receivedTags;
bool seenFinished = false;

auto isTrigger = [](std::string_view /* filterSpec */, const Tag& tag, const property_map& /* filter state */) {
const auto type = tag.get("TYPE");
return (type && std::get<std::string>(type->get()) == "TRIGGER") ? trigger::MatchResult::Matching : trigger::MatchResult::Ignore;
};
Expand All @@ -558,8 +564,8 @@ const boost::ut::suite DataSinkTests = [] {
expect(eq(dataset.signal_names[0], "test signal"s));
expect(eq(dataset.signal_units[0], "no unit"s));
expect(eq(dataset.signal_ranges[0], std::vector{-2, +2}));
expect(eq(dataset.timing_events[0].size(), 1u));
expect(eq(dataset.timing_events[0][0].index, 0));
expect(eq(dataset.timing_events[0].size(), 1UZ));
expect(eq(dataset.timing_events[0][0].first, 0));
receivedTags.insert(receivedTags.end(), dataset.timing_events[0].begin(), dataset.timing_events[0].end());
}
});
Expand Down Expand Up @@ -622,7 +628,7 @@ const boost::ut::suite DataSinkTests = [] {
expect(eq(dataset.signal_units[0], "none"s));
expect(eq(dataset.signal_ranges[0], std::vector<int32_t>{0, kSamples - 1}));
expect(eq(dataset.timing_events[0].size(), 1u));
expect(eq(dataset.timing_events[0][0].index, -5000));
expect(eq(dataset.timing_events[0][0].first, -5000));
receivedData.insert(receivedData.end(), dataset.signal_values.begin(), dataset.signal_values.end());
}
});
Expand Down Expand Up @@ -739,7 +745,7 @@ const boost::ut::suite DataSinkTests = [] {
auto& src = testGraph.emplaceBlock<gr::testing::TagSource<float>>({{"n_samples_max", kSamples}, {"mark_tag", false}});

for (std::size_t i = 0; i < kTriggers; ++i) {
src._tags.push_back(Tag{static_cast<Tag::signed_index_type>(60000 + i), {{"TYPE", "TRIGGER"}}});
src._tags.push_back(Tag{60000UZ + i, {{"TYPE", "TRIGGER"}}});
}

auto& delay = testGraph.emplaceBlock<testing::Delay<float>>({{"delay_ms", kProcessingDelayMs}});
Expand Down Expand Up @@ -768,8 +774,9 @@ const boost::ut::suite DataSinkTests = [] {
receivedData.push_back(dataset.signal_values.back());
expect(eq(dataset.timing_events.size(), 1u));
expect(eq(dataset.timing_events[0].size(), 1u));
expect(eq(dataset.timing_events[0][0].index, 3000));
receivedTags.insert(receivedTags.end(), dataset.timing_events[0].begin(), dataset.timing_events[0].end());
expect(eq(dataset.timing_events[0][0].first, 3000));
auto absolute = dataset.timing_events[0] | std::views::transform([&receivedData](const auto& t) { return gr::Tag{checkedSum(receivedData.size(), t.first), t.second}; });
receivedTags.insert(receivedTags.end(), absolute.begin(), absolute.end());
}
})) {
}
Expand All @@ -796,7 +803,7 @@ const boost::ut::suite DataSinkTests = [] {
auto& src = testGraph.emplaceBlock<gr::testing::TagSource<float>>({{"n_samples_max", kSamples}, {"mark_tag", false}});

for (std::size_t i = 0; i < kTriggers; ++i) {
src._tags.push_back(Tag{static_cast<Tag::signed_index_type>(60000 + i), {{"TYPE", "TRIGGER"}}});
src._tags.push_back(Tag{60000UZ + i, {{"TYPE", "TRIGGER"}}});
}

auto& delay = testGraph.emplaceBlock<testing::Delay<float>>({{"delay_ms", kProcessingDelayMs}});
Expand Down Expand Up @@ -883,7 +890,7 @@ const boost::ut::suite DataSinkTests = [] {
expect(eq(gr::ConnectionResult::SUCCESS, testGraph.connect<"out">(delay).to<"in">(streamToDataSet)));
expect(eq(gr::ConnectionResult::SUCCESS, testGraph.connect<"out">(streamToDataSet).to<"in">(sink)));

auto genTrigger = [](gr::Tag::signed_index_type index, std::string triggerName, std::string triggerCtx = {}) {
auto genTrigger = [](std::size_t index, std::string triggerName, std::string triggerCtx = {}) {
return Tag{index, {{gr::tag::TRIGGER_NAME.shortKey(), triggerName}, {gr::tag::TRIGGER_TIME.shortKey(), std::uint64_t(0)}, {gr::tag::TRIGGER_OFFSET.shortKey(), 0.f}, //
{gr::tag::TRIGGER_META_INFO.shortKey(), gr::property_map{{gr::tag::CONTEXT.shortKey(), triggerCtx}}}}};
};
Expand Down Expand Up @@ -932,7 +939,7 @@ const boost::ut::suite DataSinkTests = [] {
expect(eq(gr::ConnectionResult::SUCCESS, testGraph.connect<"out">(delay).to<"in">(streamToDataSet)));
expect(eq(gr::ConnectionResult::SUCCESS, testGraph.connect<"out">(streamToDataSet).to<"in">(sink)));

auto genTrigger = [](gr::Tag::signed_index_type index, std::string triggerName, std::string triggerCtx = {}) {
auto genTrigger = [](std::size_t index, std::string triggerName, std::string triggerCtx = {}) {
return Tag{index, {{gr::tag::TRIGGER_NAME.shortKey(), triggerName}, {gr::tag::TRIGGER_TIME.shortKey(), std::uint64_t(0)}, {gr::tag::TRIGGER_OFFSET.shortKey(), 0.f}, //
{gr::tag::TRIGGER_META_INFO.shortKey(), gr::property_map{{gr::tag::CONTEXT.shortKey(), triggerCtx}}}}};
};
Expand Down
Loading

0 comments on commit 3273d08

Please sign in to comment.