rocr: Support batching in InterceptQueue store (#1194)

* rocr: Support batching in InterceptQueue store

* Fix comment, loop bounds

projects/rocr-runtime/runtime/hsa-runtime/core/runtime/intercept_queue.cpp

@@ -369,27 +369,18 @@ void InterceptQueue::StoreRelaxed(hsa_signal_value_t value) {
     end = next_packet_ + amd_queue_.hsa_queue.size;
 
   uint64_t i = next_packet_;
+  uint64_t invalid_header_i = end;
+
   while (i < end) {
     // Load the packet header as atomic acquire as it may have been written by
     // another thread as atomic release. This ensures the rest of the packet
     // fields are visible. Once loaded and proven not to be INVALID, further
     // loads by this thread can be non-atomic.
     uint16_t header = atomic::Load(&ring[i & mask].packet.header, std::memory_order_acquire);
-    if (!AqlPacket::IsValid(header)) break;
-
-    // Process callbacks.
-    Cursor.interceptor_index = interceptors.size() - 1;
-    Cursor.pkt_index = i;
-    auto& handler = interceptors[Cursor.interceptor_index];
-    handler.first(&ring[i & mask], 1, i, handler.second, PacketWriter);
-    if (IsDeviceMemRingBuf() && needsPcieOrdering()) {
-      // Ensure the packet body is written as header may get reordered when writing over PCIE
-      _mm_sfence();
+    if (!AqlPacket::IsValid(header)) {
+      invalid_header_i = i;
+      break;
     }
-    // Invalidate consumed packet.
-    atomic::Store(&ring[i & mask].packet.header, kInvalidHeader, std::memory_order_release);
-
-    // Packet has now been processed so advance the read index.
     ++i;
 
     // Only allow the rewrite of one packet to be on the overflow queue. When
@@ -401,6 +392,27 @@ void InterceptQueue::StoreRelaxed(hsa_signal_value_t value) {
     if (!overflow_.empty()) break;
   }
 
+  // Process callbacks.
+  uint64_t packet_count = i - next_packet_;
+  if (packet_count) {
+    Cursor.interceptor_index = interceptors.size() - 1;
+    Cursor.pkt_index = next_packet_;
+    auto& handler = interceptors[Cursor.interceptor_index];
+    handler.first(&ring[next_packet_ & mask], packet_count, next_packet_,
+                                                handler.second, PacketWriter);
+    if (IsDeviceMemRingBuf() && needsPcieOrdering()) {
+      // Ensure the packet body is written as header may get reordered when writing over PCIE
+      _mm_sfence();
+    }
+  }
+  i = next_packet_;
+  while (i < std::min(end, invalid_header_i)) {
+    // Invalidate consumed packets.
+    atomic::Store(&ring[i & mask].packet.header, kInvalidHeader, std::memory_order_release);
+    // Packet has now been processed so advance the read index.
+    ++i;
+  }
+
   next_packet_ = i;
   Cursor.queue = nullptr;
   atomic::Store(&amd_queue_.read_dispatch_id, next_packet_, std::memory_order_release);