Don't let the PubSub internal offsets overflow + fix some warnings.

src/data/pubsub.rs

@@ -3,16 +3,28 @@
 use serde::export::PhantomData;
 use std::collections::VecDeque;
 
-/// The position of a subscriber on a pub-sub queue.
+/// A permanent subscription to a pub-sub queue.
 #[cfg_attr(feature = "serde-serialize", derive(Serialize, Deserialize))]
-pub struct PubSubCursor<T> {
-    // Index of the next message to read.
+pub struct Subscription<T> {
+    // Position on the cursor array.
     id: u32,
-    next: u32,
     _phantom: PhantomData<T>,
 }
 
-impl<T> PubSubCursor<T> {
+#[cfg_attr(feature = "serde-serialize", derive(Serialize, Deserialize))]
+struct PubSubCursor {
+    // Position on the offset array.
+    id: u32,
+    // Index of the next message to read.
+    // NOTE: Having this here is not actually necessary because
+    // this value is supposed to be equal to `offsets[self.id]`.
+    // However, we keep it because it lets us avoid one lookup
+    // on the `offsets` array inside of message-polling loops
+    // based on `read_ith`.
+    next: u32,
+}
+
+impl PubSubCursor {
     fn id(&self, num_deleted: u32) -> usize {
         (self.id - num_deleted) as usize
     }
@@ -29,6 +41,7 @@ pub struct PubSub<T> {
     deleted_offsets: u32,
     messages: VecDeque<T>,
     offsets: VecDeque<u32>,
+    cursors: Vec<PubSubCursor>,
 }
 
 impl<T> PubSub<T> {
@@ -39,9 +52,24 @@ impl<T> PubSub<T> {
             deleted_messages: 0,
             messages: VecDeque::new(),
             offsets: VecDeque::new(),
+            cursors: Vec::new(),
         }
     }
 
+    fn reset_shifts(&mut self) {
+        for offset in &mut self.offsets {
+            *offset -= self.deleted_messages;
+        }
+
+        for cursor in &mut self.cursors {
+            cursor.id -= self.deleted_offsets;
+            cursor.next -= self.deleted_messages;
+        }
+
+        self.deleted_offsets = 0;
+        self.deleted_messages = 0;
+    }
+
     /// Publish a new message.
     pub fn publish(&mut self, message: T) {
         if self.offsets.is_empty() {
@@ -55,25 +83,33 @@ impl<T> PubSub<T> {
     /// Subscribe to the queue.
     ///
     /// A subscription cannot be cancelled.
-    pub fn subscribe(&mut self) -> PubSubCursor<T> {
+    #[must_use]
+    pub fn subscribe(&mut self) -> Subscription<T> {
         let cursor = PubSubCursor {
             next: self.messages.len() as u32 + self.deleted_messages,
             id: self.offsets.len() as u32 + self.deleted_offsets,
+        };
+
+        let subscription = Subscription {
+            id: self.cursors.len() as u32,
             _phantom: PhantomData,
         };
 
         self.offsets.push_back(cursor.next);
-        cursor
+        self.cursors.push(cursor);
+        subscription
     }
 
     /// Read the i-th message not yet read by the given subsciber.
-    pub fn read_ith(&self, cursor: &PubSubCursor<T>, i: usize) -> Option<&T> {
+    pub fn read_ith(&self, sub: &Subscription<T>, i: usize) -> Option<&T> {
+        let cursor = &self.cursors[sub.id as usize];
         self.messages
             .get(cursor.next(self.deleted_messages) as usize + i)
     }
 
     /// Get the messages not yet read by the given subscriber.
-    pub fn read(&self, cursor: &PubSubCursor<T>) -> impl Iterator<Item = &T> {
+    pub fn read(&self, sub: &Subscription<T>) -> impl Iterator<Item = &T> {
+        let cursor = &self.cursors[sub.id as usize];
         let next = cursor.next(self.deleted_messages);
 
         // TODO: use self.queue.range(next..) once it is stabilised.
@@ -86,11 +122,14 @@ impl<T> PubSub<T> {
     /// Makes the given subscribe acknowledge all the messages in the queue.
     ///
     /// A subscriber cannot read acknowledged messages any more.
-    pub fn ack(&mut self, cursor: &mut PubSubCursor<T>) {
+    pub fn ack(&mut self, sub: &Subscription<T>) {
         // Update the cursor.
-        cursor.next = self.messages.len() as u32 + self.deleted_messages;
+        let cursor = &mut self.cursors[sub.id as usize];
+
         self.offsets[cursor.id(self.deleted_offsets)] = u32::MAX;
         cursor.id = self.offsets.len() as u32 + self.deleted_offsets;
+
+        cursor.next = self.messages.len() as u32 + self.deleted_messages;
         self.offsets.push_back(cursor.next);
 
         // Now clear the messages we don't need to
@@ -110,6 +149,12 @@ impl<T> PubSub<T> {
         }
 
         self.deleted_messages += num_to_delete;
+
+        if self.deleted_messages > u32::MAX / 2 || self.deleted_offsets > u32::MAX / 2 {
+            // Don't let the deleted_* shifts grow indefinitely otherwise
+            // they will end up overflowing, breaking everything.
+            self.reset_shifts();
+        }
     }
 }
 

src/dynamics/rigid_body_set.rs

@@ -2,11 +2,9 @@
 use rayon::prelude::*;
 
 use crate::data::arena::Arena;
-use crate::data::pubsub::PubSub;
 use crate::dynamics::{BodyStatus, Joint, JointSet, RigidBody};
 use crate::geometry::{ColliderHandle, ColliderSet, ContactPair, InteractionGraph};
 use crossbeam::channel::{Receiver, Sender};
-use num::Zero;
 use std::ops::{Deref, DerefMut, Index, IndexMut};
 
 /// A mutable reference to a rigid-body.

src/geometry/broad_phase_multi_sap.rs

@@ -1,6 +1,6 @@
-use crate::data::pubsub::PubSubCursor;
+use crate::data::pubsub::Subscription;
 use crate::dynamics::RigidBodySet;
-use crate::geometry::{Collider, ColliderHandle, ColliderSet, RemovedCollider};
+use crate::geometry::{ColliderHandle, ColliderSet, RemovedCollider};
 use crate::math::{Point, Vector, DIM};
 #[cfg(feature = "enhanced-determinism")]
 use crate::utils::FxHashMap32 as HashMap;
@@ -417,7 +417,7 @@ impl SAPRegion {
 pub struct BroadPhase {
     proxies: Proxies,
     regions: HashMap<Point<i32>, SAPRegion>,
-    removed_colliders: Option<PubSubCursor<RemovedCollider>>,
+    removed_colliders: Option<Subscription<RemovedCollider>>,
     deleted_any: bool,
     // We could think serializing this workspace is useless.
     // It turns out is is important to serialize at least its capacity

src/geometry/collider_set.rs

@@ -83,7 +83,7 @@ impl ColliderSet {
         /*
          * Delete the collider from its parent body.
          */
-        if let Some(mut parent) = bodies.get_mut_internal(collider.parent) {
+        if let Some(parent) = bodies.get_mut_internal(collider.parent) {
             parent.remove_collider_internal(handle, &collider);
             bodies.wake_up(collider.parent, true);
         }

src/geometry/narrow_phase.rs

@@ -14,13 +14,13 @@ use crate::geometry::proximity_detector::{
 //    proximity_detector::ProximityDetectionContextSimd, WBall,
 //};
 use crate::geometry::{
-    BroadPhasePairEvent, Collider, ColliderGraphIndex, ColliderHandle, ContactEvent,
-    ProximityEvent, ProximityPair, RemovedCollider,
+    BroadPhasePairEvent, ColliderGraphIndex, ColliderHandle, ContactEvent, ProximityEvent,
+    ProximityPair, RemovedCollider,
 };
 use crate::geometry::{ColliderSet, ContactManifold, ContactPair, InteractionGraph};
 //#[cfg(feature = "simd-is-enabled")]
 //use crate::math::{SimdFloat, SIMD_WIDTH};
-use crate::data::pubsub::PubSubCursor;
+use crate::data::pubsub::Subscription;
 use crate::ncollide::query::Proximity;
 use crate::pipeline::EventHandler;
 use std::collections::HashMap;
@@ -31,7 +31,7 @@ use std::collections::HashMap;
 pub struct NarrowPhase {
     contact_graph: InteractionGraph<ContactPair>,
     proximity_graph: InteractionGraph<ProximityPair>,
-    removed_colliders: Option<PubSubCursor<RemovedCollider>>,
+    removed_colliders: Option<Subscription<RemovedCollider>>,
     //    ball_ball: Vec<usize>,        // Workspace: Vec<*mut ContactPair>,
     //    shape_shape: Vec<usize>,      // Workspace: Vec<*mut ContactPair>,
     //    ball_ball_prox: Vec<usize>,   // Workspace: Vec<*mut ProximityPair>,

src/geometry/waabb.rs

@@ -93,10 +93,6 @@ impl<'de> serde::Deserialize<'de> for WAABB {
 }
 
 impl WAABB {
-    pub fn new(mins: Point<SimdFloat>, maxs: Point<SimdFloat>) -> Self {
-        Self { mins, maxs }
-    }
-
     pub fn new_invalid() -> Self {
         Self::splat(AABB::new_invalid())
     }
@@ -132,7 +128,7 @@ impl WAABB {
         for i in 0usize..DIM {
             let is_not_zero = ray.dir[i].simd_ne(_0);
             let is_zero_test =
-                (ray.origin[i].simd_ge(self.mins[i]) & ray.origin[i].simd_le(self.maxs[i]));
+                ray.origin[i].simd_ge(self.mins[i]) & ray.origin[i].simd_le(self.maxs[i]);
             let is_not_zero_test = {
                 let denom = _1 / ray.dir[i];
                 let mut inter_with_near_plane =

src/geometry/wquadtree.rs

@@ -379,11 +379,13 @@ impl<T: IndexedData> WQuadtree<T> {
     }
 }
 
+#[allow(dead_code)]
 struct WQuadtreeIncrementalBuilderStep {
     range: Range<usize>,
     parent: NodeIndex,
 }
 
+#[allow(dead_code)]
 struct WQuadtreeIncrementalBuilder<T> {
     quadtree: WQuadtree<T>,
     to_insert: Vec<WQuadtreeIncrementalBuilderStep>,
@@ -391,6 +393,7 @@ struct WQuadtreeIncrementalBuilder<T> {
     indices: Vec<usize>,
 }
 
+#[allow(dead_code)]
 impl<T: IndexedData> WQuadtreeIncrementalBuilder<T> {
     pub fn new() -> Self {
         Self {

src/pipeline/collision_pipeline.rs

@@ -1,6 +1,6 @@
 //! Physics pipeline structures.
 
-use crate::dynamics::{JointSet, RigidBody, RigidBodyHandle, RigidBodySet};
+use crate::dynamics::{JointSet, RigidBodySet};
 use crate::geometry::{BroadPhase, BroadPhasePairEvent, ColliderPair, ColliderSet, NarrowPhase};
 use crate::pipeline::EventHandler;
 

src/pipeline/physics_pipeline.rs

@@ -1,15 +1,13 @@
 //! Physics pipeline structures.
 
 use crate::counters::Counters;
-use crate::data::pubsub::PubSubCursor;
 #[cfg(not(feature = "parallel"))]
 use crate::dynamics::IslandSolver;
-use crate::dynamics::{IntegrationParameters, JointSet, RigidBody, RigidBodyHandle, RigidBodySet};
+use crate::dynamics::{IntegrationParameters, JointSet, RigidBodySet};
 #[cfg(feature = "parallel")]
 use crate::dynamics::{JointGraphEdge, ParallelIslandSolver as IslandSolver};
 use crate::geometry::{
-    BroadPhase, BroadPhasePairEvent, Collider, ColliderHandle, ColliderPair, ColliderSet,
-    ContactManifoldIndex, NarrowPhase, RemovedCollider,
+    BroadPhase, BroadPhasePairEvent, ColliderPair, ColliderSet, ContactManifoldIndex, NarrowPhase,
 };
 use crate::math::Vector;
 use crate::pipeline::EventHandler;

src/pipeline/query_pipeline.rs

@@ -1,9 +1,5 @@
 use crate::dynamics::RigidBodySet;
-use crate::geometry::{
-    Collider, ColliderHandle, ColliderSet, Ray, RayIntersection, WQuadtree, AABB, WAABB,
-};
-use crate::math::{Point, Vector};
-use ncollide::bounding_volume::BoundingVolume;
+use crate::geometry::{Collider, ColliderHandle, ColliderSet, Ray, RayIntersection, WQuadtree};
 
 /// A pipeline for performing queries on all the colliders of a scene.
 #[cfg_attr(feature = "serde-serialize", derive(Serialize, Deserialize))]