diff --git a/lib/image_inference_impl.cc b/lib/image_inference_impl.cc
index cb364e6a..051ccca3 100644
--- a/lib/image_inference_impl.cc
+++ b/lib/image_inference_impl.cc
@@ -262,7 +262,6 @@ void image_inference_impl::process_items_(size_t c, const input_type *&in) {
     cv::Mat new_row(cv::Size(vlen_, 1), CV_32F, (void *)in);
     points_buffer_->push_back(new_row);
   }
-  consume_each(c);
 }
 
 void image_inference_impl::create_image_() {
@@ -349,26 +348,25 @@ int image_inference_impl::general_work(int noutput_items,
 
   if (rx_freq_tags.empty()) {
     process_items_(in_count, in);
-    return 0;
-  }
+  } else {
+    for (size_t t = 0; t < rx_freq_tags.size(); ++t) {
+      const auto &tag = rx_freq_tags[t];
+      const double rx_time = rx_times[t];
+      const auto rel = tag.offset - in_first;
+      in_first += rel;
 
-  for (size_t t = 0; t < rx_freq_tags.size(); ++t) {
-    const auto &tag = rx_freq_tags[t];
-    const double rx_time = rx_times[t];
-    const auto rel = tag.offset - in_first;
-    in_first += rel;
+      if (rel > 0) {
+        process_items_(rel, in);
+      }
 
-    if (rel > 0) {
-      process_items_(rel, in);
+      uint64_t rx_freq = (uint64_t)pmt::to_double(tag.value);
+      create_image_();
+      last_rx_freq_ = rx_freq;
+      last_rx_time_ = rx_time;
     }
-
-    uint64_t rx_freq = (uint64_t)pmt::to_double(tag.value);
-    create_image_();
-    last_rx_freq_ = rx_freq;
-    last_rx_time_ = rx_time;
   }
 
-  process_items_(1, in);
+  consume_each(in_count);
   return 0;
 }
 
diff --git a/lib/retune_fft_impl.cc b/lib/retune_fft_impl.cc
index f33e56df..13736c37 100644
--- a/lib/retune_fft_impl.cc
+++ b/lib/retune_fft_impl.cc
@@ -421,7 +421,6 @@ void retune_fft_impl::process_items_(size_t c, const input_type *&in) {
       }
     }
   }
-  consume_each(c);
 }
 
 void retune_fft_impl::forecast(int noutput_items,
@@ -535,30 +534,26 @@ void retune_fft_impl::process_tags_(const input_type *in, size_t in_count,
 
   if (rx_freq_tags.empty()) {
     process_items_(in_count, in);
-    return;
-  }
-
-  for (size_t t = 0; t < rx_freq_tags.size(); ++t) {
-    const auto &tag = rx_freq_tags[t];
-    const double rx_time = rx_times[t];
-    const auto rel = tag.offset - in_first;
-    in_first += rel;
+  } else {
+    for (size_t t = 0; t < rx_freq_tags.size(); ++t) {
+      const auto &tag = rx_freq_tags[t];
+      const double rx_time = rx_times[t];
+      const auto rel = tag.offset - in_first;
+      in_first += rel;
 
-    if (rel > 0) {
-      process_items_(rel, in);
-    }
+      if (rel > 0) {
+        process_items_(rel, in);
+      }
 
-    const uint64_t rx_freq = (uint64_t)pmt::to_double(tag.value);
-    size_t range_steps = tuning_ranges_[tuning_range_].steps;
+      const uint64_t rx_freq = (uint64_t)pmt::to_double(tag.value);
 
-    d_logger->debug("new rx_freq tag: {}, last {}", rx_freq, last_rx_freq_);
-    if (pending_retune_) {
-      --pending_retune_;
+      d_logger->debug("new rx_freq tag: {}, last {}", rx_freq, last_rx_freq_);
+      if (pending_retune_) {
+        --pending_retune_;
+      }
+      process_buckets_(rx_freq, rx_time);
     }
-    process_buckets_(rx_freq, rx_time);
   }
-
-  process_items_(1, in);
 }
 
 int retune_fft_impl::general_work(int noutput_items,
@@ -579,6 +574,7 @@ int retune_fft_impl::general_work(int noutput_items,
   size_t in_count = ninput_items[0];
   size_t in_first = nitems_read(0);
   process_tags_(in, in_count, in_first);
+  consume_each(in_count);
 
   return 0;
 }
diff --git a/lib/write_freq_samples_impl.cc b/lib/write_freq_samples_impl.cc
index ffb170ee..f92bc795 100644
--- a/lib/write_freq_samples_impl.cc
+++ b/lib/write_freq_samples_impl.cc
@@ -290,7 +290,6 @@ void write_freq_samples_impl::write_samples_(size_t c, const char *&in) {
     }
     in += vlen_;
   }
-  consume_each(c);
 }
 
 int write_freq_samples_impl::general_work(
@@ -305,27 +304,26 @@ int write_freq_samples_impl::general_work(
 
   if (tags.empty()) {
     write_samples_(in_count, in);
-    return 0;
-  }
+  } else {
+    for (size_t t = 0; t < tags.size(); ++t) {
+      const auto &tag = tags[t];
+      const auto rel = tag.offset - in_first;
+      in_first += rel;
 
-  for (size_t t = 0; t < tags.size(); ++t) {
-    const auto &tag = tags[t];
-    const auto rel = tag.offset - in_first;
-    in_first += rel;
+      if (rel > 0) {
+        write_samples_(rel, in);
+      }
 
-    if (rel > 0) {
-      write_samples_(rel, in);
+      const uint64_t rx_freq = (uint64_t)pmt::to_double(tag.value);
+      d_logger->debug("new rx_freq tag: {}, last {}", rx_freq, last_rx_freq_);
+      last_rx_freq_ = rx_freq;
+      skip_tune_step_samples_count_ = skip_tune_step_samples_;
+      write_step_samples_count_ = write_step_samples_;
+      open_(1);
     }
-
-    const uint64_t rx_freq = (uint64_t)pmt::to_double(tag.value);
-    d_logger->debug("new rx_freq tag: {}, last {}", rx_freq, last_rx_freq_);
-    last_rx_freq_ = rx_freq;
-    skip_tune_step_samples_count_ = skip_tune_step_samples_;
-    write_step_samples_count_ = write_step_samples_;
-    open_(1);
   }
 
-  write_samples_(1, in);
+  consume_each(in_count);
   return 0;
 }