Rework all subthread-based workers to make notifications more robust

[pulseview.git] / pv / data / segment.cpp

diff --git a/pv/data/segment.cpp b/pv/data/segment.cpp
index 7b8e611f271999d6b3a57948166322faccaa71bf..18aabedbedd348bf0656635f60cb991757713e32 100644 (file)
--- a/pv/data/segment.cpp
+++ b/pv/data/segment.cpp
@@ -24,6 +24,8 @@
 #include <cstdlib>
 #include <cstring>
 
+#include <QDebug>
+
 using std::bad_alloc;
 using std::lock_guard;
 using std::min;
@@ -52,7 +54,7 @@ Segment::Segment(uint32_t segment_id, uint64_t samplerate, unsigned int unit_siz
        chunk_size_ = min(MaxChunkSize, (MaxChunkSize / unit_size_) * unit_size_);
 
        // Create the initial chunk
-       current_chunk_ = new uint8_t[chunk_size_];
+       current_chunk_ = new uint8_t[chunk_size_ + 7];  /* FIXME +7 is workaround for #1284 */
        data_chunks_.push_back(current_chunk_);
        used_samples_ = 0;
        unused_samples_ = chunk_size_ / unit_size_;
@@ -68,7 +70,6 @@ Segment::~Segment()
 
 uint64_t Segment::get_sample_count() const
 {
-       lock_guard<recursive_mutex> lock(mutex_);
        return sample_count_;
 }
 
@@ -100,6 +101,8 @@ uint32_t Segment::segment_id() const
 void Segment::set_complete()
 {
        is_complete_ = true;
+
+       completed();
 }
 
 bool Segment::is_complete() const
@@ -119,7 +122,7 @@ void Segment::free_unused_memory()
 
        if (current_chunk_) {
                // No more data will come in, so re-create the last chunk accordingly
-               uint8_t* resized_chunk = new uint8_t[used_samples_ * unit_size_];
+               uint8_t* resized_chunk = new uint8_t[used_samples_ * unit_size_ + 7];  /* FIXME +7 is workaround for #1284 */
                memcpy(resized_chunk, current_chunk_, used_samples_ * unit_size_);
 
                delete[] current_chunk_;
@@ -142,7 +145,7 @@ void Segment::append_single_sample(void *data)
        unused_samples_--;
 
        if (unused_samples_ == 0) {
-               current_chunk_ = new uint8_t[chunk_size_];
+               current_chunk_ = new uint8_t[chunk_size_ + 7];  /* FIXME +7 is workaround for #1284 */
                data_chunks_.push_back(current_chunk_);
                used_samples_ = 0;
                unused_samples_ = chunk_size_ / unit_size_;
@@ -188,13 +191,13 @@ void Segment::append_samples(void* data, uint64_t samples)
                                // This way, memory allocation will fail early enough to let
                                // PV remain alive. Otherwise, PV will crash in a random
                                // memory-allocating part of the application.
-                               current_chunk_ = new uint8_t[chunk_size_];
+                               current_chunk_ = new uint8_t[chunk_size_ + 7];  /* FIXME +7 is workaround for #1284 */
 
                                const int dummy_size = 2 * chunk_size_;
                                auto dummy_chunk = new uint8_t[dummy_size];
                                memset(dummy_chunk, 0xFF, dummy_size);
                                delete[] dummy_chunk;
-                       } catch (bad_alloc) {
+                       } catch (bad_alloc&) {
                                delete[] current_chunk_;  // The new may have succeeded
                                current_chunk_ = nullptr;
                                throw;
@@ -209,21 +212,34 @@ void Segment::append_samples(void* data, uint64_t samples)
        sample_count_ += samples;
 }
 
-void Segment::get_raw_samples(uint64_t start, uint64_t count,
-       uint8_t* dest) const
+const uint8_t* Segment::get_raw_sample(uint64_t sample_num) const
+{
+       assert(sample_num <= sample_count_);
+
+       uint64_t chunk_num = (sample_num * unit_size_) / chunk_size_;
+       uint64_t chunk_offs = (sample_num * unit_size_) % chunk_size_;
+
+       lock_guard<recursive_mutex> lock(mutex_);  // Because of free_unused_memory()
+
+       const uint8_t* chunk = data_chunks_[chunk_num];
+
+       return chunk + chunk_offs;
+}
+
+void Segment::get_raw_samples(uint64_t start, uint64_t count, uint8_t* dest) const
 {
        assert(start < sample_count_);
        assert(start + count <= sample_count_);
        assert(count > 0);
        assert(dest != nullptr);
 
-       lock_guard<recursive_mutex> lock(mutex_);
-
        uint8_t* dest_ptr = dest;
 
        uint64_t chunk_num = (start * unit_size_) / chunk_size_;
        uint64_t chunk_offs = (start * unit_size_) % chunk_size_;
 
+       lock_guard<recursive_mutex> lock(mutex_);  // Because of free_unused_memory()
+
        while (count > 0) {
                const uint8_t* chunk = data_chunks_[chunk_num];
 
@@ -240,9 +256,9 @@ void Segment::get_raw_samples(uint64_t start, uint64_t count,
        }
 }
 
-SegmentRawDataIterator* Segment::begin_raw_sample_iteration(uint64_t start)
+SegmentDataIterator* Segment::begin_sample_iteration(uint64_t start)
 {
-       SegmentRawDataIterator* it = new SegmentRawDataIterator;
+       SegmentDataIterator* it = new SegmentDataIterator;
 
        assert(start < sample_count_);
 
@@ -252,17 +268,12 @@ SegmentRawDataIterator* Segment::begin_raw_sample_iteration(uint64_t start)
        it->chunk_num = (start * unit_size_) / chunk_size_;
        it->chunk_offs = (start * unit_size_) % chunk_size_;
        it->chunk = data_chunks_[it->chunk_num];
-       it->value = it->chunk + it->chunk_offs;
 
        return it;
 }
 
-void Segment::continue_raw_sample_iteration(SegmentRawDataIterator* it, uint64_t increase)
+void Segment::continue_sample_iteration(SegmentDataIterator* it, uint64_t increase)
 {
-       // Fail gracefully if we are asked to deliver data we don't have
-       if (it->sample_index > sample_count_)
-               return;
-
        it->sample_index += increase;
        it->chunk_offs += (increase * unit_size_);
 
@@ -271,11 +282,9 @@ void Segment::continue_raw_sample_iteration(SegmentRawDataIterator* it, uint64_t
                it->chunk_offs -= chunk_size_;
                it->chunk = data_chunks_[it->chunk_num];
        }
-
-       it->value = it->chunk + it->chunk_offs;
 }
 
-void Segment::end_raw_sample_iteration(SegmentRawDataIterator* it)
+void Segment::end_sample_iteration(SegmentDataIterator* it)
 {
        delete it;
 
@@ -287,5 +296,19 @@ void Segment::end_raw_sample_iteration(SegmentRawDataIterator* it)
        }
 }
 
+uint8_t* Segment::get_iterator_value(SegmentDataIterator* it)
+{
+       assert(it->sample_index <= (sample_count_ - 1));
+
+       return (it->chunk + it->chunk_offs);
+}
+
+uint64_t Segment::get_iterator_valid_length(SegmentDataIterator* it)
+{
+       assert(it->sample_index <= (sample_count_ - 1));
+
+       return ((chunk_size_ - it->chunk_offs) / unit_size_);
+}
+
 } // namespace data
 } // namespace pv